on_failure_callback when the DAG (not a task) fails

[maratbakiev2]

Airflow has an option to execute a specific function when task fails (on_failure callback), but this functionality doesn't cover the following case:

1. We have a dag timeout of 2 hours
2. After timeout DAG state marked as failed while task state marked as skipped
3. So we have failed DAG but no alert

My request is to add the analogue of on_failure_callback but for the state of the DAG, not task

[Taragolis]

DAG have own on_failure_callback propery

[maratbakiev2]

You're right and we use them. But this property works the same way as if I would put it into each task of the DAG. I mean, it still alerts task failed states, not DAG failed states.

[Taragolis]

Just wondering what made you think that way?

[Taragolis]

Or maybe you tried to assign it into the default_args ???

with DAG(
    "awesome_dag",
    schedule=None,
    on_failure_callback=..., # This one related to the DAG callbacks,
    default_args={
        "on_failure_callback": ..., # This one related to the Task callbacks within the DAG
    }
):
    ...

[maratbakiev2]

Just wondering what made you think that way?

1. saw it on some forums
2. didn't get notifications on DAG failures without task failures

I set it as a separate argument of DAG class instance, not in the default args

        super().__init__(
            dag_id=dag_id, start_date=start_date, schedule_interval=schedule_interval,
            catchup=catchup, dagrun_timeout=dagrun_timeout, default_args=default_args,
            params=self.params,
                 on_failure_callback=[
                     send_slack_webhook_notification(
                         slack_webhook_conn_id='slack_webhook_alerts',
                         text=(
                            ":red_circle: DAG *{{dag.dag_id}}* failed on task *{{ dag_run.get_task_instances(state='failed')[0].task_id }}*.\n"
                            "*Data Interval End*: {{dag_run.data_interval_end}}\n"
                            "*Failure logs URL*: " + AIRFLOW_URL +"/dags/{{ dag.dag_id }}/grid?dag_run_id={{dag_run.run_id.replace(':', '%3A').replace('+', '%2B')}}&task_id={{dag_run.get_task_instances(state='failed')[0].task_id}}&tab=logs\n"
                            "*Responsible Team*: <!subteam^{{dag.params['slack_team_id']}}>\n"
                         )
                     )
                 ],
                 **kwargs)

And these are the runs when we haven't received alerts

The ones where task is skipped and dag state is failed.

[Taragolis]

If it not send it doesn’t mean it not works, it might failed to send. Try to check DAG Processor / scheduler logs might be it failed for some reason.