SLA misses and notifications in Apache Airflow - Commands & Configuration

Introduction

Sometimes tasks in Airflow take longer than expected. SLA misses help you know when this happens. Notifications alert you so you can fix problems quickly.

When you want to be alerted if a data pipeline task runs late.

When you need to monitor critical jobs that must finish on time.

When you want to send emails or messages if a task misses its deadline.

When you want to track SLA misses in the Airflow UI for reporting.

When you want to improve reliability by quickly reacting to delays.

Config File - sla_notification_dag.py

sla_notification_dag.py

from airflow import DAG
from airflow.operators.bash import BashOperator
from airflow.utils.dates import days_ago
from airflow.utils.email import send_email
from airflow.utils.session import provide_session
from datetime import timedelta

# Function to send email on SLA miss
@provide_session
def sla_miss_callback(dag, task_list, blocking_task_list, slas, session=None):
    subject = f"SLA Miss Alert for DAG: {dag.dag_id}"
    body = f"The following tasks missed their SLA: {', '.join(task_list)}"
    send_email(to=["[e-mail]"], subject=subject, html_content=body)

# Define the DAG
with DAG(
    dag_id='sla_notification_dag',
    start_date=days_ago(1),
    schedule_interval='@daily',
    sla_miss_callback=sla_miss_callback,
    catchup=False
) as dag:

    task1 = BashOperator(
        task_id='task1',
        bash_command='sleep 10',
        sla=timedelta(seconds=5)  # SLA set to 5 seconds
    )

    task2 = BashOperator(
        task_id='task2',
        bash_command='echo "Task 2 runs quickly"',
        sla=timedelta(seconds=10)  # SLA set to 10 seconds
    )

    task1 >> task2

This DAG defines two tasks with SLAs set using the sla parameter. If a task runs longer than its SLA, the sla_miss_callback function is called.

The callback sends an email alert listing the tasks that missed their SLA.

This helps notify the team immediately about delays.

Commands

This command starts the DAG named 'sla_notification_dag' to test SLA miss notifications.

Terminal

airflow dags trigger sla_notification_dag

Expected Output

{"dag_id": "sla_notification_dag", "run_id": "manual__2024-06-01T00:00:00+00:00", "state": "running"}

Lists all tasks in the DAG to verify task names and SLAs.

Terminal

airflow tasks list sla_notification_dag

Expected Output

task1

task2

Runs task1 manually for the given date to check if it respects the SLA or triggers a miss.

Terminal

airflow tasks test sla_notification_dag task1 2024-06-01

Expected Output

[2024-06-01 00:00:00,000] {bash.py:123} INFO - Running command: sleep 10

[2024-06-01 00:00:10,000] {bash.py:130} INFO - Command exited with return code 0

Runs task2 manually to verify it completes within SLA and does not trigger a miss.

Terminal

airflow tasks test sla_notification_dag task2 2024-06-01

Expected Output

[2024-06-01 00:00:00,000] {bash.py:123} INFO - Running command: echo "Task 2 runs quickly"

Task 2 runs quickly

[2024-06-01 00:00:00,100] {bash.py:130} INFO - Command exited with return code 0

Key Concept

If a task runs longer than its SLA time, Airflow calls the sla_miss_callback to notify you immediately.

Common Mistakes

Not setting the sla parameter on tasks.

Without SLA set, Airflow does not know when to trigger SLA miss notifications.

Always set the sla parameter with a timedelta on tasks you want to monitor.

Not defining sla_miss_callback in the DAG.

Without the callback, SLA misses are logged but no notifications are sent.

Define sla_miss_callback in the DAG to send alerts like emails or messages.

Setting SLA too short causing false alerts.

Tasks may often run longer than unrealistic SLAs, causing unnecessary notifications.

Set realistic SLA times based on typical task durations.

Summary

Set the sla parameter on tasks to define their maximum allowed run time.

Use sla_miss_callback in the DAG to send notifications when SLAs are missed.

Trigger and test tasks to verify SLA miss alerts work as expected.

Practice

1. What does setting an sla parameter on an Airflow task do?

easy

A. It defines a time limit; if the task runs longer, it triggers an SLA miss.

B. It sets the maximum retries for the task.

C. It schedules the task to run at a specific time.

D. It disables the task from running automatically.

Solution

1. Step 1: Understand the purpose of sla in Airflow tasks

   The sla parameter sets a time limit for task execution.

2. Step 2: Identify what happens when the limit is exceeded

   If the task runs longer than the sla time, Airflow marks it as an SLA miss.

3. Final Answer:

   It defines a time limit; if the task runs longer, it triggers an SLA miss. -> Option A

4. Quick Check:

   sla sets time limit = C [OK]

Hint: Remember SLA means max allowed time for task [OK]

Common Mistakes:

• Confusing SLA with retry count
• Thinking SLA schedules task start time
• Assuming SLA disables tasks

2. Which of the following is the correct way to define an SLA of 30 minutes on a task in Airflow?

easy

A. sla='30 minutes'

B. sla=30

C. sla=timedelta(minutes=30)

D. sla='00:30:00'

Solution

1. Step 1: Recall the correct data type for sla

   The sla parameter expects a datetime.timedelta object.

2. Step 2: Identify the correct syntax for 30 minutes timedelta

   timedelta(minutes=30) correctly creates a 30-minute duration.

3. Final Answer:

   sla=timedelta(minutes=30) -> Option C

4. Quick Check:

   sla uses timedelta object = B [OK]

Hint: Use timedelta for time durations in Airflow [OK]

Common Mistakes:

• Using string instead of timedelta
• Passing integer without timedelta
• Using incorrect string format

3. Given this DAG snippet, what will happen if task1 runs longer than 10 minutes?

from datetime import timedelta
from airflow import DAG
from airflow.operators.dummy import DummyOperator
from airflow.utils.dates import days_ago

def sla_miss_callback(dag, task_list, blocking_task_list, slas, session=None):
    print('SLA missed!')

dag = DAG('sla_example', start_date=days_ago(1))

task1 = DummyOperator(
    task_id='task1',
    dag=dag,
    sla=timedelta(minutes=10)
)

dag.sla_miss_callback = sla_miss_callback

medium

A. The sla_miss_callback function will print 'SLA missed!' when the SLA is missed.

B. Nothing happens because DummyOperator ignores SLA.

C. The task will automatically retry until it finishes within 10 minutes.

D. Airflow will fail the DAG immediately.

Solution

1. Step 1: Understand SLA miss callback usage

   The DAG-level sla_miss_callback is called when any task misses its SLA.

2. Step 2: Identify what the callback does

   The callback prints 'SLA missed!' when triggered by a task exceeding its SLA.

3. Final Answer:

   The sla_miss_callback function will print 'SLA missed!' when the SLA is missed. -> Option A

4. Quick Check:

   SLA miss triggers callback print = A [OK]

Hint: SLA miss triggers callback, not retries or failure [OK]

Common Mistakes:

• Thinking SLA causes retries automatically
• Assuming DummyOperator ignores SLA
• Believing DAG fails immediately on SLA miss

4. You wrote this code to notify on SLA misses but no notification happens:

def notify_sla_miss(dag, task_list, blocking_task_list, slas, session=None):
    send_email('[e-mail]', 'SLA Missed')

dag = DAG('sla_notify', start_date=days_ago(1), sla_miss_callback=notify_sla_miss)

task = DummyOperator(task_id='task', dag=dag, sla=timedelta(minutes=5))

What is the likely reason notifications are not sent?

medium

A. DummyOperator does not support SLA misses.

B. SLA miss callback must be set on the task, not the DAG.

C. The SLA time must be set in seconds, not minutes.

D. The send_email function is not defined or imported.

Solution

1. Step 1: Check the callback function implementation

   The callback calls send_email but no import or definition is shown.

2. Step 2: Understand Airflow's email sending

   Airflow requires proper email function or operator; undefined send_email causes failure silently.

3. Final Answer:

   The send_email function is not defined or imported. -> Option D

4. Quick Check:

   Undefined email function blocks notification = D [OK]

Hint: Ensure email functions are imported or use Airflow email operators [OK]

Common Mistakes:

• Setting callback on DAG instead of task (both valid but DAG preferred)
• Wrong SLA time units
• Assuming DummyOperator ignores SLA

5. You want to send a custom Slack message when any task misses its SLA in your DAG. Which approach correctly implements this?

hard

A. Set sla on tasks and add a SlackOperator as a downstream task to notify SLA misses.

B. Define a sla_miss_callback function that uses Slack API and assign it to the DAG's sla_miss_callback parameter.

C. Use on_failure_callback on tasks to send Slack messages when SLA is missed.

D. Configure Airflow to send Slack notifications by setting sla to the Slack channel name.

Solution

1. Step 1: Understand how SLA miss notifications work

   SLA misses trigger the sla_miss_callback function assigned to the DAG or task.

2. Step 2: Identify correct way to send Slack message on SLA miss

   Implement a callback function that calls Slack API and assign it to DAG's sla_miss_callback.

3. Final Answer:

   Define a sla_miss_callback function that uses Slack API and assign it to the DAG's sla_miss_callback parameter. -> Option B

4. Quick Check:

   SLA miss callback with Slack API = A [OK]

Hint: Use sla_miss_callback with Slack API for notifications [OK]

Common Mistakes:

• Using on_failure_callback instead of sla_miss_callback
• Trying to notify via downstream tasks
• Misconfiguring sla parameter as Slack channel