Aip 86: add deadlines to DagRunResponse

[rawwar]

closes #50913

[bbovenzi]

Oh I missed this. A dag run can have multiple deadlines?

[rawwar]

Yes, here's an example that @ferruzzi provided:

with DAG(
    dag_id="dag_with_deadlines",
    deadline=[
        DeadlineAlert(
            reference = DeadlineReference.DAGRUN_QUEUED_AT,
            interval=timedelta(hours=1),
            callback=warn_me_it_may_be_late
            ),
        DeadlineAlert(
            reference = DeadlineReference.DAGRUN_QUEUED_AT,
            interval=timedelta(hours=2),
            callback=definitely_late
        ),
        DeadlineAlert(
            reference = DeadlineReference.DAGRUN_QUEUED_AT,
            interval=timedelta(hours=3),
            callback=email_my_boss_that_it_wasnt_my_fault
        ),
    ]
)

[bbovenzi]

Ok and then on the DagResponse we will have the full DeadlineAlert with the reference, interval and callback?

[bugraoz93]

How about making this a separate endpoint and calling it for the DagRun retrieved? Simply passing the dag_run_id. This way, we don't increase the DagRunResponse too much. It already has a lot of information. We can have deadline routes as a separate if it is okay from the UI perspective

[bbovenzi]

I think keeping some deadline information on DagRunResponse is valuable because I think "show me recent dag runs with passed deadlines" would be a key feature.

[ferruzzi]

Oh I missed this. A dag run can have multiple deadlines?

To be fair, it was not part of the initial proposal but added due to overwhelming popular demand, as they say. I had several users ask me to make sure that is possible so they can do tiered/escalated response patterns and it seemed like an easy enough addition so I said I would make it happen. It may or may not have ever made it into the official AIP yet.

[ferruzzi]

I think "show me recent dag runs with passed deadlines" would be a key feature.

From my side, this may be an issue. I haven't implemented the "what happens when a deadline is missed"portion yet, but my intention is essentially:

===

On dag run creation check if there is deadline in the DAG. If yes then fetch or calculate the Reference, add the Interval, and store (the resulting timestamp, callback, callback kwargs, dag_id, and run_id) in the Deadlines table

On dagrun completion/cleanup, if there is a deadline then remove all rows matching this run_id from the table

In the scheduler loop get Deadline.MIN(timestamp) and if it is in the past then send the callback for execution and remove the row from the table [[TBD, the scheduler loop may loop here or may trigger a subprocess to loop here until MIN(timestamp) is not in the past]]

===

Which means that the way I have this system designed, it is inherently destructive. The rows are removed when they are no longer needed, so if you want that data to be persistent, we'll need some way to do that. It could just be a bool flag in the dagrun table for "deadline was missed" or maybe something else, I'm not sure. Just throwing it out there in case you were counting on the Deadline table to be a reliable source.

[rawwar]

In the scheduler loop get Deadline.MIN(timestamp) and if it is in the past then send the callback for execution and remove the row from the table [[TBD, the scheduler loop may loop here or may trigger a subprocess to loop here until MIN(timestamp) is not in the past]]

Is removing data from deadlines table necessary?

If we remove it, we have no way to show that past DagRun's have missed their SLA. Infact, there's no way for front-end to show any kind of Deadline breach alert persistently.

[pierrejeambrun]

I think we shouldn't remove rows, maybe just mark them archieved via a boolean in the db. This way we can still access the history for UI or audit. Also adding a column completed_at to know when was the deadline completed can even be an interesting option.

[ferruzzi]

If you want to change the plan at this point, please start a dev list discussion for it.