Skip to content

Commit e7b6e90

Browse files
github-actions[bot]potiuk
authored and
vatsrahul1001
committed
[v3-3-test] Restore pluggable email backend for task failure and retry alerts (#69877) (#70129)
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. (cherry picked from commit f7dec02) Co-authored-by: Jarek Potiuk <jarek@potiuk.com>
1 parent 7906cd7 commit e7b6e90

6 files changed

Lines changed: 394 additions & 14 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.

‎airflow-core/tests/unit/dag_processing/test_processor.py‎

Lines changed: 84 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -24,9 +24,9 @@
2424
import textwrap
2525
import typing
2626
import uuid
27-
from collections.abc import Callable
27+
from collections.abc import Callable, Iterable
2828
from socket import socketpair
29-
from typing import TYPE_CHECKING, BinaryIO
29+
from typing import TYPE_CHECKING, Any, BinaryIO
3030
from unittest.mock import MagicMock, patch
3131

3232
import pytest
@@ -1571,6 +1571,24 @@ def fake_collect_dags(self, *args, **kwargs):
15711571
_execute_task_callbacks(dagbag, request, log)
15721572

15731573

1574+
def _recording_email_backend(
1575+
to: list[str] | Iterable[str],
1576+
subject: str,
1577+
html_content: str,
1578+
files: list[str] | None = None,
1579+
dryrun: bool = False,
1580+
cc: str | Iterable[str] | None = None,
1581+
bcc: str | Iterable[str] | None = None,
1582+
mime_subtype: str = "mixed",
1583+
mime_charset: str = "utf-8",
1584+
conn_id: str | None = None,
1585+
custom_headers: dict[str, Any] | None = None,
1586+
**kwargs,
1587+
) -> None:
1588+
"""Legacy ``[email] email_backend`` stub, patched with a spec'd mock in tests."""
1589+
raise AssertionError("should be patched in the test")
1590+
1591+
15741592
class TestExecuteEmailCallbacks:
15751593
"""Test the email callback execution functionality."""
15761594

@@ -1923,6 +1941,70 @@ def test_parse_file_passes_bundle_name_to_dagbag(self):
19231941
call_kwargs = mock_dagbag_class.call_args.kwargs
19241942
assert call_kwargs["bundle_name"] == "test_bundle"
19251943

1944+
def test_execute_email_callbacks_uses_custom_email_backend(self):
1945+
"""The Dag-processor path honours a custom ``[email] email_backend``, like the worker path."""
1946+
backend = MagicMock(spec=_recording_email_backend)
1947+
dagbag = MagicMock(spec=DagBag)
1948+
with DAG(dag_id="test_dag") as dag:
1949+
BaseOperator(task_id="test_task", email=["test@example.com"])
1950+
dagbag.dags = {"test_dag": dag}
1951+
1952+
current_time = timezone.utcnow()
1953+
request = EmailRequest(
1954+
filepath="/path/to/dag.py",
1955+
bundle_name="test_bundle",
1956+
bundle_version="1.0.0",
1957+
ti=TIDataModel(
1958+
id=str(uuid.uuid4()),
1959+
task_id="test_task",
1960+
dag_id="test_dag",
1961+
run_id="test_run",
1962+
logical_date="2023-01-01T00:00:00Z",
1963+
try_number=1,
1964+
attempt_number=1,
1965+
state="failed",
1966+
dag_version_id=str(uuid.uuid4()),
1967+
),
1968+
context_from_server=TIRunContext(
1969+
dag_run=DRDataModel(
1970+
dag_id="test_dag",
1971+
run_id="test_run",
1972+
logical_date="2023-01-01T00:00:00Z",
1973+
data_interval_start=current_time,
1974+
data_interval_end=current_time,
1975+
run_after=current_time,
1976+
start_date=current_time,
1977+
end_date=None,
1978+
run_type="manual",
1979+
state="running",
1980+
consumed_asset_events=[],
1981+
partition_key=None,
1982+
),
1983+
max_tries=2,
1984+
),
1985+
email_type="failure",
1986+
msg="Task failed",
1987+
)
1988+
1989+
conf_overrides = {
1990+
("email", "email_backend"): f"{__name__}._recording_email_backend",
1991+
("email", "email_conn_id"): "my_smtp",
1992+
("email", "from_email"): "from@airflow",
1993+
}
1994+
with conf_vars(conf_overrides):
1995+
with patch(f"{__name__}._recording_email_backend", backend):
1996+
with patch(
1997+
"airflow.providers.smtp.notifications.smtp.SmtpNotifier", autospec=True
1998+
) as mock_smtp_notifier:
1999+
_execute_email_callbacks(dagbag, request, MagicMock(spec=FilteringBoundLogger))
2000+
2001+
mock_smtp_notifier.assert_not_called()
2002+
backend.assert_called_once()
2003+
args, kwargs = backend.call_args
2004+
assert args[0] == ["test@example.com"]
2005+
assert kwargs["conn_id"] == "my_smtp"
2006+
assert kwargs["from_email"] == "from@airflow"
2007+
19262008

19272009
class TestDagProcessingMessageTypes:
19282010
def test_message_types_in_dag_processor(self):

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

Lines changed: 92 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,92 @@
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, Protocol
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 _ErrorEmailNotifier(Protocol):
35+
"""
36+
Constructor contract shared by the notifiers used for failure and retry alerts.
37+
38+
``BaseNotifier`` itself does not describe this -- its ``__init__`` takes only ``context`` --
39+
so the shared shape is spelled out here to keep the call site type-checked.
40+
"""
41+
42+
def __call__(
43+
self,
44+
to: str | Iterable[str],
45+
from_email: str | None = ...,
46+
subject: str | None = ...,
47+
html_content: str | None = ...,
48+
) -> BaseNotifier: ...
49+
50+
51+
class _LegacyEmailBackendNotifier(BaseNotifier):
52+
"""
53+
Adapter that exposes a legacy ``[email] email_backend`` callable as a notifier.
54+
55+
Before failure and retry alerts were routed through ``BaseNotifier`` subclasses, deployments
56+
configured them through ``[email] email_backend`` -- a callable with the
57+
``airflow.utils.email.send_email`` signature, such as the Amazon SES or SendGrid senders.
58+
This adapter renders the standard email fields like any notifier, then loads and calls the
59+
configured backend, so existing ``email_backend`` setups keep working unchanged.
60+
61+
The backend is resolved from config at notify time rather than imported statically, keeping
62+
the Task SDK free of a hard dependency on ``airflow.utils.email`` (which lives in
63+
``airflow-core``).
64+
"""
65+
66+
template_fields = ("to", "from_email", "subject", "html_content")
67+
68+
def __init__(
69+
self,
70+
to: str | Iterable[str],
71+
from_email: str | None = None,
72+
subject: str | None = None,
73+
html_content: str | None = None,
74+
**kwargs: Any,
75+
) -> None:
76+
super().__init__()
77+
self.to = to
78+
self.from_email = from_email
79+
self.subject = subject
80+
self.html_content = html_content
81+
82+
def notify(self, context: Context) -> None:
83+
backend = conf.getimport("email", "email_backend", fallback=_DEFAULT_EMAIL_BACKEND)
84+
if backend is None:
85+
raise AirflowConfigException("`[email] email_backend` is not configured")
86+
backend(
87+
self.to,
88+
self.subject,
89+
self.html_content,
90+
conn_id=conf.get("email", "email_conn_id", fallback=None),
91+
from_email=self.from_email,
92+
)

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

Lines changed: 36 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -136,6 +136,11 @@
136136
get_previous_dagrun_success,
137137
set_current_context,
138138
)
139+
from airflow.sdk.execution_time.email_backend import (
140+
_DEFAULT_EMAIL_BACKEND,
141+
_ErrorEmailNotifier,
142+
_LegacyEmailBackendNotifier,
143+
)
139144
from airflow.sdk.execution_time.sentry import Sentry
140145
from airflow.sdk.execution_time.xcom import XCom
141146
from airflow.sdk.listener import get_listener_manager
@@ -1997,20 +2002,39 @@ def _send_error_email_notification(
19972002
error: BaseException | str | None,
19982003
log: Logger,
19992004
) -> None:
2000-
"""Send email notification for task errors using SmtpNotifier."""
2001-
try:
2002-
from airflow.providers.smtp.notifications.smtp import SmtpNotifier
2003-
except ImportError:
2004-
log.error(
2005-
"Failed to send task failure or retry email notification: "
2006-
"`apache-airflow-providers-smtp` is not installed. "
2007-
"Install this provider to enable email notifications."
2008-
)
2009-
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.
20102012
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+
"""
20112017
if not task.email:
20122018
return
20132019

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: _ErrorEmailNotifier = _LegacyEmailBackendNotifier
2025+
notifier_description = f"configured 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+
20142038
subject_template_file = conf.get("email", "subject_template", fallback=None)
20152039

20162040
# Read the template file if configured
@@ -2053,15 +2077,15 @@ def _send_error_email_notification(
20532077
return
20542078

20552079
try:
2056-
notifier = SmtpNotifier(
2080+
notifier = notifier_class(
20572081
to=to_emails,
20582082
subject=subject,
20592083
html_content=html_content,
20602084
from_email=conf.get("email", "from_email", fallback="airflow@airflow"),
20612085
)
20622086
notifier(email_context)
20632087
except Exception:
2064-
log.exception("Failed to send email notification")
2088+
log.exception("Failed to send email notification via %s", notifier_description)
20652089

20662090

20672091
@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=["a@b.com"],
60+
from_email="from@x.com",
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+
["a@b.com"],
75+
"Subject",
76+
"<p>body</p>",
77+
conn_id="my_conn",
78+
from_email="from@x.com",
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="a@b.com")
84+
with mock.patch.object(conf, "getimport", autospec=True, return_value=None):
85+
with pytest.raises(AirflowConfigException):
86+
notifier.notify(context={})

0 commit comments

Comments
 (0)