Skip to content

Commit 2f222f9

Browse files
potiukpotiuk
committed
Restore pluggable email backend for task failure and retry alerts
Since #57354, task email_on_failure / email_on_retry alerts were routed unconditionally through SmtpNotifier, silently ignoring the [email] email_backend configuration. Custom backends (SES, SendGrid, org-internal) stopped delivering failure/retry emails even though the deprecated email_on_* parameters still worked. This restores the old behaviour using the existing [email] email_backend option -- no new configuration is introduced: - A non-default [email] email_backend is transparently wrapped in a new LegacyEmailBackendNotifier adapter in common.compat, so existing SES / SendGrid / custom backends keep delivering alerts unchanged. The backend is loaded lazily from config at notify time, so the Task SDK keeps no static dependency on airflow.utils.email. - Otherwise the default SmtpNotifier is used, exactly as before. Both failure-email entry points (the worker task-runner path and the DAG-processor callback path) funnel through the same function, so the selected backend is used consistently regardless of how the task failed. The deprecated email_on_* parameters are not un-deprecated; this only keeps their existing behaviour pluggable until removal in Airflow 4.
1 parent 25b3b88 commit 2f222f9

4 files changed

Lines changed: 238 additions & 14 deletions

File tree

‎providers/common/compat/src/airflow/providers/common/compat/notifier/__init__.py‎

Lines changed: 54 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -17,16 +17,68 @@
1717

1818
from __future__ import annotations
1919

20-
from typing import TYPE_CHECKING
20+
from typing import TYPE_CHECKING, Any
2121

2222
from airflow.providers.common.compat.version_compat import AIRFLOW_V_3_0_PLUS
2323

2424
if TYPE_CHECKING:
25+
from collections.abc import Iterable
26+
2527
from airflow.sdk.bases.notifier import BaseNotifier
28+
from airflow.sdk.definitions.context import Context
2629
elif AIRFLOW_V_3_0_PLUS:
2730
from airflow.sdk.bases.notifier import BaseNotifier
2831
else:
2932
from airflow.notifications.basenotifier import BaseNotifier
3033

3134

32-
__all__ = ["BaseNotifier"]
35+
DEFAULT_EMAIL_BACKEND = "airflow.utils.email.send_email_smtp"
36+
37+
38+
class LegacyEmailBackendNotifier(BaseNotifier):
39+
"""
40+
Adapter that exposes a legacy ``[email] email_backend`` callable as a notifier.
41+
42+
Before failure and retry alerts were routed through ``BaseNotifier`` subclasses, deployments
43+
configured them through ``[email] email_backend`` -- a callable with the
44+
``airflow.utils.email.send_email`` signature, such as the Amazon SES or SendGrid senders.
45+
This adapter renders the standard email fields like any notifier, then loads and calls the
46+
configured backend, so existing ``email_backend`` setups keep working unchanged.
47+
48+
The backend is resolved from config at notify time rather than imported statically, keeping
49+
callers free of a hard dependency on ``airflow.utils.email`` (which lives in ``airflow-core``).
50+
"""
51+
52+
template_fields = ("to", "from_email", "subject", "html_content")
53+
54+
def __init__(
55+
self,
56+
to: str | Iterable[str],
57+
from_email: str | None = None,
58+
subject: str | None = None,
59+
html_content: str | None = None,
60+
**kwargs: Any,
61+
) -> None:
62+
super().__init__()
63+
self.to = to
64+
self.from_email = from_email
65+
self.subject = subject
66+
self.html_content = html_content
67+
68+
def notify(self, context: Context) -> None:
69+
from airflow.providers.common.compat.sdk import AirflowConfigException, conf
70+
71+
backend = conf.getimport("email", "email_backend", fallback=DEFAULT_EMAIL_BACKEND)
72+
if backend is None:
73+
raise AirflowConfigException("`[email] email_backend` is not configured")
74+
conn_id = conf.get("email", "email_conn_id", fallback=None)
75+
backend(
76+
self.to,
77+
self.subject,
78+
self.html_content,
79+
conn_id=conn_id,
80+
from_email=self.from_email,
81+
)
82+
83+
84+
__all__ = ["BaseNotifier", "LegacyEmailBackendNotifier"]

‎providers/common/compat/tests/unit/common/compat/test_notifier.py‎

