Airflow DAG Monitor

Автоматизированная система мониторинга Apache Airflow DAG-ов с интеграцией в Zabbix. Обнаруживает проблемные DAG-запуски (ошибки, зависания), выполняет автоматический перезапуск и эскалирует нерешённые инциденты на дашборд администратора.

---

Оглавление

• Возможности
• Архитектура
• Требования
• Установка
• Автономная сборка (offline)
• Auto-discovery Docker Airflow
• Конфигурация
• Запуск
• Настройка Zabbix
• Логика работы
• Логирование и диагностика
• Безопасность
• Структура проекта
• Лицензия

---

Возможности

• Auto-discovery Docker: автоматическое определение URL, порта и учётных данных Airflow из Docker-контейнера, .env и docker-compose.yml
• Универсальность: один и тот же архив работает в разных регионах/контурах без ручной правки подключения к Airflow
• Автоматическое обнаружение всех активных (не приостановленных) DAG-ов через Airflow REST API
• Детекция проблем: упавшие (failed) и зависшие (выполнение > 30 мин) DAG-запуски
• Самовосстановление: однократная попытка перезапуска с контролем результата
• Эскалация в Zabbix: при неуспешном перезапуске данные экспортируются для Zabbix Agent
• Heartbeat-мониторинг: Zabbix отслеживает работоспособность самого монитора
• Идемпотентность: персистентное состояние исключает дублирование перезапусков и алертов
• Автодетекция API: поддержка Airflow REST API v1 и experimental API
• Graceful shutdown: корректное завершение по SIGTERM/SIGINT с сохранением состояния

---

Архитектура

+------------------------------------------------------------------+
|                      Airflow DAG Monitor                         |
|                                                                  |
|  +----------+   +----------+   +--------------+   +----------+  |
|  | Airflow  |-->| Analyzer |-->| ActionHandler |-->|   Data   |  |
|  | Client   |   |          |   |              |   | Exporter |  |
|  +----------+   +----------+   +--------------+   +----------+  |
|       |                |               |               |         |
|       v                v               v               v         |
|  +--------------------------------------------------------+     |
|  |                  State Manager                         |     |
|  |            (retry counts, alert flags)                 |     |
|  +--------------------------------------------------------+     |
+------------------------------------------------------------------+
       |                                                 |
       v                                                 v
  +---------+   +----------+   +---------+   +----------------+
  | Airflow |   | JSON     |   | Zabbix  |   | Zabbix Server  |
  |   API   |   | files    |-->| Agent   |-->| Dashboard      |
  +---------+   +----------+   +---------+   +----------------+

---

Требования

Компонент	Версия
Python	3.10+
Apache Airflow	2.x (REST API v1)
Zabbix Agent	5.x / 6.x / 7.x (уже установлен на хосте)
Docker	для auto-discovery (опционально)
ОС	Linux (systemd)

---

Установка

Автоматическая

git clone https://git.totmin.ru/en2zmax/AirflowDAGMonitor.git /opt/airflow-monitor
cd /opt/airflow-monitor
./setup.sh

Скрипт setup.sh выполнит:

1. Копирование файлов в /opt/airflow-monitor (если запущен из другого каталога)
2. Создание Python virtual environment в /opt/airflow-monitor/venv
3. Установку зависимостей из requirements.txt (требуется интернет)
4. Создание рабочих директорий (/var/lib/airflow-monitor, /var/log/airflow-monitor)
5. Установку systemd unit файла

Ручная

cd /opt/airflow-monitor

# Виртуальное окружение
python3 -m venv venv
source venv/bin/activate
pip install -r requirements.txt

# Директории
sudo mkdir -p /var/lib/airflow-monitor /var/log/airflow-monitor
sudo chown $(whoami):$(id -gn) /var/lib/airflow-monitor /var/log/airflow-monitor

# Systemd (опционально)
sudo cp airflow-monitor.service /etc/systemd/system/
sudo systemctl daemon-reload

---

Автономная сборка (offline)

Для развёртывания в закрытом контуре без доступа в интернет используется скрипт build.sh. Он собирает ZIP-архив, содержащий весь исходный код, wheel-пакеты зависимостей и скрипт офлайн-установки.

Сборка архива

На машине с доступом в интернет:

./build.sh

Результат: dist/airflow-monitor-1.0.0.tar.gz

Содержимое архива

airflow-monitor-1.0.0/
├── airflow_monitor/            # Исходный код
├── wheels/                     # Wheel-пакеты всех зависимостей
│   ├── requests-2.32.3-py3-none-any.whl
│   ├── PyYAML-6.0.2-cp312-...-linux_x86_64.whl
│   ├── urllib3-2.2.3-py3-none-any.whl
│   ├── charset_normalizer-...whl
│   ├── idna-3.10-py3-none-any.whl
│   └── certifi-2024.8.30-py3-none-any.whl
├── config.yaml                 # Шаблон конфигурации
├── requirements.txt
├── airflow-monitor.service     # Systemd unit
├── install.sh                  # Скрипт офлайн-установки
└── README.md

Развёртывание на целевом сервере

На машине без интернета:

# 1. Скопировать архив на сервер
scp dist/airflow-monitor-1.0.0.tar.gz user@target:/tmp/

# 2. Распаковать
tar xzf /tmp/airflow-monitor-1.0.0.tar.gz -C /tmp/

# 3. Установить (по умолчанию в /opt/airflow-monitor)
cd /tmp/airflow-monitor-1.0.0
bash install.sh

Сервис будет установлен в /opt/airflow-monitor.

Параметры install.sh

bash install.sh [--prefix INSTALL_DIR] [--user SERVICE_USER]

  --prefix   Каталог установки (по умолчанию: /opt/airflow-monitor)
  --user     Пользователь для запуска сервиса (по умолчанию: текущий)

Примеры:

# Стандартная установка в /opt/airflow-monitor
bash install.sh

# Указать пользователя для запуска сервиса
bash install.sh --user airflow

# Установить в альтернативный каталог
bash install.sh --prefix /srv/airflow-monitor --user airflow

Скрипт install.sh выполнит:

1. Копирование файлов в /opt/airflow-monitor (или --prefix)
2. Создание Python virtual environment
3. Установку зависимостей из локальных wheel-файлов (pip install --no-index --find-links wheels/)
4. Создание рабочих директорий (/var/lib/airflow-monitor, /var/log/airflow-monitor)
5. Установку systemd unit с автоподстановкой путей и пользователя

Важно: На целевом сервере должен быть установлен Python 3.10+ и Zabbix Agent. Доступ в интернет не требуется — все зависимости включены в архив.

---

Auto-discovery Docker Airflow

Монитор умеет автоматически находить Airflow в Docker на текущем сервере. Флаг --discover делает скрипт универсальным: один и тот же архив разворачивается в любом регионе/контуре без ручной правки URL, порта и учётных данных.

Что делает discovery

docker ps                          → найти контейнер airflow-webserver
docker inspect <container>         → получить compose project directory
docker port <container> 8080       → получить host-порт webserver
<compose_dir>/.env                 → прочитать переменные окружения
<compose_dir>/docker-compose.yml   → прочитать конфигурацию сервисов

Из этих данных извлекаются:

• URL: http://localhost:<host_port> (из docker port)
• Username: из .env (_AIRFLOW_WWW_USER_USERNAME или AIRFLOW_WWW_USER_USERNAME)
• Password: из .env (_AIRFLOW_WWW_USER_PASSWORD или AIRFLOW_WWW_USER_PASSWORD), с разрешением ${VAR:-default} из docker-compose

Использование

# Проверить что discovery находит (без запуска монитора)
/opt/airflow-monitor/venv/bin/python -m airflow_monitor --discover --dry-run

# Запуск с auto-discovery (секция airflow в config.yaml перезаписывается)
/opt/airflow-monitor/venv/bin/python -m airflow_monitor --discover -c /opt/airflow-monitor/config.yaml

# Discovery без config.yaml (все параметры по умолчанию + Zabbix не настроен)
/opt/airflow-monitor/venv/bin/python -m airflow_monitor --discover

Пример вывода --dry-run:

Discovered Airflow at http://localhost:80
  Compose dir: /opt/airflow
  Container:   airflow-airflow-webserver-1
  User:        airflow

--- Effective Airflow config ---
{
  "base_url": "http://localhost:80",
  "username": "airflow",
  "password": "P@ssw0rd1",
  "api_version": "v1",
  "timeout": 30,
  "verify_ssl": false,
  "request_delay": 0.5
}

Приоритет настроек

При запуске с --discover --config config.yaml:

1. Загружается config.yaml (секции monitor, zabbix, logging)
2. Секция airflow перезаписывается данными из Docker discovery
3. Это позволяет настроить Zabbix и пороги в конфиге, а подключение к Airflow определять автоматически

Требования для discovery

• Docker должен быть доступен текущему пользователю (docker ps без sudo)
• Airflow webserver контейнер должен быть запущен
• .env файл должен быть в директории docker-compose проекта

---

Конфигурация

Все параметры задаются в config.yaml. Скопируйте шаблон и отредактируйте:

cp config.yaml config.yaml.bak
vim config.yaml

Основные параметры

airflow — подключение к Airflow

Параметр	По умолчанию	Описание
base_url	http://localhost:8080	URL Airflow webserver
username	airflow	Логин для Basic Auth
password	airflow	Пароль для Basic Auth
api_version	auto	Версия API: auto, v1, experimental
timeout	30	Таймаут HTTP-запросов (секунды)
verify_ssl	true	Проверка SSL-сертификатов
request_delay	0.5	Задержка между запросами к API (секунды)

monitor — параметры мониторинга

Параметр	По умолчанию	Описание
cycle_interval	300	Интервал между циклами мониторинга (секунды)
long_running_threshold	1800	Порог длительности выполнения DAG (секунды)
retry_wait	120	Ожидание после перезапуска перед проверкой
max_retries	1	Максимум попыток перезапуска на один DAG run
state_file	/var/lib/airflow-monitor/state.json	Файл персистентного состояния
state_max_age	86400	Время жизни записей состояния (секунды)

zabbix — интеграция с Zabbix Agent

Параметр	По умолчанию	Описание
enabled	true	Включить/выключить экспорт данных в файлы
data_dir	/var/lib/airflow-monitor	Каталог для файлов данных
problems_file	problems.json	Файл со списком проблем (JSON)
heartbeat_file	heartbeat	Файл с timestamp последнего цикла
status_file	status.json	Файл со статусом монитора (JSON)

logging — логирование

Параметр	По умолчанию	Описание
level	INFO	Уровень: DEBUG, INFO, WARNING, ERROR
file	/var/log/airflow-monitor/monitor.log	Путь к файлу логов
max_bytes	10485760	Размер файла до ротации (10 МБ)
backup_count	5	Количество ротированных файлов

---

Запуск

Тестовый запуск (ручной)

/opt/airflow-monitor/venv/bin/python -m airflow_monitor --config /opt/airflow-monitor/config.yaml

Остановка: Ctrl+C

Production (systemd)

# Включить автозапуск и стартовать
sudo systemctl enable --now airflow-monitor

# Проверить статус
sudo systemctl status airflow-monitor

# Просмотр логов в реальном времени
journalctl -u airflow-monitor -f

# Перезапуск после изменения config.yaml
sudo systemctl restart airflow-monitor

# Остановка
sudo systemctl stop airflow-monitor

Параметры CLI

usage: python -m airflow_monitor [-h] [--config CONFIG] [--discover] [--dry-run]

  --config, -c    Путь к config.yaml (по умолчанию: config.yaml в текущей директории)
  --discover, -d  Автообнаружение Airflow Docker (перезаписывает секцию airflow)
  --dry-run       С --discover: показать найденную конфигурацию и выйти

---

Настройка Zabbix

Схема доставки данных

Прямого доступа к Zabbix Server с сервера Airflow нет. Данные передаются через локальный Zabbix Agent, который уже подключён к серверу (напрямую или через Zabbix Proxy).

+------------------+   файлы    +--------------+   сеть    +--------------+   сеть    +---------------+
| airflow-monitor  |---------->| Zabbix Agent |--------->| Zabbix Proxy |--------->| Zabbix Server |
| (systemd)        | /var/lib/ | UserParameter|          | (если есть)  |          | Dashboard     |
+------------------+           +--------------+          +--------------+          +---------------+
                                                                |                         |
                                                                +---- или напрямую -------+

1. Монитор пишет результаты в файлы /var/lib/airflow-monitor/
2. Zabbix Agent читает файлы через UserParameter (конфиг airflow-monitor.conf)
3. Zabbix Agent передаёт данные на Zabbix Server/Proxy по своему стандартному каналу

1. Конфиг Zabbix Agent (устанавливается автоматически)

Файл /etc/zabbix/zabbix_agentd.conf.d/airflow-monitor.conf устанавливается при запуске setup.sh или install.sh. Содержимое:

# Список проблемных DAG-ов (JSON)
UserParameter=airflow.dag.problems,cat /var/lib/airflow-monitor/problems.json 2>/dev/null || echo '[]'

# Количество проблемных DAG-ов
UserParameter=airflow.dag.problems.count,python3 -c "import json,sys; print(len(json.load(open('/var/lib/airflow-monitor/problems.json'))))" 2>/dev/null || echo 0

# Heartbeat — epoch timestamp
UserParameter=airflow.monitor.heartbeat,cat /var/lib/airflow-monitor/heartbeat 2>/dev/null || echo 0

# Статус монитора (JSON)
UserParameter=airflow.monitor.status,cat /var/lib/airflow-monitor/status.json 2>/dev/null || echo '{}'

# Сервис запущен (1/0)
UserParameter=airflow.monitor.alive,systemctl is-active airflow-monitor >/dev/null 2>&1 && echo 1 || echo 0

2. Проверка на сервере

После установки проверьте, что Zabbix Agent корректно читает данные:

# Тест UserParameter через агент
zabbix_agentd -t airflow.dag.problems
zabbix_agentd -t airflow.dag.problems.count
zabbix_agentd -t airflow.monitor.heartbeat
zabbix_agentd -t airflow.monitor.alive

3. Импорт шаблона (рекомендуется)

Вместо ручного создания items и триггеров импортируйте готовый шаблон:

Файл	Zabbix версия
zabbix/zbx_template_airflow_monitor.yaml	6.x+
zabbix/zbx_template_airflow_monitor_5x.xml	5.x

Импорт: Configuration → Templates → Import → выбрать файл.

После импорта привяжите шаблон "Airflow DAG Monitor" к хосту.

Шаблон содержит:

• 8 items (5 основных + 3 dependent из JSON)
• 4 триггера (проблемы DAG, heartbeat, сервис, cycle time)
• 2 графика (проблемы, время цикла)
• 2 макроса (пороги, переопределяются на уровне хоста)

4. Ручное создание items (если без шаблона)

На Zabbix Server создайте items для хоста (хост определяется по HostnameItem=system.hostname из zabbix_agentd.conf):

Item	Тип	Key	Тип данных	Интервал
Airflow DAG Problems	Zabbix agent	airflow.dag.problems	Text	1m
Airflow DAG Problems Count	Zabbix agent	airflow.dag.problems.count	Numeric (unsigned)	1m
Airflow Monitor Heartbeat	Zabbix agent	airflow.monitor.heartbeat	Numeric (unsigned)	1m
Airflow Monitor Status	Zabbix agent	airflow.monitor.status	Text	5m
Airflow Monitor Alive	Zabbix agent	airflow.monitor.alive	Numeric (unsigned)	1m

4. Создание триггеров

Есть проблемные DAG-и:

Имя:       Airflow: Обнаружены проблемные DAG-запуски ({ITEM.LASTVALUE1})
Выражение: last(/host/airflow.dag.problems.count)>0
Важность:  High

Монитор не обновляет данные (heartbeat устарел более чем на 10 минут):

Имя:       Airflow Monitor: Данные устарели
Выражение: (now()-last(/host/airflow.monitor.heartbeat))>600
Важность:  Disaster

Сервис мониторинга остановлен:

Имя:       Airflow Monitor: Сервис не запущен
Выражение: last(/host/airflow.monitor.alive)=0
Важность:  High

5. Формат данных

Файл problems.json содержит JSON-массив:

[
  {
    "dag_id": "etl_daily_pipeline",
    "dag_run_id": "scheduled__2026-04-08T00:00:00+00:00",
    "issue_type": "failed",
    "status": "failed",
    "duration_seconds": 145.3,
    "error_info": "Task 'load_data' failed: ConnectionError: Connection refused",
    "retry_count": 1
  },
  {
    "dag_id": "report_generator",
    "dag_run_id": "scheduled__2026-04-08T06:00:00+00:00",
    "issue_type": "long_running",
    "status": "running",
    "duration_seconds": 2415.7,
    "error_info": "",
    "retry_count": 1
  }
]

Поле	Описание
dag_id	Идентификатор DAG
dag_run_id	Идентификатор запуска
issue_type	Тип проблемы: failed или long_running
status	Состояние в Airflow
duration_seconds	Длительность выполнения (секунды)
error_info	Информация об ошибке из task instance (для failed)
retry_count	Количество выполненных перезапусков

При отсутствии проблем файл содержит [] — триггер автоматически снимается.

Файл status.json:

{
  "timestamp": "2026-04-08T10:15:00+00:00",
  "cycle_count": 42,
  "dag_count": 15,
  "issue_count": 0,
  "cycle_time_seconds": 3.45
}

---

Логика работы

Цикл мониторинга

        +------------------------+
        |  Загрузка состояния    |
        +-----------+------------+
                    v
        +----------------------------+
        | Получение списка DAG-ов    |
        | GET /api/v1/dags           |
        +-----------+----------------+
                    v
        +----------------------------+
        | Для каждого DAG:           |
        | GET /api/v1/dags/{id}/runs |
        | (фильтр: running, failed)  |
        +-----------+----------------+
                    v
          +--------------------+
          | Анализ: проблемы?  |
          +---+------------+---+
          нет |            | да
              v            v
          +------+   +-----------------+
          | Done |   | Перезапуск (x1) |
          +------+   +--------+--------+
                              v
                    +-------------------+
                    | Ожидание 2 мин    |
                    +--------+----------+
                             v
                    +-------------------+
                    | Повторная проверка |
                    +----+----------+---+
                  ОК     |          | Проблема
                         v          v
                    +------+   +------------------+
                    | Done |   | Экспорт в файлы  |
                    +------+   | для Zabbix Agent |
                               +--------+---------+
                                        v
                              +-------------------+
                              | Heartbeat+cleanup  |
                              | Сохранение состоян.|
                              +-------------------+

Обработка ошибок

Ситуация	Поведение
Airflow API недоступен	Пропуск цикла, повтор через cycle_interval
Таймаут запроса к API	Пропуск конкретного DAG, продолжение с остальными
Ошибка авторизации (401/403)	Логирование, продолжение цикла
Ошибка записи файлов данных	Логирование ошибки, повтор в следующем цикле
Zabbix Agent не читает данные	Проверить конфиг в /etc/zabbix/zabbix_agentd.conf.d/
Повреждённый файл состояния	Пересоздание с пустым состоянием
Необработанное исключение в цикле	Логирование traceback, переход к следующему циклу

Защита от дублирования

Персистентное состояние хранит для каждого DAG-запуска:

• Счётчик перезапусков — исключает повторный retry
• Флаг alerted — исключает повторный экспорт в Zabbix
• Ключ {dag_id}::{dag_run_id} — уникальная идентификация
• Автоочистка записей старше 24 часов

---

Логирование и диагностика

Файлы логов

Путь	Описание
/var/log/airflow-monitor/monitor.log	Лог приложения (с ротацией)
journalctl -u airflow-monitor	Лог через systemd journal

Уровни логирования

• DEBUG — детали API-запросов, содержимое state файла
• INFO — начало/конец цикла, количество DAG-ов, выполненные действия
• WARNING — проблемные DAG-и, ошибки отдельных API-вызовов
• ERROR — ошибки Zabbix, невозможность сохранить состояние
• CRITICAL — невозможность подключиться к Airflow API

Проверка работоспособности

# Статус сервиса
systemctl status airflow-monitor

# Последние логи
journalctl -u airflow-monitor --since "1 hour ago"

# Содержимое state файла
cat /var/lib/airflow-monitor/state.json | python3 -m json.tool

# Проверка что Zabbix Agent читает данные
zabbix_agentd -t airflow.dag.problems
zabbix_agentd -t airflow.dag.problems.count
zabbix_agentd -t airflow.monitor.heartbeat

---

Безопасность

Учётные данные

Файл config.yaml содержит пароль Airflow в открытом виде. Рекомендации:

# Ограничить доступ к конфигурации
chmod 600 config.yaml
chown root:root config.yaml

Systemd hardening

Сервис запускается с ограничениями:

Директива	Значение	Назначение
NoNewPrivileges	true	Запрет эскалации привилегий
ProtectSystem	strict	Файловая система только для чтения
ReadWritePaths	явный список	Доступ на запись только в рабочие каталоги
PrivateTmp	true	Изолированный /tmp
ProtectHome	read-only	Домашняя директория только для чтения

Lock file

Файл блокировки /var/run/airflow-monitor.lock предотвращает одновременный запуск нескольких экземпляров. Блокировка снимается автоматически при завершении процесса.

---

Структура проекта

air-flow-monitor/
├── airflow_monitor/              # Основной Python-пакет
│   ├── __init__.py               # Версия пакета
│   ├── __main__.py               # Точка входа, CLI, сигналы, lock file
│   ├── config.py                 # Загрузка YAML-конфигурации в dataclasses
│   ├── discovery.py              # Auto-discovery Airflow Docker инфраструктуры
│   ├── client.py                 # HTTP-клиент Airflow REST API
│   ├── analyzer.py               # Анализ DAG-запусков, классификация проблем
│   ├── state.py                  # Персистентное состояние (JSON, атомарная запись)
│   ├── actions.py                # Перезапуск DAG-ов, экспорт данных для Zabbix Agent
│   └── monitor.py                # Оркестрация цикла мониторинга
├── config.yaml                   # Конфигурация (Airflow, Zabbix, пороги, логи)
├── requirements.txt              # Python-зависимости
├── setup.sh                      # Скрипт установки (с интернетом)
├── build.sh                      # Сборка автономного ZIP-архива
├── airflow-monitor.service       # Systemd unit файл
├── zabbix/
│   └── airflow-monitor.conf      # UserParameter конфиг для Zabbix Agent
├── dist/                         # Собранные архивы (после ./build.sh)
└── README.md

Зависимости

Пакет	Версия	Назначение
requests	>=2.31, <3	HTTP-клиент для Airflow API
PyYAML	>=6.0, <7	Парсинг конфигурации

Все остальные модули (json, logging, subprocess, signal, threading, fcntl, pathlib, dataclasses, datetime) входят в стандартную библиотеку Python.

---

Лицензия

Внутренний проект. Все права защищены.