Configure Airflow and Pulse to Monitor Standalone Airflow on Kubernetes

This page describes how to configure both Airflow and Pulse to enable Airflow to emit metrics and events, and for Pulse to collect, process, and visualize them from a standalone Airflow deployment running on Kubernetes.

This is supported in Pulse 4.1.0 and later.

For details, see the following sections.

Configure Airflow on Kubernetes to Emit Metrics

This configuration enables Airflow to emit events and metrics that Pulse can collect and visualize.

Step 1: Install the Acceldata Event Listener Plugin

Copy the Acceldata event listener JAR into the Airflow event listener plugin directory.
Update config.ini with the correct cluster name and Events endpoint.

Example config.ini:

Bash
    
 
[pulse]ad.cluster = airflowonk8sad.events.url = nats://host:19009
Copy

Verify the plugin is loaded:

Bash
    
 
airflow plugins
Copy

Step 2: Create a ConfigMap for the Plugin and Mount It to the Scheduler

Create a ConfigMap from the event listener plugin files:

Bash
    
 
kubectl create configmap airflow-plugins \  --from-file=event_logging_plugin/ \  -n airflow
Copy

Mount the ConfigMap in the Airflow scheduler deployment:

Bash
    
​x
 
volumes:  - name: airflow-event-logging-plugin-volume    configMap:      name: airflow-event-logging-plugin      defaultMode: 420​volumeMounts:  - name: airflow-event-logging-plugin-volume    mountPath: /opt/airflow/plugins/event_logging_plugin
Copy

Step 3: Install `nats-py` in the Scheduler Container

Update the scheduler container args to install the nats-py dependency:

Bash
    
 
containers:  - args:      - bash      - -c      - pip install apache-airflow[statsd]==2.8.3 && pip install nats-py && exec airflow scheduler
Copy

Step 4: Distribute the Plugin to All Task Pods

Ensure all task pods receive the event listener plugin by using a pod-template.yaml.

Example pod-template.yaml:

Bash
    
 
spec:  initContainers:    - name: install-deps      image: apache/airflow:2.8.3      imagePullPolicy: IfNotPresent      env:        - name: HOME          value: /home/airflow      command: ["python", "-m", "pip", "install", "--user", "nats-py"]      volumeMounts:        - name: event-logging          mountPath: /opt/airflow/plugins/event_logging_plugin          readOnly: true​  volumes:    - name: event-logging      configMap:        name: airflow-event-logging-plugin
Copy

Step 5: Validate Event Streaming to NATS

Trigger a DAG run from the Airflow UI, then verify Airflow streams are created in NATS:

Bash
    
 
nats stream ls | grep airflow
Copy

Example output:

airflow_events_airflowonk8s
processed_airflow_events_0

Configure Pulse to Monitor Airflow on Kubernetes

This configuration enables Pulse to collect Airflow events and StatsD metrics from Kubernetes for monitoring and visualization.

Prerequisites

PulseNode DaemonSet deployed on the Kubernetes cluster

Configure PulseNode to Collect Airflow StatsD Metrics

PulseNode (Telegraf) collects Airflow StatsD metrics over UDP port 8125 and writes them to the VM DB.

On the Pulse node, update:

Path: /opt/pulse/node/config/node.conf

Add the following Telegraf StatsD input configuration:

Bash
    
 
[[inputs.statsd]]  service_address = ":8125"  templates = [    "*.dag.*.*.duration measurement.measurement.dag_id.task_id.measurement",    "*.dagrun.dependency-check.* measurement.measurement.measurement.dag_id",    "*.dag_processing.last_duration.* measurement.measurement.measurement.dag_id",    "*.dagrun.duration.*.* measurement.measurement.measurement.status.dag_id",    "*.dagrun.schedule_delay.* measurement.measurement.measurement.dag_id",    "*.dag_processing.last_runtime.* measurement.measurement.measurement.dag_id",    "*.dag.loading-duration.* measurement.measurement.measurement.dag_id",    "*.operator.*.* measurement.measurement.status.operator",    "*.dag_processing.last_run.seconds_ago.* measurement.measurement.measurement.measurement.dag_file",    "*.dag.*.*.queued_duration measurement.measurement.dag_id.task_id.measurement",    "*.ti.finish.*.*.* measurement.measurement.measurement.dag_id.task_id.state",    "*.dagrun.*.first_task_scheduling_delay measurement.measurement.dag_id.measurement"  ]
Copy

After completing these steps, Pulse collects Airflow metrics through StatsD and receives Airflow events through NATS for monitoring and analysis.

Last updated on

Was this page helpful?

Configure Airflow and Pulse to Monitor Standalone Airflow on Kubernetes

Configure Airflow on Kubernetes to Emit Metrics

Step 1: Install the Acceldata Event Listener Plugin

Step 2: Create a ConfigMap for the Plugin and Mount It to the Scheduler

Step 3: Install nats-py in the Scheduler Container

Step 4: Distribute the Plugin to All Task Pods

Step 5: Validate Event Streaming to NATS

Configure Pulse to Monitor Airflow on Kubernetes

Prerequisites

Configure PulseNode to Collect Airflow StatsD Metrics

Step 3: Install `nats-py` in the Scheduler Container