Skip to content

Commit 816add6

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, so existing SES / SendGrid / custom backends keep delivering alerts unchanged. The backend is resolved 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. The notifier lives in the Task SDK (airflow.sdk.execution_time.email_backend) next to its only caller, so no extra provider needs to be installed for a custom email backend to keep working. 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 816add6

5 files changed

Lines changed: 297 additions & 12 deletions

File tree

‎airflow-core/newsfragments/69877.bugfix.rst‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1 @@
1+
Restore delivery of ``email_on_failure`` and ``email_on_retry`` task alerts through a custom ``[email] email_backend``. These alerts were routed unconditionally through ``SmtpNotifier``, so deployments using an Amazon SES, SendGrid or org-internal backend stopped receiving them. The default remains ``SmtpNotifier``. Note that a configured ``email_backend`` which cannot be imported now fails with a logged error instead of silently falling back to SMTP -- check that ``[email] email_backend`` still resolves before upgrading.

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

Lines changed: 75 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,75 @@
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+
18+
from __future__ import annotations
19+
20+
from typing import TYPE_CHECKING, Any
21+
22+
from airflow.sdk.bases.notifier import BaseNotifier
23+
from airflow.sdk.configuration import conf
24+
from airflow.sdk.exceptions import AirflowConfigException
25+
26+
if TYPE_CHECKING:
27+
from collections.abc import Iterable
28+
29+
from airflow.sdk.definitions.context import Context
30+
31+
DEFAULT_EMAIL_BACKEND = "airflow.utils.email.send_email_smtp"
32+
33+
34+
class _LegacyEmailBackendNotifier(BaseNotifier):
35+
"""
36+
Adapter that exposes a legacy ``[email] email_backend`` callable as a notifier.
37+
38+
Before failure and retry alerts were routed through ``BaseNotifier`` subclasses, deployments
39+
configured them through ``[email] email_backend`` -- a callable with the
40+
``airflow.utils.email.send_email`` signature, such as the Amazon SES or SendGrid senders.
41+
This adapter renders the standard email fields like any notifier, then loads and calls the
42+
configured backend, so existing ``email_backend`` setups keep working unchanged.
43+
44+
The backend is resolved from config at notify time rather than imported statically, keeping
45+
the Task SDK free of a hard dependency on ``airflow.utils.email`` (which lives in
46+
``airflow-core``).
47+
"""
48+
49+
template_fields = ("to", "from_email", "subject", "html_content")
50+
51+
def __init__(
52+
self,
53+
to: str | Iterable[str],
54+
from_email: str | None = None,
55+
subject: str | None = None,
56+
html_content: str | None = None,
57+
**kwargs: Any,
58+
) -> None:
59+
super().__init__()
60+
self.to = to
61+
self.from_email = from_email
62+
self.subject = subject
63+
self.html_content = html_content
64+
65+
def notify(self, context: Context) -> None:
66+
backend = conf.getimport("email", "email_backend", fallback=DEFAULT_EMAIL_BACKEND)
67+
if backend is None:
68+
raise AirflowConfigException("`[email] email_backend` is not configured")
69+
backend(
70+
self.to,
71+
self.subject,
72+
self.html_content,
73+
conn_id=conf.get("email", "email_conn_id", fallback=None),
74+
from_email=self.from_email,
75+
)

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

Lines changed: 35 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -136,6 +136,10 @@
136136
get_previous_dagrun_success,
137137
set_current_context,
138138
)
139+
from airflow.sdk.execution_time.email_backend import (
140+
DEFAULT_EMAIL_BACKEND,
141+
_LegacyEmailBackendNotifier,
142+
)
139143
from airflow.sdk.execution_time.sentry import Sentry
140144
from airflow.sdk.execution_time.xcom import XCom
141145
from airflow.sdk.listener import get_listener_manager
@@ -1998,20 +2002,39 @@ def _send_error_email_notification(
19982002
error: BaseException | str | None,
19992003
log: Logger,
20002004
) -> 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
2005+
"""
2006+
Send email notification for task errors through the configured email backend.
2007+
2008+
A non-default ``[email] email_backend`` (an SES, SendGrid or org-internal callable with the
2009+
``airflow.utils.email.send_email`` signature) is wrapped in
2010+
:class:`~airflow.sdk.execution_time.email_backend._LegacyEmailBackendNotifier`; otherwise the
2011+
default :class:`~airflow.providers.smtp.notifications.smtp.SmtpNotifier` is used.
20112012
2013+
Both the worker task-runner path (:func:`finalize`) and the DAG-processor callback path
2014+
(``_execute_email_callbacks``) funnel through this function, so the resolved backend is used
2015+
consistently regardless of how the task failed.
2016+
"""
20122017
if not task.email:
20132018
return
20142019

