Integration with PagerDuty and Slack in Apache Airflow - Commands & Configuration

Introduction

Sometimes workflows fail or need attention. Integrating Airflow with PagerDuty and Slack helps you get alerts quickly so you can fix problems fast.

When a critical Airflow task fails and you want to get an immediate alert on Slack.

When you want to create an incident in PagerDuty automatically from Airflow failures.

When your team uses Slack for communication and PagerDuty for incident management.

When you want to reduce downtime by getting notified instantly about workflow issues.

When you want to keep track of Airflow alerts in one place without checking logs manually.

Config File - my_dag.py

my_dag.py

from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.utils.dates import days_ago
from airflow.providers.slack.operators.slack_webhook import SlackWebhookOperator
from airflow.providers.pagerduty.operators.pagerduty import PagerDutyIncidentOperator

def task_that_fails():
    raise Exception('This task fails to trigger alerts')

with DAG(
    dag_id='pagerduty_slack_integration',
    start_date=days_ago(1),
    schedule_interval='@daily',
    catchup=False
) as dag:

    fail_task = PythonOperator(
        task_id='fail_task',
        python_callable=task_that_fails
    )

    slack_alert = SlackWebhookOperator(
        task_id='slack_alert',
        http_conn_id='slack_connection',
        message=':warning: Task failed in DAG pagerduty_slack_integration',
        channel='#alerts'
    )

    pagerduty_alert = PagerDutyIncidentOperator(
        task_id='pagerduty_alert',
        routing_key='your_pagerduty_integration_key',
        summary='Airflow task failure in pagerduty_slack_integration',
        source='airflow',
        severity='error'
    )

    fail_task >> [slack_alert, pagerduty_alert]

This DAG defines a failing task to simulate an error.

The SlackWebhookOperator sends a message to a Slack channel using a configured connection.

The PagerDutyIncidentOperator creates an incident in PagerDuty using your integration key.

The failing task triggers both alerts in parallel when it fails.

Commands

This command adds a connection in Airflow to Slack's webhook URL so the SlackWebhookOperator can send messages.

Terminal

airflow connections add 'slack_connection' --conn-type 'http' --conn-host 'hooks.slack.com' --conn-extra '{"webhook_token": "/services/T00000000/B00000000/XXXXXXXXXXXXXXXXXXXXXXXX"}'

Expected Output

Connection `slack_connection` added successfully

→

--conn-type - Specifies the connection type, here 'http' for Slack webhook

→

--conn-host - Sets the Slack webhook host

→

--conn-extra - Sets the Slack webhook token path

This command manually triggers the DAG to run and test the integration with PagerDuty and Slack.

Terminal

airflow dags trigger pagerduty_slack_integration

Expected Output

Created <DagRun pagerduty_slack_integration @ 2024-06-01T00:00:00+00:00: manual__2024-06-01T00:00:00+00:00, externally triggered: True>

This command lists all tasks in the DAG to verify the task names and operators.

Terminal

airflow tasks list pagerduty_slack_integration

Expected Output

fail_task

slack_alert

pagerduty_alert

This command tests the failing task locally to see the error and trigger alerts manually.

Terminal

airflow tasks test pagerduty_slack_integration fail_task 2024-06-01

Expected Output

Traceback (most recent call last):

  File "/path/to/airflow/cli/commands/task_command.py", line 123, in test

    result = task.execute(context=context)

  File "/path/to/airflow/operators/python.py", line 114, in execute

    return self.execute_callable()

  File "/path/to/airflow/operators/python.py", line 130, in execute_callable

    return self.python_callable(*self.op_args, **self.op_kwargs)

  File "/path/to/my_dag.py", line 6, in task_that_fails

    raise Exception('This task fails to trigger alerts')

Exception: This task fails to trigger alerts

Key Concept

If you remember nothing else from this pattern, remember: configure Airflow connections and use operators to send alerts automatically on task failures.

Common Mistakes

Not setting the Slack webhook URL correctly in Airflow connections.

Without the correct webhook URL, Slack messages will not be sent and you won't get alerts.

Use the exact Slack webhook URL when adding the connection with airflow connections add.

Using the wrong PagerDuty integration key or missing it in the PagerDutyIncidentOperator.

PagerDuty will reject the incident creation without a valid routing key, so no alert is created.

Copy the correct integration key from PagerDuty and set it in the operator's routing_key parameter.

Not linking the failing task to alert tasks with >> operator.

If the alert tasks are not downstream of the failing task, they won't run on failure.

Use fail_task >> [slack_alert, pagerduty_alert] to trigger alerts after failure.

Summary

Add Airflow connections for Slack webhook with the correct URL.

Create a DAG with a failing task and alert tasks using SlackWebhookOperator and PagerDutyIncidentOperator.

Trigger the DAG and test tasks to verify alerts are sent on failure.

Practice

1. What is the main purpose of integrating Airflow with PagerDuty and Slack?

easy

A. To store Airflow logs in PagerDuty and Slack

B. To schedule Airflow DAGs automatically

C. To replace Airflow's scheduler with PagerDuty

D. To receive fast alerts and notifications about workflow issues

Solution

1. Step 1: Understand the role of PagerDuty and Slack in Airflow

   PagerDuty and Slack are used to send alerts and notifications when Airflow workflows have issues or important events.

2. Step 2: Identify the main benefit of integration

   The integration helps teams respond quickly by notifying them immediately about failures or important status updates.

3. Final Answer:

   To receive fast alerts and notifications about workflow issues -> Option D

4. Quick Check:

   Alerts and notifications = C [OK]

Hint: Think: Why notify teams fast? Alerts! [OK]

Common Mistakes:

• Confusing scheduling with alerting
• Thinking logs are stored in PagerDuty or Slack
• Assuming PagerDuty replaces Airflow scheduler

2. Which Airflow operator is used to send alerts to PagerDuty within a DAG?

easy

A. PagerDutyIncidentOperator

B. SlackAPIPostOperator

C. EmailOperator

D. HttpSensor

Solution

1. Step 1: Identify operators for PagerDuty and Slack

   SlackAPIPostOperator sends messages to Slack, PagerDutyIncidentOperator sends incidents to PagerDuty.

2. Step 2: Match operator to PagerDuty alerts

   PagerDutyIncidentOperator is designed specifically for PagerDuty integration.

3. Final Answer:

   PagerDutyIncidentOperator -> Option A

4. Quick Check:

   PagerDuty alerts = PagerDutyIncidentOperator A [OK]

Hint: PagerDuty operator name includes 'PagerDuty' [OK]

Common Mistakes:

• Choosing SlackAPIPostOperator for PagerDuty alerts
• Confusing EmailOperator with PagerDuty integration
• Selecting HttpSensor which is unrelated

3. Given this Airflow DAG snippet, what will be the output when the task notify_slack runs successfully?

notify_slack = SlackAPIPostOperator(
    task_id='notify_slack',
    token='xoxb-1234',
    channel='#alerts',
    text='DAG completed successfully!'
)

medium

A. No message is sent because the channel name is incorrect

B. An error occurs because the token is invalid

C. A message 'DAG completed successfully!' is posted to the #alerts Slack channel

D. The DAG fails to run because SlackAPIPostOperator is deprecated

Solution

1. Step 1: Understand SlackAPIPostOperator parameters

   The operator sends a message to the specified Slack channel using the provided token and text.

2. Step 2: Analyze the given snippet

   The token and channel are set, and the text is 'DAG completed successfully!'. If the token is valid, the message posts successfully.

3. Final Answer:

   A message 'DAG completed successfully!' is posted to the #alerts Slack channel -> Option C

4. Quick Check:

   Slack message sent = D [OK]

Hint: SlackAPIPostOperator posts text to channel if token valid [OK]

Common Mistakes:

• Assuming token is invalid without evidence
• Thinking SlackAPIPostOperator is deprecated
• Confusing channel name format

4. You wrote this task to send a PagerDuty alert but it fails with an authentication error:

pagerduty_alert = PagerDutyIncidentOperator(
    task_id='pagerduty_alert',
    routing_key='invalid_key',
    title='Airflow failure',
    severity='critical'
)

What is the most likely cause of the failure?

medium

A. The routing_key is incorrect or invalid

B. The severity value must be 'low' instead of 'critical'

C. PagerDutyIncidentOperator does not support the title parameter

D. The task_id must be 'pagerduty_incident' exactly

Solution

1. Step 1: Check the routing_key parameter

   The routing_key is a secret token that must be valid for PagerDuty API authentication.

2. Step 2: Understand authentication errors

   An invalid routing_key causes authentication failure, blocking alert creation.

3. Final Answer:

   The routing_key is incorrect or invalid -> Option A

4. Quick Check:

   Authentication error = invalid routing_key A [OK]

Hint: Authentication errors usually mean bad keys or tokens [OK]

Common Mistakes:

• Changing severity to fix authentication
• Removing title parameter incorrectly
• Changing task_id unnecessarily

5. You want to create an Airflow DAG that sends a Slack message only if a PagerDuty incident is triggered. Which approach correctly integrates both operators to achieve this?

hard

A. Use PagerDutyIncidentOperator first, then SlackAPIPostOperator with a trigger rule 'all_done' to send Slack message regardless of PagerDuty success

B. Use PagerDutyIncidentOperator first, then SlackAPIPostOperator with trigger rule 'all_success' to send Slack message only if PagerDuty alert succeeds

C. Use SlackAPIPostOperator first, then PagerDutyIncidentOperator with trigger rule 'one_failed' to send PagerDuty alert only if Slack fails

D. Use both operators in parallel without dependencies to send alerts simultaneously

Solution

1. Step 1: Define the desired workflow

   Slack message should be sent only if PagerDuty incident is successfully triggered.

2. Step 2: Set operator order and trigger rules

   Run PagerDutyIncidentOperator first. Set SlackAPIPostOperator to run after it with trigger_rule='all_success' to ensure Slack message only sends if PagerDuty alert succeeds.

3. Final Answer:

   Use PagerDutyIncidentOperator first, then SlackAPIPostOperator with trigger rule 'all_success' to send Slack message only if PagerDuty alert succeeds -> Option B

4. Quick Check:

   Slack after PagerDuty success = B [OK]

Hint: Slack depends on PagerDuty success with 'all_success' trigger [OK]

Common Mistakes:

• Sending Slack message regardless of PagerDuty result
• Reversing operator order
• Running both operators in parallel without dependency