ferruzzi commented on code in PR #68917:
URL: https://github.com/apache/airflow/pull/68917#discussion_r3694083320
##########
airflow-core/src/airflow/serialization/definitions/dag.py:
##########
@@ -741,51 +743,93 @@ 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,
- },
- }
- )
+ # Deadline creation is best-effort. A failure here must not
prevent the DagRun
+ # itself from being created. Use a plain try/except rather than
+ # ``session.begin_nested()`` since ``create_dagrun`` runs under
+ # ``prohibit_commit`` and releasing a SAVEPOINT would trip that
guard.
+ 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:
+ log.exception(
+ "Failed to create deadline for alert %s on DagRun %s
(dag_id=%s); "
+ "skipping this deadline, the DagRun is unaffected",
+ getattr(deadline_alert, "id", "<unknown>"),
+ orm_dagrun.run_id,
+ self.dag_id,
+ )
+ stats.incr("deadline_alerts.deadline_creation_failed",
tags={"dag_id": self.dag_id})
+
+ @staticmethod
+ def _resolve_variable_interval(interval: VariableInterval, *, session:
Session) -> datetime.timedelta:
+ """
+ Resolve a ``VariableInterval`` to a concrete ``timedelta`` at DagRun
creation.
+
+ The Variable is resolved using the standard secrets lookup order. The
scheduler
+ session is passed to the metastore backend to avoid creating a new
session
+ during DagRun creation.
+
+ :param interval: The ``VariableInterval`` to resolve.
+ :param session: Scheduler session used for metadata database lookups.
+ :return: The resolved ``timedelta``.
+ :raises ValueError: If the Variable cannot be resolved or converted to
a valid ``timedelta``.
+ """
+ for backend in ensure_secrets_loaded():
+ value = call_secrets_backend_method(
+ backend.get_variable,
+ team_name=None,
Review Comment:
Should `team_name` be hardcoded to `None` here?? Seems like it should check
if there is a team unless I;'m missing something.
--
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]
