ferruzzi commented on code in PR #68917:
URL: https://github.com/apache/airflow/pull/68917#discussion_r4010427616
##########
airflow-core/src/airflow/serialization/definitions/dag.py:
##########
@@ -715,51 +715,125 @@ def _process_dagrun_deadline_alerts(
if not deadline_alert:
continue
- deserialized_deadline_alert = decode_deadline_alert(
- {
- Encoding.TYPE: DAT.DEADLINE_ALERT,
- Encoding.VAR: {
- DeadlineAlertFields.REFERENCE:
deadline_alert.reference,
- DeadlineAlertFields.INTERVAL: deadline_alert.interval,
- DeadlineAlertFields.CALLBACK:
deadline_alert.callback_def,
- },
- }
- )
+ # Isolate each deadline alert: creating a deadline is auxiliary to
creating the
+ # DagRun itself, and must never prevent the DagRun from being
created. A single bad
+ # alert -- e.g. a ``VariableInterval`` whose backing Variable is
missing / non-integer
+ # / <= 0 (``coerce_to_timedelta`` raises ``ValueError``), or a
reference whose
+ # ``evaluate_with`` fails -- would otherwise propagate out of
``create_dagrun`` and
+ # abort the whole run, silently stopping the DAG from scheduling.
+ #
+ # Isolation is done with a plain ``try``/``except`` and MUST NOT
use
+ # ``session.begin_nested()``: ``create_dagrun`` runs on the
scheduler session inside
+ # the ``prohibit_commit`` guard, and releasing a SAVEPOINT issues
a commit, which trips
+ # that guard (``RuntimeError("UNEXPECTED COMMIT ...")``) and --
because this very block
+ # swallows it -- silently skips deadline creation for *every*
scheduled DagRun. The
+ # try/except alone is sufficient: the only DB mutation in the loop
body is the final
+ # ``session.add`` (everything before it is a decode, an in-memory
resolution, or a
+ # read-only ``evaluate_with`` query), so an exception leaves no
partial state to undo,
+ # and the pending ``Deadline`` is persisted by the caller's outer
transaction.
+ try:
+ deserialized_deadline_alert = decode_deadline_alert(
+ {
+ Encoding.TYPE: DAT.DEADLINE_ALERT,
+ Encoding.VAR: {
+ DeadlineAlertFields.REFERENCE:
deadline_alert.reference,
+ DeadlineAlertFields.INTERVAL:
deadline_alert.interval,
+ DeadlineAlertFields.CALLBACK:
deadline_alert.callback_def,
+ },
+ }
+ )
- interval = deserialized_deadline_alert.interval
+ interval = deserialized_deadline_alert.interval
- if isinstance(interval, VariableInterval):
- interval = interval.resolve()
+ if isinstance(interval, VariableInterval):
+ interval = self._resolve_variable_interval(interval,
session=session)
- 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,
- )
+ 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,
+ )
- 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,
+ 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}),
- )
+ 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}),
+ )
+ except Exception:
Review Comment:
Sorry for the churn. I should have waited until i was done reviewing before
I posted that. You already have a `prohibit_commit` test, so you likely
already knew most of that. The request for reraising the RuntimeError stands,
though.
##########
airflow-core/tests/unit/models/test_dagrun.py:
##########
@@ -1610,30 +1610,136 @@ def
test_dagrun_deadline_does_not_warn_for_average_runtime_without_history(
"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.timezone.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.timezone.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"),
+ [
+ pytest.param(datetime.timedelta(seconds=5), True,
id="deadline_created"),
+ pytest.param(VariableInterval("missing_key"), False,
id="deadline_skipped"),
+ ],
+ )
+ @mock.patch.object(Deadline, "prune_deadlines")
+ def test_dagrun_deadline_handling_does_not_commit(
Review Comment:
This test almost catches the problem in the other thread, and I think one
more case would be great.
Both of these param cases avoid resolving a `VariableInterval` successfully
under the guard. `deadline_created` uses a plain `timedelta` so `resolve()`
is never called, and `deadline_skipped` has `Variable.get` raise a `KeyError`
before anything can commit. Since the `mock.patch.object(Variable, "get",
side_effect=KeyError)` on line 1736 applies to both, there is no path here
where the `Variable` lookup actually succeeds inside `prohibit_commit`.
Can you add a third case along the lines of
`pytest.param(VariableInterval("some_key"), True,
id="variable_interval_resolved")` with `Variable.get` returning some
stringified integer ("4200" or whatever) rather than raising? That should hit
the session-collision section and would fail if anyone drops the `session`
forwarding.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