Lines changed: 62 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,62 @@
1+
# Licensed to the Apache Software Foundation (ASF) under one
2+
# or more contributor license agreements. See the NOTICE file
3+
# distributed with this work for additional information
4+
# regarding copyright ownership. The ASF licenses this file
5+
# to you under the Apache License, Version 2.0 (the
6+
# "License"); you may not use this file except in compliance
7+
# with the License. You may obtain a copy of the License at
8+
#
9+
# http://www.apache.org/licenses/LICENSE-2.0
10+
#
11+
# Unless required by applicable law or agreed to in writing,
12+
# software distributed under the License is distributed on an
13+
# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
14+
# KIND, either express or implied. See the License for the
15+
# specific language governing permissions and limitations
16+
# under the License.
17+
from __future__ import annotations
18+
19+
from unittest import mock
20+
21+
import pytest
22+
23+
from airflow.providers.common.compat.notifier import LegacyEmailBackendNotifier
24+
25+
26+
class TestLegacyEmailBackendNotifier:
27+
def test_notify_calls_configured_backend(self):
28+
"""notify() loads the legacy backend from config and calls it with the standard fields."""
29+
from airflow.providers.common.compat.sdk import conf
30+
31+
backend = mock.MagicMock()
32+
notifier = LegacyEmailBackendNotifier(
33+
to=["a@b.com"],
34+
from_email="from@x.com",
35+
subject="Subject",
36+
html_content="<p>body</p>",
37+
)
38+
with (
39+
mock.patch.object(conf, "getimport", return_value=backend) as getimport,
40+
mock.patch.object(conf, "get", return_value="my_conn"),
41+
):
42+
notifier.notify(context={})
43+
44+
getimport.assert_called_once_with(
45+
"email", "email_backend", fallback="airflow.utils.email.send_email_smtp"
46+
)
47+
backend.assert_called_once_with(
48+
["a@b.com"],
49+
"Subject",
50+
"<p>body</p>",
51+
conn_id="my_conn",
52+
from_email="from@x.com",
53+
)
54+
55+
def test_notify_raises_when_backend_unresolvable(self):
56+
"""An empty/unloadable backend raises rather than silently doing nothing."""
57+
from airflow.providers.common.compat.sdk import AirflowConfigException, conf
58+
59+
notifier = LegacyEmailBackendNotifier(to="a@b.com")
60+
with mock.patch.object(conf, "getimport", return_value=None):
61+
with pytest.raises(AirflowConfigException):
62+
notifier.notify(context={})

‎task-sdk/src/airflow/sdk/execution_time/task_runner.py‎

