Skip to content

Commit fd671dc

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 fd671dc

5 files changed

Lines changed: 246 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: 60 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,60 @@
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.sdk.configuration import conf
24+
from airflow.sdk.exceptions import AirflowConfigException
25+
from airflow.sdk.execution_time.email_backend import LegacyEmailBackendNotifier
26+
27+
28+
class TestLegacyEmailBackendNotifier:
29+
def test_notify_calls_configured_backend(self):
30+
"""notify() loads the legacy backend from config and calls it with the standard fields."""
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+
notifier = LegacyEmailBackendNotifier(to="a@b.com")
58+
with mock.patch.object(conf, "getimport", return_value=None):
59+
with pytest.raises(AirflowConfigException):
60+
notifier.notify(context={})

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

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

38443844

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

@@ -4008,6 +4013,76 @@ def execute(self, context):
40084013
)
40094014
assert kwargs["from_email"] == self.FROM
40104015

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

0 commit comments

Comments
 (0)