获取有关Airflow on_failure_callback上下文的异常详细信息

Question

有没有办法在气流上获得异常细节on_failure_callback？

我注意到它不属于你的一部分context.我想创建一个通用的异常处理机制,它向Slack发布有关错误的信息,包括有关异常的详细信息.我现在设法触发/执行回调并发布到Slack,但无法发布异常详细信息.

谢谢.

Answer 1

可以将Anon_failure_callback提供给 DAG 和/或单个任务。

在第一种情况（提供给 DAG）中， （Airflow 调用您的参数）'exception'中没有。contexton_failure_callback

在第二种情况（提供任务）中，有。

包含的对象应该是 python Exception。从中获得诸如堆栈跟踪之类的东西是令人惊讶的不直观，但从这个答案中我使用以下内容来获得相当可读的堆栈跟踪：

import traceback

...

exception = context.get('exception')
formatted_exception = ''.join(
   traceback.format_exception(etype=type(exception), 
     value=exception, tb=exception.__traceback__
   )
).strip()

Answer 2

我认为可能无法在回调中获取异常详细信息。看源码

# Handling callbacks pessimistically
try:
    if self.state == State.UP_FOR_RETRY and task.on_retry_callback:
        task.on_retry_callback(context)
    if self.state == State.FAILED and task.on_failure_callback:
        task.on_failure_callback(context)
except Exception as e3:
    logging.error("Failed at executing callback")
    logging.exception(e3)