Lines changed: 45 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -1991,27 +1991,60 @@ def _run_task_state_change_callbacks(
19911991
log.exception("Failed to run task callback", kind=kind, index=i, callback=callback)
19921992

19931993

1994+
DEFAULT_EMAIL_BACKEND = "airflow.utils.email.send_email_smtp"
1995+
1996+
19941997
def _send_error_email_notification(
19951998
task: BaseOperator | MappedOperator,
19961999
ti: RuntimeTaskInstance,
19972000
context: Context,
19982001
error: BaseException | str | None,
19992002
log: Logger,
20002003
) -> None:
2001-
"""Send email notification for task errors using SmtpNotifier."""
2002-
try:
2003-
from airflow.providers.smtp.notifications.smtp import SmtpNotifier
2004-
except ImportError:
2005-
log.error(
2006-
"Failed to send task failure or retry email notification: "
2007-
"`apache-airflow-providers-smtp` is not installed. "
2008-
"Install this provider to enable email notifications."
2009-
)
2010-
return
2004+
"""
2005+
Send email notification for task errors through the configured email backend.
20112006
2007+
A non-default ``[email] email_backend`` (an SES, SendGrid or org-internal callable with the
2008+
``airflow.utils.email.send_email`` signature) is wrapped in
2009+
:class:`~airflow.providers.common.compat.notifier.LegacyEmailBackendNotifier`; otherwise the
2010+
default :class:`~airflow.providers.smtp.notifications.smtp.SmtpNotifier` is used.
2011+
2012+
Both the worker task-runner path (:func:`finalize`) and the DAG-processor callback path
2013+
(``_execute_email_callbacks``) funnel through this function, so the resolved backend is used
2014+
consistently regardless of how the task failed.
2015+
"""
20122016
if not task.email:
20132017
return
20142018

2019+
email_backend = conf.get("email", "email_backend", fallback=DEFAULT_EMAIL_BACKEND)
2020+
notifier_description = "SmtpNotifier"
2021+
2022+
if email_backend and email_backend != DEFAULT_EMAIL_BACKEND:
2023+
try:
2024+
from airflow.providers.common.compat.notifier import LegacyEmailBackendNotifier
2025+
except ImportError:
2026+
log.error(
2027+
"Failed to send task failure or retry email notification: a custom "
2028+
"`[email] email_backend` (%r) is configured but "
2029+
"`apache-airflow-providers-common-compat` is not installed. Install or upgrade "
2030+
"that provider to keep using a custom email backend.",
2031+
email_backend,
2032+
)
2033+
return
2034+
notifier_class: type = LegacyEmailBackendNotifier
2035+
notifier_description = f"email_backend {email_backend!r}"
2036+
else:
2037+
try:
2038+
from airflow.providers.smtp.notifications.smtp import SmtpNotifier
2039+
except ImportError:
2040+
log.error(
2041+
"Failed to send task failure or retry email notification: "
2042+
"`apache-airflow-providers-smtp` is not installed. "
2043+
"Install this provider to enable email notifications."
2044+
)
2045+
return
2046+
notifier_class = SmtpNotifier
2047+
20152048
subject_template_file = conf.get("email", "subject_template", fallback=None)
20162049

20172050
# Read the template file if configured
@@ -2054,15 +2087,15 @@ def _send_error_email_notification(
20542087
return
20552088

20562089
try:
2057-
notifier = SmtpNotifier(
2090+
notifier = notifier_class(
20582091
to=to_emails,
20592092
subject=subject,
20602093
html_content=html_content,
20612094
from_email=conf.get("email", "from_email", fallback="airflow@airflow"),
20622095
)
20632096
notifier(email_context)
20642097
except Exception:
2065-
log.exception("Failed to send email notification")
2098+
log.exception("Failed to send email notification via %s", notifier_description)
20662099

20672100

20682101
@detail_span("task.execute")

‎task-sdk/tests/task_sdk/execution_time/test_task_runner.py‎

Lines changed: 77 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -22,6 +22,7 @@
2222
import functools
2323
import json
2424
import os
25+
import sys
2526
import textwrap
2627
import time
2728
from collections.abc import Iterable
@@ -3842,6 +3843,11 @@ def mock_send_side_effect(*args, **kwargs):
38423843
)
38433844

38443845

3846+
def _recording_email_backend(*args, **kwargs):
3847+
"""Legacy ``[email] email_backend`` stub; patched with a mock in tests to record calls."""
3848+
raise AssertionError("should be patched in the test")
3849+
3850+
38453851
class TestEmailNotifications:
38463852
FROM = "from@airflow"
38473853

@@ -4008,6 +4014,77 @@ def execute(self, context):
40084014
)
40094015
assert kwargs["from_email"] == self.FROM
40104016

4017+
def test_custom_email_backend_is_used(self, create_runtime_ti, mock_supervisor_comms):
4018+
"""A custom ``[email] email_backend`` is wrapped and invoked with rendered fields."""
4019+
from airflow.sdk.exceptions import AirflowFailException
4020+
from airflow.sdk.execution_time.task_runner import finalize, run
4021+
4022+
backend = mock.MagicMock()
4023+
4024+
class FailingOperator(BaseOperator):
4025+
def execute(self, context):
4026+
raise AirflowFailException("Task failed on purpose")
4027+
4028+
task = FailingOperator(
4029+
task_id="legacy_backend_task",
4030+
email=["test@example.com"],
4031+
email_on_failure=True,
4032+
)
4033+
4034+
runtime_ti = create_runtime_ti(task=task)
4035+
context = runtime_ti.get_template_context()
4036+
log = mock.MagicMock()
4037+
4038+
with conf_vars(
4039+
{
4040+
("email", "email_backend"): f"{__name__}._recording_email_backend",
4041+
("email", "email_conn_id"): "my_smtp",
4042+
("email", "from_email"): self.FROM,
4043+
}
4044+
):
4045+
with mock.patch(f"{__name__}._recording_email_backend", backend):
4046+
with mock.patch(
4047+
"airflow.providers.smtp.notifications.smtp.SmtpNotifier"
4048+
) as mock_smtp_notifier:
4049+
state, _, error = run(runtime_ti, context, log)
4050+
finalize(runtime_ti, state, context, log, error)
4051+
4052+
# The default SMTP notifier must not be used when a custom backend is configured.
4053+
mock_smtp_notifier.assert_not_called()
4054+
backend.assert_called_once()
4055+
args, kwargs = backend.call_args
4056+
# send_email(to, subject, html_content, conn_id=..., from_email=...)
4057+
assert args[0] == ["test@example.com"]
4058+
assert kwargs["conn_id"] == "my_smtp"
4059+
assert kwargs["from_email"] == self.FROM
4060+
4061+
def test_missing_compat_provider_is_logged(self, create_runtime_ti, mock_supervisor_comms):
4062+
"""A custom backend without `common.compat` installed is logged and does not raise."""
4063+
from airflow.sdk.exceptions import AirflowFailException
4064+
from airflow.sdk.execution_time.task_runner import finalize, run
4065+
4066+
class FailingOperator(BaseOperator):
4067+
def execute(self, context):
4068+
raise AirflowFailException("Task failed on purpose")
4069+
4070+
task = FailingOperator(
4071+
task_id="missing_compat_task",
4072+
email=["test@example.com"],
4073+
email_on_failure=True,
4074+
)
4075+
4076+
runtime_ti = create_runtime_ti(task=task)
4077+
context = runtime_ti.get_template_context()
4078+
log = mock.MagicMock()
4079+
4080+
with conf_vars({("email", "email_backend"): f"{__name__}._recording_email_backend"}):
4081+
with mock.patch.dict(sys.modules, {"airflow.providers.common.compat.notifier": None}):
4082+
state, _, error = run(runtime_ti, context, log)
4083+
# Must not raise even though the compat notifier cannot be imported.
4084+
finalize(runtime_ti, state, context, log, error)
4085+
4086+
log.error.assert_called()
4087+
40114088
@pytest.mark.enable_redact
40124089
def test_rendered_templates_mask_secrets(self, create_runtime_ti, mock_supervisor_comms):
40134090
"""Test that secrets registered with mask_secret() are redacted in rendered template fields."""

0 commit comments

Comments
 (0)