如何在Airflow的DAG级别实现SLA？现有方案存在局限

Airflow原生default_args中的sla参数是任务级别的，无法直接实现DAG整体的SLA监控。针对你遇到的「任务均满足SLA但整体超时无告警」的问题，以下是几种可行的实现方案：

方案一：添加收尾任务并配置DAG级SLA #

这是最直接的方法，利用Airflow原生SLA机制，通过一个依赖所有任务的收尾任务来监控整个DAG的执行时长。

实现思路 #

1. 在DAG末尾添加一个空任务（或轻量校验任务），让它依赖DAG中所有其他任务。
2. 给这个收尾任务设置DAG级的SLA时长——由于它只有在所有任务完成后才会启动，它的SLA超时会直接反映整个DAG的总耗时是否超标。

示例代码 #

from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime, timedelta

def task_a():
    # 模拟耗时1小时的任务
    pass

def task_b():
    pass

def task_c():
    pass

def dag_final_check():
    # 空任务，仅用于触发SLA告警
    pass

default_args = {
    'owner': 'airflow',
    'depends_on_past': False,
    'email_on_failure': True,
    'email_on_sla_miss': True,  # 必须开启SLA告警开关
    'email': 'noreply@astronomer.io',
    'email_on_retry': False
}

with DAG(
    'dag_level_sla_demo',
    default_args=default_args,
    schedule_interval=timedelta(days=1),
    start_date=datetime(2023, 1, 1),
    catchup=False,
) as dag:
    a = PythonOperator(task_id='task_a', python_callable=task_a)
    b = PythonOperator(task_id='task_b', python_callable=task_b)
    c = PythonOperator(task_id='task_c', python_callable=task_c)
    
    # 收尾任务，设置DAG级SLA
    final_task = PythonOperator(
        task_id='dag_final_sla_check',
        python_callable=dag_final_check,
        sla=timedelta(hours=2)  # 这里设置DAG的总SLA时长
    )

    # 所有任务完成后执行收尾任务
    [a, b, c] >> final_task

关键说明 #

• 必须在default_args中开启email_on_sla_miss: True，否则SLA超时不会触发邮件告警。
• 收尾任务的execution_date与DAG一致，因此它的SLA是从DAG启动时间开始计算的，能准确反映整体耗时。

方案二：利用DAG Run回调函数自定义监控 #

通过DAG的成功/失败回调函数，直接查询DAG Run的启动和结束时间，判断总耗时是否超过阈值，然后触发自定义告警。

实现思路 #

1. 编写一个回调函数，从上下文获取DAG Run的start_date和end_date。
2. 计算两者的时间差，若超过设定的SLA时长，则执行告警逻辑（如发邮件、通知即时通讯工具）。
3. 将该函数配置为DAG的on_success_callback和on_failure_callback，确保无论DAG成功或失败都能检查时长。

示例代码 #

from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime, timedelta
import smtplib
from email.mime.text import MIMEText

def task_a():
    pass

def check_dag_sla(context):
    dag_run = context['dag_run']
    sla_duration = timedelta(hours=2)
    total_duration = dag_run.end_date - dag_run.start_date

    if total_duration > sla_duration:
        # 自定义告警逻辑，这里以邮件为例
        sender = 'noreply@astronomer.io'
        receivers = ['noreply@astronomer.io']
        msg_content = f"""DAG {dag_run.dag_id} 执行超时！
        DAG Run ID: {dag_run.run_id}
        实际耗时: {total_duration}
        SLA时长: {sla_duration}"""
        message = MIMEText(msg_content)
        message['Subject'] = f"[Airflow SLA告警] {dag_run.dag_id} 超时"
        message['From'] = sender
        message['To'] = ', '.join(receivers)

        # 发送邮件（需提前配置Airflow的SMTP服务）
        try:
            with smtplib.SMTP('your_smtp_server', 587) as smtp:
                smtp.starttls()
                smtp.login('smtp_user', 'smtp_pass')
                smtp.send_message(message)
        except Exception as e:
            print(f"告警邮件发送失败: {str(e)}")

default_args = {
    'owner': 'airflow',
    'depends_on_past': False,
    'email_on_failure': True,
    'email': 'noreply@astronomer.io',
    'email_on_retry': False
}

with DAG(
    'dag_level_sla_callback',
    default_args=default_args,
    schedule_interval=timedelta(days=1),
    start_date=datetime(2023, 1, 1),
    catchup=False,
    on_success_callback=check_dag_sla,
    on_failure_callback=check_dag_sla
) as dag:
    a = PythonOperator(task_id='task_a', python_callable=task_a)
    b = PythonOperator(task_id='task_b', python_callable=task_a)
    c = PythonOperator(task_id='task_c', python_callable=task_a)

    a >> b >> c

关键说明 #

• 该方法无需额外任务，逻辑灵活，可根据需求扩展告警方式（如发送到Slack、企业微信）。
• 需确保Airflow具备查询自身元数据的权限，且SMTP服务配置正确。

方案三：通过监控系统告警（无侵入式） #

如果你的团队已有监控体系（如Prometheus+Grafana），可以通过Airflow导出的dag_run_duration指标设置告警阈值，无需修改DAG代码。

实现思路 #

1. 确保Airflow配置了Metrics导出（如StatsD或Prometheus Exporter）。
2. 在监控系统中创建告警规则：当dag_run_duration指标超过设定的SLA时长（如2小时）时触发告警。
3. 配置告警通知方式（邮件、即时通讯工具等）。

适用场景 #

适合已有成熟监控体系的团队，无需侵入DAG代码，实现全局统一的SLA监控。

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