Airflow中dagrun_timeout未生效：如何强制超时DAG运行失败？

我正在使用Airflow 2.0.2版本，尝试通过配置dagrun_timeout参数为DAG运行设置时间限制，期望当DAG运行达到时限时，无论任务处于什么状态都强制让DAG失败。

根据相关资料，超时失败不会在任务运行中触发，但即使仅在任务完成后检查超时，我仍未看到预期的超时失败效果。

在下方的DAG示例中，我设置了dagrun_timeout: timedelta(seconds=10)，理论上DAG在t1执行后就应该超时，但实际它仍继续运行并成功完成所有任务。我已确保所有任务总耗时超过下一次调度时间，但仍未触发超时。

示例代码：

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

default_args = {
    'owner': 'me',
    'depends_on_past': False,
    'start_date': datetime(2019, 6, 1),
    'retries': 0,
    'execution_timeout':timedelta(seconds=65),
    'dagrun_timeout': timedelta(seconds=10),
    'retry_delay': timedelta(seconds=5)
}

dag = DAG(
    'min_timeout',
    default_args=default_args,
    schedule_interval=timedelta(minutes=1),
    max_active_runs=1
)

t1 = BashOperator(
    task_id='fast_task',
    bash_command='sleep 60',
    dag=dag)

t2 = BashOperator(
    task_id='slow_task',
    bash_command='sleep 5',
    dag=dag)

t3 = BashOperator(
    task_id='fast_task_2',
    bash_command='sleep 5',
    dag=dag)

t1 >> t2 >> t3

问题原因分析 #

1. 版本逻辑缺陷：Airflow 2.0.x中dagrun_timeout仅在DAG运行的最终结束阶段才会触发超时判断，不会在任务执行过程中或单个任务完成后检查。这意味着只有当所有任务都执行完毕，系统才会对比DAG总运行时长是否超时，此时即使超时也不会回溯标记失败。
2. 任务超时优先级覆盖：示例中t1的execution_timeout设为65秒，远大于dagrun_timeout的10秒。任务自身的执行超时优先级高于DAG级超时，且t1执行完成后后续任务继续运行，直到全部结束才触发DAG超时检查，但此时Airflow 2.0.x的逻辑不会触发失败操作。

解决方案 #

方案1：升级Airflow版本 #

Airflow 2.1.0及后续版本修复了dagrun_timeout的检查逻辑，增加了周期性主动检查机制，会在任务执行过程中判断DAG是否超时，并及时标记DAG失败。升级到2.1.0或更高版本是最彻底的解决方式。

方案2：自定义超时检查逻辑（适用于无法升级的场景） #

如果无法升级，可以通过自定义代码实现主动超时监控：

• 添加一个并行执行的PythonOperator，定期检查当前DAGRun的运行时长，一旦超过阈值，主动标记DAG为失败并终止所有运行中的任务。

示例代码：

from airflow import DAG
from airflow.operators.bash_operator import BashOperator
from airflow.operators.python_operator import PythonOperator
from datetime import datetime, timedelta
from airflow.models import DagRun, TaskInstance
from airflow.utils.state import State
import time

def check_dag_timeout(**context):
    dag_run = context['dag_run']
    start_time = dag_run.start_date
    timeout_seconds = 10
    while True:
        elapsed = (datetime.now() - start_time).total_seconds()
        if elapsed > timeout_seconds:
            # 标记DAG为失败
            dag_run.set_state(State.FAILED)
            # 终止所有运行中的任务
            running_tasks = TaskInstance.find(dag_id=dag_run.dag_id, execution_date=dag_run.execution_date, state=State.RUNNING)
            for task in running_tasks:
                task.set_state(State.FAILED)
            break
        time.sleep(2)
        # DAG已结束则退出检查
        if dag_run.state != State.RUNNING:
            break

default_args = {
    'owner': 'me',
    'depends_on_past': False,
    'start_date': datetime(2019, 6, 1),
    'retries': 0,
    'execution_timeout':timedelta(seconds=65),
    'retry_delay': timedelta(seconds=5)
}

dag = DAG(
    'min_timeout_fixed',
    default_args=default_args,
    schedule_interval=timedelta(minutes=1),
    max_active_runs=1
)

timeout_check = PythonOperator(
    task_id='timeout_check',
    python_callable=check_dag_timeout,
    provide_context=True,
    dag=dag,
    execution_timeout=timedelta(seconds=120)
)

t1 = BashOperator(
    task_id='fast_task',
    bash_command='sleep 60',
    dag=dag)

t2 = BashOperator(
    task_id='slow_task',
    bash_command='sleep 5',
    dag=dag)

t3 = BashOperator(
    task_id='fast_task_2',
    bash_command='sleep 5',
    dag=dag)

# 超时检查任务与t1并行启动
timeout_check >> t1 >> t2 >> t3

方案3：调整超时参数配合（仅适用于事后标记） #

如果不需要中途终止任务，仅希望在DAG结束后标记超时失败，可以将dagrun_timeout设置为大于所有任务execution_timeout之和。这样当所有任务完成后，系统会正确检查DAG总时长是否超时，但无法实现中途终止任务的需求。

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