ferruzzi commented on code in PR #68917:
URL: https://github.com/apache/airflow/pull/68917#discussion_r4010294598
##########
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:
I agree on principle but there's one hitch here. We need to re-raise the
`RuntimeError` that a session collision can raise here. This is part of what
I've been trying to untangle, but basically this code falls inside the
scheduler's `prohibit_commit` guard which can raise `RuntimeError("UNEXPECTED
COMMIT - THIS WILL BREAK HA LOCKS!")` and we need that one specific message
bubbling up or you'll be swallowing legitimate database safety issues.
It won't cost you the DagRun (which you rightly used to push back earlier)
since the guard fires inside `session.commit()`, so by the time your `except`
sees it, the scheduler has already rolled it back and closed the session. So
in this case, the DagRun is already shot and we should capture why.
To be clear, this isn't something you could have necessarily seen since you
forward session to both `resolve()` and `evaluate_with()`, but it becomes an
issue once #70370 lands. I'm still (and I'm sorry for how long it's taking)
just paving the way for easier rebases/merges later and you're next in the
chain behind #70370.
--
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]
