Airflow - 2 alerts are send on_failure

user_3723311 0
Viewed 2502

I have a strange bug on Airflow 1.10. I wanted to try to send email and a notification on Microsoft Teams. I made a small dumb DAG to try it out. Everything is working well but I got 2 notifications on a row. 2 Emails and 2 Messages on Teams.

I used this for Teams : https://github.com/mendhak/Airflow-MS-Teams-Operator

Here the Dags :

from datetime import datetime

from airflow import DAG
from airflow.operators.dummy_operator import DummyOperator
from airflow.operators.python_operator import PythonOperator
from operators.ms_teams_webhook_operator import MSTeamsWebhookOperator
from airflow.utils.email import send_email_smtp

default_args = {
    "owner": "me",
    "depends_on_past": False,
    "start_date": datetime(2020, 6, 15),
    'email_on_failure': False
}


def on_failure(context):
    dag_id = context['dag_run'].dag_id

    task_id = context['task_instance'].task_id
    # context['task_instance'].xcom_push(key=dag_id, value=True)

    logs_url = f"https://myairflow/admin/airflow/log?dag_id={dag_id}&task_id={task_id}&execution_date={context['ts']}"

    teams_notification = MSTeamsWebhookOperator(
        task_id="msteams_notify_failure",
        trigger_rule="all_done",
        message=f"{dag_id} has failed on task: {task_id}",
        button_text="View log",
        button_url=logs_url,
        theme_color="FF0000",
        http_conn_id='msteams-python-webhook')
    teams_notification.execute(context)

    title = f"Titre {dag_id} - {task_id}"
    body = title

    send_email_smtp("gil.felot@lisea.fr", title, body)


def print_fail():
    print("Hello !")
    exit(1)


with DAG(
        "test_email2",  # ICI
        default_args=default_args,
        schedule_interval=None
) as dag:
    preprocessing_started = DummyOperator(
        task_id="go_email_go"
    )

    python_fail = PythonOperator(
        task_id="pyhton_def",
        python_callable=print_fail,
        on_failure_callback=on_failure,
        email_on_failure=False
    )

preprocessing_started >> python_fail

EDIT :

Using the Hook instead. Now nothing is trigerred

from datetime import datetime

from airflow import DAG
from airflow.operators.dummy_operator import DummyOperator
from airflow.operators.python_operator import PythonOperator
from airflow.utils.email import send_email_smtp
from hooks.ms_teams_webhook_hook import MSTeamsWebhookHook


def on_failure(context):
    dag_id = context['dag_run'].dag_id

    task_id = context['task_instance'].task_id
    # context['task_instance'].xcom_push(key=dag_id, value=True)

    logs_url = f"https://myairflow/admin/airflow/log?dag_id={dag_id}&task_id={task_id}&execution_date={context['ts']}"

    teams_notification_hook = MSTeamsWebhookHook(
        http_conn_id='msteams-python-webhook',
        message=f"Le DAG {dag_id} a échoué sur la tâche : {task_id}",
        subtitle="Voir les logs ?",
        button_text="Logs",
        button_url=logs_url,
        theme_color="FF0000"
    )
    teams_notification_hook.execute()

    title = f"Titre {dag_id} - {task_id}"
    body = title

    send_email_smtp("my@email.fr", title, body)


def on_success(context):
    print("OK callback")
    dag_id = context['dag_run'].dag_id

    for i in context.items():
        print(i)

    teams_notification_hook = MSTeamsWebhookHook(
        http_conn_id='msteams-python-webhook',
        message=f"Le DAG {dag_id} s'est terminé avec succès",
        theme_color="00EE00"
    )
    teams_notification_hook.execute(context)

    title = f"Titre {dag_id} - Success"
    body = title

    send_email_smtp("my@email.fr", title, body)


default_args = {
    "owner": "lisea-mesea",
    "depends_on_past": False,
    "start_date": datetime(2020, 6, 15),
    "email_on_failure": False,
    "on_failure_callback": on_success
    # "on_failure_callback": on_failure
}


def print_fail():
    print("Hello !")
    exit(1)


with DAG(
        "test_email2",  # ICI
        default_args=default_args,
        schedule_interval=None
) as dag:
    preprocessing_started = DummyOperator(
        task_id="go_email_go"
    )

    python_fail = PythonOperator(
        task_id="pyhton_def",
        python_callable=print_fail,
        # on_failure_callback=on_failure,
        email_on_failure=False
    )

preprocessing_started >> python_fail
airflow email
1 Answers

I recommend strongly to use the MSTeamsWebhookHook in the on_failure_callback instead of the MSTeamsWebhookOperator.

Without going much into the weeds, the BaseOperator from which MSTeamsWebhookOperator inherits tries to rig it to the current dag when it's instantiated.

The dag property setter is can be seen to register the task instance on the dag.

This means that asides from manual execution of the operator taking place in the on_failure_callback, a task instance for MSTeamsWebhookOperator is scheduled. The later is not intended as we only care is for the capability to send notifications provided in the Hook when a task fails.

This doesn't give an explanation for why emails are sent out twice as that doesn't appear to be using an Operator. That requires a separate investigation.

user_51898110
Related