Post cover image

June 1, 2024

📊 Monitoring Made Easy With ATP: Leveraging Airflow Plugins for DAG and Task Analytics 🔍

Introduction

By Ravi Kant Yadav

4 min read

Introduction

As a software developer and data enthusiast, I constantly seek ways to enhance monitoring and analytics for complex workflows. Apache Airflow is a powerful tool for orchestrating these workflows, but in my experience managing over 3,000 DAGs, I encountered challenges in tracking and analyzing task execution without making changes to the existing DAGs. To address this, I developed an Airflow Telemetry Plugin (ATP). Leveraging Airflow's plugin feature, this plugin integrates telemetry data from Airflow with various data platforms like Elasticsearch and Kafka, allowing for seamless monitoring and analytics without altering the DAGs. In this article, I'll guide you on how to use this plugin, the problems it solves, and provide insights on creating similar plugins.

Why Airflow Telemetry?

The Problem

Managing a large number of DAGs in Apache Airflow posed specific challenges for me:

  • Limited Visibility: Tracking task execution and workflow statuses across a vast number of DAGs was difficult.
  • Fragmented Data: Scattered metrics and logs made it hard to get a unified view of the system's performance.
  • Custom Requirements: I needed to add additional fields like domain and system names based on specific DAG names, without modifying the existing DAGs. Create new fields dynamically like if any one give key value in dag tag then it create new field example:
 tags=["impact: critical","version: 1.1.1"])

To address these challenges, I leveraged Airflow's powerful plugin feature to create the Airflow Telemetry Plugin (ATP). This plugin enables comprehensive monitoring and analytics by pushing telemetry data to external platforms like Elasticsearch, all without requiring changes to the existing DAGs.

The Solution

The Airflow Telemetry Plugin (ATP) solves these issues by pushing telemetry data to external platforms, enabling better monitoring and insights into Airflow's operations. This plugin offers:

  • Centralised Monitoring: Aggregates task execution and workflow status data in one place.
  • Scalable Insights: Easily scalable to accommodate growing workflows.
  • Enhanced Visibility: Provides detailed metrics and logs for better tracking and troubleshooting.

Key Features

Here are some standout features of the Airflow Telemetry Plugin:

  • 🌐 Data Push to Different Platforms: Users can push telemetry data to platforms like Elasticsearch, with future support for Kafka.
  • ⚙️ Configurable Connection Settings: Easily configure connection settings for the target platform directly from the Airflow UI.
  • 📊 Comprehensive Data Metrics: Collects various metrics related to task execution and workflow status, including success, failure, and running statuses.
  • 🔄 Flexible Data Mapping: Allows customizable mapping of Airflow metadata to data fields in the target platform.

How to Use the Airflow Telemetry Plugin

Installation

Installing the Airflow Telemetry Plugin is straightforward. Follow these steps:

  1. Clone the Repository:
git clone https://github.com/...../airflow-telemetry-plugin.git

  1. Move the Plugin Code: Move the cloned code into your Airflow plugin directory.

  2. Install Dependencies: Install any necessary dependencies listed in the repository.

Configuration

Configuring the plugin from the Airflow UI is simple and user-friendly:

  1. Navigate to the Airflow Web Interface:
  • Open your web browser and enter the URL for your Airflow instance.

2. Access the Admin Panel:

  • Once logged in, navigate to the Admin section of the Airflow UI.

3. Navigate to Connections

  • In the Admin section, click on the "Connections" option.

4. Add a New Connection:

  • Click on "Add Connection" and fill out the form with the necessary details:
  1. Connection ID/Name: A unique identifier or name for the connection.
  2. Connection Type: Select the appropriate connection type, such as Elasticsearch.
  3. Connection Parameters:
  • HOST: Enter the hostname or IP address of the target platform.
  • Schema: Specify the protocol (e.g., http or https).
  • PORT: Input the port number for connecting to the target platform
  • LOGIN: Enter the username or access key if authentication is required.
  • PASSWORD: Enter the password or secret key for authentication.
  • EXTRA: Optionally, provide additional parameters:

{
  "index_key": "Specify the key for indexing data."
  "max_retries": "Set the maximum number of retry attempts. "
  "timeout": "Specify the connection timeout duration. "
  "lookup_file":"Optionally, specify a lookup file. "
  "create_field_from_tag": "Optionally, specify whether to create a field from a tag. "
  "required_key": "if key exites then check if doc not push this key then get error message in ERRER_MESSAGE key. "
}

  1. Save the Connection: Save the connection configuration.

  2. Configure the Plugin: After saving, configure the plugin to push telemetry data to the desired platform by specifying the connection ID/name within the plugin configuration

Usage

Once installed and configured, the Airflow Telemetry Plugin (ATP) automatically collects telemetry data from Airflow components and pushes it to the configured platform. Users can access the collected data from the target platform for monitoring and analysis purposes.

Telemetry Data Fields

The following telemetry data is pushed to the platform :

Key Names:

{
    "queued_at": "",
    "dag_id": "",
    "run_id": "",
    "dag_name": "",
    "dag_owner": "",
    "dag_tags": [],
    "external_trigger": ,
    "start_date": "",
    "logical_date": "",
    "execution_date": "",
    "end_date": "",
    "email_id": "",
    "dag_status": "",
    "dag_file_location": "",
    "dag_schedule_interval": "",
    "airflow_base_url": "",
    "task_id": "",
    "task_status": "",
    "error_message": "",
    "...any new key from dag_tag or lookup file..":""
}

Flowchart of the Airflow Telemetry Plugin

Below is a flowchart illustrating the architecture and flow of the Airflow Telemetry Plugin (ATP):

None

Explanation of the flowchart:

  1. Airflow Telemetry Initialisation: The plugin initialises and prepares to collect telemetry data.
  2. Get Event and Push Docs: The plugin monitors Airflow events:
  • On DAG Running
  • On DAG Success
  • On DAG Failed
  • On Task Running
  • On Task Success
  • On Task Failed

  1. Prepare Airflow Telemetry Docs: The plugin extracts all required key-value pairs from the task instance and DAG run object to create a new telemetry document.

  2. Add New Key from Lookup:

  • Check Lookup File: If a lookup file is configured, the plugin reads it and checks for matching conditions. If conditions match, new key-value pairs are added to the document.
  1. Add New Key from Tags:
  • Check Tags: If create_field_from_tag is set to true and DAG tags contain any key-value pairs, these are added to the document.
  1. Validate Documents:
  • Document Validity Check: The plugin ensures document validity by checking for mandatory key-value pairs. If any mandatory key is missing, an "error_message" key is added with a corresponding message.
  1. Push Updated Document into Elastic: Finally, the updated and validated document is pushed into Elasticsearch for storage and analysis.

Contact

For questions or inquiries about the Airflow Telemetry Plugin, please contact: Ravi Kant Yadav email : ravikantyadav1918@gmail.com.