Airflow Slack通知关联错误task_id与日志链接问题求助

开发者阿泽

2026-8-2

解决Airflow Slack通知关联错误任务上下文的问题

明确获取当前任务实例
别直接依赖context['task_instance']，改用context['ti']（当前TaskInstance对象）来提取任务名称和日志链接，避免上游任务的上下文干扰。

显式获取当前上下文（推荐方案）
要是担心外部传入的context被污染，直接在通知函数里主动获取当前任务的真实上下文：

from airflow.operators.python import get_current_context

def slack_failure_notification():
    current_ctx = get_current_context()
    ti = current_ctx['ti']
    task_name = ti.task_id
    log_url = ti.log_url
    # 这里编写Slack消息的构建与发送逻辑

给每个任务单独绑定回调
不要在DAG全局层面设置on_failure_callback，而是给每个需要失败通知的任务单独绑定回调函数，确保触发通知时获取的是自身的任务上下文：
```
task1 = PythonOperator(
    task_id='task1',
    python_callable=your_task_func,
    on_failure_callback=slack_failure_notification,
    dag=dag
)
```
添加日志验证上下文准确性
在通知函数里加入日志打印，确认拿到的任务ID是不是实际失败的任务：
```
import logging

def slack_failure_notification(context):
    ti = context['ti']
    logging.info(f"触发通知的任务ID: {ti.task_id}")
    # 后续的Slack通知逻辑
```
去Airflow日志里查看这条打印内容，就能快速验证上下文是否正确。