Status

Summary

The existing SLA implementation before Airflow 3.0 has been problematic and a constant point of debate.  In the Airflow 3 planning meetings, it was discussed and decided that the move to a new major version is a good time to tear out the old implementation and replace it with something new.   Since AIP-57 has been abandoned, this proposal will replace and supersede AIP-57.  That AIP and the discussion around it are not to be considered as part of this proposal and, unless mentioned here, nothing in those discussions or that AIP is considered to be part of this one.

The old SLA feature is removed in Airflow 3.0 (PR here) and in 3.1 it will be replaced with Airflow Alerts.  The purpose of this AIP is to discuss the implementation of the new Alerts feature to replace the SLA feature.

One of the main points of contention of the SLA feature revolves around when you start counting.  If you set an SLA of "1 hour", is that one hour after it was scheduled, one hour after it starts running, one hour after it was queued, or something else entirely.   This leads to frequent questions and confusion.  This AIP revolves around the idea of allowing the user to define that starting point and treat the new SLA as a calculated "need-by date".   To avoid confusion, the new feature will be referred to as Alerts rather than SLA and the implementation will allow a much more versatile experience for the user.

Motivation

Problems in the Old Implementation

(credit for this section to Sung Yun in AIP-57)

1. Sla_miss was evaluated only if the task_id had ever been in SUCCESS or SKIPPED state.  This means that if the task had not succeeded in the first place - due to it being blocked, still running, or yet to be scheduled -  the sla_miss_callback was not executed.  This is a detrimental misfeature of the old SLA implementation, as users often want to use SLA for delay detection, to enable them to investigate any possible issues in their workflow pre-emptively. 
2. If a dag is ever changed to rename the task_ids or add new task_ids, the first runs of the new task_ids will never have their SLAs evaluated. 
3. SLA was defined as a timedelta relative to the dagrun.data_interval_end based on the timetable of the scheduled job.  Some DAGs are triggered externally - either manually or via a Dataset - meaning they don’t have a fixed schedule or start time preventing the evaluation of an SLA miss. 
4. Sla_miss_callback was an attribute of the DAG, but the SLAs were defined as attributes of the individual tasks. This is counterintuitive, and does not allow users to run callbacks that are customized at the task level if deadlines are missed.
5. The old implementation triggered the parsing of dag files when an sla_miss_callback was sent to the DagFileProcessor. This means that if an sla was defined within a dag, we may end up parsing the dag file more frequently than at the frequency defined by min_file_process_interval because the sla_miss_callbacks were sent to the DagFileProcessor every time dag_run.update_state was called. This is especially bad for sla_miss_callbacks compared to the other dag level callbacks, because sla_miss_callback was requested every time we examine and attempt to schedule the dagrun, even if there were no state changes within the dag.

Considerations

As stated above, this AIP intentionally ignores all previous discussion on the topic or fixing, changing, redefining, or re-implementing "SLA".   Previous discussions have been ongoing for a very long time, across many different mediums, and have gone through multiple proposals.  It is impossible to summarize or collate all those discussions, so treat this as a "blank slate" proposal and do not assume any other changes outside of what is explicitly stated herein.

Definitions

As this is a topic that has been discussed many times, in an effort to avoid confusion, I am using the following definitions in this project.

Anchor

An Enum will contain a selection of options, similar to how a user currently provides the TriggerRule.   The implementation will include an interface which allows future work to easily add new anchors, but they will not be a freeform input for the user.  These options will resolve to a datetime() representing the possible points in time to start counting from.  This value can be dynamic (examples include NEXT_DAGRUN_EXECUTION_DATE  or DAGRUN_QUEUED_AT, which map to their respective database fields), or a static  date/time value provided as a string and cast to a datetime() object.

Delta

A datetime.timedelta.  When added to the anchor you get the “need-by date”.  (see below)

Need-by

“Need-by” will be defined as the anchor time (see above) plus some delta.  For example,  `need_by = AlertAnchors.NEXT_DAGRUN_EXECUTION_DATE + 1_hour`

Callback

A Callable (or, by extension, a functools.partial) to execute when a miss is detected.  This implementation will not be limited to picking either "email" or "notify", it will allow any user-provided callable and include those as built-in options/examples.

Why is it needed?

Before diving into the challenges with the current SLA feature, let's consider what our users are expecting from an SLA feature in Airflow:

Scenario 1: The Mission-Critical Daily Report

Elmo, a data engineer at a large e-commerce company, manages a DAG that generates a critical daily sales report. This report must be ready by 9 AM every day for the executive team's business review.

Elmo expects:

• To set a DAG-level deadline of 9 AM for the entire workflow
• To receive immediate notifications if the DAG is running behind schedule and might miss the deadline
• To easily adjust the Alert for weekends and holidays, while having ability to define deadline in local timezone
• To see a historical view of how often the DAG meets its SLA, helping him identify trends and potential improvements

Scenario 2: The Long-Running Processing Task

Rosita, a data scientist at a finance firm, works with a DAG that includes a data processing task which transforms the data using EMR extracted from an external API. This task occasionally runs much longer than expected, delaying downstream processes.

Rosita needs:

• To set a maximum execution time for the extract task
• To receive notifications when the task exceeds its expected duration
• To log detailed information when the expected time limit is breached, helping her investigate and optimize the task

Scenario 3: The ETL Pipeline

Zoe, an engineer at a tech startup, manages a DAG with multiple stages: data ingestion, preparation, processing, and reporting. She needs to ensure each stage completes within a specific timeframe.

Zoe requires:

• To set individual Alerts for different tasks within her DAG
• To receive notifications if any task in the pipeline exceeds its timeframe
• To easily view which parts of her pipeline are consistently meeting or missing their deadlines

What change do you propose to make?

1. In dag_processing/processor.py add "save next SLA": 
1. If this DAG has an Alert configured, then calculate the “need-by” and store the timestamp in a sorted-by-closest-timestamp object (or db table like the old SLAMiss table?)
2. In the scheduler loop add “check for Alert misses”:
1. If the first/soonest sla has passed: pop the timestamp, queue the callback, recurse.
3. Add the Alert handler back in where the previous SLA callback handler used to live.
   
4. When a dagrun ends, attempt to remove its SLA timestamp.

What problem does it solve?

Allowing the user to decide which point int he DAG's lifecycle the need-by time is based off will help clear up much of the confusion around the old SLA implementation.

Are there any downsides to this change?

Which users are affected by the change?

All users who currently use the SLA feature will need to migrate.  The SLA feature has been removed in 3.0 and replaced with a notification to this effect, and the documentation for the Alerts will include examples to set an Alertthat works the same way as the old SLA to help minimize the transition effort for those users who liked the old way.

How are users affected by the change? (e.g. DB upgrade required?)

The SLAMiss table has been removed in 3.0 and this implementation will likely replace it with a similar Alerts table which maps dagrun_id::timestamp.

What is the level of migration effort (manual and automated) needed for the users to adapt to the breaking changes? (especially in context of Airflow 3)

DAGs which implement the old SLA feature will need to be manually migrated.  Documentation will include examples showing the old vs new configurations to make that as easy as possible

What defines this AIP as "done"?

This AIP will only cover DAG-level Alerts. 

Future work can add:

• Task-level Alerts (Task within a DAG did not finish before a given time)
  
• Dataset Alerts (Dataset has not been modified before a given time)
  
• Update the Alerts view on the frontend to include a user-defined failure code or message.
• A new sidecar process, equivalent to a Triggerer, which can offload the processing of the Alert callbacks may be future work.