commit bc437cbf8caa8b4c1180cba585a1184fa5b01e5f Author: Maksim Totmin Date: Wed Apr 8 17:23:03 2026 +0700 feat: добавил систему мониторинга Airflow DAG-ов с интеграцией в Zabbix - автообнаружение Docker Airflow через docker inspect и .env - мониторинг failed и long-running DAG-запусков с автоматическим retry - экспорт данных в файлы для Zabbix Agent через UserParameter - офлайн-сборка ZIP-архива для закрытых контуров - Zabbix шаблоны для 5.x (XML) и 6.x+ (YAML) - systemd сервис с graceful shutdown и lock file diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..1bcf1b0 --- /dev/null +++ b/.gitignore @@ -0,0 +1,8 @@ +__pycache__/ +*.py[cod] +*.egg-info/ +venv/ +build/ +dist/ +*.tmp +.env diff --git a/README.md b/README.md new file mode 100644 index 0000000..00d1421 --- /dev/null +++ b/README.md @@ -0,0 +1,734 @@ +# Airflow DAG Monitor + +Автоматизированная система мониторинга Apache Airflow DAG-ов с интеграцией в Zabbix. +Обнаруживает проблемные DAG-запуски (ошибки, зависания), выполняет автоматический перезапуск +и эскалирует нерешённые инциденты на дашборд администратора. + +--- + +## Оглавление + +- [Возможности](#возможности) +- [Архитектура](#архитектура) +- [Требования](#требования) +- [Установка](#установка) +- [Автономная сборка (offline)](#автономная-сборка-offline) +- [Auto-discovery Docker Airflow](#auto-discovery-docker-airflow) +- [Конфигурация](#конфигурация) +- [Запуск](#запуск) +- [Настройка Zabbix](#настройка-zabbix) +- [Логика работы](#логика-работы) +- [Логирование и диагностика](#логирование-и-диагностика) +- [Безопасность](#безопасность) +- [Структура проекта](#структура-проекта) + +--- + +## Возможности + +- **Auto-discovery Docker**: автоматическое определение URL, порта и учётных данных Airflow из Docker-контейнера, `.env` и `docker-compose.yml` +- **Универсальность**: один и тот же архив работает в разных регионах/контурах без ручной правки подключения к Airflow +- **Автоматическое обнаружение** всех активных (не приостановленных) DAG-ов через Airflow REST API +- **Детекция проблем**: упавшие (`failed`) и зависшие (выполнение > 30 мин) DAG-запуски +- **Самовосстановление**: однократная попытка перезапуска с контролем результата +- **Эскалация в Zabbix**: при неуспешном перезапуске данные отправляются через `zabbix_sender` +- **Heartbeat-мониторинг**: Zabbix отслеживает работоспособность самого монитора +- **Идемпотентность**: персистентное состояние исключает дублирование перезапусков и алертов +- **Автодетекция API**: поддержка Airflow REST API v1 и experimental API +- **Graceful shutdown**: корректное завершение по SIGTERM/SIGINT с сохранением состояния + +--- + +## Архитектура + +``` +┌──────────────────────────────────────────────────────────────────┐ +│ Airflow DAG Monitor │ +│ │ +│ ┌─────────┐ ┌──────────┐ ┌──────────────┐ ┌────────┐ │ +│ │ Airflow │───>│ Analyzer │───>│ActionHandler │───>│ Zabbix │ │ +│ │ Client │ │ │ │ │ │ Sender │ │ +│ └─────────┘ └──────────┘ └──────────────┘ └────────┘ │ +│ │ │ │ │ +│ v v v │ +│ ┌──────────────────────────────────────────────────────────┐ │ +│ │ State Manager │ │ +│ │ (retry counts, alert flags) │ │ +│ └──────────────────────────────────────────────────────────┘ │ +└──────────────────────────────────────────────────────────────────┘ + │ │ + v v + ┌─────────┐ ┌──────────────┐ + │ Airflow │ │Zabbix Server │ + │ API │ │ Dashboard │ + └─────────┘ └──────────────┘ +``` + +--- + +## Требования + +| Компонент | Версия | +|--------------------|---------------------| +| Python | 3.10+ | +| Apache Airflow | 2.x (REST API v1) | +| Zabbix Agent | 5.x / 6.x / 7.x (уже установлен на хосте) | +| Docker | для auto-discovery (опционально) | +| ОС | Linux (systemd) | + +--- + +## Установка + +### Автоматическая + +```bash +git clone /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 файла + +### Ручная + +```bash +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-пакеты зависимостей и скрипт +офлайн-установки. + +### Сборка архива + +На машине **с доступом в интернет**: + +```bash +# Сборка для текущей платформы +./build.sh + +# Сборка для конкретной платформы (если целевой сервер отличается) +./build.sh --platform manylinux2014_x86_64 +``` + +Результат: `dist/airflow-monitor-1.0.0.zip` + +### Содержимое архива + +``` +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 +``` + +### Развёртывание на целевом сервере + +На машине **без интернета**: + +```bash +# 1. Скопировать архив на сервер +scp dist/airflow-monitor-1.0.0.zip user@target:/tmp/ + +# 2. Распаковать +unzip /tmp/airflow-monitor-1.0.0.zip -d /tmp/ + +# 3. Установить (по умолчанию в /opt/airflow-monitor) +cd /tmp/airflow-monitor-1.0.0 +./install.sh +``` + +Сервис будет установлен в `/opt/airflow-monitor`. + +### Параметры install.sh + +``` +./install.sh [--prefix INSTALL_DIR] [--user SERVICE_USER] + + --prefix Каталог установки (по умолчанию: /opt/airflow-monitor) + --user Пользователь для запуска сервиса (по умолчанию: текущий) +``` + +Примеры: + +```bash +# Стандартная установка в /opt/airflow-monitor +./install.sh + +# Указать пользователя для запуска сервиса +./install.sh --user airflow + +# Установить в альтернативный каталог +./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_sender`. +> Доступ в интернет **не требуется** — все зависимости включены в архив. + +### Кросс-платформенная сборка + +Если архитектура машины сборки отличается от целевого сервера: + +```bash +# Целевой сервер: x86_64 +./build.sh --platform manylinux2014_x86_64 + +# Целевой сервер: aarch64 +./build.sh --platform manylinux2014_aarch64 +``` + +При использовании `--platform` скачиваются только бинарные wheel-пакеты для указанной +платформы. Если зависимость доступна только в виде source distribution, сборка завершится +ошибкой — в таком случае соберите на машине с аналогичной архитектурой. + +--- + +## Auto-discovery Docker Airflow + +Монитор умеет автоматически находить Airflow в Docker на текущем сервере. +Флаг `--discover` делает скрипт **универсальным**: один и тот же архив разворачивается +в любом регионе/контуре без ручной правки URL, порта и учётных данных. + +### Что делает discovery + +``` +docker ps → найти контейнер airflow-webserver +docker inspect → получить compose project directory +docker port 8080 → получить host-порт webserver +/.env → прочитать переменные окружения +/docker-compose.yml → прочитать конфигурацию сервисов +``` + +Из этих данных извлекаются: +- **URL**: `http://localhost:` (из `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 + +### Использование + +```bash +# Проверить что 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`. Скопируйте шаблон и отредактируйте: + +```bash +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` | Количество ротированных файлов | + +--- + +## Запуск + +### Тестовый запуск (ручной) + +```bash +/opt/airflow-monitor/venv/bin/python -m airflow_monitor --config /opt/airflow-monitor/config.yaml +``` + +Остановка: `Ctrl+C` + +### Production (systemd) + +```bash +# Включить автозапуск и стартовать +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 Server │ +│ (systemd) │ /var/lib/... │ UserParameter│ │ Dashboard │ +└──────────────────┘ └──────────────┘ └──────────────┘ + или + ┌──────────────┐ + │ Zabbix Proxy │ + └──────────────┘ +``` + +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`. Содержимое: + +```ini +# Список проблемных 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 корректно читает данные: + +```bash +# Тест 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-массив: + +```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`: + +```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│ │ Перезапуск (1 раз)│ + └────┘ └────────┬─────────┘ + v + ┌─────────────────────┐ + │ Ожидание 2 мин │ + └──────────┬──────────┘ + v + ┌─────────────────────┐ + │ Повторная проверка │ + └───────┬──────┬──────┘ + ОК │ │ Проблема + v v + ┌────┐ ┌──────────────────┐ + │Done│ │ Алерт в Zabbix │ + └────┘ └──────────────────┘ + │ + v + ┌─────────────────────┐ + │ Heartbeat + cleanup │ + │ Сохранение состояния │ + └─────────────────────┘ +``` + +### Обработка ошибок + +| Ситуация | Поведение | +|------------------------------------|----------------------------------------------------| +| Airflow API недоступен | Пропуск цикла, повтор через `cycle_interval` | +| Таймаут запроса к API | Пропуск конкретного DAG, продолжение с остальными | +| Ошибка авторизации (401/403) | Логирование, продолжение цикла | +| `zabbix_sender` не найден | Отключение Zabbix, продолжение мониторинга | +| `zabbix_sender` вернул ошибку | Повтор отправки в следующем цикле | +| Повреждённый файл состояния | Пересоздание с пустым состоянием | +| Необработанное исключение в цикле | Логирование 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 + +### Проверка работоспособности + +```bash +# Статус сервиса +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 +zabbix_sender -z zabbix.example.com -s airflow-server -k airflow.dag.problems -o '[]' +``` + +--- + +## Безопасность + +### Учётные данные + +Файл `config.yaml` содержит пароль Airflow в открытом виде. Рекомендации: + +```bash +# Ограничить доступ к конфигурации +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 +│ └── 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. + +--- + +## Лицензия + +Внутренний проект. Все права защищены. diff --git a/airflow-monitor.service b/airflow-monitor.service new file mode 100644 index 0000000..a60bbda --- /dev/null +++ b/airflow-monitor.service @@ -0,0 +1,30 @@ +[Unit] +Description=Airflow DAG Monitor +Documentation=file:///opt/airflow-monitor/README.md +After=network-online.target +Wants=network-online.target + +[Service] +Type=simple +User=mat +Group=users +WorkingDirectory=/opt/airflow-monitor +ExecStart=/opt/airflow-monitor/venv/bin/python -m airflow_monitor --discover --config /opt/airflow-monitor/config.yaml +Restart=on-failure +RestartSec=30 +TimeoutStopSec=15 + +# Logging +StandardOutput=journal +StandardError=journal +SyslogIdentifier=airflow-monitor + +# Security hardening +NoNewPrivileges=true +ProtectSystem=strict +ReadWritePaths=/var/lib/airflow-monitor /var/log/airflow-monitor /var/run +PrivateTmp=true +ProtectHome=read-only + +[Install] +WantedBy=multi-user.target diff --git a/airflow_monitor/__init__.py b/airflow_monitor/__init__.py new file mode 100644 index 0000000..bcfc52b --- /dev/null +++ b/airflow_monitor/__init__.py @@ -0,0 +1,3 @@ +"""Airflow DAG Monitor - monitors DAG runs and alerts via Zabbix.""" + +__version__ = "1.0.0" diff --git a/airflow_monitor/__main__.py b/airflow_monitor/__main__.py new file mode 100644 index 0000000..ed105f5 --- /dev/null +++ b/airflow_monitor/__main__.py @@ -0,0 +1,187 @@ +"""Entry point for Airflow DAG Monitor. + +Usage: + python -m airflow_monitor --config /path/to/config.yaml + python -m airflow_monitor --discover # auto-detect Docker Airflow + python -m airflow_monitor --discover --dry-run # show discovered config +""" + +import argparse +import dataclasses +import fcntl +import json +import logging +import logging.handlers +import os +import pathlib +import signal +import sys +import threading + +from .config import AppConfig, load_config +from .discovery import DiscoveryError, discover_airflow +from .monitor import Monitor + + +def setup_logging(config) -> None: + """Configure logging with rotating file handler and stdout.""" + root_logger = logging.getLogger() + root_logger.setLevel(getattr(logging, config.level.upper(), logging.INFO)) + + formatter = logging.Formatter( + "%(asctime)s [%(levelname)s] %(name)s: %(message)s", + datefmt="%Y-%m-%d %H:%M:%S", + ) + + # Stdout handler (captured by journald when running as service) + stdout_handler = logging.StreamHandler(sys.stdout) + stdout_handler.setFormatter(formatter) + root_logger.addHandler(stdout_handler) + + # Rotating file handler + log_path = pathlib.Path(config.file) + log_path.parent.mkdir(parents=True, exist_ok=True) + try: + file_handler = logging.handlers.RotatingFileHandler( + config.file, + maxBytes=config.max_bytes, + backupCount=config.backup_count, + encoding="utf-8", + ) + file_handler.setFormatter(formatter) + root_logger.addHandler(file_handler) + except OSError as e: + root_logger.warning("Cannot open log file %s: %s (using stdout only)", config.file, e) + + +def acquire_lock(lock_path: str): + """Acquire an exclusive lock to prevent multiple instances. + + Returns the file descriptor (must be kept open for lock duration). + Exits with code 1 if another instance is running. + """ + lock_dir = pathlib.Path(lock_path).parent + lock_dir.mkdir(parents=True, exist_ok=True) + + try: + fd = open(lock_path, "w") + fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB) + fd.write(str(os.getpid())) + fd.flush() + return fd + except BlockingIOError: + print( + f"Another instance is already running (lock: {lock_path})", + file=sys.stderr, + ) + sys.exit(1) + except OSError as e: + print(f"Cannot acquire lock file {lock_path}: {e}", file=sys.stderr) + sys.exit(1) + + +def _apply_discovery(config: AppConfig, discovered: dict) -> AppConfig: + """Override airflow connection settings with discovered values.""" + airflow = dataclasses.replace( + config.airflow, + base_url=discovered["base_url"], + username=discovered["username"], + password=discovered["password"], + api_version=discovered.get("api_version", "v1"), + verify_ssl=False, # Docker HTTP, not HTTPS + ) + return dataclasses.replace(config, airflow=airflow) + + +def main(): + parser = argparse.ArgumentParser( + description="Airflow DAG Monitor - monitors DAG runs and alerts via Zabbix", + ) + parser.add_argument( + "--config", "-c", + default="config.yaml", + help="Path to config.yaml (default: config.yaml in current directory)", + ) + parser.add_argument( + "--discover", "-d", + action="store_true", + help="Auto-discover Airflow Docker setup (overrides airflow section in config)", + ) + parser.add_argument( + "--dry-run", + action="store_true", + help="With --discover: show discovered config and exit without starting monitor", + ) + args = parser.parse_args() + + # Load config + try: + config = load_config(args.config) + except (FileNotFoundError, ValueError) as e: + if args.discover: + # Config file optional when using discovery — use defaults + from .config import AirflowConfig, MonitorConfig, ZabbixConfig, LoggingConfig + config = AppConfig( + airflow=AirflowConfig(), + monitor=MonitorConfig(), + zabbix=ZabbixConfig(), + logging=LoggingConfig(), + ) + else: + print(f"Configuration error: {e}", file=sys.stderr) + sys.exit(1) + + # Auto-discovery + if args.discover: + try: + discovered = discover_airflow() + config = _apply_discovery(config, discovered) + print(f"Discovered Airflow at {discovered['base_url']}") + print(f" Compose dir: {discovered['compose_dir']}") + print(f" Container: {discovered['container_name']}") + print(f" User: {discovered['username']}") + except DiscoveryError as e: + print(f"Discovery failed: {e}", file=sys.stderr) + sys.exit(1) + + if args.dry_run: + print("\n--- Effective Airflow config ---") + print(json.dumps(dataclasses.asdict(config.airflow), indent=2)) + print("\n--- Effective Zabbix config ---") + print(json.dumps(dataclasses.asdict(config.zabbix), indent=2)) + sys.exit(0) + + # Setup logging + setup_logging(config.logging) + logger = logging.getLogger(__name__) + logger.info("Airflow DAG Monitor starting") + + # Acquire lock + lock_fd = acquire_lock(config.monitor.lock_file) + logger.info("Lock acquired: %s (PID %d)", config.monitor.lock_file, os.getpid()) + + # Setup graceful shutdown + shutdown_event = threading.Event() + + def handle_signal(signum, frame): + sig_name = signal.Signals(signum).name + logger.info("Received %s, initiating shutdown", sig_name) + shutdown_event.set() + + signal.signal(signal.SIGTERM, handle_signal) + signal.signal(signal.SIGINT, handle_signal) + + # Run monitor + try: + monitor = Monitor(config, shutdown_event) + monitor.run() + except Exception: + logger.exception("Fatal error") + sys.exit(1) + finally: + lock_fd.close() + logger.info("Monitor stopped") + + +if __name__ == "__main__": + main() diff --git a/airflow_monitor/actions.py b/airflow_monitor/actions.py new file mode 100644 index 0000000..1a9647d --- /dev/null +++ b/airflow_monitor/actions.py @@ -0,0 +1,244 @@ +"""Action handlers: retry DAG runs and export data for Zabbix agent.""" + +import json +import logging +import os +import pathlib +import tempfile +import time +from datetime import datetime, timezone + +from .analyzer import DagIssue +from .client import AirflowClient +from .config import MonitorConfig, ZabbixConfig +from .state import StateManager + +logger = logging.getLogger(__name__) + + +class DataExporter: + """Exports monitoring data to files for Zabbix agent UserParameter. + + Files written: + - problems.json — JSON array of problematic DAG runs (or []) + - heartbeat — epoch timestamp (updated every cycle) + - status.json — overall monitor status (cycle count, DAG count, etc.) + + Zabbix agent reads these via UserParameter defined in + /etc/zabbix/zabbix_agentd.conf.d/airflow-monitor.conf + """ + + def __init__(self, config: ZabbixConfig): + self._config = config + self._data_dir = pathlib.Path(config.data_dir) + self._problems_path = self._data_dir / config.problems_file + self._heartbeat_path = self._data_dir / config.heartbeat_file + self._status_path = self._data_dir / config.status_file + + logger.debug( + "DataExporter initialized: enabled=%s, data_dir=%s, " + "problems=%s, heartbeat=%s, status=%s", + config.enabled, self._data_dir, + self._problems_path, self._heartbeat_path, self._status_path, + ) + + if config.enabled: + self._data_dir.mkdir(parents=True, exist_ok=True) + + def _atomic_write(self, path: pathlib.Path, content: str): + """Write content to file atomically (write tmp → rename).""" + try: + fd, tmp_path = tempfile.mkstemp( + dir=str(path.parent), suffix=".tmp", + ) + with os.fdopen(fd, "w", encoding="utf-8") as f: + f.write(content) + os.replace(tmp_path, str(path)) + logger.debug( + "File written: %s (%d bytes)", + path, len(content), + ) + except OSError as e: + logger.error("Failed to write %s: %s", path, e) + try: + os.unlink(tmp_path) + except OSError: + pass + + def export_problems(self, problems: list[dict]) -> bool: + """Write problem list as JSON file for Zabbix agent. + + Writes '[]' if no problems (Zabbix trigger auto-clears). + """ + if not self._config.enabled: + logger.debug("DataExporter disabled, skipping export") + return True + + payload = json.dumps(problems, indent=2, ensure_ascii=False) + logger.debug( + "Exporting problems: count=%d, size=%d bytes → %s", + len(problems), len(payload), self._problems_path, + ) + self._atomic_write(self._problems_path, payload) + return True + + def export_heartbeat(self): + """Write current epoch timestamp for Zabbix agent heartbeat check.""" + if not self._config.enabled: + return + + epoch = str(int(time.time())) + logger.debug("Exporting heartbeat: %s → %s", epoch, self._heartbeat_path) + self._atomic_write(self._heartbeat_path, epoch) + + def export_status(self, cycle_count: int, dag_count: int, + issue_count: int, cycle_time: float): + """Write overall monitor status for Zabbix agent.""" + if not self._config.enabled: + return + + status = { + "timestamp": datetime.now(timezone.utc).isoformat(), + "cycle_count": cycle_count, + "dag_count": dag_count, + "issue_count": issue_count, + "cycle_time_seconds": round(cycle_time, 2), + } + payload = json.dumps(status, indent=2, ensure_ascii=False) + logger.debug( + "Exporting status: cycle=#%d, dags=%d, issues=%d, time=%.1fs → %s", + cycle_count, dag_count, issue_count, cycle_time, self._status_path, + ) + self._atomic_write(self._status_path, payload) + + +class ActionHandler: + """Handles issue resolution: retry or escalate via data export.""" + + def __init__( + self, + client: AirflowClient, + state: StateManager, + exporter: DataExporter, + config: MonitorConfig, + ): + self._client = client + self._state = state + self._exporter = exporter + self._config = config + logger.debug( + "ActionHandler initialized: max_retries=%d, retry_wait=%ds", + config.max_retries, config.retry_wait, + ) + + def handle_issue(self, issue: DagIssue) -> str: + """Process a single issue. + + Returns: + "already_handled" - previously alerted, skip + "retried" - retry initiated, needs re-check + "needs_alert" - retry exhausted, alert needed + """ + dag_id = issue.dag_id + run_id = issue.dag_run_id + + # Ensure entry exists in state + self._state.ensure_entry(dag_id, run_id) + + retry_count = self._state.get_retry_count(dag_id, run_id) + is_alerted = self._state.is_alerted(dag_id, run_id) + + logger.debug( + "Handling issue: %s/%s type=%s state=%s duration=%.0fs " + "retry_count=%d/%d alerted=%s", + dag_id, run_id, issue.issue_type, issue.state, + issue.duration_seconds, retry_count, self._config.max_retries, + is_alerted, + ) + + # Already alerted? Skip. + if is_alerted: + logger.debug(" → already_handled: alert was sent previously") + return "already_handled" + + # Can we retry? + if retry_count < self._config.max_retries: + logger.debug( + " → attempting retry %d/%d via clear_dag_run", + retry_count + 1, self._config.max_retries, + ) + success = self._client.clear_dag_run(dag_id, run_id) + if success: + self._state.increment_retry(dag_id, run_id) + logger.info( + "Retry %d/%d initiated for %s/%s (%s, duration=%.0fs)", + retry_count + 1, self._config.max_retries, + dag_id, run_id, issue.issue_type, issue.duration_seconds, + ) + return "retried" + else: + logger.warning( + "Retry failed for %s/%s (API error), escalating to alert", + dag_id, run_id, + ) + return "needs_alert" + + # Retries exhausted + logger.debug( + " → needs_alert: retries exhausted (%d/%d)", + retry_count, self._config.max_retries, + ) + return "needs_alert" + + def collect_and_alert(self, issues: list[DagIssue]): + """Export alerts for issues that need alerting. + + Filters to only unalerted issues, builds JSON payload, + writes to file for Zabbix agent, and marks as alerted in state. + """ + to_alert = [] + for issue in issues: + if not self._state.is_alerted(issue.dag_id, issue.dag_run_id): + entry = self._state.ensure_entry(issue.dag_id, issue.dag_run_id) + alert_data = { + "dag_id": issue.dag_id, + "dag_run_id": issue.dag_run_id, + "issue_type": issue.issue_type, + "status": issue.state, + "duration_seconds": issue.duration_seconds, + "error_info": issue.error_info, + "retry_count": entry["retry_count"], + } + to_alert.append(alert_data) + logger.debug( + "Issue queued for alert: %s/%s type=%s retries=%d", + issue.dag_id, issue.dag_run_id, + issue.issue_type, entry["retry_count"], + ) + + if not to_alert: + # Write empty list to clear Zabbix trigger + self._exporter.export_problems([]) + logger.info("No issues to alert, exported empty problem list") + return + + logger.warning( + "Exporting %d problematic DAG runs for Zabbix: %s", + len(to_alert), + ", ".join( + f"{p['dag_id']}/{p['dag_run_id']}({p['issue_type']})" + for p in to_alert + ), + ) + + # Log full alert payload at debug level + logger.debug("Alert payload:\n%s", json.dumps(to_alert, indent=2, ensure_ascii=False)) + + self._exporter.export_problems(to_alert) + + for issue in issues: + if not self._state.is_alerted(issue.dag_id, issue.dag_run_id): + self._state.mark_alerted(issue.dag_id, issue.dag_run_id) + logger.debug("Marked as alerted: %s/%s", issue.dag_id, issue.dag_run_id) + + logger.info("Exported %d issues to problems file for Zabbix agent", len(to_alert)) diff --git a/airflow_monitor/analyzer.py b/airflow_monitor/analyzer.py new file mode 100644 index 0000000..276b9d5 --- /dev/null +++ b/airflow_monitor/analyzer.py @@ -0,0 +1,180 @@ +"""DAG run analyzer - classifies runs and detects issues.""" + +import dataclasses +import logging +from datetime import datetime, timezone +from typing import Callable + +logger = logging.getLogger(__name__) + + +@dataclasses.dataclass +class DagIssue: + """Represents a detected problem with a DAG run.""" + + dag_id: str + dag_run_id: str + issue_type: str # "failed" or "long_running" + state: str # airflow state string + start_time: str # ISO 8601 UTC + duration_seconds: float + error_info: str = "" + + def to_dict(self) -> dict: + return dataclasses.asdict(self) + + +class DagAnalyzer: + """Analyzes DAG runs and produces a list of issues.""" + + def __init__(self, long_running_threshold: int): + self._threshold = long_running_threshold + logger.debug( + "DagAnalyzer initialized: long_running_threshold=%ds (%.1f min)", + long_running_threshold, long_running_threshold / 60, + ) + + def analyze_dag_runs( + self, + dag_id: str, + runs: list[dict], + task_instances_fn: Callable[[str, str], list[dict]], + ) -> list[DagIssue]: + """Analyze DAG runs for a single DAG. + + Args: + dag_id: DAG identifier. + runs: List of DAG run dicts from Airflow API. + task_instances_fn: Callable(dag_id, dag_run_id) -> list of task instances. + Called lazily only for failed runs to extract error details. + + Returns: + List of DagIssue objects for problematic runs. + """ + issues = [] + now = datetime.now(timezone.utc) + logger.debug("Analyzing %d runs for DAG '%s' (now=%s)", len(runs), dag_id, now.isoformat()) + + for run in runs: + run_id = run.get("dag_run_id", "") + state = run.get("state", "") + start_date_str = run.get("start_date") or run.get("execution_date", "") + + if not start_date_str: + logger.warning("DAG run %s/%s has no start_date, skipping", dag_id, run_id) + continue + + start_date = self._parse_datetime(start_date_str) + if start_date is None: + logger.warning( + "Cannot parse start_date '%s' for %s/%s", + start_date_str, dag_id, run_id, + ) + continue + + duration = (now - start_date).total_seconds() + + logger.debug( + " Run %s/%s: state=%s, start=%s, duration=%.0fs (%.1f min), threshold=%ds", + dag_id, run_id, state, start_date_str, + duration, duration / 60, self._threshold, + ) + + if state == "failed": + error_info = self._extract_error(dag_id, run_id, task_instances_fn) + issue = DagIssue( + dag_id=dag_id, + dag_run_id=run_id, + issue_type="failed", + state=state, + start_time=start_date.isoformat(), + duration_seconds=round(duration, 1), + error_info=error_info, + ) + issues.append(issue) + logger.debug(" → ISSUE DETECTED: %s (error: %s)", issue.issue_type, error_info or "n/a") + + elif state == "running" and duration > self._threshold: + issue = DagIssue( + dag_id=dag_id, + dag_run_id=run_id, + issue_type="long_running", + state=state, + start_time=start_date.isoformat(), + duration_seconds=round(duration, 1), + ) + issues.append(issue) + logger.debug( + " → ISSUE DETECTED: long_running (%.0fs > %ds threshold)", + duration, self._threshold, + ) + + elif state == "running": + logger.debug(" → OK: running within threshold (%.0fs <= %ds)", duration, self._threshold) + + else: + logger.debug(" → SKIP: state=%s (not actionable)", state) + + logger.debug("Analysis complete for DAG '%s': %d issues found", dag_id, len(issues)) + return issues + + def _extract_error( + self, + dag_id: str, + dag_run_id: str, + task_instances_fn: Callable[[str, str], list[dict]], + ) -> str: + """Extract error info from failed task instances.""" + logger.debug("Extracting error info for %s/%s", dag_id, dag_run_id) + try: + tasks = task_instances_fn(dag_id, dag_run_id) + except Exception as e: + logger.warning("Failed to fetch task instances for %s/%s: %s", dag_id, dag_run_id, e) + return "" + + failed_tasks = [t for t in tasks if t.get("state") == "failed"] + logger.debug( + "Task instances for %s/%s: total=%d, failed=%d", + dag_id, dag_run_id, len(tasks), len(failed_tasks), + ) + + if not failed_tasks: + return "" + + # Return info about the first failed task + task = failed_tasks[0] + task_id = task.get("task_id", "unknown") + # Try to get the error message from different possible fields + error = ( + task.get("rendered_fields", {}).get("error", "") + or task.get("note", "") + or "" + ) + # Truncate long error messages + if len(error) > 500: + error = error[:500] + "..." + + result = f"Task '{task_id}' failed" + (f": {error}" if error else "") + logger.debug("Error extracted for %s/%s: %s", dag_id, dag_run_id, result) + return result + + @staticmethod + def _parse_datetime(dt_string: str) -> datetime | None: + """Parse ISO 8601 datetime string from Airflow API. + + Handles both 'Z' suffix and '+00:00' timezone offset. + """ + if not dt_string: + return None + + # Normalize 'Z' to '+00:00' for fromisoformat + cleaned = dt_string.replace("Z", "+00:00") + + try: + dt = datetime.fromisoformat(cleaned) + # Ensure timezone-aware (UTC) + if dt.tzinfo is None: + dt = dt.replace(tzinfo=timezone.utc) + return dt + except ValueError: + return None diff --git a/airflow_monitor/client.py b/airflow_monitor/client.py new file mode 100644 index 0000000..b865a01 --- /dev/null +++ b/airflow_monitor/client.py @@ -0,0 +1,291 @@ +"""Airflow REST API client with v1/experimental auto-detection.""" + +import json +import logging +import time +from urllib.parse import quote + +import requests + +from .config import AirflowConfig + +logger = logging.getLogger(__name__) + + +class AirflowAPIError(Exception): + """Raised when Airflow API returns an unexpected error.""" + + +class AirflowClient: + """Client for Apache Airflow REST API. + + Supports both stable API v1 (/api/v1/) and experimental API + (/api/experimental/). Auto-detects which is available at startup. + """ + + def __init__(self, config: AirflowConfig): + self._config = config + self._session = requests.Session() + self._session.auth = (config.username, config.password) + self._session.verify = config.verify_ssl + self._session.headers.update({"Content-Type": "application/json"}) + self._base_url = config.base_url.rstrip("/") + self._api_prefix: str | None = None + self._api_version: str | None = None + self._last_request_time: float = 0 + logger.debug( + "AirflowClient initialized: url=%s, user=%s, timeout=%d, ssl=%s", + self._base_url, config.username, config.timeout, config.verify_ssl, + ) + + def detect_api_version(self): + """Detect available API version. + + Tries v1 stable API first, falls back to experimental. + Raises AirflowAPIError if neither is available. + """ + if self._config.api_version != "auto": + if self._config.api_version == "v1": + self._api_prefix = f"{self._base_url}/api/v1" + self._api_version = "v1" + else: + self._api_prefix = f"{self._base_url}/api/experimental" + self._api_version = "experimental" + logger.info("Using configured API version: %s", self._api_version) + return + + # Try v1 stable API + url = f"{self._base_url}/api/v1/health" + logger.debug("Probing API v1: GET %s", url) + try: + resp = self._session.get(url, timeout=self._config.timeout) + logger.debug("Probe v1 response: status=%d, body=%s", resp.status_code, resp.text[:200]) + if resp.status_code == 200: + self._api_prefix = f"{self._base_url}/api/v1" + self._api_version = "v1" + logger.info("Detected Airflow API v1 (stable)") + return + except requests.RequestException as e: + logger.debug("Probe v1 failed: %s", e) + + # Try experimental API + url = f"{self._base_url}/api/experimental/test" + logger.debug("Probing experimental API: GET %s", url) + try: + resp = self._session.get(url, timeout=self._config.timeout) + logger.debug("Probe experimental response: status=%d", resp.status_code) + if resp.status_code == 200: + self._api_prefix = f"{self._base_url}/api/experimental" + self._api_version = "experimental" + logger.info("Detected Airflow experimental API") + return + except requests.RequestException as e: + logger.debug("Probe experimental failed: %s", e) + + raise AirflowAPIError( + f"Cannot connect to Airflow API at {self._base_url}. " + "Tried /api/v1/health and /api/experimental/test" + ) + + def _rate_limit(self): + """Enforce minimum delay between API requests.""" + elapsed = time.time() - self._last_request_time + if elapsed < self._config.request_delay: + wait = self._config.request_delay - elapsed + logger.debug("Rate limiting: waiting %.2fs", wait) + time.sleep(wait) + + def _request(self, method: str, path: str, **kwargs) -> requests.Response: + """Make an API request with error handling and rate limiting.""" + if not self._api_prefix: + raise AirflowAPIError("API version not detected. Call detect_api_version() first") + + self._rate_limit() + url = f"{self._api_prefix}{path}" + + # Log request details + params = kwargs.get("params") + body = kwargs.get("json") + logger.debug( + "API request: %s %s params=%s body=%s", + method, url, params, json.dumps(body) if body else None, + ) + + t0 = time.monotonic() + try: + resp = self._session.request( + method, url, timeout=self._config.timeout, **kwargs + ) + self._last_request_time = time.time() + elapsed_ms = (time.monotonic() - t0) * 1000 + + # Log response details + logger.debug( + "API response: %s %s → %d (%dms) body=%s", + method, path, resp.status_code, elapsed_ms, + resp.text[:500] if resp.text else "", + ) + + resp.raise_for_status() + return resp + except requests.ConnectionError as e: + raise AirflowAPIError(f"Connection error to {url}: {e}") from e + except requests.Timeout as e: + elapsed_ms = (time.monotonic() - t0) * 1000 + raise AirflowAPIError(f"Timeout after {elapsed_ms:.0f}ms calling {url}: {e}") from e + except requests.HTTPError as e: + status = resp.status_code + body = resp.text[:500] + raise AirflowAPIError(f"HTTP {status} from {url}: {body}") from e + + def get_enabled_dags(self) -> list[dict]: + """Get list of enabled (active and not paused) DAGs. + + Returns list of dicts with at least 'dag_id' key. + """ + logger.debug("Fetching enabled DAGs (api_version=%s)", self._api_version) + if self._api_version == "v1": + return self._get_enabled_dags_v1() + return self._get_enabled_dags_experimental() + + def _get_enabled_dags_v1(self) -> list[dict]: + """Fetch DAGs via stable v1 API with pagination.""" + dags = [] + offset = 0 + limit = 100 + + while True: + logger.debug("Fetching DAGs page: offset=%d, limit=%d", offset, limit) + resp = self._request( + "GET", "/dags", + params={"limit": limit, "offset": offset, "only_active": True}, + ) + data = resp.json() + page_dags = data.get("dags", []) + total = data.get("total_entries", 0) + + active_count = 0 + paused_count = 0 + for dag in page_dags: + if not dag.get("is_paused", True): + dags.append(dag) + active_count += 1 + else: + paused_count += 1 + + logger.debug( + "DAGs page result: total_entries=%d, page_size=%d, " + "active=%d, paused=%d", + total, len(page_dags), active_count, paused_count, + ) + + offset += limit + if offset >= total or not page_dags: + break + + logger.debug( + "All enabled DAGs: %s", + [d["dag_id"] for d in dags], + ) + return dags + + def _get_enabled_dags_experimental(self) -> list[dict]: + """Fetch DAGs via experimental API (limited info).""" + resp = self._request("GET", "/dags") + return resp.json() if resp.status_code == 200 else [] + + def get_dag_runs(self, dag_id: str, states: list[str] | None = None) -> list[dict]: + """Get recent DAG runs, optionally filtered by state.""" + encoded_dag_id = quote(dag_id, safe="") + logger.debug("Fetching DAG runs: dag_id=%s, states=%s", dag_id, states) + + if self._api_version == "v1": + if states: + resp = self._request( + "GET", + f"/dags/{encoded_dag_id}/dagRuns", + params=[("limit", 25), ("order_by", "-start_date")] + + [("state", s) for s in states], + ) + else: + resp = self._request( + "GET", + f"/dags/{encoded_dag_id}/dagRuns", + params={"order_by": "-start_date", "limit": 25}, + ) + runs = resp.json().get("dag_runs", []) + else: + resp = self._request("GET", f"/dags/{dag_id}/dag_runs") + runs = resp.json() if isinstance(resp.json(), list) else [] + if states: + runs = [r for r in runs if r.get("state") in states] + + logger.debug( + "DAG runs for %s: count=%d, runs=[%s]", + dag_id, len(runs), + ", ".join( + f"{r.get('dag_run_id', '?')}({r.get('state', '?')})" + for r in runs + ), + ) + return runs + + def get_task_instances(self, dag_id: str, dag_run_id: str) -> list[dict]: + """Get task instances for a specific DAG run.""" + encoded_dag_id = quote(dag_id, safe="") + encoded_run_id = quote(dag_run_id, safe="") + + logger.debug("Fetching task instances: %s/%s", dag_id, dag_run_id) + + if self._api_version == "v1": + resp = self._request( + "GET", + f"/dags/{encoded_dag_id}/dagRuns/{encoded_run_id}/taskInstances", + ) + tasks = resp.json().get("task_instances", []) + logger.debug( + "Task instances for %s/%s: count=%d, states=[%s]", + dag_id, dag_run_id, len(tasks), + ", ".join( + f"{t.get('task_id', '?')}({t.get('state', '?')})" + for t in tasks + ), + ) + return tasks + else: + logger.warning( + "Task instances not supported in experimental API for %s/%s", + dag_id, dag_run_id, + ) + return [] + + def clear_dag_run(self, dag_id: str, dag_run_id: str) -> bool: + """Clear (retry) a DAG run by resetting failed task instances. + + Returns True on success, False on failure. + """ + encoded_dag_id = quote(dag_id, safe="") + encoded_run_id = quote(dag_run_id, safe="") + + if self._api_version != "v1": + logger.warning( + "Clearing DAG runs not supported in experimental API for %s/%s", + dag_id, dag_run_id, + ) + return False + + logger.debug( + "Clearing DAG run: %s/%s (only_failed=True)", + dag_id, dag_run_id, + ) + try: + self._request( + "POST", + f"/dags/{encoded_dag_id}/dagRuns/{encoded_run_id}/clear", + json={"dry_run": False, "only_failed": True}, + ) + logger.info("Cleared DAG run %s/%s for retry", dag_id, dag_run_id) + return True + except AirflowAPIError as e: + logger.error("Failed to clear DAG run %s/%s: %s", dag_id, dag_run_id, e) + return False diff --git a/airflow_monitor/config.py b/airflow_monitor/config.py new file mode 100644 index 0000000..3aabe32 --- /dev/null +++ b/airflow_monitor/config.py @@ -0,0 +1,100 @@ +"""Configuration loader for Airflow DAG Monitor.""" + +import dataclasses +import pathlib + +import yaml + + +@dataclasses.dataclass +class AirflowConfig: + """Airflow API connection settings.""" + + base_url: str = "http://localhost:8080" + username: str = "airflow" + password: str = "airflow" + api_version: str = "auto" # "auto", "v1", or "experimental" + timeout: int = 30 + verify_ssl: bool = True + request_delay: float = 0.5 # delay between API calls (seconds) + + +@dataclasses.dataclass +class MonitorConfig: + """Monitoring behavior settings.""" + + cycle_interval: int = 300 # seconds between full cycles + long_running_threshold: int = 1800 # 30 minutes + retry_wait: int = 120 # wait after retry before re-check + max_retries: int = 1 + state_file: str = "/var/lib/airflow-monitor/state.json" + lock_file: str = "/var/run/airflow-monitor.lock" + state_max_age: int = 86400 # 24 hours + + +@dataclasses.dataclass +class ZabbixConfig: + """Zabbix agent integration settings. + + Monitor writes data to files, Zabbix agent reads via UserParameter. + """ + + enabled: bool = True + data_dir: str = "/var/lib/airflow-monitor" + problems_file: str = "problems.json" + heartbeat_file: str = "heartbeat" + status_file: str = "status.json" + + +@dataclasses.dataclass +class LoggingConfig: + """Logging settings.""" + + level: str = "INFO" + file: str = "/var/log/airflow-monitor/monitor.log" + max_bytes: int = 10_485_760 # 10 MB + backup_count: int = 5 + + +@dataclasses.dataclass +class AppConfig: + """Top-level application configuration.""" + + airflow: AirflowConfig + monitor: MonitorConfig + zabbix: ZabbixConfig + logging: LoggingConfig + + +def _build_dataclass(cls, data: dict): + """Build a dataclass instance from a dict, ignoring unknown keys.""" + if data is None: + return cls() + fields = {f.name for f in dataclasses.fields(cls)} + filtered = {k: v for k, v in data.items() if k in fields} + return cls(**filtered) + + +def load_config(path: str) -> AppConfig: + """Load configuration from a YAML file. + + Missing sections fall back to defaults. + Raises FileNotFoundError if the file does not exist. + Raises ValueError on invalid YAML. + """ + config_path = pathlib.Path(path) + if not config_path.exists(): + raise FileNotFoundError(f"Config file not found: {path}") + + with open(config_path, "r", encoding="utf-8") as f: + raw = yaml.safe_load(f) + + if not isinstance(raw, dict): + raise ValueError(f"Config file must be a YAML mapping, got {type(raw).__name__}") + + return AppConfig( + airflow=_build_dataclass(AirflowConfig, raw.get("airflow")), + monitor=_build_dataclass(MonitorConfig, raw.get("monitor")), + zabbix=_build_dataclass(ZabbixConfig, raw.get("zabbix")), + logging=_build_dataclass(LoggingConfig, raw.get("logging")), + ) diff --git a/airflow_monitor/discovery.py b/airflow_monitor/discovery.py new file mode 100644 index 0000000..3dc59f1 --- /dev/null +++ b/airflow_monitor/discovery.py @@ -0,0 +1,333 @@ +"""Auto-discovery of Airflow Docker infrastructure. + +Finds running Airflow containers, locates docker-compose project directory, +reads .env and docker-compose.yml to extract connection parameters. +""" + +import json +import logging +import os +import re +import subprocess +from pathlib import Path + +import yaml + +logger = logging.getLogger(__name__) + + +class DiscoveryError(Exception): + """Raised when Airflow Docker infrastructure cannot be found.""" + + +def discover_airflow() -> dict: + """Auto-discover Airflow Docker setup and return connection parameters. + + Discovery steps: + 1. Find airflow-webserver container via `docker ps` + 2. Extract compose project directory from container labels + 3. Read .env from compose directory + 4. Read docker-compose.yml for port mappings and env vars + 5. Build connection config dict + + Returns: + dict with keys: base_url, username, password, api_version, + compose_dir, container_name + """ + logger.info("Starting Airflow Docker auto-discovery") + + # Step 1: Find webserver container + container = _find_webserver_container() + container_name = container["Names"] + logger.info("Found Airflow webserver container: %s", container_name) + + # Step 2: Get compose project directory + compose_dir = _get_compose_dir(container) + logger.info("Compose project directory: %s", compose_dir) + + # Step 3: Read .env file + env_vars = _read_env_file(compose_dir) + + # Step 4: Read docker-compose.yml + compose_config = _read_compose_file(compose_dir) + + # Step 5: Extract connection parameters + host_port = _extract_webserver_port(container) + credentials = _extract_credentials(env_vars, compose_config) + + result = { + "base_url": f"http://localhost:{host_port}", + "username": credentials["username"], + "password": credentials["password"], + "api_version": "v1", + "compose_dir": str(compose_dir), + "container_name": container_name, + } + + logger.info( + "Discovery complete: url=%s, user=%s, compose_dir=%s", + result["base_url"], result["username"], result["compose_dir"], + ) + return result + + +def _run_cmd(cmd: list[str], timeout: int = 15) -> str: + """Run a shell command and return stdout. Raises DiscoveryError on failure.""" + logger.debug("Running command: %s", " ".join(cmd)) + try: + result = subprocess.run( + cmd, capture_output=True, text=True, timeout=timeout, + ) + logger.debug( + "Command result: rc=%d, stdout=%d bytes, stderr=%s", + result.returncode, len(result.stdout), + result.stderr.strip()[:200] if result.stderr.strip() else "", + ) + if result.returncode != 0: + raise DiscoveryError( + f"Command failed: {' '.join(cmd)}\n" + f"stderr: {result.stderr.strip()}" + ) + return result.stdout.strip() + except FileNotFoundError: + raise DiscoveryError( + f"Command not found: {cmd[0]}. Is Docker installed?" + ) + except subprocess.TimeoutExpired: + raise DiscoveryError(f"Command timed out: {' '.join(cmd)}") + + +def _find_webserver_container() -> dict: + """Find running Airflow webserver container. + + Searches for containers with 'airflow' in the image and 'webserver' + in the name or command. + """ + output = _run_cmd([ + "docker", "ps", "--format", "{{json .}}", + "--filter", "status=running", + ]) + + if not output: + raise DiscoveryError("No running Docker containers found") + + candidates = [] + for line in output.splitlines(): + try: + container = json.loads(line) + except json.JSONDecodeError: + continue + + image = container.get("Image", "").lower() + names = container.get("Names", "").lower() + command = container.get("Command", "").lower() + status = container.get("Status", "") + + logger.debug( + " Container: name=%s, image=%s, command=%s, status=%s", + names, image, command[:60], status, + ) + + # Match airflow webserver by multiple signals + is_airflow = "airflow" in image or "airflow" in names + is_webserver = ( + "webserver" in names + or "webserver" in command + or ("airflow" in command and "webserver" in command) + ) + + if is_airflow and is_webserver: + logger.debug(" → MATCH: airflow webserver candidate") + candidates.append(container) + + if not candidates: + raise DiscoveryError( + "No running Airflow webserver container found. " + "Checked: image contains 'airflow' AND name/command contains 'webserver'" + ) + + if len(candidates) > 1: + logger.warning( + "Found %d webserver containers, using first: %s", + len(candidates), candidates[0]["Names"], + ) + + return candidates[0] + + +def _get_compose_dir(container: dict) -> Path: + """Get docker-compose project directory from container labels.""" + container_name = container["Names"] + + # Inspect container for compose labels + output = _run_cmd([ + "docker", "inspect", + "--format", '{{index .Config.Labels "com.docker.compose.project.working_dir"}}', + container_name, + ]) + + if output and output != "": + compose_dir = Path(output) + if compose_dir.exists(): + return compose_dir + + # Fallback: try to find compose file via container's bind mounts + inspect_json = _run_cmd(["docker", "inspect", container_name]) + try: + data = json.loads(inspect_json) + if data: + mounts = data[0].get("Mounts", []) + for mount in mounts: + source = Path(mount.get("Source", "")) + # Look for parent directory containing docker-compose.yml + for candidate in [source.parent, source.parent.parent, source]: + for compose_name in ["docker-compose.yml", "docker-compose.yaml", "compose.yml", "compose.yaml"]: + if (candidate / compose_name).exists(): + return candidate + except (json.JSONDecodeError, IndexError, KeyError): + pass + + raise DiscoveryError( + f"Cannot determine compose directory for container {container_name}. " + "Label 'com.docker.compose.project.working_dir' not found." + ) + + +def _extract_webserver_port(container: dict) -> int: + """Extract host port mapped to webserver's 8080.""" + container_name = container["Names"] + + # docker port gives us the exact mapping + try: + output = _run_cmd(["docker", "port", container_name, "8080"]) + # Output: "0.0.0.0:80" or "0.0.0.0:80\n:::80" + for line in output.splitlines(): + match = re.search(r":(\d+)$", line.strip()) + if match: + port = int(match.group(1)) + logger.info("Webserver port: %d (from docker port)", port) + return port + except DiscoveryError: + pass + + # Fallback: parse Ports field from docker ps + ports_str = container.get("Ports", "") + # Format: "0.0.0.0:80->8080/tcp" + match = re.search(r"(\d+)->8080", ports_str) + if match: + port = int(match.group(1)) + logger.info("Webserver port: %d (from docker ps)", port) + return port + + logger.warning("Cannot determine webserver port, defaulting to 8080") + return 8080 + + +def _read_env_file(compose_dir: Path) -> dict: + """Read .env file from compose directory.""" + env_file = compose_dir / ".env" + env_vars = {} + + if not env_file.exists(): + logger.warning("No .env file found in %s", compose_dir) + return env_vars + + with open(env_file, "r", encoding="utf-8") as f: + for line in f: + line = line.strip() + if not line or line.startswith("#"): + continue + if "=" in line: + key, _, value = line.partition("=") + key = key.strip() + value = value.strip().strip("'\"") + env_vars[key] = value + + logger.info("Read %d variables from .env", len(env_vars)) + # Log variable names (not values) for debugging + logger.debug( + ".env variables: %s", + ", ".join(sorted(env_vars.keys())), + ) + return env_vars + + +def _read_compose_file(compose_dir: Path) -> dict: + """Read docker-compose.yml from compose directory.""" + for name in ["docker-compose.yml", "docker-compose.yaml", "compose.yml", "compose.yaml"]: + path = compose_dir / name + if path.exists(): + with open(path, "r", encoding="utf-8") as f: + data = yaml.safe_load(f) + logger.info("Read compose file: %s", path) + return data if isinstance(data, dict) else {} + + logger.warning("No docker-compose.yml found in %s", compose_dir) + return {} + + +def _extract_credentials(env_vars: dict, compose_config: dict) -> dict: + """Extract Airflow webserver credentials from .env and compose config. + + Checks multiple variable names in priority order since different + setups use different naming conventions. + """ + # Username: check .env, then compose env, then default + username = ( + env_vars.get("_AIRFLOW_WWW_USER_USERNAME") + or env_vars.get("AIRFLOW_WWW_USER_USERNAME") + or _get_compose_env(compose_config, "_AIRFLOW_WWW_USER_USERNAME") + or "airflow" + ) + + # Password: check .env, then compose env, then default + password = ( + env_vars.get("_AIRFLOW_WWW_USER_PASSWORD") + or env_vars.get("AIRFLOW_WWW_USER_PASSWORD") + or _get_compose_env(compose_config, "_AIRFLOW_WWW_USER_PASSWORD") + or "airflow" + ) + + # Resolve ${VAR:-default} references in compose values + password = _resolve_env_ref(password, env_vars) + username = _resolve_env_ref(username, env_vars) + + logger.info("Credentials: user=%s, password=%s", username, "***") + return {"username": username, "password": password} + + +def _get_compose_env(compose_config: dict, var_name: str) -> str | None: + """Extract environment variable value from docker-compose services. + + Looks in airflow-init and airflow-webserver services. + """ + services = compose_config.get("services", {}) + + for service_name in ["airflow-init", "airflow-webserver"]: + service = services.get(service_name, {}) + env = service.get("environment", {}) + + if isinstance(env, dict): + value = env.get(var_name) + if value is not None: + return str(value) + elif isinstance(env, list): + for item in env: + if isinstance(item, str) and item.startswith(f"{var_name}="): + return item.split("=", 1)[1] + + return None + + +def _resolve_env_ref(value: str, env_vars: dict) -> str: + """Resolve ${VAR:-default} or ${VAR} references in a string.""" + if not isinstance(value, str): + return str(value) if value is not None else "" + + # Pattern: ${VAR_NAME:-default_value} or ${VAR_NAME} + def replacer(match): + var_name = match.group(1) + default = match.group(3) if match.group(3) is not None else "" + return env_vars.get(var_name, default) + + return re.sub(r"\$\{([^:}]+)(?::-(.*?))?\}", replacer, value) diff --git a/airflow_monitor/monitor.py b/airflow_monitor/monitor.py new file mode 100644 index 0000000..659cd82 --- /dev/null +++ b/airflow_monitor/monitor.py @@ -0,0 +1,250 @@ +"""Main monitoring loop - orchestrates the full monitoring cycle.""" + +import logging +import threading +import time + +from .actions import ActionHandler, DataExporter +from .analyzer import DagAnalyzer +from .client import AirflowAPIError, AirflowClient +from .config import AppConfig +from .state import StateManager + +logger = logging.getLogger(__name__) + + +class Monitor: + """Orchestrates the Airflow DAG monitoring cycle. + + Cycle flow: + 1. Load state + 2. Fetch enabled DAGs + 3. Get active runs for each DAG + 4. Analyze for issues (failed, long-running) + 5. Retry where possible + 6. Wait and re-check retried runs + 7. Alert via Zabbix for unresolved issues + 8. Send heartbeat, cleanup, save state + """ + + def __init__(self, config: AppConfig, shutdown_event: threading.Event): + self._config = config + self._shutdown = shutdown_event + self._client = AirflowClient(config.airflow) + self._state = StateManager(config.monitor.state_file, config.monitor.state_max_age) + self._analyzer = DagAnalyzer(config.monitor.long_running_threshold) + self._exporter = DataExporter(config.zabbix) + self._actions = ActionHandler( + self._client, self._state, self._exporter, config.monitor, + ) + self._cycle_count = 0 + + def run(self): + """Main loop. Runs until shutdown_event is set.""" + logger.info( + "Starting Airflow DAG monitor (cycle=%ds, threshold=%ds, retries=%d, retry_wait=%ds)", + self._config.monitor.cycle_interval, + self._config.monitor.long_running_threshold, + self._config.monitor.max_retries, + self._config.monitor.retry_wait, + ) + + try: + self._client.detect_api_version() + except AirflowAPIError as e: + logger.critical("Cannot connect to Airflow API: %s", e) + return + + while not self._shutdown.is_set(): + try: + self._cycle() + except Exception: + logger.exception("Unhandled error in monitoring cycle") + + # Interruptible sleep between cycles + logger.debug( + "Sleeping %ds until next cycle", + self._config.monitor.cycle_interval, + ) + self._shutdown.wait(timeout=self._config.monitor.cycle_interval) + + logger.info("Shutting down gracefully") + self._state.save() + + def _cycle(self): + """Execute one complete monitoring cycle.""" + self._cycle_count += 1 + cycle_start = time.monotonic() + logger.info("=== Monitoring cycle #%d started ===", self._cycle_count) + self._state.load() + + # --- Phase 1: Collect issues --- + phase_start = time.monotonic() + all_issues = [] + try: + dags = self._client.get_enabled_dags() + except AirflowAPIError as e: + logger.error("Failed to fetch DAG list: %s", e) + self._exporter.export_heartbeat() + return + + logger.info( + "Phase 1/5 [Collect]: found %d enabled DAGs (%.1fs)", + len(dags), time.monotonic() - phase_start, + ) + + for i, dag in enumerate(dags, 1): + if self._shutdown.is_set(): + return + + dag_id = dag.get("dag_id", "") + if not dag_id: + continue + + logger.debug("Processing DAG %d/%d: %s", i, len(dags), dag_id) + + try: + runs = self._client.get_dag_runs(dag_id, states=["running", "failed"]) + except AirflowAPIError as e: + logger.warning("Failed to fetch runs for DAG '%s': %s", dag_id, e) + continue + + if not runs: + logger.debug(" No active/failed runs for %s", dag_id) + continue + + logger.debug(" Found %d active/failed runs for %s", len(runs), dag_id) + + issues = self._analyzer.analyze_dag_runs( + dag_id, runs, + task_instances_fn=self._client.get_task_instances, + ) + all_issues.extend(issues) + + if all_issues: + logger.warning( + "Phase 1 result: %d issues found: %s", + len(all_issues), + ", ".join( + f"{i.dag_id}/{i.dag_run_id}({i.issue_type}, {i.duration_seconds:.0f}s)" + for i in all_issues + ), + ) + else: + logger.info("Phase 1 result: no issues found across %d DAGs", len(dags)) + + # --- Phase 2: Handle issues (retry or mark for alert) --- + phase_start = time.monotonic() + needs_recheck = [] + for issue in all_issues: + action = self._actions.handle_issue(issue) + logger.info( + "Phase 2/5 [Handle]: DAG %s run %s → %s (type=%s, duration=%.0fs)", + issue.dag_id, issue.dag_run_id, action, + issue.issue_type, issue.duration_seconds, + ) + if action == "retried": + needs_recheck.append(issue) + + logger.info( + "Phase 2/5 [Handle]: processed %d issues, %d retried, (%.1fs)", + len(all_issues), len(needs_recheck), + time.monotonic() - phase_start, + ) + + # --- Phase 3: Wait and re-check retried runs --- + if needs_recheck and not self._shutdown.is_set(): + wait_time = self._config.monitor.retry_wait + logger.info( + "Phase 3/5 [Wait]: waiting %ds to re-check %d retried runs", + wait_time, len(needs_recheck), + ) + self._shutdown.wait(timeout=wait_time) + + if not self._shutdown.is_set(): + phase_start = time.monotonic() + still_failing = self._recheck_retried(needs_recheck) + # Replace all_issues with only the still-failing ones for alerting + # Keep non-retried issues that need alerting + non_retried_issues = [ + i for i in all_issues if i not in needs_recheck + ] + all_issues = non_retried_issues + still_failing + logger.info( + "Phase 3/5 [Recheck]: %d/%d still failing after retry (%.1fs)", + len(still_failing), len(needs_recheck), + time.monotonic() - phase_start, + ) + else: + logger.debug("Phase 3/5 [Wait]: skipped (no retried runs)") + + # --- Phase 4: Alert via Zabbix --- + phase_start = time.monotonic() + self._actions.collect_and_alert(all_issues) + logger.info( + "Phase 4/5 [Alert]: alert phase complete (%.1fs)", + time.monotonic() - phase_start, + ) + + # --- Phase 5: Housekeeping --- + phase_start = time.monotonic() + self._exporter.export_heartbeat() + self._state.purge_old_entries() + self._state.save() + + cycle_elapsed = time.monotonic() - cycle_start + self._exporter.export_status( + self._cycle_count, len(dags), len(all_issues), cycle_elapsed, + ) + logger.info( + "=== Monitoring cycle #%d complete: %d DAGs checked, " + "%d issues, cycle_time=%.1fs ===", + self._cycle_count, len(dags), len(all_issues), cycle_elapsed, + ) + + def _recheck_retried(self, retried_issues: list) -> list: + """Re-check DAG runs that were retried. + + Returns list of DagIssue objects that are still failing. + """ + still_failing = [] + + for issue in retried_issues: + if self._shutdown.is_set(): + break + + logger.debug("Re-checking retried run: %s/%s", issue.dag_id, issue.dag_run_id) + + try: + runs = self._client.get_dag_runs( + issue.dag_id, states=["running", "failed"], + ) + except AirflowAPIError as e: + logger.warning( + "Failed to re-check DAG '%s': %s. Treating as still failing.", + issue.dag_id, e, + ) + still_failing.append(issue) + continue + + recheck_issues = self._analyzer.analyze_dag_runs( + issue.dag_id, runs, + task_instances_fn=self._client.get_task_instances, + ) + + # Check if the same dag_run_id still has problems + for ri in recheck_issues: + if ri.dag_run_id == issue.dag_run_id: + logger.warning( + "DAG %s/%s still failing after retry: type=%s, duration=%.0fs", + ri.dag_id, ri.dag_run_id, ri.issue_type, ri.duration_seconds, + ) + still_failing.append(ri) + break + else: + logger.info( + "DAG %s/%s recovered after retry", + issue.dag_id, issue.dag_run_id, + ) + + return still_failing diff --git a/airflow_monitor/state.py b/airflow_monitor/state.py new file mode 100644 index 0000000..55acc46 --- /dev/null +++ b/airflow_monitor/state.py @@ -0,0 +1,164 @@ +"""Persistent state manager for tracking DAG run retries and alerts.""" + +import json +import logging +import os +import pathlib +import tempfile +import time +from datetime import datetime, timezone + +logger = logging.getLogger(__name__) + + +class StateManager: + """Manages persistent state in a JSON file. + + State tracks retry counts and alert flags per DAG run to ensure + idempotent behavior across monitoring cycles. + """ + + def __init__(self, state_file: str, max_age: int = 86400): + self._path = pathlib.Path(state_file) + self._max_age = max_age + self._data: dict = {"version": 1, "entries": {}} + logger.debug( + "StateManager initialized: file=%s, max_age=%ds", + state_file, max_age, + ) + + def load(self): + """Load state from disk. Start fresh if missing or corrupt.""" + if not self._path.exists(): + logger.debug("State file not found at %s, starting with empty state", self._path) + self._data = {"version": 1, "entries": {}} + return + + try: + with open(self._path, "r", encoding="utf-8") as f: + raw = f.read() + self._data = json.loads(raw) + if not isinstance(self._data.get("entries"), dict): + raise ValueError("Invalid state structure: 'entries' is not a dict") + entry_count = len(self._data["entries"]) + logger.debug( + "State loaded: %d entries, file_size=%d bytes", + entry_count, len(raw), + ) + if entry_count > 0: + logger.debug( + "State entries: %s", + ", ".join( + f"{k}(retries={v.get('retry_count', 0)}, alerted={v.get('alerted', False)})" + for k, v in self._data["entries"].items() + ), + ) + except (json.JSONDecodeError, ValueError, KeyError) as e: + logger.warning("Corrupt state file %s, starting fresh: %s", self._path, e) + self._data = {"version": 1, "entries": {}} + + def save(self): + """Save state to disk atomically (write tmp -> rename).""" + self._path.parent.mkdir(parents=True, exist_ok=True) + try: + content = json.dumps(self._data, indent=2, ensure_ascii=False) + fd, tmp_path = tempfile.mkstemp( + dir=str(self._path.parent), suffix=".tmp" + ) + with os.fdopen(fd, "w", encoding="utf-8") as f: + f.write(content) + os.replace(tmp_path, str(self._path)) + logger.debug( + "State saved: %d entries, %d bytes → %s", + len(self._data["entries"]), len(content), self._path, + ) + except OSError as e: + logger.error("Failed to save state to %s: %s", self._path, e) + # Clean up temp file if it exists + try: + os.unlink(tmp_path) + except OSError: + pass + + @staticmethod + def _key(dag_id: str, dag_run_id: str) -> str: + return f"{dag_id}::{dag_run_id}" + + @staticmethod + def _now_iso() -> str: + return datetime.now(timezone.utc).isoformat() + + def get_entry(self, dag_id: str, dag_run_id: str) -> dict | None: + """Get stored entry for a DAG run, or None.""" + return self._data["entries"].get(self._key(dag_id, dag_run_id)) + + def ensure_entry(self, dag_id: str, dag_run_id: str) -> dict: + """Get or create an entry for a DAG run.""" + key = self._key(dag_id, dag_run_id) + if key not in self._data["entries"]: + self._data["entries"][key] = { + "dag_id": dag_id, + "dag_run_id": dag_run_id, + "retry_count": 0, + "last_retry_time": None, + "alerted": False, + "first_seen": self._now_iso(), + "last_seen": self._now_iso(), + } + logger.debug("State: new entry created for %s", key) + else: + self._data["entries"][key]["last_seen"] = self._now_iso() + logger.debug("State: updated last_seen for %s", key) + return self._data["entries"][key] + + def get_retry_count(self, dag_id: str, dag_run_id: str) -> int: + """Get current retry count for a DAG run.""" + entry = self.get_entry(dag_id, dag_run_id) + count = entry["retry_count"] if entry else 0 + logger.debug("State: retry_count for %s/%s = %d", dag_id, dag_run_id, count) + return count + + def increment_retry(self, dag_id: str, dag_run_id: str): + """Increment retry counter and record time.""" + entry = self.ensure_entry(dag_id, dag_run_id) + old_count = entry["retry_count"] + entry["retry_count"] += 1 + entry["last_retry_time"] = self._now_iso() + logger.debug( + "State: retry_count for %s/%s incremented %d → %d", + dag_id, dag_run_id, old_count, entry["retry_count"], + ) + + def mark_alerted(self, dag_id: str, dag_run_id: str): + """Mark a DAG run as alerted (avoid duplicate alerts).""" + entry = self.ensure_entry(dag_id, dag_run_id) + entry["alerted"] = True + logger.debug("State: marked alerted for %s/%s", dag_id, dag_run_id) + + def is_alerted(self, dag_id: str, dag_run_id: str) -> bool: + """Check if alert was already sent for this DAG run.""" + entry = self.get_entry(dag_id, dag_run_id) + return entry["alerted"] if entry else False + + def purge_old_entries(self): + """Remove entries older than max_age seconds.""" + cutoff = time.time() - self._max_age + keys_to_remove = [] + + for key, entry in self._data["entries"].items(): + last_seen = entry.get("last_seen", entry.get("first_seen", "")) + try: + ts = datetime.fromisoformat(last_seen).timestamp() + if ts < cutoff: + keys_to_remove.append(key) + except (ValueError, TypeError): + keys_to_remove.append(key) + + for key in keys_to_remove: + logger.debug("State: purging old entry %s", key) + del self._data["entries"][key] + + if keys_to_remove: + logger.info("Purged %d old state entries (max_age=%ds)", len(keys_to_remove), self._max_age) + else: + logger.debug("State: no entries to purge") diff --git a/build.sh b/build.sh new file mode 100755 index 0000000..02df05f --- /dev/null +++ b/build.sh @@ -0,0 +1,322 @@ +#!/usr/bin/env bash +# ============================================================================= +# Airflow DAG Monitor - Offline Build Script +# +# Собирает автономный ZIP-архив со всеми зависимостями (wheel-пакеты), +# исходным кодом и скриптами установки. Архив можно перенести в закрытый +# контур без доступа в интернет и развернуть через install.sh. +# +# Использование: +# ./build.sh # сборка для текущей платформы +# ./build.sh --platform manylinux2014_x86_64 # явное указание платформы +# ============================================================================= +set -euo pipefail + +SCRIPT_DIR="$(cd "$(dirname "$0")" && pwd)" +VERSION=$(python3 -c " +import re, pathlib +text = pathlib.Path('$SCRIPT_DIR/airflow_monitor/__init__.py').read_text() +m = re.search(r'__version__\s*=\s*[\\x22\\x27]([^\\x22\\x27]+)', text) +print(m.group(1) if m else '0.0.0') +") +BUILD_DIR="$SCRIPT_DIR/build" +DIST_DIR="$SCRIPT_DIR/dist" +WHEELS_DIR="$BUILD_DIR/wheels" +STAGE_DIR="$BUILD_DIR/airflow-monitor-${VERSION}" +PLATFORM="" + +# --- Parse arguments --- +while [[ $# -gt 0 ]]; do + case "$1" in + --platform) + PLATFORM="$2" + shift 2 + ;; + -h|--help) + echo "Usage: $0 [--platform PLATFORM]" + echo "" + echo "Options:" + echo " --platform Target platform for wheels (e.g., manylinux2014_x86_64)" + echo " Default: auto-detect from current system" + echo "" + echo "Output: dist/airflow-monitor-VERSION.zip" + exit 0 + ;; + *) + echo "Unknown option: $1" >&2 + exit 1 + ;; + esac +done + +echo "=== Airflow DAG Monitor — Offline Build ===" +echo "Version: ${VERSION}" +echo "Build dir: ${BUILD_DIR}" +echo "" + +# --- Cleanup previous build --- +rm -rf "$BUILD_DIR" +mkdir -p "$WHEELS_DIR" "$STAGE_DIR" "$DIST_DIR" + +# --- Step 1: Download wheel packages --- +echo "[1/4] Downloading dependencies as wheel packages..." + +PIP_ARGS=( + download + -r "$SCRIPT_DIR/requirements.txt" + --dest "$WHEELS_DIR" + --quiet +) + +if [ -n "$PLATFORM" ]; then + PIP_ARGS+=(--platform "$PLATFORM" --only-binary=:all:) + echo " Target platform: $PLATFORM" +else + echo " Target platform: current ($(python3 -c 'import platform; print(platform.machine())'))" +fi + +python3 -m pip "${PIP_ARGS[@]}" + +WHEEL_COUNT=$(find "$WHEELS_DIR" -type f | wc -l) +echo " Downloaded $WHEEL_COUNT packages:" +find "$WHEELS_DIR" -type f -printf " %f\n" | sort + +# --- Step 2: Stage project files --- +echo "[2/4] Staging project files..." + +# Application code +cp -r "$SCRIPT_DIR/airflow_monitor" "$STAGE_DIR/airflow_monitor" +# Remove __pycache__ +find "$STAGE_DIR/airflow_monitor" -type d -name __pycache__ -exec rm -rf {} + 2>/dev/null || true + +# Configuration and docs +cp "$SCRIPT_DIR/config.yaml" "$STAGE_DIR/" +cp "$SCRIPT_DIR/requirements.txt" "$STAGE_DIR/" +cp "$SCRIPT_DIR/airflow-monitor.service" "$STAGE_DIR/" +cp "$SCRIPT_DIR/README.md" "$STAGE_DIR/" + +# Zabbix agent config +cp -r "$SCRIPT_DIR/zabbix" "$STAGE_DIR/zabbix" + +# Wheels directory +cp -r "$WHEELS_DIR" "$STAGE_DIR/wheels" + +echo " Staged files:" +find "$STAGE_DIR" -type f -printf " %P\n" | head -30 + +# --- Step 3: Generate offline install script --- +echo "[3/4] Generating install.sh..." + +cat > "$STAGE_DIR/install.sh" << 'INSTALL_SCRIPT' +#!/usr/bin/env bash +# ============================================================================= +# Airflow DAG Monitor - Offline Installer +# +# Устанавливает сервис из автономного архива без доступа в интернет. +# Все зависимости включены в каталог wheels/. +# +# Использование: +# ./install.sh # установка в /opt/airflow-monitor +# ./install.sh --prefix /srv/monitor # установка в указанный каталог +# ============================================================================= +set -euo pipefail + +SCRIPT_DIR="$(cd "$(dirname "$0")" && pwd)" +INSTALL_DIR="/opt/airflow-monitor" +SERVICE_USER="$(whoami)" +SERVICE_NAME="airflow-monitor" + +# --- Parse arguments --- +while [[ $# -gt 0 ]]; do + case "$1" in + --prefix) + INSTALL_DIR="$2" + shift 2 + ;; + --user) + SERVICE_USER="$2" + shift 2 + ;; + -h|--help) + echo "Usage: $0 [--prefix INSTALL_DIR] [--user SERVICE_USER]" + echo "" + echo "Options:" + echo " --prefix Installation directory (default: /opt/airflow-monitor)" + echo " --user User to run the service as (default: current user)" + exit 0 + ;; + *) + echo "Unknown option: $1" >&2 + exit 1 + ;; + esac +done + +VENV_DIR="$INSTALL_DIR/venv" + +echo "=== Airflow DAG Monitor — Offline Install ===" +echo "Source: $SCRIPT_DIR" +echo "Install dir: $INSTALL_DIR" +echo "Service user: $SERVICE_USER" +echo "" + +# --- Step 1: Copy files if installing to a different directory --- +if [ "$INSTALL_DIR" != "$SCRIPT_DIR" ]; then + echo "[1/5] Copying files to $INSTALL_DIR..." + mkdir -p "$INSTALL_DIR" + cp -r "$SCRIPT_DIR/airflow_monitor" "$INSTALL_DIR/" + cp -r "$SCRIPT_DIR/wheels" "$INSTALL_DIR/" + cp -r "$SCRIPT_DIR/zabbix" "$INSTALL_DIR/" + cp "$SCRIPT_DIR/config.yaml" "$INSTALL_DIR/" + cp "$SCRIPT_DIR/requirements.txt" "$INSTALL_DIR/" + cp "$SCRIPT_DIR/airflow-monitor.service" "$INSTALL_DIR/" + cp "$SCRIPT_DIR/README.md" "$INSTALL_DIR/" + echo " Done" +else + echo "[1/5] Installing in place, skipping copy" +fi + +# --- Step 2: Create virtual environment --- +echo "[2/5] Creating Python virtual environment..." +if [ -d "$VENV_DIR" ]; then + rm -rf "$VENV_DIR" +fi +python3 -m venv "$VENV_DIR" +echo " Created: $VENV_DIR" + +# --- Step 3: Install dependencies from local wheels --- +echo "[3/5] Installing dependencies from local wheels (offline)..." +"$VENV_DIR/bin/pip" install --upgrade pip --quiet 2>/dev/null || true +"$VENV_DIR/bin/pip" install \ + --no-index \ + --find-links "$INSTALL_DIR/wheels" \ + -r "$INSTALL_DIR/requirements.txt" \ + --quiet +echo " Installed packages:" +"$VENV_DIR/bin/pip" list --format=columns 2>/dev/null | grep -vE '^(Package|---)' +echo "" + +# --- Step 4: Create required directories --- +echo "[4/5] Creating directories..." + +STATE_DIR="/var/lib/airflow-monitor" +LOG_DIR="/var/log/airflow-monitor" + +for dir in "$STATE_DIR" "$LOG_DIR"; do + if [ ! -d "$dir" ]; then + sudo mkdir -p "$dir" + sudo chown "${SERVICE_USER}:$(id -gn "$SERVICE_USER")" "$dir" + echo " Created: $dir" + else + echo " Exists: $dir" + fi +done + +# --- Step 5: Install systemd service --- +echo "[5/5] Installing systemd service..." + +# Patch service file with actual paths and user +SERVICE_FILE="$INSTALL_DIR/airflow-monitor.service" +if [ -f "$SERVICE_FILE" ]; then + TEMP_SERVICE=$(mktemp) + sed \ + -e "s|User=.*|User=${SERVICE_USER}|" \ + -e "s|Group=.*|Group=$(id -gn "$SERVICE_USER")|" \ + -e "s|WorkingDirectory=.*|WorkingDirectory=${INSTALL_DIR}|" \ + -e "s|ExecStart=.*|ExecStart=${VENV_DIR}/bin/python -m airflow_monitor --config ${INSTALL_DIR}/config.yaml|" \ + "$SERVICE_FILE" > "$TEMP_SERVICE" + + sudo cp "$TEMP_SERVICE" /etc/systemd/system/${SERVICE_NAME}.service + rm -f "$TEMP_SERVICE" + sudo systemctl daemon-reload + echo " Installed: /etc/systemd/system/${SERVICE_NAME}.service" +else + echo " WARNING: service file not found, skipping" +fi + +# --- Step 6: Install Zabbix agent config --- +echo "[6/6] Installing Zabbix agent config..." +ZABBIX_CONF_SRC="$INSTALL_DIR/zabbix/airflow-monitor.conf" +ZABBIX_CONF_DST="/etc/zabbix/zabbix_agentd.conf.d/airflow-monitor.conf" + +if [ -f "$ZABBIX_CONF_SRC" ]; then + if [ -d "/etc/zabbix/zabbix_agentd.conf.d" ]; then + sudo cp "$ZABBIX_CONF_SRC" "$ZABBIX_CONF_DST" + sudo chmod 644 "$ZABBIX_CONF_DST" + echo " Installed: $ZABBIX_CONF_DST" + if systemctl is-active zabbix-agent >/dev/null 2>&1; then + sudo systemctl restart zabbix-agent + echo " Restarted: zabbix-agent" + elif systemctl is-active zabbix-agent2 >/dev/null 2>&1; then + sudo systemctl restart zabbix-agent2 + echo " Restarted: zabbix-agent2" + else + echo " WARNING: zabbix-agent not running, restart manually" + fi + else + echo " WARNING: /etc/zabbix/zabbix_agentd.conf.d/ not found" + echo " Copy manually: sudo cp $ZABBIX_CONF_SRC $ZABBIX_CONF_DST" + fi +else + echo " WARNING: zabbix config not found" +fi + +# Ensure data dir is readable by zabbix agent +sudo chmod 755 /var/lib/airflow-monitor 2>/dev/null || true + +echo "" +echo "=== Install Complete ===" +echo "" +echo "Next steps:" +echo " 1. Edit config.yaml:" +echo " vim $INSTALL_DIR/config.yaml" +echo "" +echo " 2. Test manually:" +echo " $VENV_DIR/bin/python -m airflow_monitor --discover -c $INSTALL_DIR/config.yaml" +echo "" +echo " 3. Enable and start:" +echo " sudo systemctl enable --now $SERVICE_NAME" +echo "" +echo " 4. Check logs:" +echo " journalctl -u $SERVICE_NAME -f" +echo "" +echo " 5. Verify Zabbix agent reads data:" +echo " zabbix_agentd -t airflow.monitor.heartbeat" +echo " zabbix_agentd -t airflow.dag.problems.count" +INSTALL_SCRIPT + +chmod +x "$STAGE_DIR/install.sh" +echo " Generated: install.sh" + +# --- Step 4: Create ZIP archive --- +echo "[4/4] Creating ZIP archive..." + +ARCHIVE_NAME="airflow-monitor-${VERSION}.zip" +ARCHIVE_PATH="$DIST_DIR/$ARCHIVE_NAME" + +(cd "$BUILD_DIR" && zip -r "$ARCHIVE_PATH" "airflow-monitor-${VERSION}" -q) + +ARCHIVE_SIZE=$(du -h "$ARCHIVE_PATH" | cut -f1) +FILE_COUNT=$(unzip -l "$ARCHIVE_PATH" | tail -1 | awk '{print $2}') + +echo " Archive: $ARCHIVE_PATH" +echo " Size: $ARCHIVE_SIZE" +echo " Files: $FILE_COUNT" + +# --- Cleanup build directory --- +rm -rf "$BUILD_DIR" + +echo "" +echo "=== Build Complete ===" +echo "" +echo "Архив для переноса в закрытый контур:" +echo " $ARCHIVE_PATH" +echo "" +echo "Развёртывание на целевом сервере:" +echo " 1. Скопировать: scp $ARCHIVE_PATH user@target:/tmp/" +echo " 2. Распаковать: unzip /tmp/$ARCHIVE_NAME -d /tmp/" +echo " 3. Установить: cd /tmp/airflow-monitor-${VERSION} && ./install.sh" +echo "" +echo "По умолчанию устанавливается в /opt/airflow-monitor." +echo "Для смены каталога или пользователя:" +echo " ./install.sh --prefix /srv/airflow-monitor --user airflow" diff --git a/config.yaml b/config.yaml new file mode 100644 index 0000000..96b48fa --- /dev/null +++ b/config.yaml @@ -0,0 +1,82 @@ +# ============================================================================= +# Airflow DAG Monitor - Configuration +# ============================================================================= + +# -------------------------------------------------------------------------- +# Секция airflow не обязательна при запуске с флагом --discover. +# В режиме auto-discovery параметры base_url, username, password +# определяются автоматически из Docker-контейнера и .env файла. +# -------------------------------------------------------------------------- +airflow: + # Airflow webserver URL + # При --discover определяется автоматически из docker port + base_url: "http://localhost:8080" + + # Basic Auth credentials + # При --discover определяются из .env / docker-compose.yml + username: "airflow" + password: "airflow" + + # API version: "auto", "v1", or "experimental" + api_version: "v1" + + # HTTP request timeout (seconds) + timeout: 30 + + # Verify SSL certificates + verify_ssl: false + + # Delay between API requests to avoid overwhelming Airflow (seconds) + request_delay: 0.5 + +monitor: + # Interval between monitoring cycles (seconds) + cycle_interval: 300 # 5 minutes + + # DAG run duration threshold to consider as "long running" (seconds) + long_running_threshold: 1800 # 30 minutes + + # Time to wait after retry before re-checking (seconds) + retry_wait: 120 # 2 minutes + + # Maximum number of retries per DAG run before alerting + max_retries: 1 + + # Persistent state file path + state_file: "/var/lib/airflow-monitor/state.json" + + # Lock file to prevent multiple instances + lock_file: "/var/run/airflow-monitor.lock" + + # Purge state entries older than this (seconds) + state_max_age: 86400 # 24 hours + +zabbix: + # Enable/disable Zabbix integration (экспорт данных в файлы) + enabled: true + + # Каталог для файлов данных, которые читает Zabbix agent + data_dir: "/var/lib/airflow-monitor" + + # Имя файла со списком проблемных DAG-ов (JSON) + problems_file: "problems.json" + + # Имя файла с heartbeat (epoch timestamp) + heartbeat_file: "heartbeat" + + # Имя файла со статусом монитора (JSON) + status_file: "status.json" + +logging: + # Log level: DEBUG, INFO, WARNING, ERROR, CRITICAL + # DEBUG рекомендуется на этапе внедрения, затем переключить на INFO + level: "DEBUG" + + # Log file path + file: "/var/log/airflow-monitor/monitor.log" + + # Maximum log file size before rotation (bytes) + max_bytes: 10485760 # 10 MB + + # Number of rotated log files to keep + backup_count: 5 diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..17bac78 --- /dev/null +++ b/requirements.txt @@ -0,0 +1,2 @@ +requests>=2.31,<3 +PyYAML>=6.0,<7 diff --git a/setup.sh b/setup.sh new file mode 100755 index 0000000..2f0b7f5 --- /dev/null +++ b/setup.sh @@ -0,0 +1,126 @@ +#!/usr/bin/env bash +# ============================================================================= +# Airflow DAG Monitor - Setup Script +# +# Creates virtual environment, installs dependencies, prepares directories, +# and installs systemd service. +# +# По умолчанию устанавливает в /opt/airflow-monitor. +# Если запущен из другого каталога — копирует файлы в /opt/airflow-monitor. +# ============================================================================= +set -euo pipefail + +SCRIPT_DIR="$(cd "$(dirname "$0")" && pwd)" +INSTALL_DIR="/opt/airflow-monitor" + +# Copy project files to /opt/airflow-monitor if running from another location +if [ "$SCRIPT_DIR" != "$INSTALL_DIR" ]; then + echo "Copying project to $INSTALL_DIR..." + sudo mkdir -p "$INSTALL_DIR" + sudo chown "$(whoami):$(id -gn)" "$INSTALL_DIR" + cp -r "$SCRIPT_DIR/airflow_monitor" "$INSTALL_DIR/" + cp "$SCRIPT_DIR/config.yaml" "$INSTALL_DIR/" + cp "$SCRIPT_DIR/requirements.txt" "$INSTALL_DIR/" + cp "$SCRIPT_DIR/airflow-monitor.service" "$INSTALL_DIR/" + cp "$SCRIPT_DIR/README.md" "$INSTALL_DIR/" + [ -f "$SCRIPT_DIR/setup.sh" ] && cp "$SCRIPT_DIR/setup.sh" "$INSTALL_DIR/" + [ -f "$SCRIPT_DIR/build.sh" ] && cp "$SCRIPT_DIR/build.sh" "$INSTALL_DIR/" + [ -d "$SCRIPT_DIR/zabbix" ] && cp -r "$SCRIPT_DIR/zabbix" "$INSTALL_DIR/" +fi + +VENV_DIR="$INSTALL_DIR/venv" +SERVICE_NAME="airflow-monitor" +SERVICE_FILE="$INSTALL_DIR/${SERVICE_NAME}.service" + +echo "=== Airflow DAG Monitor Setup ===" +echo "Install directory: $INSTALL_DIR" +echo "" + +# --- Step 1: Create virtual environment --- +echo "[1/4] Creating Python virtual environment..." +if [ -d "$VENV_DIR" ]; then + echo " Virtual environment already exists, recreating..." + rm -rf "$VENV_DIR" +fi +python3 -m venv "$VENV_DIR" +echo " Created: $VENV_DIR" + +# --- Step 2: Install dependencies --- +echo "[2/4] Installing dependencies..." +"$VENV_DIR/bin/pip" install --upgrade pip --quiet +"$VENV_DIR/bin/pip" install -r "$INSTALL_DIR/requirements.txt" --quiet +echo " Installed: $(cat "$INSTALL_DIR/requirements.txt" | grep -v '^#' | tr '\n' ' ')" + +# --- Step 3: Create required directories --- +echo "[3/4] Creating directories..." + +STATE_DIR="/var/lib/airflow-monitor" +LOG_DIR="/var/log/airflow-monitor" + +for dir in "$STATE_DIR" "$LOG_DIR"; do + if [ ! -d "$dir" ]; then + sudo mkdir -p "$dir" + sudo chown "$(whoami):$(id -gn)" "$dir" + echo " Created: $dir" + else + echo " Exists: $dir" + fi +done + +# --- Step 4: Install systemd service --- +echo "[4/5] Installing systemd service..." +if [ -f "$SERVICE_FILE" ]; then + sudo cp "$SERVICE_FILE" /etc/systemd/system/ + sudo systemctl daemon-reload + echo " Installed: /etc/systemd/system/${SERVICE_NAME}.service" +else + echo " WARNING: ${SERVICE_FILE} not found, skipping service install" +fi + +# --- Step 5: Install Zabbix agent config --- +echo "[5/5] Installing Zabbix agent config..." +ZABBIX_CONF_SRC="$INSTALL_DIR/zabbix/airflow-monitor.conf" +ZABBIX_CONF_DST="/etc/zabbix/zabbix_agentd.conf.d/airflow-monitor.conf" + +if [ -f "$ZABBIX_CONF_SRC" ]; then + if [ -d "/etc/zabbix/zabbix_agentd.conf.d" ]; then + sudo cp "$ZABBIX_CONF_SRC" "$ZABBIX_CONF_DST" + # Zabbix agent must be able to read data files + sudo chmod 644 "$ZABBIX_CONF_DST" + echo " Installed: $ZABBIX_CONF_DST" + # Restart zabbix agent to pick up new UserParameters + if systemctl is-active zabbix-agent >/dev/null 2>&1; then + sudo systemctl restart zabbix-agent + echo " Restarted: zabbix-agent" + elif systemctl is-active zabbix-agent2 >/dev/null 2>&1; then + sudo systemctl restart zabbix-agent2 + echo " Restarted: zabbix-agent2" + else + echo " WARNING: zabbix-agent not running, restart it manually" + fi + else + echo " WARNING: /etc/zabbix/zabbix_agentd.conf.d/ not found" + echo " Copy manually: cp $ZABBIX_CONF_SRC $ZABBIX_CONF_DST" + fi +else + echo " WARNING: zabbix config not found at $ZABBIX_CONF_SRC" +fi + +# Ensure data directory is readable by zabbix agent +sudo chmod 755 "$STATE_DIR" + +echo "" +echo "=== Setup Complete ===" +echo "" +echo "Next steps:" +echo " 1. Edit config.yaml:" +echo " vim $INSTALL_DIR/config.yaml" +echo " 2. Test manually:" +echo " $VENV_DIR/bin/python -m airflow_monitor --discover -c $INSTALL_DIR/config.yaml" +echo " 3. Enable and start:" +echo " sudo systemctl enable --now $SERVICE_NAME" +echo " 4. Check logs:" +echo " journalctl -u $SERVICE_NAME -f" +echo " 5. Verify Zabbix agent reads data:" +echo " zabbix_agentd -t airflow.monitor.heartbeat" +echo " zabbix_agentd -t airflow.dag.problems.count" diff --git a/zabbix/airflow-monitor.conf b/zabbix/airflow-monitor.conf new file mode 100644 index 0000000..cfd0b6f --- /dev/null +++ b/zabbix/airflow-monitor.conf @@ -0,0 +1,32 @@ +# ============================================================================= +# Zabbix Agent UserParameter — Airflow DAG Monitor +# +# Устанавливается в /etc/zabbix/zabbix_agentd.conf.d/airflow-monitor.conf +# Zabbix agent читает файлы, записанные сервисом airflow-monitor, +# и передаёт данные на Zabbix Server/Proxy через стандартный канал агента. +# +# Файлы данных: +# /var/lib/airflow-monitor/problems.json — JSON массив проблемных DAG-ов +# /var/lib/airflow-monitor/heartbeat — epoch timestamp последнего цикла +# /var/lib/airflow-monitor/status.json — статус монитора (JSON) +# ============================================================================= + +# Список проблемных DAG-ов (JSON text) +# Zabbix item: type=Zabbix agent, key=airflow.dag.problems, type of info=Text +UserParameter=airflow.dag.problems,cat /var/lib/airflow-monitor/problems.json 2>/dev/null || echo '[]' + +# Количество проблемных DAG-ов (для простых триггеров) +# Zabbix item: type=Zabbix agent, key=airflow.dag.problems.count, type of info=Numeric (unsigned) +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 последнего успешного цикла +# Zabbix item: type=Zabbix agent, key=airflow.monitor.heartbeat, type of info=Numeric (unsigned) +UserParameter=airflow.monitor.heartbeat,cat /var/lib/airflow-monitor/heartbeat 2>/dev/null || echo 0 + +# Статус монитора (JSON: cycle_count, dag_count, issue_count, cycle_time) +# Zabbix item: type=Zabbix agent, key=airflow.monitor.status, type of info=Text +UserParameter=airflow.monitor.status,cat /var/lib/airflow-monitor/status.json 2>/dev/null || echo '{}' + +# Проверка что сервис запущен (1 = работает, 0 = остановлен) +# Zabbix item: type=Zabbix agent, key=airflow.monitor.alive, type of info=Numeric (unsigned) +UserParameter=airflow.monitor.alive,systemctl is-active airflow-monitor >/dev/null 2>&1 && echo 1 || echo 0 diff --git a/zabbix/zbx_template_airflow_monitor.yaml b/zabbix/zbx_template_airflow_monitor.yaml new file mode 100644 index 0000000..92d306a --- /dev/null +++ b/zabbix/zbx_template_airflow_monitor.yaml @@ -0,0 +1,288 @@ +zabbix_export: + version: '6.4' + template_groups: + - uuid: 846977d1dfed4968bc5f8bdb363285bc + name: 'Templates/Applications' + templates: + - uuid: a1b2c3d4e5f6a7b8c9d0e1f2a3b4c5d6 + template: 'Airflow DAG Monitor' + name: 'Airflow DAG Monitor' + description: | + Мониторинг Apache Airflow DAG-ов через сервис airflow-monitor. + Сервис пишет данные в файлы, Zabbix Agent читает через UserParameter. + + Требования: + - Установлен airflow-monitor (/opt/airflow-monitor) + - Установлен конфиг /etc/zabbix/zabbix_agentd.conf.d/airflow-monitor.conf + - Перезапущен zabbix-agent после установки конфига + + Версия шаблона: 1.0.0 + groups: + - name: 'Templates/Applications' + items: + # ================================================================ + # Основные items + # ================================================================ + - uuid: 11111111111111111111111111111001 + name: 'Airflow: Problematic DAGs (JSON)' + key: airflow.dag.problems + type: ZABBIX_PASSIVE + value_type: TEXT + delay: 1m + history: 7d + description: 'JSON-массив проблемных DAG-запусков. [] = нет проблем.' + tags: + - tag: component + value: airflow + - tag: component + value: dag-monitor + + - uuid: 11111111111111111111111111111002 + name: 'Airflow: Problematic DAGs count' + key: airflow.dag.problems.count + type: ZABBIX_PASSIVE + value_type: UNSIGNED + delay: 1m + history: 90d + trends: 365d + description: 'Количество DAG-запусков с проблемами (failed или long_running).' + tags: + - tag: component + value: airflow + + - uuid: 11111111111111111111111111111003 + name: 'Airflow Monitor: Heartbeat' + key: airflow.monitor.heartbeat + type: ZABBIX_PASSIVE + value_type: UNSIGNED + delay: 1m + history: 7d + description: 'Epoch timestamp последнего успешного цикла мониторинга.' + tags: + - tag: component + value: dag-monitor + + - uuid: 11111111111111111111111111111004 + name: 'Airflow Monitor: Status (JSON)' + key: airflow.monitor.status + type: ZABBIX_PASSIVE + value_type: TEXT + delay: 5m + history: 7d + description: 'JSON со статусом монитора: cycle_count, dag_count, issue_count, cycle_time.' + tags: + - tag: component + value: dag-monitor + + - uuid: 11111111111111111111111111111005 + name: 'Airflow Monitor: Service alive' + key: airflow.monitor.alive + type: ZABBIX_PASSIVE + value_type: UNSIGNED + delay: 1m + history: 90d + trends: 365d + description: '1 = сервис airflow-monitor запущен, 0 = остановлен.' + tags: + - tag: component + value: dag-monitor + + # ================================================================ + # Вычисляемые items (из heartbeat) + # ================================================================ + - uuid: 11111111111111111111111111111006 + name: 'Airflow Monitor: Data age (seconds)' + key: airflow.monitor.data_age + type: CALCULATED + value_type: UNSIGNED + delay: 1m + history: 7d + params: 'now()-last(//airflow.monitor.heartbeat)' + description: 'Сколько секунд прошло с последнего обновления данных.' + tags: + - tag: component + value: dag-monitor + + # ================================================================ + # Dependent items (извлечение из status JSON) + # ================================================================ + - uuid: 11111111111111111111111111111010 + name: 'Airflow Monitor: DAG count' + key: airflow.monitor.dag_count + type: DEPENDENT + value_type: UNSIGNED + history: 90d + trends: 365d + description: 'Количество отслеживаемых DAG-ов (из status.json).' + master_item: + key: airflow.monitor.status + preprocessing: + - type: JSONPATH + parameters: + - '$.dag_count' + - type: DISCARD_UNCHANGED_HEARTBEAT + parameters: + - 1h + tags: + - tag: component + value: airflow + + - uuid: 11111111111111111111111111111011 + name: 'Airflow Monitor: Cycle time (seconds)' + key: airflow.monitor.cycle_time + type: DEPENDENT + value_type: FLOAT + history: 90d + trends: 365d + description: 'Время выполнения последнего цикла мониторинга (секунды).' + master_item: + key: airflow.monitor.status + preprocessing: + - type: JSONPATH + parameters: + - '$.cycle_time_seconds' + tags: + - tag: component + value: dag-monitor + + - uuid: 11111111111111111111111111111012 + name: 'Airflow Monitor: Cycle count' + key: airflow.monitor.cycle_count + type: DEPENDENT + value_type: UNSIGNED + history: 90d + trends: 365d + description: 'Номер текущего цикла мониторинга (с момента запуска сервиса).' + master_item: + key: airflow.monitor.status + preprocessing: + - type: JSONPATH + parameters: + - '$.cycle_count' + - type: DISCARD_UNCHANGED_HEARTBEAT + parameters: + - 1h + tags: + - tag: component + value: dag-monitor + + # ================================================================== + # Triggers + # ================================================================== + triggers: + - uuid: 22222222222222222222222222222001 + expression: 'last(/Airflow DAG Monitor/airflow.dag.problems.count)>0' + name: 'Airflow: {ITEM.LASTVALUE1} problematic DAG run(s) detected' + priority: HIGH + description: | + Обнаружены проблемные DAG-запуски (failed или long_running). + Подробности в item "Airflow: Problematic DAGs (JSON)". + Триггер снимается автоматически, когда все проблемы разрешены. + tags: + - tag: scope + value: availability + - tag: component + value: airflow + + - uuid: 22222222222222222222222222222002 + expression: 'last(/Airflow DAG Monitor/airflow.monitor.data_age)>600' + name: 'Airflow Monitor: Data is stale (no update for {ITEM.LASTVALUE1}s)' + priority: DISASTER + description: | + Данные мониторинга не обновлялись более 10 минут. + Возможные причины: сервис остановлен, Airflow API недоступен, + ошибка в цикле мониторинга. + dependencies: + - name: 'Airflow Monitor: Service is not running' + expression: 'last(/Airflow DAG Monitor/airflow.monitor.alive)=0' + tags: + - tag: scope + value: availability + - tag: component + value: dag-monitor + + - uuid: 22222222222222222222222222222003 + expression: 'last(/Airflow DAG Monitor/airflow.monitor.alive)=0' + name: 'Airflow Monitor: Service is not running' + priority: HIGH + description: | + Сервис airflow-monitor не запущен (systemctl is-active вернул не-active). + Запустить: sudo systemctl start airflow-monitor + tags: + - tag: scope + value: availability + - tag: component + value: dag-monitor + + - uuid: 22222222222222222222222222222004 + expression: 'last(/Airflow DAG Monitor/airflow.monitor.cycle_time)>60' + name: 'Airflow Monitor: Cycle time is high ({ITEM.LASTVALUE1}s)' + priority: WARNING + description: | + Цикл мониторинга выполняется дольше 60 секунд. + Возможные причины: большое количество DAG-ов, медленный API, + проблемы с сетью до Airflow. + tags: + - tag: scope + value: performance + - tag: component + value: dag-monitor + + - uuid: 22222222222222222222222222222005 + expression: 'last(/Airflow DAG Monitor/airflow.dag.problems.count)>0 and length(last(/Airflow DAG Monitor/airflow.dag.problems))>2' + recovery_mode: RECOVERY_EXPRESSION + recovery_expression: 'last(/Airflow DAG Monitor/airflow.dag.problems.count)=0' + name: 'Airflow: DAG failures require attention' + priority: AVERAGE + description: | + Есть проблемные DAG-запуски после попытки автоматического перезапуска. + Требуется ручное вмешательство. + tags: + - tag: scope + value: notice + - tag: component + value: airflow + + # ================================================================== + # Graphs + # ================================================================== + graphs: + - uuid: 33333333333333333333333333333001 + name: 'Airflow: Problem DAG count' + graph_items: + - color: FF0000 + item: + host: 'Airflow DAG Monitor' + key: airflow.dag.problems.count + - color: 00AA00 + yaxisside: RIGHT + item: + host: 'Airflow DAG Monitor' + key: airflow.monitor.dag_count + + - uuid: 33333333333333333333333333333002 + name: 'Airflow Monitor: Cycle time' + graph_items: + - color: 0000FF + item: + host: 'Airflow DAG Monitor' + key: airflow.monitor.cycle_time + + - uuid: 33333333333333333333333333333003 + name: 'Airflow Monitor: Data freshness' + graph_items: + - color: FF6600 + item: + host: 'Airflow DAG Monitor' + key: airflow.monitor.data_age + + # ================================================================== + # Macros (для тюнинга через host-level overrides) + # ================================================================== + macros: + - macro: '{$AIRFLOW.STALE_THRESHOLD}' + value: '600' + description: 'Порог устаревания данных (секунды). По умолчанию 10 минут.' + - macro: '{$AIRFLOW.CYCLE_TIME_WARN}' + value: '60' + description: 'Порог предупреждения о долгом цикле (секунды).' diff --git a/zabbix/zbx_template_airflow_monitor_5x.xml b/zabbix/zbx_template_airflow_monitor_5x.xml new file mode 100644 index 0000000..c9a0cca --- /dev/null +++ b/zabbix/zbx_template_airflow_monitor_5x.xml @@ -0,0 +1,347 @@ + + + 5.4 + + + Templates/Applications + + + + + + + + + Service state + + + 0 + Stopped + + + 1 + Running + + + + +