Airflow dagrun_timeout超时时如何触发on_failure_callback或DAG级失败回调

原因确认 #

你的推测完全正确：on_failure_callback 默认属于任务级配置，仅在单个Task运行失败时触发；dagrun_timeout 是DAG运行实例（DagRun）级别的超时规则，超时后直接将整个DagRun标记为失败，不会主动触发任务级的失败回调。

可用解决方案 #

方案1：使用Airflow 2.x原生DAG级失败回调（最优） #

Airflow 2.0及以上版本已支持直接在DAG构造参数中配置DAG级别的on_failure_callback，只要DagRun最终状态变为失败（包括dagrun_timeout超时、任意核心任务失败等所有场景），都会自动触发该回调。
示例代码如下：

from datetime import datetime, timedelta
from airflow import DAG
from airflow.operators.bash import BashOperator

# DAG失败回调函数，可自定义告警逻辑
def dag_failure_alert(context):
    dag_run = context["dag_run"]
    # 此处可替换为邮件、企业微信、钉钉等告警逻辑
    print(f"DAG运行失败：{dag_run.dag_id}，执行时间：{dag_run.execution_date}，失败原因：{dag_run.state}")

default_args = {
    "owner": "airflow",
    "retries": 1
}

with DAG(
    dag_id="long_running_dag",
    default_args=default_args,
    start_date=datetime(2024, 1, 1),
    schedule_interval=None,
    # DAG级超时配置
    dagrun_timeout=timedelta(minutes=60),
    # 关键：配置DAG级失败回调，不要写在default_args中
    on_failure_callback=dag_failure_alert,
    catchup=False
) as dag:
    long_task = BashOperator(
        task_id="long_time_task",
        bash_command="sleep 3600"
    )

方案2：Airflow 1.x版本兼容方案 #

如果使用的是不支持DAG级回调的Airflow 1.x版本，可以新增一个兜底告警任务，设置触发规则为all_done，在该任务中判断DagRun状态后触发告警逻辑：

from airflow.operators.python import PythonOperator
from airflow.utils.trigger_rule import TriggerRule

# 原有任务定义省略
def check_dag_status(context):
    dag_run = context["dag_run"]
    if dag_run.state == "failed":
        # 执行你的告警逻辑
        pass

alert_task = PythonOperator(
    task_id="dag_status_check",
    python_callable=check_dag_status,
    provide_context=True,
    # 不管上游任务成功/失败都执行
    trigger_rule=TriggerRule.ALL_DONE
)

# 所有原有任务的下游挂载该告警任务
long_task >> alert_task

实现思路调整建议 #

无需修改现有dagrun_timeout的超时判定逻辑，只需要将原来写在default_args中的任务级失败回调，拆分出DAG层面的通用告警逻辑，放在DAG级回调中即可，原有任务级的失败回调仍然可以保留用于单个任务异常的场景。

内容的提问来源于stack exchange，提问作者Anton