如何通过SlackAPIPostOperator发送Airflow任务错误日志？

问题原因 #

在Airflow的on_failure_callback中，直接调用context['exception']可能因版本差异、任务失败场景不同返回空值。异常信息更可靠的存储位置是任务实例对象的exception或traceback属性。

解决方案 #

修改你的task_failure_alert函数，从context['task_instance']中获取异常信息，同时处理异常为空的边界情况：

def task_failure_alert(context):
    env = Variable.get("Environment", default_var="Sandbox")
    ti = context['task_instance']
    # 获取异常信息，为空时显示默认文本
    exception_msg = str(ti.exception) if ti.exception else "未捕获到具体异常信息"
    # 可选：如需完整错误栈，可启用下面的代码
    # traceback_msg = ti.traceback if ti.traceback else "无错误栈信息"
    
    failed_alert = SlackAPIPostOperator(
        task_id="DAG_failed",
        channel="#data-alerts",
        slack_conn_id="slack",
        text=f":red_circle: <!subteam^::redacted::> Task {context['task_instance_key_str']} failed for {context['dag']}.\n错误信息: {exception_msg}\n查看日志: {ti.log_url}",
    )
    if env == "Production":
        return failed_alert.execute(context)
    else:
        print(f"Task has failed, task_instance_key_str: {context['task_instance_key_str']}, error: {exception_msg}")

关键说明 #

• task_instance.exception：存储任务失败时的异常对象，转成字符串后即可得到简洁的错误描述。
• task_instance.traceback：如果需要完整的错误栈信息用于排查复杂问题，可以替换或补充到通知内容中。
• 必须处理异常为空的情况，避免因ti.exception为None导致字符串拼接报错。

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