Airflow sla_miss_callback 函数未触发

Question

我一直在尝试获得一个 slack 消息回调来触发 SLA 未命中。我注意到:

SLA 未命中在 Airflow Web UI 中成功注册slamiss/list/

on_failure_callback成功运作

然而， sla_miss_callback函数本身永远不会被触发。
我试过的:

不同组合相加sla和 sla_miss_callback在default_args级别、DAG 级别和任务级别

检查我们的调度程序和工作人员的日志以获取 SLA 相关消息(另见 here)，但我们什么也没看到

如果从任何其他地方调用，松弛消息回调函数将起作用
基本任务或功能

default_args = {
    "owner": "airflow",
    "depends_on_past": False,
    'start_date': airflow.utils.dates.days_ago(n=0,minute=1),
    'on_failure_callback': send_task_failed_msg_to_slack,
    'sla': timedelta(minutes=1),
    "retries": 0, 
    "pool": 'canary',
    'priority_weight': 1
}

dag = airflow.DAG(
    dag_id='sla_test',
    default_args=default_args,
    sla_miss_callback=send_sla_miss_message_to_slack,
    schedule_interval='*/5 * * * *',
    catchup=False,
    max_active_runs=1,
    dagrun_timeout=timedelta(minutes=5)
)

def sleep():
    """ Sleep for 2 minutes """
    time.sleep(90)
    LOGGER.info("Slept for 2 minutes")

def simple_print(**context):
    """ Prints a message """
    print("Hello World!")


sleep = PythonOperator(
    task_id="sleep",
    python_callable=sleep,
    dag=dag
)

simple_task = PythonOperator(
    task_id="simple_task",
    python_callable=simple_print,
    provide_context=True,
    dag=dag
)

sleep >> simple_task

Answer 1

我曾经处于类似的情况。
在调查调度程序日志时，我发现了以下错误:

[2020-07-08 09:14:32,781] {scheduler_job.py:534} INFO -  --------------> ABOUT TO CALL SLA MISS CALL BACK  
[2020-07-08 09:14:32,781] {scheduler_job.py:541} ERROR - Could not call sla_miss_callback for DAG 
sla_miss_alert() takes 1 positional arguments but 5 were given

问题是你的 sla_miss_callback函数只需要 1 个参数，但实际上这应该是这样的:

def sla_miss_alert(dag, task_list, blocking_task_list, slas, blocking_tis):
    """Function that alerts me that dag_id missed sla"""
    # <function code here>

作为引用，请查看 Airflow source code .
注意:不要放 sla_miss_callback=sla_miss_alert在 default_args .它应该在 DAG 定义本身中定义。