2020+
email_backend = conf.get("email", "email_backend", fallback=DEFAULT_EMAIL_BACKEND)
2021+
notifier_description = "SmtpNotifier"
2022+
2023+
if email_backend and email_backend != DEFAULT_EMAIL_BACKEND:
2024+
notifier_class: type = _LegacyEmailBackendNotifier
2025+
notifier_description = f"email_backend {email_backend!r}"
2026+
else:
2027+
try:
2028+
from airflow.providers.smtp.notifications.smtp import SmtpNotifier
2029+
except ImportError:
2030+
log.error(
2031+
"Failed to send task failure or retry email notification: "
2032+
"`apache-airflow-providers-smtp` is not installed. "
2033+
"Install this provider to enable email notifications."
2034+
)
2035+
return
2036+
notifier_class = SmtpNotifier
2037+
20152038
subject_template_file = conf.get("email", "subject_template", fallback=None)
20162039

20172040
# Read the template file if configured
@@ -2054,15 +2077,15 @@ def _send_error_email_notification(
20542077
return
20552078

20562079
try:
2057-
notifier = SmtpNotifier(
2080+
notifier = notifier_class(
20582081
to=to_emails,
20592082
subject=subject,
20602083
html_content=html_content,
20612084
from_email=conf.get("email", "from_email", fallback="airflow@airflow"),
20622085
)
20632086
notifier(email_context)
20642087
except Exception:
2065-
log.exception("Failed to send email notification")
2088+
log.exception("Failed to send email notification via %s", notifier_description)
20662089

20672090

20682091
@detail_span("task.execute")

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

Lines changed: 86 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,86 @@
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 collections.abc import Iterable
20+
from typing import Any
21+
from unittest import mock
22+
23+
import pytest
24+
25+
from airflow.sdk.configuration import conf
26+
from airflow.sdk.exceptions import AirflowConfigException
27+
from airflow.sdk.execution_time.email_backend import _LegacyEmailBackendNotifier
28+
29+
30+
def _legacy_email_backend(
31+
to: list[str] | Iterable[str],
32+
subject: str,
33+
html_content: str,
34+
files: list[str] | None = None,
35+
dryrun: bool = False,
36+
cc: str | Iterable[str] | None = None,
37+
bcc: str | Iterable[str] | None = None,
38+
mime_subtype: str = "mixed",
39+
mime_charset: str = "utf-8",
40+
conn_id: str | None = None,
41+
custom_headers: dict[str, Any] | None = None,
42+
**kwargs,
43+
) -> None:
44+
"""
45+
Spec for an ``[email] email_backend`` callable.
46+
47+
Mirrors the ``airflow.utils.email.send_email`` signature so the autospec enforces the
48+
calling convention the notifier has to honour. Duplicated here rather than imported
49+
because the Task SDK must not depend on ``airflow-core``.
50+
"""
51+
raise AssertionError("spec only; never called")
52+
53+
54+
class TestLegacyEmailBackendNotifier:
55+
def test_notify_calls_configured_backend(self):
56+
"""notify() loads the legacy backend from config and calls it with the standard fields."""
57+
backend = mock.create_autospec(_legacy_email_backend)
58+
notifier = _LegacyEmailBackendNotifier(
59+
to=["[email protected]"],
60+
from_email="[email protected]",
61+
subject="Subject",
62+
html_content="<p>body</p>",
63+
)
64+
with (
65+
mock.patch.object(conf, "getimport", autospec=True, return_value=backend) as getimport,
66+
mock.patch.object(conf, "get", autospec=True, return_value="my_conn"),
67+
):
68+
notifier.notify(context={})
69+
70+
getimport.assert_called_once_with(
71+
"email", "email_backend", fallback="airflow.utils.email.send_email_smtp"
72+
)
73+
backend.assert_called_once_with(
74+
["[email protected]"],
75+
"Subject",
76+
"<p>body</p>",
77+
conn_id="my_conn",
78+
from_email="[email protected]",
79+
)
80+
81+
def test_notify_raises_when_backend_unresolvable(self):
82+
"""An empty/unloadable backend raises rather than silently doing nothing."""
83+
notifier = _LegacyEmailBackendNotifier(to="[email protected]")
84+
with mock.patch.object(conf, "getimport", autospec=True, return_value=None):
85+
with pytest.raises(AirflowConfigException):
86+
notifier.notify(context={})

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

Lines changed: 100 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -3842,6 +3842,30 @@ def mock_send_side_effect(*args, **kwargs):
38423842
)
38433843

38443844

3845+
def _recording_email_backend(
3846+
to: list[str] | Iterable[str],
3847+
subject: str,
3848+
html_content: str,
3849+
files: list[str] | None = None,
3850+
dryrun: bool = False,
3851+
cc: str | Iterable[str] | None = None,
3852+
bcc: str | Iterable[str] | None = None,
3853+
mime_subtype: str = "mixed",
3854+
mime_charset: str = "utf-8",
3855+
conn_id: str | None = None,
3856+
custom_headers: dict[str, Any] | None = None,
3857+
**kwargs,
3858+
) -> None:
3859+
"""
3860+
Legacy ``[email] email_backend`` stub, patched with an autospecced mock in tests.
3861+
3862+
Mirrors the ``airflow.utils.email.send_email`` signature so the autospec enforces the
3863+
calling convention ``_LegacyEmailBackendNotifier`` has to honour. Duplicated here rather
3864+
than imported because the Task SDK must not depend on ``airflow-core``.
3865+
"""
3866+
raise AssertionError("should be patched in the test")
3867+
3868+
38453869
class TestEmailNotifications:
38463870
FROM = "from@airflow"
38473871

@@ -4008,6 +4032,82 @@ def execute(self, context):
40084032
)
40094033
assert kwargs["from_email"] == self.FROM
40104034

