How to fix Airflow SLA

A great feature of Airflow is the fact that you can automate the execution of processes and also deal with possible failures without the need to sit in front of your computer, staring at the screen until the processes finish or until an error comes up. Instead, you can use SLAs and callback functions to automatically restart the process, warn the developers via Slack or Teams or send an email to the CEO if a DAG/task takes longer than you expect it to, or if it completely fails.

What is an SLA?

In I.T., a service-level agreement (SLA) defines the level of service you expect from a given vendor or process. SLAs can exist between companies and external suppliers, or between two departments within a company. In the Airflow context, the SLA is closely linked with the execution time of a process and its final state (Success, Failure, Skipped…)

Callback functions

A valuable part of process monitoring is to act upon unwanted states of a given task, or across all tasks in a given DAG. For example, you may wish to alert managers when certain tasks are taking longer than expected to finish, or have the last task in your DAG invoke a callback when it succeeds.

Currently, there are four types of callbacks:

The callback function needs to be defined on the DAG level and will be called if any task gets into a state that matches the callback trigger state. For this example if any task on the DAG fails, sla_missed!!!! will be printed:

How SLA works VS How it should work

In Airflow’s context, SLA can be seen as “for how long your DAG can run before you need to do something about it”. You can set any datetime interval you want, and if the execution time of your DAG exceeds that, Airflow will automatically trigger a callback function defined by you. However, Airflow`s SLA has some issues:

• SLA only works for scheduled DAGs
  If your DAG has no schedule, SLAs will not work. Also, even if your DAG has a schedule but you trigger it manually, Airflow will ignore your run’s start_date and pretend as if your run was scheduled. This happens because Airflow only considers the schedule execution time of each DAG run and the interval between runs. If your DAG does not have an interval between runs, meaning there is no schedule, Airflow fails to calculate the execution dates, leading it to miss any triggered runs.
• SLA for tasks is counter intuitive
  You would expect that, by setting a SLA for a task, you’re defining the expected duration for that task. However, instead of using the start time of the task, Airflow uses the start time of the DAG.
  

Here is an example:

• The first task calls the sleep function with a duration of 2 minutes.
• The second task calls the same function with a duration of 60 seconds.

You would expect Airflow would not trigger any SLA for this DAG, as the duration of each task did not exceed their respective SLAs, as shown is figure 1:

But this is what really happens:

1. Airflow triggers this DAG at X hours and 0 minutes
2. The first task finishes at X hours and 2 minutes
3. Even if you define a task level SLA, Airflow will use as a reference the start_time of the DAG run, not the start_time of each task. When the second task is triggered, the SLA for this task has already exceed(2 minutes have already passed since the DAG started), which will make Airflow trigger the miss SLA callback function

The solve all these problems, we need to:

• Instead of calculating when a DAG should run based on its schedule, we get all the tasks that are running at a given moment. This already solves the schedule issue. It does not matter whether your DAG has a schedule or not, if your task is running, we will look at it
• Instead of calculating the SLA from the DAG’s start time, we will do it from each task’s start time. By doing so, defining the SLA at task level will be intuitive
• We will add some conditionals so that both DAG level SLA and task level SLA work simultaneously.
• Dynamic task SLA feature: If the DAG has an SLA but the tasks don’t. The manage_slas function will treat each task SLA as being de DAG SLA minus the execution time of all previous tasks.
  • The first task SLA will be the same as the DAG SLA.
  • The subsequent tasks SLA will be the DAG SLA minus the execution time of all previous tasks.

Airflow’s Scheduler possible issues

Once per minute, by default, the scheduler collects DAG parsing results and checks whether any active tasks can be triggered. It also monitors states and runs the manage_slas subprocess responsible for checking DAGs and tasks SLAs. Because it runs on an interval, quick tasks (less than a minute with default settings) are prone to omit SLA misses, this has to be taken into consideration when dealing with fast and time sensitive tasks.

How to fix Airflow’s SLA

• To change the airflow original SLA behavior is necessary to change how the function manage_slas calculates the execution time of a task.
• In Airflow’s base code, find the definition of the function “manage_sla”. If you are using version 2.3.x, it will be in Airflow/dag_processing/processor.py. If you are using previous versions of Airflow, it might be in Airflow/jobs/scheduler_job.py.

In this Blogpost we did not get too deep into coding the modifications made to the Airflow library, The patch code that changes  the SLA behavior, in order to reflect this more coherent treatment of SLAs in the Airflow DAGS and tasks can be found in our Githubs for Airflow versions 1.10.15 and 2.3.0.