ferruzzi commented on code in PR #68917:
URL: https://github.com/apache/airflow/pull/68917#discussion_r3884902298
##########
airflow-core/src/airflow/serialization/definitions/dag.py:
##########
@@ -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)
Review Comment:
You are going to end up having to rebase on top of a couple of other PRs
that need to land before this one. I'm trying to make that future rebase a bit
easier here.
#71767 has an elif branch right after the stats.incr you have on L800. If
you can add that here, it should ease your merge pain later:
```python
stats.incr("deadline_alerts.deadline_created", tags=metrics_tags)
elif required_dagrun_column := {
SerializedReferenceModels.DagRunLogicalDateDeadline: "logical_date",
SerializedReferenceModels.DagRunQueuedAtDeadline: "queued_at",
}.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=orm_dagrun.run_id,
deadline_alert_id=deadline_alert.id,
reference_type=deserialized_deadline_alert.reference.reference_name,
required_dagrun_column=required_dagrun_column,
)
```
That PR also adds two tests in
`airflow-core/tests/unit/models/test_dagrun.py` that you'll want to pull over
which cover that branch. That can wait till the merge but may be easier to
just paste it over now since you are copying this over.
--
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]
