Re: [PR] Handle missed deadline alerts when DagRuns fail [airflow]

["ferruzzi (via GitHub)" <gi...@apache.org>]

ferruzzi commented on code in PR #68961:
URL: https://github.com/apache/airflow/pull/68961#discussion_r3786270181


##########
airflow-core/src/airflow/serialization/encoders.py:
##########
@@ -251,6 +251,7 @@ def encode_deadline_alert(d: DeadlineAlert | SerializedDeadlineAlert) -> dict[st
         "reference": encode_deadline_reference(d.reference),
         "interval": serialize(d.interval),
         "callback": serialize(d.callback),
+        "fire_on_failure": d.fire_on_failure,

Review Comment:
   Sorry I missed this earlier.  This is going to be a back-compat issue in older SDK versions which don't have `d.fire_on_failure` yet.  Easy fix though, set a fallback to be the existing current behavior if `d.fire_on_failure` doesn't exist :
   
   ```
   "fire_on_failure": getattr(d, "fire_on_failure", False)
   ```



##########
airflow-core/src/airflow/models/dagrun.py:
##########
@@ -1240,6 +1241,42 @@ def _emit_dagrun_span(self, state: DagRunState):
             span.set_status(status_code)
             span.end()
 
+    def _handle_missed_deadlines(self, *, session: Session) -> None:
+        """Handle pending deadlines that opt in to firing when their DagRun fails."""
+        deadline_query = (
+            select(Deadline)
+            .join(DeadlineAlertModel, Deadline.deadline_alert_id == DeadlineAlertModel.id)
+            .where(Deadline.dagrun_id == self.id)
+            .where(~Deadline.missed)
+            .where(DeadlineAlertModel.fire_on_failure.is_(True))
+            .options(
+                selectinload(Deadline.callback),
+                selectinload(Deadline.dagrun),
+                selectinload(Deadline.deadline_alert),
+            )
+        )
+        for deadline in session.scalars(
+            with_row_locks(
+                deadline_query,
+                of=Deadline,
+                session=session,
+                skip_locked=True,
+                key_share=False,
+            )
+        ):
+            deadline_id = deadline.id
+            try:
+                deadline.handle_miss(session)
+            except DBAPIError:

Review Comment:
   Sorry I missed this earlier, this should likely be `SQLAlchemyError` as used elsewhere.  If you disagree, feel free to just resolve this, but I noticed that one was already imported and you had to add an import for this new one



##########
airflow-core/src/airflow/migrations/versions/0128_3_4_0_add_fire_on_failure_to_deadline_alert.py:
##########
@@ -0,0 +1,48 @@
+#

Review Comment:
   Don't forget you'll need to rename this file when you fix the db.py conflicts, we're on migration 0131



-- 
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: commits-unsubscribe@airflow.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org