4035+
def test_custom_email_backend_is_used(self, create_runtime_ti, mock_supervisor_comms):
4036+
"""A custom ``[email] email_backend`` is wrapped and invoked with rendered fields."""
4037+
from airflow.sdk.exceptions import AirflowFailException
4038+
from airflow.sdk.execution_time.task_runner import finalize, run
4039+
4040+
backend = mock.create_autospec(_recording_email_backend)
4041+
4042+
class FailingOperator(BaseOperator):
4043+
def execute(self, context):
4044+
raise AirflowFailException("Task failed on purpose")
4045+
4046+
task = FailingOperator(
4047+
task_id="legacy_backend_task",
4048+
email=["[email protected]"],
4049+
email_on_failure=True,
4050+
)
4051+
4052+
runtime_ti = create_runtime_ti(task=task)
4053+
context = runtime_ti.get_template_context()
4054+
log = mock.MagicMock()
4055+
4056+
with conf_vars(
4057+
{
4058+
("email", "email_backend"): f"{__name__}._recording_email_backend",
4059+
("email", "email_conn_id"): "my_smtp",
4060+
("email", "from_email"): self.FROM,
4061+
}
4062+
):
4063+
with mock.patch(f"{__name__}._recording_email_backend", backend):
4064+
with mock.patch(
4065+
"airflow.providers.smtp.notifications.smtp.SmtpNotifier", autospec=True
4066+
) as mock_smtp_notifier:
4067+
state, _, error = run(runtime_ti, context, log)
4068+
finalize(runtime_ti, state, context, log, error)
4069+
4070+
# The default SMTP notifier must not be used when a custom backend is configured.
4071+
mock_smtp_notifier.assert_not_called()
4072+
backend.assert_called_once()
4073+
args, kwargs = backend.call_args
4074+
# send_email(to, subject, html_content, conn_id=..., from_email=...)
4075+
assert args[0] == ["[email protected]"]
4076+
assert kwargs["conn_id"] == "my_smtp"
4077+
assert kwargs["from_email"] == self.FROM
4078+
4079+
def test_unresolvable_email_backend_is_logged(self, create_runtime_ti, mock_supervisor_comms):
4080+
"""An unimportable custom backend is logged, and does not silently fall back to SMTP."""
4081+
from airflow.sdk.exceptions import AirflowFailException
4082+
from airflow.sdk.execution_time.task_runner import finalize, run
4083+
4084+
class FailingOperator(BaseOperator):
4085+
def execute(self, context):
4086+
raise AirflowFailException("Task failed on purpose")
4087+
4088+
task = FailingOperator(
4089+
task_id="bad_backend_task",
4090+
email=["[email protected]"],
4091+
email_on_failure=True,
4092+
)
4093+
4094+
runtime_ti = create_runtime_ti(task=task)
4095+
context = runtime_ti.get_template_context()
4096+
log = mock.MagicMock()
4097+
4098+
with conf_vars({("email", "email_backend"): "airflow.does.not.Exist"}):
4099+
# SmtpNotifier is patched to succeed, so a logged exception can only come from
4100+
# resolving the custom backend -- without that, this would pass pre-fix too.
4101+
with mock.patch(
4102+
"airflow.providers.smtp.notifications.smtp.SmtpNotifier", autospec=True
4103+
) as mock_smtp_notifier:
4104+
state, _, error = run(runtime_ti, context, log)
4105+
# Must not raise even though the backend cannot be loaded.
4106+
finalize(runtime_ti, state, context, log, error)
4107+
4108+
mock_smtp_notifier.assert_not_called()
4109+
log.exception.assert_called()
4110+
40114111
@pytest.mark.enable_redact
40124112
def test_rendered_templates_mask_secrets(self, create_runtime_ti, mock_supervisor_comms):
40134113
"""Test that secrets registered with mask_secret() are redacted in rendered template fields."""

0 commit comments

Comments
 (0)