This is an automated email from the ASF dual-hosted git repository.
ferruzzi pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new f792bd60a98 Do not abort Dag run creation when a deadline alert fails
(#68917)
f792bd60a98 is described below
commit f792bd60a9852dd91f1281f825e8c92e28f386de
Author: Sean Ghaeli <[email protected]>
AuthorDate: Tue Oct 13 12:53:21 2026 -0700
Do not abort Dag run creation when a deadline alert fails (#68917)
* Do not abort Dag run creation when a deadline alert fails
Deadline alerts are an observability feature layered on top of a Dag run, so
they must never be able to stop one from existing. Decoding a stored alert,
resolving its interval and evaluating its reference all happen inside
create_dagrun, and any of them could raise straight out through Dag run
creation -- so one malformed or unresolvable alert took the whole Dag run
down
with it, and on a Dag with several alerts the first failure starved the
rest of
their deadlines.
The new deadline_alerts.deadline_creation_failed counter keeps these skips
visible rather than silent.
---
airflow-core/newsfragments/68917.bugfix.rst | 1 +
.../src/airflow/serialization/definitions/dag.py | 122 ++++++++++------
.../airflow/serialization/definitions/deadline.py | 9 +-
airflow-core/src/airflow/utils/sqlalchemy.py | 5 +-
airflow-core/tests/unit/models/test_dagrun.py | 158 ++++++++++++++++++---
.../serialization/definitions/test_deadline.py | 1 +
.../observability/metrics/metrics_template.yaml | 6 +
7 files changed, 237 insertions(+), 65 deletions(-)
diff --git a/airflow-core/newsfragments/68917.bugfix.rst
b/airflow-core/newsfragments/68917.bugfix.rst
new file mode 100644
index 00000000000..d966d7ba5b9
--- /dev/null
+++ b/airflow-core/newsfragments/68917.bugfix.rst
@@ -0,0 +1 @@
+A deadline alert that cannot be decoded, or whose interval or reference fails
to resolve, no longer aborts creation of the Dag run it belongs to. The failing
alert is logged and skipped, the remaining alerts still get their deadlines,
and a new ``deadline_alerts.deadline_creation_failed`` counter is emitted for
it.
diff --git a/airflow-core/src/airflow/serialization/definitions/dag.py
b/airflow-core/src/airflow/serialization/definitions/dag.py
index f95692aad50..c529bf95ade 100644
--- a/airflow-core/src/airflow/serialization/definitions/dag.py
+++ b/airflow-core/src/airflow/serialization/definitions/dag.py
@@ -50,12 +50,13 @@ from airflow.models.deadline import Deadline
from airflow.models.deadline_alert import DeadlineAlert as DeadlineAlertModel
from airflow.models.task_coordinates import public_map_index_expression
from airflow.models.tasklog import LogTemplate
-from airflow.serialization.decoders import decode_deadline_alert_model,
resolve_deadline_alert_interval
-from airflow.serialization.definitions.deadline import
SerializedReferenceModels
+from airflow.serialization.decoders import decode_deadline_alert,
resolve_deadline_alert_interval
+from airflow.serialization.definitions.deadline import DeadlineAlertFields,
SerializedReferenceModels
from airflow.serialization.definitions.param import SerializedParamsDict
from airflow.timetables.base import DagRunInfo, DataInterval, TimeRestriction
from airflow.utils.helpers import prune_dict
from airflow.utils.session import NEW_SESSION, provide_session
+from airflow.utils.sqlalchemy import PROHIBIT_COMMIT_ERROR_MESSAGE
from airflow.utils.state import DagRunState, TaskInstanceState
from airflow.utils.types import DagRunType
@@ -754,54 +755,87 @@ class SerializedDAG:
select(DeadlineAlertModel).where(DeadlineAlertModel.serialized_dag_id ==
serialized_dag_id)
).all()
- for deadline_alert in deadline_alert_records:
- if not deadline_alert:
- continue
-
- deserialized_deadline_alert =
decode_deadline_alert_model(deadline_alert)
- interval =
resolve_deadline_alert_interval(deserialized_deadline_alert, session=session)
+ if not deadline_alert_records:
+ return
- if isinstance(deserialized_deadline_alert.reference,
SerializedReferenceModels.TYPES.DAGRUN):
- deadline_time =
deserialized_deadline_alert.reference.evaluate_with(
- session=session,
- interval=interval,
- # TODO : Pretty sure we can drop these last two; verify
after testing is complete
- dag_id=self.dag_id,
- run_id=orm_dagrun.run_id,
+ team_name = (
+ DagModel.get_team_name(self.dag_id, session=session)
+ if airflow_conf.getboolean("core", "multi_team")
+ else None
+ )
+ metrics_tags = prune_dict({"dag_id": self.dag_id, "team_name":
team_name})
+
+ # Read every value off the ORM up front. A failing alert can leave the
session rolled back
+ # and its instances detached, so nothing after this point may touch
ORM state -- otherwise
+ # the skip path raises DetachedInstanceError and takes the DagRun down
anyway.
+ run_id = orm_dagrun.run_id
+ dagrun_id = orm_dagrun.id
+ dagrun_dag_id = orm_dagrun.dag_id
+ bundle_name = orm_dagrun.dag_model.bundle_name
+ stored_alerts = [
+ (alert.id, alert.name, alert.reference, alert.interval,
alert.callback_def)
+ for alert in deadline_alert_records
+ if alert
+ ]
+
+ for alert_id, alert_name, alert_reference, alert_interval,
alert_callback_def in stored_alerts:
+ # Deadline creation is best-effort. Catch per alert so one bad
alert cannot starve the rest or abort the DagRun.
+ try:
+ deserialized_deadline_alert = decode_deadline_alert(
+ {
+ DeadlineAlertFields.NAME: alert_name,
+ DeadlineAlertFields.REFERENCE: alert_reference,
+ DeadlineAlertFields.INTERVAL: alert_interval,
+ DeadlineAlertFields.CALLBACK: alert_callback_def,
+ }
)
+ interval =
resolve_deadline_alert_interval(deserialized_deadline_alert, session=session)
- if deadline_time is not None:
- session.add(
- Deadline(
- deadline_time=deadline_time,
- callback=deserialized_deadline_alert.callback,
- dagrun_id=orm_dagrun.id,
- deadline_alert_id=deadline_alert.id,
- dag_id=orm_dagrun.dag_id,
- bundle_name=orm_dagrun.dag_model.bundle_name,
- )
- )
- team_name = (
- DagModel.get_team_name(self.dag_id, session=session)
- if airflow_conf.getboolean("core", "multi_team")
- else None
- )
- stats.incr(
- "deadline_alerts.deadline_created",
- tags=prune_dict({"dag_id": self.dag_id, "team_name":
team_name}),
- )
- elif required_dagrun_column :=
_DAGRUN_REFERENCE_REQUIRED_COLUMNS.get(
- type(deserialized_deadline_alert.reference)
- ):
- log.warning(
- "skipping deadline alert because the deadline
reference evaluated to None",
+ if isinstance(deserialized_deadline_alert.reference,
SerializedReferenceModels.TYPES.DAGRUN):
+ deadline_time =
deserialized_deadline_alert.reference.evaluate_with(
+ session=session,
+ interval=interval,
+ # TODO : Pretty sure we can drop these last two;
verify after testing is complete
dag_id=self.dag_id,
- run_id=orm_dagrun.run_id,
- deadline_alert_id=deadline_alert.id,
-
reference_type=deserialized_deadline_alert.reference.reference_name,
- required_dagrun_column=required_dagrun_column,
+ run_id=run_id,
)
+ if deadline_time is not None:
+ session.add(
+ Deadline(
+ deadline_time=deadline_time,
+ callback=deserialized_deadline_alert.callback,
+ dagrun_id=dagrun_id,
+ deadline_alert_id=alert_id,
+ dag_id=dagrun_dag_id,
+ bundle_name=bundle_name,
+ )
+ )
+ stats.incr("deadline_alerts.deadline_created",
tags=metrics_tags)
+ elif required_dagrun_column :=
_DAGRUN_REFERENCE_REQUIRED_COLUMNS.get(
+ type(deserialized_deadline_alert.reference)
+ ):
+ log.warning(
+ "skipping deadline alert because the deadline
reference evaluated to None",
+ dag_id=self.dag_id,
+ run_id=run_id,
+ deadline_alert_id=alert_id,
+
reference_type=deserialized_deadline_alert.reference.reference_name,
+ required_dagrun_column=required_dagrun_column,
+ )
+ except Exception as exc:
+ if isinstance(exc, RuntimeError) and
PROHIBIT_COMMIT_ERROR_MESSAGE in str(exc):
+ # A rejected commit means the scheduler's HA locking is
already broken. Recording
+ # that as a skipped alert would hide it, so let it escape
instead.
+ raise
+ log.exception(
+ "skipping deadline alert because creating its deadline
failed",
+ dag_id=self.dag_id,
+ run_id=run_id,
+ deadline_alert_id=alert_id,
+ )
+ stats.incr("deadline_alerts.deadline_creation_failed",
tags=metrics_tags)
+
@provide_session
def set_task_instance_state(
self,
diff --git a/airflow-core/src/airflow/serialization/definitions/deadline.py
b/airflow-core/src/airflow/serialization/definitions/deadline.py
index e4362338cc0..25399aabc62 100644
--- a/airflow-core/src/airflow/serialization/definitions/deadline.py
+++ b/airflow-core/src/airflow/serialization/definitions/deadline.py
@@ -406,7 +406,14 @@ class SerializedVariableInterval:
f"VariableInterval '{self.key}' must be an integer (seconds),
got: {value!r}"
) from e
- return timedelta(seconds=seconds)
+ try:
+ return timedelta(seconds=seconds)
+ except OverflowError as e:
+ # ``timedelta`` raises OverflowError, which is not a ValueError,
so callers
+ # inspecting for a bad interval would otherwise miss it.
+ raise ValueError(
+ f"VariableInterval '{self.key}' is too large to be a valid
interval: {seconds} seconds"
+ ) from e
@attrs.define
diff --git a/airflow-core/src/airflow/utils/sqlalchemy.py
b/airflow-core/src/airflow/utils/sqlalchemy.py
index 6b7db82b8ef..5d90ef63ec0 100644
--- a/airflow-core/src/airflow/utils/sqlalchemy.py
+++ b/airflow-core/src/airflow/utils/sqlalchemy.py
@@ -682,6 +682,9 @@ def with_db_lock_timeout(session: Session, lock_timeout:
int = 30) -> Generator[
session.execute(text(f"SET SESSION innodb_lock_wait_timeout =
{old_mysql_timeout}"))
+PROHIBIT_COMMIT_ERROR_MESSAGE = "UNEXPECTED COMMIT - THIS WILL BREAK HA LOCKS!"
+
+
class CommitProhibitorGuard:
"""Context manager class that powers prohibit_commit."""
@@ -694,7 +697,7 @@ class CommitProhibitorGuard:
if self.expected_commit:
self.expected_commit = False
return
- raise RuntimeError("UNEXPECTED COMMIT - THIS WILL BREAK HA LOCKS!")
+ raise RuntimeError(PROHIBIT_COMMIT_ERROR_MESSAGE)
def __enter__(self) -> Self:
event.listen(self.session, "before_commit", self._validate_commit)
diff --git a/airflow-core/tests/unit/models/test_dagrun.py
b/airflow-core/tests/unit/models/test_dagrun.py
index 25795e40d8e..4e00a1392ea 100644
--- a/airflow-core/tests/unit/models/test_dagrun.py
+++ b/airflow-core/tests/unit/models/test_dagrun.py
@@ -21,7 +21,7 @@ import datetime
from collections import defaultdict
from collections.abc import Mapping
from concurrent.futures import ThreadPoolExecutor
-from contextlib import contextmanager
+from contextlib import contextmanager, nullcontext
from functools import partial, reduce
from typing import TYPE_CHECKING
from unittest import mock
@@ -1954,30 +1954,150 @@ class TestDagRun:
"log_level": "warning",
} not in caplog
+ @pytest.mark.parametrize(
+ ("interval", "failure"),
+ [
+ pytest.param(
+ VariableInterval("missing_key"),
+ mock.patch.object(Variable, "get", side_effect=KeyError),
+ id="unresolvable_interval",
+ ),
+ pytest.param(
+ datetime.timedelta(hours=1),
+ mock.patch(
+
"airflow.serialization.definitions.dag.decode_deadline_alert",
+ autospec=True,
+ side_effect=ValueError("corrupt deadline alert blob"),
+ ),
+ id="undecodable_alert",
+ ),
+ pytest.param(
+ datetime.timedelta(hours=1),
+ mock.patch.object(
+ SerializedReferenceModels.FixedDatetimeDeadline,
+ "evaluate_with",
+ autospec=True,
+ side_effect=RuntimeError("evaluate_with failed"),
+ ),
+ id="unevaluable_reference",
+ ),
+ ],
+ )
+ @mock.patch("airflow._shared.observability.metrics.stats.incr")
@mock.patch.object(Deadline, "prune_deadlines")
- def test_dagrun_deadline_variable_interval_missing_variable_fails(self, _,
session, deadline_test_dag):
+ def test_dagrun_deadline_failure_does_not_abort_dagrun(
+ self, _, mock_stats_incr, interval, failure, session, deadline_test_dag
+ ):
+ future_date = datetime.datetime(2037, 1, 1, tzinfo=datetime.UTC)
- with mock.patch.object(
- Variable,
- "get",
- side_effect=KeyError,
+ scheduler_dag = deadline_test_dag(
+ deadline=DeadlineAlert(
+ reference=DeadlineReference.FIXED_DATETIME(future_date),
+ interval=interval,
+ callback=AsyncCallback(empty_callback_for_deadline),
+ ),
+ )
+
+ with (
+ conf_vars({("core", "multi_team"): "true"}),
+ mock.patch("airflow.models.dag.DagModel.get_team_name",
return_value="team_alpha"),
+ failure,
):
- future_date = datetime.datetime.now() +
datetime.timedelta(days=365)
+ dag_run = self.create_dag_run(
+ dag=scheduler_dag,
+ task_states={"task_1": TaskInstanceState.SUCCESS},
+ session=session,
+ )
+
+ assert dag_run is not None
+ assert session.execute(select(Deadline)).scalars().one_or_none() is
None
+ mock_stats_incr.assert_any_call(
+ "deadline_alerts.deadline_creation_failed",
+ tags={"dag_id": scheduler_dag.dag_id, "team_name": "team_alpha"},
+ )
- scheduler_dag = deadline_test_dag(
- deadline=DeadlineAlert(
- reference=DeadlineReference.FIXED_DATETIME(future_date),
- interval=VariableInterval("missing_key"),
- callback=AsyncCallback(empty_callback_for_deadline),
+ @mock.patch("airflow._shared.observability.metrics.stats.incr")
+ @mock.patch.object(Deadline, "prune_deadlines")
+ @mock.patch.object(Variable, "get")
+ def test_dagrun_deadline_failure_that_detaches_orm_objects_still_skips(
+ self, mock_variable_get, _, mock_stats_incr, session, deadline_test_dag
+ ):
+ """A failing interval resolution can roll back and close the caller's
session underneath
+ the handler (``create_session`` reusing the scoped session does
exactly that), detaching
+ every ORM instance. The skip path must still log and count without
touching ORM state."""
+ scheduler_dag = deadline_test_dag(
+ deadline=DeadlineAlert(
+ reference=DeadlineReference.FIXED_DATETIME(
+ datetime.datetime(2037, 1, 1, tzinfo=datetime.UTC)
),
- )
+ interval=VariableInterval("missing_key"),
+ callback=AsyncCallback(empty_callback_for_deadline),
+ ),
+ )
+ # The alert rows must be persistent from an earlier transaction (as in
the live
+ # scheduler, where the dag processor committed them) so the rollback
below expires
+ # them and the close detaches them instead of returning them to
transient.
+ session.commit()
- with pytest.raises(ValueError, match="not found"):
- self.create_dag_run(
- dag=scheduler_dag,
- task_states={"task_1": TaskInstanceState.SUCCESS},
- session=session,
- )
+ def teardown_session_and_raise(*args, **kwargs):
+ session.rollback()
+ session.close()
+ raise KeyError("missing_key")
+
+ mock_variable_get.side_effect = teardown_session_and_raise
+
+ dag_run = self.create_dag_run(dag=scheduler_dag, session=session)
+
+ assert dag_run is not None
+ mock_stats_incr.assert_any_call(
+ "deadline_alerts.deadline_creation_failed",
+ tags={"dag_id": scheduler_dag.dag_id},
+ )
+
+ @pytest.mark.parametrize(
+ ("interval", "expect_deadline", "variable_value"),
+ [
+ pytest.param(datetime.timedelta(seconds=5), True, None,
id="deadline_created"),
+ pytest.param(VariableInterval("missing_key"), False, None,
id="deadline_skipped"),
+ pytest.param(VariableInterval("resolvable_key"), True, "4200",
id="variable_interval_resolved"),
+ ],
+ )
+ @mock.patch.object(Deadline, "prune_deadlines")
+ def test_dagrun_deadline_handling_does_not_commit(
+ self, _, interval, expect_deadline, variable_value, session,
deadline_test_dag
+ ):
+ """DagRuns are created under ``prohibit_commit``, so creating a
deadline, skipping one whose
+ interval will not resolve, and resolving a ``VariableInterval`` out of
the metadata database
+ all have to leave the scheduler's session uncommitted."""
+ future_date = datetime.datetime(2037, 1, 1, tzinfo=datetime.UTC)
+ scheduler_dag = deadline_test_dag(
+ deadline=DeadlineAlert(
+ reference=DeadlineReference.FIXED_DATETIME(future_date),
+ interval=interval,
+ callback=AsyncCallback(empty_callback_for_deadline),
+ ),
+ )
+
+ if variable_value is None:
+ variable_lookup = mock.patch.object(Variable, "get",
side_effect=KeyError)
+ else:
+ # A real row committed before the guard opens, so resolve() has to
go through
+ # MetastoreBackend. Drop the session forwarding and the backend
opens the same scoped
+ # session and commits it, which the guard rejects and the lookup
then reports as missing.
+ Variable.set(key=interval.key, value=variable_value,
session=session)
+ session.commit()
+ variable_lookup = nullcontext()
+
+ # No task_states: the helper's get_task_instance() opens its own
session and would trip the guard.
+ with prohibit_commit(session), variable_lookup:
+ dag_run = self.create_dag_run(dag=scheduler_dag, session=session)
+ session.flush()
+
+ assert dag_run is not None
+ deadline = session.execute(select(Deadline)).scalars().one_or_none()
+ assert (deadline is not None) is expect_deadline
+ if variable_value is not None:
+ assert deadline.deadline_time == future_date +
datetime.timedelta(seconds=int(variable_value))
@pytest.mark.parametrize(
diff --git a/airflow-core/tests/unit/serialization/definitions/test_deadline.py
b/airflow-core/tests/unit/serialization/definitions/test_deadline.py
index f0fad39a37e..643d03bbb95 100644
--- a/airflow-core/tests/unit/serialization/definitions/test_deadline.py
+++ b/airflow-core/tests/unit/serialization/definitions/test_deadline.py
@@ -61,6 +61,7 @@ class TestVariableInterval:
(None, True, "not found"),
("abc", False, "must be an integer"),
("", False, "must be an integer"),
+ ("99999999999999", False, "too large to be a valid interval"),
],
)
def test_resolve_invalid(self, mocker, value, raise_missing, match):
diff --git
a/shared/observability/src/airflow_shared/observability/metrics/metrics_template.yaml
b/shared/observability/src/airflow_shared/observability/metrics/metrics_template.yaml
index 9aa2b7a4a9d..4782f69b7d7 100644
---
a/shared/observability/src/airflow_shared/observability/metrics/metrics_template.yaml
+++
b/shared/observability/src/airflow_shared/observability/metrics/metrics_template.yaml
@@ -326,6 +326,12 @@ metrics:
legacy_name: "-"
name_variables: []
+ - name: "deadline_alerts.deadline_creation_failed"
+ description: "Number of deadline alerts skipped because their deadline
could not be created"
+ type: "counter"
+ legacy_name: "-"
+ name_variables: []
+
- name: "deadline_alerts.deadline_missed"
description: "Number of deadline alerts that fired because a Dag run
missed its deadline"
type: "counter"
