How to get reason for failure using slack in airflow2.0

Airflow
1

Hi All,
How to get the reason for the failure of an operator, without going into logs. As I want to post the reason as a notification through slack?

My on_failure_callback function:


def task_fail_slack_alert(context):
    SLACK_CONN_ID = 'slack'
    slack_webhook_token = BaseHook.get_connection(SLACK_CONN_ID).password
    slack_msg = """
            :red_circle: Task Failed. 
            *Task*: {task}  
            *Dag*: {dag} 
            *Execution Time*: {exec_date}  
            *Log Url*: {log_url} 
  
            """.format(
            task=context.get('task_instance').task_id,
            dag=context.get('task_instance').dag_id,
            ti=context.get('task_instance'),
            exec_date=context.get('execution_date'),
            log_url=context.get('task_instance').log_url,
        )
    failed_alert = SlackWebhookOperator(
        task_id='slack_test',
        http_conn_id='slack',
        webhook_token=slack_webhook_token,
        message=slack_msg,
        username='airflow')
    return failed_alert.execute(context=context)

I want the error why it has failed. How ca I get that?

Thanks, Xi

2

Hi @kommu -

Similar to the other items you are retrieving from context, you can use:
exception = context.get('exception')

An example of a failure callback with this implemented:

def failure_callback(context):
    slack_conn_id = "slack_callbacks"
    slack_webhook_token = BaseHook.get_connection(slack_conn_id).password
    log_url = context.get("task_instance").log_url
    exception = context.get('exception')
    formatted_exception = ''.join(
                traceback.format_exception(etype=type(exception),
                                           value=exception,
                                           tb=exception.__traceback__)
    ).strip()
    slack_msg = f"""
            :x: Task has failed. 
            *Task*: {context.get('task_instance').task_id}
            *DAG*: {context.get('task_instance').dag_id} 
            *Execution Time*: {context.get('execution_date')}  
            *Exception*: {formatted_exception}
            <{log_url}| *Log URL*>
            """
    slack_alert = SlackWebhookOperator(
        task_id="slack_test",
        http_conn_id="slack_callbacks",
        webhook_token=slack_webhook_token,
        message=slack_msg,
        username="airflow",
    )
    return slack_alert.execute(context=context)

See more examples here: