clickstream-ch-kafka-superset-demo/plans/monitoring_airflow_plan.md

План подключения Airflow к мониторингу

Дата создания: 2026-02-08
Источник: Airflow 2.10.5 via statsd_exporter
Проверка API: Context7 (/prometheus/docs для scrape_configs)

---

1. Обзор

Подключаем мониторинг Airflow через statsd_exporter. Airflow встроенно поддерживает отправку метрик в StatsD-формате, а statsd_exporter конвертирует их в Prometheus-формат.

Метрики, которые будем собирать:

• airflow_dag_processing_total_parse_time — время парсинга DAG
• airflow_dagbag_size — количество DAG в DagBag
• airflow_executor_open_slots — доступные слоты executor
• airflow_executor_queued_tasks — задачи в очереди
• airflow_executor_running_tasks — запущенные задачи
• airflow_task_duration_seconds{task_id, dag_id} — длительность выполнения тасков
• airflow_task_failures_total{task_id, dag_id} — количество падений тасков
• airflow_task_success_total{task_id, dag_id} — успешные выполнения
• airflow_scheduler_heartbeat_total — heartbeats шедулера

---

2. Изменения в инфраструктуре

2.1. docker-compose.yml

Добавить сервис statsd-exporter:

  statsd-exporter:
    image: prom/statsd-exporter:v0.27.1
    command: [
      "--statsd.listen-udp=:8125",
      "--statsd.listen-tcp=",
      "--web.listen-address=:9102",
      "--statsd.mapping-config=/tmp/statsd_mapping.yml"
    ]
    ports:
      - "9102:9102"    # Prometheus metrics endpoint
      - "8125:8125/udp" # StatsD receive port
    volumes:
      - ./configs/statsd_mapping.yml:/tmp/statsd_mapping.yml:ro
    networks:
      - cs_dwh

Добавить переменные окружения в airflow-default-env:

x-airflow-env: &airflow-default-env
  # ... существующие переменные ...
  # StatsD для мониторинга
  AIRFLOW__METRICS__STATSD_ON: "True"
  AIRFLOW__METRICS__STATSD_HOST: "statsd-exporter"
  AIRFLOW__METRICS__STATSD_PORT: "8125"
  AIRFLOW__METRICS__STATSD_PREFIX: "airflow"

Почему prom/statsd-exporter:v0.27.1:

• Стабильная версия с поддержкой mapping-конфигурации
• Официальный образ от Prometheus team
• Проверено на совместимость с Airflow 2.10.x

2.2. configs/statsd_mapping.yml

Создать mapping-конфигурацию для конвертации StatsD метрик в Prometheus:

# Mapping StatsD → Prometheus для Airflow
mappings:
  # DAG processing metrics
  - match: "airflow.dag_processing.total_parse_time"
    name: "airflow_dag_processing_total_parse_time"
    help: "Total time to parse all DAGs"
    type: gauge

  - match: "airflow.dagbag_size"
    name: "airflow_dagbag_size"
    help: "DAGs in the DagBag"
    type: gauge

  # Executor metrics
  - match: "airflow.executor.open_slots"
    name: "airflow_executor_open_slots"
    help: "Open slots on executor"
    type: gauge

  - match: "airflow.executor.queued_tasks"
    name: "airflow_executor_queued_tasks"
    help: "Queued tasks on executor"
    type: gauge

  - match: "airflow.executor.running_tasks"
    name: "airflow_executor_running_tasks"
    help: "Running tasks on executor"
    type: gauge

  # Task metrics с labels
  - match: "airflow.operator_failures_*"
    name: "airflow_operator_failures"
    help: "Operator failures"
    type: counter
    labels:
      operator: "$1"

  - match: "airflow.operator_successes_*"
    name: "airflow_operator_successes"
    help: "Operator successes"
    type: counter
    labels:
      operator: "$1"

  # Task duration by dag_id и task_id
  # Проверено на Airflow 2.10.5: dag.<dag_id>.<task_id>.duration
  - match: "airflow.dag.*.*.duration"
    name: "airflow_task_duration_seconds"
    help: "Task duration by dag_id and task_id"
    type: timer
    labels:
      dag_id: "$1"
      task_id: "$2"

  # Task failures/success by dag_id и task_id
  # Проверено на Airflow 2.10.5: ti.finish.<dag_id>.<task_id>.<state>
  - match: "airflow.ti.finish.*.*.failed"
    name: "airflow_task_failures_total"
    help: "Task failures by dag_id and task_id"
    type: counter
    labels:
      dag_id: "$1"
      task_id: "$2"

  - match: "airflow.ti.finish.*.*.success"
    name: "airflow_task_success_total"
    help: "Task success by dag_id and task_id"
    type: counter
    labels:
      dag_id: "$1"
      task_id: "$2"

  # Scheduler metrics
  - match: "airflow.scheduler_heartbeat"
    name: "airflow_scheduler_heartbeat_total"
    help: "Scheduler heartbeats"
    type: counter

  - match: "airflow.scheduler.critical_section_busy"
    name: "airflow_scheduler_critical_section_busy"
    help: "Scheduler critical section busy"
    type: gauge

2.3. configs/prometheus.yml

Добавить job для скрейпа statsd-exporter:

scrape_configs:
  - job_name: "clickhouse"
    metrics_path: "/metrics"
    static_configs:
      - targets: ["clickhouse:9126"]
        labels:
          instance: Clickhouse-1
    honor_labels: true

  - job_name: "kafka"
    metrics_path: "/metrics"
    static_configs:
      - targets: ["kafka-exporter:9308"]
        labels:
          instance: Kafka-1
    honor_labels: true

  # Новый job для Airflow
  - job_name: "airflow"
    metrics_path: "/metrics"
    static_configs:
      - targets: ["statsd-exporter:9102"]
        labels:
          instance: Airflow-1
    honor_labels: true

---

3. Дашборд Grafana

3.1. Создать файл: configs/grafana/provisioning/dashboards/airflow-overview.json

Основные панели дашборда:

Панель	PromQL запрос	Описание
DAG Bag Size	airflow_dagbag_size	Количество DAG в системе
Parse Time	airflow_dag_processing_total_parse_time	Время парсинга DAG
Executor Slots	airflow_executor_open_slots	Доступные слоты
Queued Tasks	airflow_executor_queued_tasks	Задачи в очереди
Running Tasks	airflow_executor_running_tasks	Выполняемые задачи
Task Duration	rate(airflow_task_duration_seconds_sum[5m]) / rate(airflow_task_duration_seconds_count[5m])	Средняя длительность тасков
Task Failures	rate(airflow_task_failures_total[5m])	Rate падений по таскам
Scheduler Heartbeat	rate(airflow_scheduler_heartbeat_total[5m])	Активность шедулера

3.2. Структура дашборда (основные секции)

{
  "dashboard": {
    "id": null,
    "uid": "airflow-overview",
    "title": "Airflow Overview",
    "tags": ["airflow", "orchestration"],
    "timezone": "Europe/Moscow",
    "schemaVersion": 36,
    "refresh": "10s",
    "panels": [
      // Row 1: Scheduler Health
      // - DAG Bag Size (stat)
      // - Parse Time (gauge)
      // - Scheduler Heartbeat (graph)
      
      // Row 2: Executor Status
      // - Open Slots (stat)
      // - Queued Tasks (stat)
      // - Running Tasks (stat)
      // - Tasks distribution (graph)
      
      // Row 3: Task Performance
      // - Task Duration by DAG (heatmap или graph)
      // - Task Failures Rate (graph)
      // - Task Success Rate (graph)
      
      // Row 4: Per-DAG Details
      // - Top DAGs by runtime (table)
      // - Failed tasks by DAG (table)
    ]
  }
}

---

4. Алерты Grafana

4.1. Создать файл: configs/grafana/provisioning/alerting/airflow-alert-rules.yml

