Resolve VariableInterval deadlines safely at DagRun creation

airflow-core/newsfragments/68917.bugfix.rst

@@ -0,0 +1 @@
+Deadline alerts using a ``VariableInterval`` no longer risk aborting DagRun creation. The interval is now resolved through the full secrets chain (environment variables, configured secrets backends, then the metadata database) on the scheduler's own session, so ``AIRFLOW_VAR_*`` and secrets-backend-backed Variables resolve correctly and the read does not commit inside the scheduler's ``prohibit_commit`` guard. Each deadline alert is also isolated: a single unresolvable or undecodable alert is logged and skipped instead of preventing the DagRun from being created.

airflow-core/src/airflow/serialization/definitions/dag.py

@@ -31,8 +31,9 @@
 from sqlalchemy import func, or_, select, tuple_
 
 from airflow._shared.observability.metrics import stats
+from airflow._shared.secrets_backend.base import call_secrets_backend_method
 from airflow._shared.timezones.timezone import coerce_datetime
-from airflow.configuration import conf as airflow_conf
+from airflow.configuration import conf as airflow_conf, ensure_secrets_loaded
 from airflow.exceptions import (
     AirflowException,
     DagNotPartitionedError,
@@ -50,7 +51,9 @@
 from airflow.models.deadline_alert import DeadlineAlert as DeadlineAlertModel
 from airflow.models.taskinstancekey import TaskInstanceKey
 from airflow.models.tasklog import LogTemplate
+from airflow.models.variable import AirflowSecretsBackendAccessDenied
 from airflow.sdk.definitions.deadline import VariableInterval
+from airflow.secrets.metastore import MetastoreBackend
 from airflow.serialization.decoders import decode_deadline_alert
 from airflow.serialization.definitions.deadline import DeadlineAlertFields, SerializedReferenceModels
 from airflow.serialization.definitions.param import SerializedParamsDict
@@ -737,55 +740,118 @@ def _process_dagrun_deadline_alerts(
             select(DeadlineAlertModel).where(DeadlineAlertModel.serialized_dag_id == serialized_dag_id)
         ).all()
 
+        if not deadline_alert_records:
+            return
+
+        # Resolve the DagRun's team once, so a team-scoped VariableInterval is looked up
+        # against the right team (not the global scope) and the stats tags are consistent.
+        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})
+
         for deadline_alert in deadline_alert_records:
             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, team_name=team_name, 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}),
-                    )
+                        stats.incr("deadline_alerts.deadline_created", tags=metrics_tags)
+            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=metrics_tags)
+
+    @staticmethod
+    def _resolve_variable_interval(
+        interval: VariableInterval, *, team_name: str | None, 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 team_name: Team owning the DagRun, forwarded to scope the Variable lookup.
+        :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``.
+        :raises AirflowSecretsBackendAccessDenied: If a backend authoritatively denies access.
+        """
+        # Collapse this loop into Variable.get_variable_from_secrets once it can reuse the
+        # caller's session; tracked at https://github.com/apache/airflow/issues/71801
+        for backend in ensure_secrets_loaded():
+            try:
+                value = call_secrets_backend_method(
+                    backend.get_variable,
+                    team_name=team_name,
+                    key=interval.key,
+                    **({"session": session} if isinstance(backend, MetastoreBackend) else {}),
+                )
+            except AirflowSecretsBackendAccessDenied:
+                # Authoritative deny — must NOT fall through to a less-restrictive backend.
+                raise
+            except Exception:
+                log.exception(
+                    "Unable to retrieve variable from secrets backend (%s). "
+                    "Checking subsequent secrets backend.",
+                    type(backend).__name__,
+                )
+                continue
+            if value is not None:
+                return interval.coerce_to_timedelta(value)
+        raise ValueError(
+            f"VariableInterval '{interval.key}' could not be resolved from any "
+            f"secrets backend, environment variable, or the metadata database"
+        )
 
     @provide_session
     def set_task_instance_state(