Upgrade email notifications to use SmtpNotifier

[amoghrajesh]

closes: #54712

What?

Replace legacy email_on_failure and email_on_retry attributes with SmtpNotifier from apache-airflow-providers-smtp. This change automatically converts the old email configuration to the new notifier-based system while maintaining backward compatibility.

The intention behind doing this is to maintain ONE way of doing things compared to having copies of doing the same thing in multiple ways across core, task sdk, providers.

Testing

Setup

Using docker image of mailpit to set up a SMTP server in same network as breeze

➜  ~ docker run -d \
  --network breeze_default \
  -p 8025:8025 \
  --name mailpit \
  axllent/mailpit

Setup a connection for SMTP

The UI can be accessed at localhost:8025 now

Route 1: Testing email functionality from tasks

DAG used:

from datetime import datetime
from airflow import DAG
from airflow.providers.standard.operators.bash import BashOperator
from airflow.providers.smtp.notifications.smtp import send_smtp_notification
from airflow.providers.standard.operators.python import PythonOperator

def retryable_func():
    1//0

with DAG(
    dag_id="smtp_notifier",
    schedule=None,
    start_date=datetime(2023, 1, 1),
    catchup=False,
):
    op1 = BashOperator(
        task_id="mytask",
        email="bash-task@example.com",
        email_on_failure=True,
        bash_command="exit 1",
    )

    op2 = PythonOperator(
        task_id="mytask2",
        email_on_retry=True,
        python_callable=retryable_func,
        email="python-task@example.com",
        retries=3,
    )

    [op1, op2]

One of the tasks has email_on_retry defined, the other has email_on_failure.

The bash task fails and python tasks retries.

Bash Task

Python Task

Emails

For retried task (observe try number)

For failed task

Route 2: SMTP provider not installed at all

Uninstall smtp provider on workers

root@f7bf900b8f64:/opt/airflow# pip freeze | grep smtp
aiosmtplib==4.0.2
# Editable install with no version control (apache-airflow-providers-smtp==2.3.0)
-e /opt/airflow/providers/smtp
root@f7bf900b8f64:/opt/airflow# uv pip uninstall apache-airflow-providers-smtp
Using Python 3.10.18 environment at: /usr/python
Uninstalled 1 package in 23ms
 - apache-airflow-providers-smtp==2.3.0 (from file:///opt/airflow/providers/smtp)
root@f7bf900b8f64:/opt/airflow# pip freeze | grep smtp
aiosmtplib==4.0.2

Bash Task

Python Task

Route 3: Testing callbacks with email because it uses the same interface

DAG:

from datetime import datetime
from airflow import DAG
from airflow.providers.smtp.notifications.smtp import send_smtp_notification
from airflow.providers.standard.operators.bash import BashOperator
from airflow.providers.standard.operators.python import PythonOperator

def retryable_func():
    1//0

with DAG(
    dag_id="smtp_notifier",
    schedule=None,
    start_date=datetime(2023, 1, 1),
    catchup=False,
):
    op1 = BashOperator(
        task_id="mytask",
        on_failure_callback=[
            send_smtp_notification(
                from_email="from@mail.com",
                to="to@mail.com",
                subject="[Error] The Task {{ ti.task_id }} failed",
                html_content="debug logs",
            )
        ],
        # email="bash-task@example.com",
        # email_on_failure=True,
        bash_command="exit 1",
    )

    op2 = PythonOperator(
        task_id="mytask2",
        # email_on_retry=True,
        python_callable=retryable_func,
        # email="python-task@example.com",
        on_retry_callback=[
            send_smtp_notification(
                from_email="from2@mail.com",
                to="to2@mail.com",
                subject="[Error] The Task Python {{ ti.task_id }} failed",
                html_content="debug logs",
            )
        ],
        retries=3,
    )

    [op1, op2]

One of the task has a retry call back and the other has failure callback.

Task failures

Failure callback:

Retry callback:

And one for each retry:

Route 4: Testing with custom subject and html content templates

Subject template:

Custom Subject: {{ti.task_id}} failed in DAG {{ti.dag_id}}

Content template:

<!DOCTYPE html>
<html>
<head>
    <meta charset="UTF-8">
    <title>Custom Email Template</title>
</head>
<body>
    <h2>Task Failure Notification</h2>
    <p><strong>Task:</strong> {{ti.task_id}}</p>
    <p><strong>DAG:</strong> {{ti.dag_id}}</p>
    <p><strong>Run ID:</strong> {{ti.run_id}}</p>
    <p><strong>Try Number:</strong> {{try_number}} of {{max_tries + 1}}</p>
    <p><strong>Host:</strong> {{ti.hostname}}</p>
    
    <h3>Exception Details:</h3>
    <pre>{{exception_html}}</pre>
    
    <p>
        <a href="{{ti.log_url}}">View Logs</a> | 
        <a href="{{ti.mark_success_url}}">Mark Success</a>
    </p>
</body>
</html>

Defined these env vars:

export AIRFLOW__EMAIL__SUBJECT_TEMPLATE=/files/custom_email_subject.jinja2
export AIRFLOW__EMAIL__HTML_CONTENT_TEMPLATE=/files/custom_email_body.html

This is how it renders:

---

^ Add meaningful description above
Read the Pull Request Guidelines for more information.
In case of fundamental code changes, an Airflow Improvement Proposal (AIP) is needed.
In case of a new dependency, check compliance with the ASF 3rd Party License Policy.
In case of backwards incompatible changes please leave a note in a newsfragment file, named {pr_number}.significant.rst or {issue_number}.significant.rst, in airflow-core/newsfragments.

[amoghrajesh]

Needs much more testing + cleanup but roughly this is where I am heading with moving email notifications to use SmtpNotifier

[ashb]

I don't think we use AIRFLOW__EMAIL__HTML_CONTENT_TEMPLATE anymore -- we probably should?

[amoghrajesh]

@ashb @eladkal good call above thanks.

I tested with subject template as well as html body template and these are the results:

Subject template:

Custom Subject: {{ti.task_id}} failed in DAG {{ti.dag_id}}

Content template:

<!DOCTYPE html>
<html>
<head>
    <meta charset="UTF-8">
    <title>Custom Email Template</title>
</head>
<body>
    <h2>Task Failure Notification</h2>
    <p><strong>Task:</strong> {{ti.task_id}}</p>
    <p><strong>DAG:</strong> {{ti.dag_id}}</p>
    <p><strong>Run ID:</strong> {{ti.run_id}}</p>
    <p><strong>Try Number:</strong> {{try_number}} of {{max_tries + 1}}</p>
    <p><strong>Host:</strong> {{ti.hostname}}</p>
    
    <h3>Exception Details:</h3>
    <pre>{{exception_html}}</pre>
    
    <p>
        <a href="{{ti.log_url}}">View Logs</a> | 
        <a href="{{ti.mark_success_url}}">Mark Success</a>
    </p>
</body>
</html>

Defined these env vars:

export AIRFLOW__EMAIL__SUBJECT_TEMPLATE=/files/custom_email_subject.jinja2
export AIRFLOW__EMAIL__HTML_CONTENT_TEMPLATE=/files/custom_email_body.html

This is how it renders:

[amoghrajesh]

Added it to 3.2

[amoghrajesh]

@ashb @uranusjr @ephraimbuddy gentle bump for a review request on this one when you have time

[amoghrajesh]

cc: @atul-astronomer functional testing can be performed on this one by having a dag with email on failure (check PR desc)