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.

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

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

Example config.ini:

[pulse]
ad.cluster = airflowonk8s
ad.events.url = nats://host:19009
Verify the plugin is loaded:

airflow plugins

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

1. Create a ConfigMap from the event listener plugin files:

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

1. Mount the ConfigMap in the Airflow scheduler deployment:

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

Step 3: Install nats-py in the Scheduler Container

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

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

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:

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

Step 5: Validate Event Streaming to NATS

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

nats stream ls | grep airflow
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.

1. On the Pulse node, update:

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

1. Add the following Telegraf StatsD input configuration:

[[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"
]

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