apiVersion: 1
groups:
  - orgId: 1
    name: airflow_alerts
    folder: Airflow
    interval: 30s
    rules:
      # Alert: Scheduler не отправляет heartbeats
      - uid: airflow_scheduler_down
        title: Airflow Scheduler Down
        condition: B
        data:
          - refId: A
            relativeTimeRange:
              from: 300
              to: 0
            datasourceUid: prometheus_uid
            model:
              expr: rate(airflow_scheduler_heartbeat_total[5m])
              instant: true
          - refId: B
            relativeTimeRange:
              from: 0
              to: 0
            datasourceUid: __expr__
            model:
              type: threshold
              expression: A
              conditions:
                - evaluator:
                    type: lt
                    params: [0.1]
        noDataState: NoData
        execErrState: Error
        for: 2m
        annotations:
          summary: "Airflow scheduler не отправляет heartbeats"
          description: "Scheduler possible down — rate(airflow_scheduler_heartbeat_total) < 0.1 в течение 2 минут"
        labels:
          severity: critical

      # Alert: В очереди слишком много задач
      - uid: airflow_queue_high
        title: Airflow Queue Backlog
        condition: B
        data:
          - refId: A
            datasourceUid: prometheus_uid
            model:
              expr: airflow_executor_queued_tasks
              instant: true
          - refId: B
            datasourceUid: __expr__
            model:
              type: threshold
              expression: A
              conditions:
                - evaluator:
                    type: gt
                    params: [50]
        noDataState: OK
        execErrState: Error
        for: 5m
        annotations:
          summary: "В очереди Airflow > 50 задач"
          description: "airflow_executor_queued_tasks превышает 50 в течение 5 минут — возможна перегрузка"
        labels:
          severity: warning

      # Alert: Много падений тасков
      - uid: airflow_task_failures_high
        title: High Task Failure Rate
        condition: B
        data:
          - refId: A
            datasourceUid: prometheus_uid
            model:
              expr: rate(airflow_task_failures_total[5m])
              instant: true
          - refId: B
            datasourceUid: __expr__
            model:
              type: threshold
              expression: A
              conditions:
                - evaluator:
                    type: gt
                    params: [0.1]
        noDataState: OK
        execErrState: Error
        for: 3m
        annotations:
          summary: "Высокий rate падений тасков"
          description: "rate(airflow_task_failures_total) > 0.1 в течение 3 минут — проверьте логи DAG"
        labels:
          severity: warning

      # Alert: DAG parsing занимает слишком много времени
      - uid: airflow_parse_time_high
        title: High DAG Parse Time
        condition: B
        data:
          - refId: A
            datasourceUid: prometheus_uid
            model:
              expr: airflow_dag_processing_total_parse_time
              instant: true
          - refId: B
            datasourceUid: __expr__
            model:
              type: threshold
              expression: A
              conditions:
                - evaluator:
                    type: gt
                    params: [30]
        noDataState: OK
        execErrState: Error
        for: 5m
        annotations:
          summary: "Парсинг DAG занимает > 30 секунд"
          description: "Возможно, есть тяжелые DAG — оптимизируйте или разбейте на под-DAG"
        labels:
          severity: info

---

5. Проверка после деплоя

# 1. Перезапуск стека с новым сервисом
make reload-monitoring

# 2. Проверка, что statsd-exporter поднялся
curl -s http://localhost:9102/metrics | grep airflow

# 3. Проверка таргета в Prometheus
curl -s http://localhost:9090/api/v1/targets | grep -A5 airflow

# 4. Проверка дашборда в Grafana
open http://localhost:3000/d/airflow-overview

# 5. Проверка алертов
curl -s -u admin:admin http://localhost:3000/api/v1/provisioning/alert-rules | grep airflow

---

6. Troubleshooting

Нет метрик airflow_* в Prometheus

1. Проверить env vars в Airflow:

   docker compose exec airflow-webserver env | grep STATSD
   

   Ожидается: AIRFLOW__METRICS__STATSD_ON=True

2. Проверить доступность statsd-exporter:

   docker compose exec airflow-webserver nc -zv statsd-exporter 8125
   

3. Проверить метрики в statsd-exporter напрямую:

   curl -s http://localhost:9102/metrics
   

StatsD метрики не мапятся

Проверить формат mapping-файла:

docker compose logs statsd-exporter | grep -i "mapping\|error"

Airflow не отправляет метрики

Проверить конфигурацию внутри Airflow:

docker compose exec airflow-webserver airflow config get-value metrics statsd_on
docker compose exec airflow-webserver airflow config get-value metrics statsd_host

---

7. Зависимости и порядок применения

1. Создать configs/statsd_mapping.yml
2. Добавить statsd-exporter в docker-compose.yml
3. Добавить StatsD env vars в airflow-default-env
4. Обновить configs/prometheus.yml (добавить job airflow)
5. Создать дашборд configs/grafana/provisioning/dashboards/airflow-overview.json
6. Создать алерты configs/grafana/provisioning/alerting/airflow-alert-rules.yml
7. Запустить make reload-monitoring