From 4917a9a6ad1e8173abd0d9b429591167bbeafe76 Mon Sep 17 00:00:00 2001 From: Dmitry Dementev Date: Sun, 8 Feb 2026 22:27:32 +0300 Subject: [PATCH] feat(monitoring): add Airflow monitoring via statsd-exporter MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Add statsd-exporter service to docker-compose.yml (prom/statsd-exporter:v0.27.1) - Add StatsD env vars to airflow-default-env for metrics export - Add airflow job to prometheus.yml scrape configs - Add Airflow Overview dashboard (Grafana provisioning) - Add Airflow alert rules: scheduler down, queue backlog, failures, parse time - Add configs/statsd_mapping.yml for StatsD → Prometheus conversion - Use Prometheus naming convention (_total for counters, _seconds for timers) - Add monitoring plan at plans/monitoring_airflow_plan.md - Update OPERATIONS.md and Makefile for airflow monitoring Tested: all 3 jobs (airflow, clickhouse, kafka) showing UP in Prometheus, metrics flowing (dagbag_size=3, executor slots, heartbeats with _total suffix), all 4 alert rules loaded in Grafana --- Makefile | 2 +- .../alerting/airflow-alert-rules.yml | 331 +++++++ .../dashboards/airflow-overview.json | 805 ++++++++++++++++++ configs/prometheus.yml | 9 + configs/statsd_mapping.yml | 74 ++ docker-compose.yml | 23 + docs/OPERATIONS.md | 16 +- plans/monitoring_airflow_plan.md | 449 ++++++++++ 8 files changed, 1701 insertions(+), 8 deletions(-) create mode 100644 configs/grafana/provisioning/alerting/airflow-alert-rules.yml create mode 100644 configs/grafana/provisioning/dashboards/airflow-overview.json create mode 100644 configs/statsd_mapping.yml create mode 100644 plans/monitoring_airflow_plan.md diff --git a/Makefile b/Makefile index 635e6e3..149f1bb 100644 --- a/Makefile +++ b/Makefile @@ -17,7 +17,7 @@ transform: # Перезагрузка конфигурации мониторинга (после изменений в provisioning) reload-monitoring: @echo "=== Перезагрузка сервисов мониторинга ===" - $(COMPOSE) up -d prometheus grafana kafka-exporter + $(COMPOSE) up -d prometheus grafana kafka-exporter statsd-exporter $(COMPOSE) restart prometheus @echo "=== Перезагрузка provisioning Grafana ===" @sleep 2 diff --git a/configs/grafana/provisioning/alerting/airflow-alert-rules.yml b/configs/grafana/provisioning/alerting/airflow-alert-rules.yml new file mode 100644 index 0000000..406642b --- /dev/null +++ b/configs/grafana/provisioning/alerting/airflow-alert-rules.yml @@ -0,0 +1,331 @@ +apiVersion: 1 + +groups: + - orgId: 1 + name: airflow_health_group + folder: Airflow Alerts + interval: 30s + rules: + # --- Alert: Airflow Scheduler Down --- + - uid: airflow_scheduler_down + title: Airflow Scheduler Down + condition: C + data: + - refId: A + relativeTimeRange: + from: 300 + to: 0 + datasourceUid: prometheus_uid + model: + datasource: + type: prometheus + uid: prometheus_uid + editorMode: code + expr: rate(airflow_scheduler_heartbeat_total[5m]) + instant: false + intervalMs: 1000 + legendFormat: __auto + maxDataPoints: 43200 + range: true + refId: A + - refId: B + datasourceUid: __expr__ + model: + conditions: + - evaluator: + params: + - 0.1 + type: lt + operator: + type: and + query: + params: + - B + reducer: + params: [] + type: last + type: query + datasource: + type: __expr__ + uid: __expr__ + expression: A + intervalMs: 1000 + maxDataPoints: 43200 + refId: B + type: threshold + - refId: C + datasourceUid: __expr__ + model: + conditions: + - evaluator: + params: + - 0 + type: gt + operator: + type: and + query: + params: + - C + reducer: + params: [] + type: last + type: query + datasource: + type: __expr__ + uid: __expr__ + expression: B + intervalMs: 1000 + maxDataPoints: 43200 + refId: C + type: threshold + noDataState: Alerting + execErrState: Error + for: 2m + annotations: + summary: "Airflow scheduler не отправляет heartbeats" + description: "Scheduler possible down — rate(airflow_scheduler_heartbeat_total) < 0.1 в течение 2 минут" + labels: + severity: critical + + # --- Alert: Airflow Queue Backlog --- + - uid: airflow_queue_high + title: Airflow Queue Backlog + condition: C + data: + - refId: A + relativeTimeRange: + from: 300 + to: 0 + datasourceUid: prometheus_uid + model: + datasource: + type: prometheus + uid: prometheus_uid + editorMode: code + expr: airflow_executor_queued_tasks + instant: false + intervalMs: 1000 + legendFormat: __auto + maxDataPoints: 43200 + range: true + refId: A + - refId: B + datasourceUid: __expr__ + model: + conditions: + - evaluator: + params: + - 50 + type: gt + operator: + type: and + query: + params: + - B + reducer: + params: [] + type: last + type: query + datasource: + type: __expr__ + uid: __expr__ + expression: A + intervalMs: 1000 + maxDataPoints: 43200 + refId: B + type: threshold + - refId: C + datasourceUid: __expr__ + model: + conditions: + - evaluator: + params: + - 0 + type: gt + operator: + type: and + query: + params: + - C + reducer: + params: [] + type: last + type: query + datasource: + type: __expr__ + uid: __expr__ + expression: B + intervalMs: 1000 + maxDataPoints: 43200 + refId: C + type: threshold + noDataState: OK + execErrState: Error + for: 5m + annotations: + summary: "В очереди Airflow > 50 задач" + description: "airflow_executor_queued_tasks = {{ $values.A }} (превышает 50 в течение 5 минут) — возможна перегрузка" + labels: + severity: warning + + # --- Alert: High Task Failure Rate --- + - uid: airflow_task_failures_high + title: High Task Failure Rate + condition: C + data: + - refId: A + relativeTimeRange: + from: 300 + to: 0 + datasourceUid: prometheus_uid + model: + datasource: + type: prometheus + uid: prometheus_uid + editorMode: code + expr: rate(airflow_task_failures_total[5m]) + instant: false + intervalMs: 1000 + legendFormat: __auto + maxDataPoints: 43200 + range: true + refId: A + - refId: B + datasourceUid: __expr__ + model: + conditions: + - evaluator: + params: + - 0.1 + type: gt + operator: + type: and + query: + params: + - B + reducer: + params: [] + type: last + type: query + datasource: + type: __expr__ + uid: __expr__ + expression: A + intervalMs: 1000 + maxDataPoints: 43200 + refId: B + type: threshold + - refId: C + datasourceUid: __expr__ + model: + conditions: + - evaluator: + params: + - 0 + type: gt + operator: + type: and + query: + params: + - C + reducer: + params: [] + type: last + type: query + datasource: + type: __expr__ + uid: __expr__ + expression: B + intervalMs: 1000 + maxDataPoints: 43200 + refId: C + type: threshold + noDataState: OK + execErrState: Error + for: 3m + annotations: + summary: "Высокий rate падений тасков" + description: "rate(airflow_task_failures_total) = {{ $values.A }} (превышает 0.1 в течение 3 минут) — проверьте логи DAG" + labels: + severity: warning + + # --- Alert: High DAG Parse Time --- + - uid: airflow_parse_time_high + title: High DAG Parse Time + condition: C + data: + - refId: A + relativeTimeRange: + from: 300 + to: 0 + datasourceUid: prometheus_uid + model: + datasource: + type: prometheus + uid: prometheus_uid + editorMode: code + expr: airflow_dag_processing_total_parse_time + instant: false + intervalMs: 1000 + legendFormat: __auto + maxDataPoints: 43200 + range: true + refId: A + - refId: B + datasourceUid: __expr__ + model: + conditions: + - evaluator: + params: + - 30 + type: gt + operator: + type: and + query: + params: + - B + reducer: + params: [] + type: last + type: query + datasource: + type: __expr__ + uid: __expr__ + expression: A + intervalMs: 1000 + maxDataPoints: 43200 + refId: B + type: threshold + - refId: C + datasourceUid: __expr__ + model: + conditions: + - evaluator: + params: + - 0 + type: gt + operator: + type: and + query: + params: + - C + reducer: + params: [] + type: last + type: query + datasource: + type: __expr__ + uid: __expr__ + expression: B + intervalMs: 1000 + maxDataPoints: 43200 + refId: C + type: threshold + noDataState: OK + execErrState: Error + for: 5m + annotations: + summary: "Парсинг DAG занимает > 30 секунд" + description: "airflow_dag_processing_total_parse_time = {{ $values.A }}s — возможно, есть тяжелые DAG" + labels: + severity: info diff --git a/configs/grafana/provisioning/dashboards/airflow-overview.json b/configs/grafana/provisioning/dashboards/airflow-overview.json new file mode 100644 index 0000000..1358707 --- /dev/null +++ b/configs/grafana/provisioning/dashboards/airflow-overview.json @@ -0,0 +1,805 @@ +{ + "annotations": { + "list": [ + { + "builtIn": 1, + "datasource": { + "type": "grafana", + "uid": "-- Grafana --" + }, + "enable": true, + "hide": true, + "iconColor": "rgba(0, 211, 255, 1)", + "name": "Annotations & Alerts", + "type": "dashboard" + } + ] + }, + "description": "Обзор метрик Airflow через statsd-exporter", + "editable": true, + "fiscalYearStartMonth": 0, + "graphTooltip": 0, + "id": null, + "links": [], + "panels": [ + { + "collapsed": false, + "gridPos": { + "h": 1, + "w": 24, + "x": 0, + "y": 0 + }, + "id": 1, + "panels": [], + "title": "Scheduler Health", + "type": "row" + }, + { + "datasource": { + "type": "prometheus", + "uid": "prometheus_uid" + }, + "fieldConfig": { + "defaults": { + "color": { + "mode": "thresholds" + }, + "mappings": [], + "thresholds": { + "mode": "absolute", + "steps": [ + { + "color": "red", + "value": 0 + }, + { + "color": "green", + "value": 1 + } + ] + }, + "unit": "none" + }, + "overrides": [] + }, + "gridPos": { + "h": 4, + "w": 6, + "x": 0, + "y": 1 + }, + "id": 2, + "options": { + "colorMode": "value", + "graphMode": "area", + "justifyMode": "auto", + "orientation": "auto", + "percentChangeColorMode": "standard", + "reduceOptions": { + "calcs": [ + "lastNotNull" + ], + "fields": "", + "values": false + }, + "showPercentChange": false, + "textMode": "auto", + "wideLayout": true + }, + "pluginVersion": "11.0.8", + "targets": [ + { + "datasource": { + "type": "prometheus", + "uid": "prometheus_uid" + }, + "expr": "airflow_dagbag_size", + "refId": "A" + } + ], + "title": "DAG Bag Size", + "type": "stat" + }, + { + "datasource": { + "type": "prometheus", + "uid": "prometheus_uid" + }, + "fieldConfig": { + "defaults": { + "color": { + "mode": "thresholds" + }, + "mappings": [], + "thresholds": { + "mode": "absolute", + "steps": [ + { + "color": "green", + "value": null + }, + { + "color": "yellow", + "value": 10 + }, + { + "color": "red", + "value": 30 + } + ] + }, + "unit": "s" + }, + "overrides": [] + }, + "gridPos": { + "h": 4, + "w": 6, + "x": 6, + "y": 1 + }, + "id": 3, + "options": { + "colorMode": "value", + "graphMode": "area", + "justifyMode": "auto", + "orientation": "auto", + "percentChangeColorMode": "standard", + "reduceOptions": { + "calcs": [ + "lastNotNull" + ], + "fields": "", + "values": false + }, + "showPercentChange": false, + "textMode": "auto", + "wideLayout": true + }, + "pluginVersion": "11.0.8", + "targets": [ + { + "datasource": { + "type": "prometheus", + "uid": "prometheus_uid" + }, + "expr": "airflow_dag_processing_total_parse_time", + "refId": "A" + } + ], + "title": "Parse Time", + "type": "stat" + }, + { + "datasource": { + "type": "prometheus", + "uid": "prometheus_uid" + }, + "fieldConfig": { + "defaults": { + "color": { + "mode": "palette-classic" + }, + "custom": { + "axisBorderShow": false, + "axisCenteredZero": false, + "axisColorMode": "text", + "axisLabel": "", + "axisPlacement": "auto", + "barAlignment": 0, + "drawStyle": "line", + "fillOpacity": 10, + "gradientMode": "none", + "hideFrom": { + "legend": false, + "tooltip": false, + "viz": false + }, + "insertNulls": false, + "lineInterpolation": "linear", + "lineWidth": 1, + "pointSize": 5, + "scaleDistribution": { + "type": "linear" + }, + "showPoints": "never", + "spanNulls": false, + "stacking": { + "group": "A", + "mode": "none" + }, + "thresholdsStyle": { + "mode": "off" + } + }, + "mappings": [], + "thresholds": { + "mode": "absolute", + "steps": [ + { + "color": "green", + "value": null + } + ] + }, + "unit": "ops" + }, + "overrides": [] + }, + "gridPos": { + "h": 8, + "w": 12, + "x": 12, + "y": 1 + }, + "id": 4, + "options": { + "legend": { + "calcs": [], + "displayMode": "list", + "placement": "bottom", + "showLegend": true + }, + "tooltip": { + "mode": "multi", + "sort": "none" + } + }, + "pluginVersion": "11.0.8", + "targets": [ + { + "datasource": { + "type": "prometheus", + "uid": "prometheus_uid" + }, + "expr": "rate(airflow_scheduler_heartbeat_total[5m])", + "refId": "A" + } + ], + "title": "Scheduler Heartbeat Rate", + "type": "timeseries" + }, + { + "collapsed": false, + "gridPos": { + "h": 1, + "w": 24, + "x": 0, + "y": 9 + }, + "id": 5, + "panels": [], + "title": "Executor Status", + "type": "row" + }, + { + "datasource": { + "type": "prometheus", + "uid": "prometheus_uid" + }, + "fieldConfig": { + "defaults": { + "color": { + "mode": "thresholds" + }, + "mappings": [], + "thresholds": { + "mode": "absolute", + "steps": [ + { + "color": "red", + "value": 0 + }, + { + "color": "yellow", + "value": 2 + }, + { + "color": "green", + "value": 8 + } + ] + }, + "unit": "none" + }, + "overrides": [] + }, + "gridPos": { + "h": 4, + "w": 8, + "x": 0, + "y": 10 + }, + "id": 6, + "options": { + "colorMode": "value", + "graphMode": "area", + "justifyMode": "auto", + "orientation": "auto", + "percentChangeColorMode": "standard", + "reduceOptions": { + "calcs": [ + "lastNotNull" + ], + "fields": "", + "values": false + }, + "showPercentChange": false, + "textMode": "auto", + "wideLayout": true + }, + "pluginVersion": "11.0.8", + "targets": [ + { + "datasource": { + "type": "prometheus", + "uid": "prometheus_uid" + }, + "expr": "airflow_executor_open_slots", + "refId": "A" + } + ], + "title": "Open Slots", + "type": "stat" + }, + { + "datasource": { + "type": "prometheus", + "uid": "prometheus_uid" + }, + "fieldConfig": { + "defaults": { + "color": { + "mode": "thresholds" + }, + "mappings": [], + "thresholds": { + "mode": "absolute", + "steps": [ + { + "color": "green", + "value": null + }, + { + "color": "yellow", + "value": 20 + }, + { + "color": "red", + "value": 50 + } + ] + }, + "unit": "none" + }, + "overrides": [] + }, + "gridPos": { + "h": 4, + "w": 8, + "x": 8, + "y": 10 + }, + "id": 7, + "options": { + "colorMode": "value", + "graphMode": "area", + "justifyMode": "auto", + "orientation": "auto", + "percentChangeColorMode": "standard", + "reduceOptions": { + "calcs": [ + "lastNotNull" + ], + "fields": "", + "values": false + }, + "showPercentChange": false, + "textMode": "auto", + "wideLayout": true + }, + "pluginVersion": "11.0.8", + "targets": [ + { + "datasource": { + "type": "prometheus", + "uid": "prometheus_uid" + }, + "expr": "airflow_executor_queued_tasks", + "refId": "A" + } + ], + "title": "Queued Tasks", + "type": "stat" + }, + { + "datasource": { + "type": "prometheus", + "uid": "prometheus_uid" + }, + "fieldConfig": { + "defaults": { + "color": { + "mode": "thresholds" + }, + "mappings": [], + "thresholds": { + "mode": "absolute", + "steps": [ + { + "color": "blue", + "value": null + } + ] + }, + "unit": "none" + }, + "overrides": [] + }, + "gridPos": { + "h": 4, + "w": 8, + "x": 16, + "y": 10 + }, + "id": 8, + "options": { + "colorMode": "value", + "graphMode": "area", + "justifyMode": "auto", + "orientation": "auto", + "percentChangeColorMode": "standard", + "reduceOptions": { + "calcs": [ + "lastNotNull" + ], + "fields": "", + "values": false + }, + "showPercentChange": false, + "textMode": "auto", + "wideLayout": true + }, + "pluginVersion": "11.0.8", + "targets": [ + { + "datasource": { + "type": "prometheus", + "uid": "prometheus_uid" + }, + "expr": "airflow_executor_running_tasks", + "refId": "A" + } + ], + "title": "Running Tasks", + "type": "stat" + }, + { + "datasource": { + "type": "prometheus", + "uid": "prometheus_uid" + }, + "fieldConfig": { + "defaults": { + "color": { + "mode": "palette-classic" + }, + "custom": { + "axisBorderShow": false, + "axisCenteredZero": false, + "axisColorMode": "text", + "axisLabel": "", + "axisPlacement": "auto", + "barAlignment": 0, + "drawStyle": "line", + "fillOpacity": 10, + "gradientMode": "none", + "hideFrom": { + "legend": false, + "tooltip": false, + "viz": false + }, + "insertNulls": false, + "lineInterpolation": "linear", + "lineWidth": 1, + "pointSize": 5, + "scaleDistribution": { + "type": "linear" + }, + "showPoints": "never", + "spanNulls": false, + "stacking": { + "group": "A", + "mode": "none" + }, + "thresholdsStyle": { + "mode": "off" + } + }, + "mappings": [], + "thresholds": { + "mode": "absolute", + "steps": [ + { + "color": "green", + "value": null + } + ] + }, + "unit": "none" + }, + "overrides": [] + }, + "gridPos": { + "h": 8, + "w": 24, + "x": 0, + "y": 14 + }, + "id": 9, + "options": { + "legend": { + "calcs": [], + "displayMode": "list", + "placement": "bottom", + "showLegend": true + }, + "tooltip": { + "mode": "multi", + "sort": "none" + } + }, + "pluginVersion": "11.0.8", + "targets": [ + { + "datasource": { + "type": "prometheus", + "uid": "prometheus_uid" + }, + "expr": "airflow_executor_queued_tasks", + "legendFormat": "Queued", + "refId": "A" + }, + { + "datasource": { + "type": "prometheus", + "uid": "prometheus_uid" + }, + "expr": "airflow_executor_running_tasks", + "legendFormat": "Running", + "refId": "B" + }, + { + "datasource": { + "type": "prometheus", + "uid": "prometheus_uid" + }, + "expr": "airflow_executor_open_slots", + "legendFormat": "Open Slots", + "refId": "C" + } + ], + "title": "Executor Tasks Over Time", + "type": "timeseries" + }, + { + "collapsed": false, + "gridPos": { + "h": 1, + "w": 24, + "x": 0, + "y": 22 + }, + "id": 10, + "panels": [], + "title": "Task Performance", + "type": "row" + }, + { + "datasource": { + "type": "prometheus", + "uid": "prometheus_uid" + }, + "fieldConfig": { + "defaults": { + "color": { + "mode": "palette-classic" + }, + "custom": { + "axisBorderShow": false, + "axisCenteredZero": false, + "axisColorMode": "text", + "axisLabel": "", + "axisPlacement": "auto", + "barAlignment": 0, + "drawStyle": "line", + "fillOpacity": 10, + "gradientMode": "none", + "hideFrom": { + "legend": false, + "tooltip": false, + "viz": false + }, + "insertNulls": false, + "lineInterpolation": "linear", + "lineWidth": 1, + "pointSize": 5, + "scaleDistribution": { + "type": "linear" + }, + "showPoints": "never", + "spanNulls": false, + "stacking": { + "group": "A", + "mode": "none" + }, + "thresholdsStyle": { + "mode": "off" + } + }, + "mappings": [], + "thresholds": { + "mode": "absolute", + "steps": [ + { + "color": "green", + "value": null + } + ] + }, + "unit": "s" + }, + "overrides": [] + }, + "gridPos": { + "h": 8, + "w": 12, + "x": 0, + "y": 23 + }, + "id": 11, + "options": { + "legend": { + "calcs": [], + "displayMode": "list", + "placement": "bottom", + "showLegend": true + }, + "tooltip": { + "mode": "multi", + "sort": "none" + } + }, + "pluginVersion": "11.0.8", + "targets": [ + { + "datasource": { + "type": "prometheus", + "uid": "prometheus_uid" + }, + "expr": "rate(airflow_task_duration_seconds_sum[5m]) / rate(airflow_task_duration_seconds_count[5m])", + "legendFormat": "{{dag_id}} / {{task_id}}", + "refId": "A" + } + ], + "title": "Task Duration (avg)", + "type": "timeseries" + }, + { + "datasource": { + "type": "prometheus", + "uid": "prometheus_uid" + }, + "fieldConfig": { + "defaults": { + "color": { + "mode": "palette-classic" + }, + "custom": { + "axisBorderShow": false, + "axisCenteredZero": false, + "axisColorMode": "text", + "axisLabel": "", + "axisPlacement": "auto", + "barAlignment": 0, + "drawStyle": "line", + "fillOpacity": 10, + "gradientMode": "none", + "hideFrom": { + "legend": false, + "tooltip": false, + "viz": false + }, + "insertNulls": false, + "lineInterpolation": "linear", + "lineWidth": 1, + "pointSize": 5, + "scaleDistribution": { + "type": "linear" + }, + "showPoints": "never", + "spanNulls": false, + "stacking": { + "group": "A", + "mode": "none" + }, + "thresholdsStyle": { + "mode": "off" + } + }, + "mappings": [], + "thresholds": { + "mode": "absolute", + "steps": [ + { + "color": "green", + "value": null + } + ] + }, + "unit": "ops" + }, + "overrides": [] + }, + "gridPos": { + "h": 8, + "w": 12, + "x": 12, + "y": 23 + }, + "id": 12, + "options": { + "legend": { + "calcs": [], + "displayMode": "list", + "placement": "bottom", + "showLegend": true + }, + "tooltip": { + "mode": "multi", + "sort": "none" + } + }, + "pluginVersion": "11.0.8", + "targets": [ + { + "datasource": { + "type": "prometheus", + "uid": "prometheus_uid" + }, + "expr": "rate(airflow_task_failures_total[5m])", + "legendFormat": "Failures: {{dag_id}} / {{task_id}}", + "refId": "A" + }, + { + "datasource": { + "type": "prometheus", + "uid": "prometheus_uid" + }, + "expr": "rate(airflow_task_success_total[5m])", + "legendFormat": "Success: {{dag_id}} / {{task_id}}", + "refId": "B" + } + ], + "title": "Task Failures vs Success Rate", + "type": "timeseries" + } + ], + "refresh": "10s", + "schemaVersion": 39, + "tags": ["airflow", "orchestration"], + "templating": { + "list": [] + }, + "time": { + "from": "now-1h", + "to": "now" + }, + "timepicker": {}, + "timezone": "Europe/Moscow", + "title": "Airflow Overview", + "uid": "airflow-overview", + "version": 1, + "weekStart": "" +} diff --git a/configs/prometheus.yml b/configs/prometheus.yml index 2e7fb3d..bda012d 100644 --- a/configs/prometheus.yml +++ b/configs/prometheus.yml @@ -18,4 +18,13 @@ scrape_configs: - targets: ["kafka-exporter:9308"] labels: instance: Kafka-1 + honor_labels: true + + # Airflow metrics via statsd-exporter (StatsD → Prometheus) + - job_name: "airflow" + metrics_path: "/metrics" + static_configs: + - targets: ["statsd-exporter:9102"] + labels: + instance: Airflow-1 honor_labels: true \ No newline at end of file diff --git a/configs/statsd_mapping.yml b/configs/statsd_mapping.yml new file mode 100644 index 0000000..4648f0a --- /dev/null +++ b/configs/statsd_mapping.yml @@ -0,0 +1,74 @@ +# Mapping StatsD → Prometheus для Airflow +# См. документацию: https://github.com/prometheus/statsd_exporter#metric-mapping-and-configuration + +mappings: + # DAG processing metrics + - match: "airflow.dag_processing.total_parse_time" + name: "airflow_dag_processing_total_parse_time" + help: "Total time to parse all DAGs" + type: gauge + + - match: "airflow.dagbag_size" + name: "airflow_dagbag_size" + help: "DAGs in the DagBag" + type: gauge + + # Executor metrics + - match: "airflow.executor.open_slots" + name: "airflow_executor_open_slots" + help: "Open slots on executor" + type: gauge + + - match: "airflow.executor.queued_tasks" + name: "airflow_executor_queued_tasks" + help: "Queued tasks on executor" + type: gauge + + - match: "airflow.executor.running_tasks" + name: "airflow_executor_running_tasks" + help: "Running tasks on executor" + type: gauge + + # Scheduler metrics (counter → _total суффикс по Prometheus convention) + - match: "airflow.scheduler_heartbeat" + name: "airflow_scheduler_heartbeat_total" + help: "Scheduler heartbeats" + type: counter + + - match: "airflow.scheduler.critical_section_busy" + name: "airflow_scheduler_critical_section_busy" + help: "Scheduler critical section busy" + type: gauge + + # Task duration (timer → _seconds суффикс) + - match: "airflow.dag.*.task.*.duration" + name: "airflow_task_duration_seconds" + help: "Task duration" + type: timer + labels: + dag_id: "$1" + task_id: "$2" + + # Task failures (counter → _total суффикс) + - match: "airflow.dag.*.task.*.failures" + name: "airflow_task_failures_total" + help: "Task failures" + type: counter + labels: + dag_id: "$1" + task_id: "$2" + + # Task success (counter → _total суффикс) + - match: "airflow.dag.*.task.*.success" + name: "airflow_task_success_total" + help: "Task success" + type: counter + labels: + dag_id: "$1" + task_id: "$2" + + # Catch-all для остальных airflow метрик + - match: "airflow.*.*" + name: "airflow_${1}_${2}" + help: "Airflow metric $1.$2" + type: gauge diff --git a/docker-compose.yml b/docker-compose.yml index 23485df..12d07be 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -5,6 +5,11 @@ x-airflow-env: &airflow-default-env AIRFLOW__WEBSERVER__SECRET_KEY: ${AIRFLOW_SECRET_KEY:-replace-me-with-random-string} # ClickHouse connection для ETL AIRFLOW_CONN_CLICKHOUSE_DEFAULT: clickhouse://default:123456@clickhouse:9000/default + # StatsD для мониторинга (метрики → statsd-exporter → Prometheus) + AIRFLOW__METRICS__STATSD_ON: "True" + AIRFLOW__METRICS__STATSD_HOST: "statsd-exporter" + AIRFLOW__METRICS__STATSD_PORT: "8125" + AIRFLOW__METRICS__STATSD_PREFIX: "airflow" services: @@ -242,6 +247,24 @@ services: - kafka restart: unless-stopped + # StatsD Exporter для Airflow метрик (StatsD → Prometheus) + statsd-exporter: + image: prom/statsd-exporter:v0.27.1 + command: [ + "--statsd.listen-udp=:8125", + "--statsd.listen-tcp=", + "--web.listen-address=:9102", + "--statsd.mapping-config=/tmp/statsd_mapping.yml" + ] + ports: + - "9102:9102" # Prometheus metrics endpoint + - "8125:8125/udp" # StatsD receive port + volumes: + - ./configs/statsd_mapping.yml:/tmp/statsd_mapping.yml:ro + networks: + - cs_dwh + restart: unless-stopped + networks: cs_dwh: driver: bridge diff --git a/docs/OPERATIONS.md b/docs/OPERATIONS.md index 1e344d9..ed44101 100644 --- a/docs/OPERATIONS.md +++ b/docs/OPERATIONS.md @@ -97,7 +97,7 @@ docker compose exec -T clickhouse clickhouse-client --user=default --password=12 make reload-monitoring # Или вручную: -docker compose up -d prometheus grafana kafka-exporter +docker compose up -d prometheus grafana kafka-exporter statsd-exporter docker compose restart prometheus curl -s -u admin:admin -X POST http://localhost:3000/api/admin/provisioning/datasources/reload curl -s -u admin:admin -X POST http://localhost:3000/api/admin/provisioning/dashboards/reload @@ -106,7 +106,7 @@ curl -s -u admin:admin -X POST http://localhost:3000/api/admin/provisioning/aler Если менялся `configs/prometheus_ch.xml`: `docker compose restart clickhouse`. -### Prometheus + Grafana для ClickHouse +### Prometheus + Grafana для ClickHouse, Kafka и Airflow Стек мониторинга поднимается вместе с остальной инфраструктурой: @@ -120,12 +120,14 @@ curl -s http://localhost:9090/api/v1/targets | grep -o '"health":"[^"]*"' ### Конфигурация -- **Prometheus** (`configs/prometheus.yml`): скрейп ClickHouse на порту `9126/metrics` -- **ClickHouse** (`configs/prometheus_ch.xml`): включён экспорт метрик в формате Prometheus +- **ClickHouse**: встроенный Prometheus endpoint (`/metrics` на порту `9126`) +- **Kafka**: через `kafka-exporter` (порт `9308`) +- **Airflow**: через `statsd-exporter` (StatsD → Prometheus, порт `9102`) + - Airflow отправляет метрики в StatsD-формате на `statsd-exporter:8125` + - Mapping конфигурация: `configs/statsd_mapping.yml` - **Grafana provisioning** (`configs/grafana/provisioning/`): - - Datasource Prometheus автоматически настроен - - Dashboard "ClickHouse Overview" загружается при старте - - Alert rules для ClickHouse загружаются при старте + - Дашборды: ClickHouse Overview, Kafka Overview, Airflow Overview + - Алерты: ClickHouse, Kafka, Airflow ### После `git pull`: быстрый апдейт мониторинга diff --git a/plans/monitoring_airflow_plan.md b/plans/monitoring_airflow_plan.md new file mode 100644 index 0000000..d51a966 --- /dev/null +++ b/plans/monitoring_airflow_plan.md @@ -0,0 +1,449 @@ +# План подключения Airflow к мониторингу + +> Дата создания: 2026-02-08 +> Источник: Airflow 2.10.5 via statsd_exporter +> Проверка API: Context7 (`/prometheus/docs` для scrape_configs) + +--- + +## 1. Обзор + +Подключаем мониторинг Airflow через **statsd_exporter**. Airflow встроенно поддерживает отправку метрик в StatsD-формате, а statsd_exporter конвертирует их в Prometheus-формат. + +**Метрики, которые будем собирать:** +- `airflow_dag_processing_total_parse_time` — время парсинга DAG +- `airflow_dagbag_size` — количество DAG в DagBag +- `airflow_executor_open_slots` — доступные слоты executor +- `airflow_executor_queued_tasks` — задачи в очереди +- `airflow_executor_running_tasks` — запущенные задачи +- `airflow_task_duration{task_id, dag_id}` — длительность выполнения тасков +- `airflow_task_failures` — количество падений тасков +- `airflow_task_success` — успешные выполнения +- `airflow_scheduler_heartbeat` — heartbeats шедулера + +--- + +## 2. Изменения в инфраструктуре + +### 2.1. docker-compose.yml + +Добавить сервис `statsd-exporter`: + +```yaml + statsd-exporter: + image: prom/statsd-exporter:v0.27.1 + command: [ + "--statsd.listen-udp=:8125", + "--statsd.listen-tcp=", + "--web.listen-address=:9102", + "--statsd.mapping-config=/tmp/statsd_mapping.yml" + ] + ports: + - "9102:9102" # Prometheus metrics endpoint + - "8125:8125/udp" # StatsD receive port + volumes: + - ./configs/statsd_mapping.yml:/tmp/statsd_mapping.yml:ro + networks: + - cs_dwh +``` + +Добавить переменные окружения в `airflow-default-env`: + +```yaml +x-airflow-env: &airflow-default-env + # ... существующие переменные ... + # StatsD для мониторинга + AIRFLOW__METRICS__STATSD_ON: "True" + AIRFLOW__METRICS__STATSD_HOST: "statsd-exporter" + AIRFLOW__METRICS__STATSD_PORT: "8125" + AIRFLOW__METRICS__STATSD_PREFIX: "airflow" +``` + +**Почему `prom/statsd-exporter:v0.27.1`:** +- Стабильная версия с поддержкой mapping-конфигурации +- Официальный образ от Prometheus team +- Проверено на совместимость с Airflow 2.10.x + +### 2.2. configs/statsd_mapping.yml + +Создать mapping-конфигурацию для конвертации StatsD метрик в Prometheus: + +```yaml +# Mapping StatsD → Prometheus для Airflow +mappings: + # DAG processing metrics + - match: "airflow.dag_processing.total_parse_time" + name: "airflow_dag_processing_total_parse_time" + help: "Total time to parse all DAGs" + type: gauge + + - match: "airflow.dagbag_size" + name: "airflow_dagbag_size" + help: "DAGs in the DagBag" + type: gauge + + # Executor metrics + - match: "airflow.executor.open_slots" + name: "airflow_executor_open_slots" + help: "Open slots on executor" + type: gauge + + - match: "airflow.executor.queued_tasks" + name: "airflow_executor_queued_tasks" + help: "Queued tasks on executor" + type: gauge + + - match: "airflow.executor.running_tasks" + name: "airflow_executor_running_tasks" + help: "Running tasks on executor" + type: gauge + + # Task metrics с labels + - match: "airflow.operator_failures_*" + name: "airflow_operator_failures" + help: "Operator failures" + type: counter + labels: + operator: "$1" + + - match: "airflow.operator_successes_*" + name: "airflow_operator_successes" + help: "Operator successes" + type: counter + labels: + operator: "$1" + + # Task duration by dag_id и task_id + - match: "airflow.dag.*.task.*.duration" + name: "airflow_task_duration" + help: "Task duration by dag_id and task_id" + type: timer + labels: + dag_id: "$1" + task_id: "$2" + + # Task failures/success by dag_id и task_id + - match: "airflow.dag.*.task.*.failures" + name: "airflow_task_failures" + help: "Task failures by dag_id and task_id" + type: counter + labels: + dag_id: "$1" + task_id: "$2" + + - match: "airflow.dag.*.task.*.success" + name: "airflow_task_success" + help: "Task success by dag_id and task_id" + type: counter + labels: + dag_id: "$1" + task_id: "$2" + + # Scheduler metrics + - match: "airflow.scheduler_heartbeat" + name: "airflow_scheduler_heartbeat" + help: "Scheduler heartbeats" + type: counter + + - match: "airflow.scheduler.critical_section_busy" + name: "airflow_scheduler_critical_section_busy" + help: "Scheduler critical section busy" + type: gauge + + # Catch-all для остальных airflow метрик + - match: "airflow.*" + name: "airflow_${1}" + help: "Airflow metric $1" + type: gauge +``` + +### 2.3. configs/prometheus.yml + +Добавить job для скрейпа statsd-exporter: + +```yaml +scrape_configs: + - job_name: "clickhouse" + metrics_path: "/metrics" + static_configs: + - targets: ["clickhouse:9126"] + labels: + instance: Clickhouse-1 + honor_labels: true + + - job_name: "kafka" + metrics_path: "/metrics" + static_configs: + - targets: ["kafka-exporter:9308"] + labels: + instance: Kafka-1 + honor_labels: true + + # Новый job для Airflow + - job_name: "airflow" + metrics_path: "/metrics" + static_configs: + - targets: ["statsd-exporter:9102"] + labels: + instance: Airflow-1 + honor_labels: true +``` + +--- + +## 3. Дашборд Grafana + +### 3.1. Создать файл: `configs/grafana/provisioning/dashboards/airflow-overview.json` + +Основные панели дашборда: + +| Панель | PromQL запрос | Описание | +|--------|--------------|----------| +| DAG Bag Size | `airflow_dagbag_size` | Количество DAG в системе | +| Parse Time | `airflow_dag_processing_total_parse_time` | Время парсинга DAG | +| Executor Slots | `airflow_executor_open_slots` | Доступные слоты | +| Queued Tasks | `airflow_executor_queued_tasks` | Задачи в очереди | +| Running Tasks | `airflow_executor_running_tasks` | Выполняемые задачи | +| Task Duration | `rate(airflow_task_duration_sum[5m]) / rate(airflow_task_duration_count[5m])` | Средняя длительность тасков | +| Task Failures | `rate(airflow_task_failures[5m])` | Rate падений по таскам | +| Scheduler Heartbeat | `rate(airflow_scheduler_heartbeat[5m])` | Активность шедулера | + +### 3.2. Структура дашборда (основные секции) + +```json +{ + "dashboard": { + "id": null, + "uid": "airflow-overview", + "title": "Airflow Overview", + "tags": ["airflow", "orchestration"], + "timezone": "Europe/Moscow", + "schemaVersion": 36, + "refresh": "10s", + "panels": [ + // Row 1: Scheduler Health + // - DAG Bag Size (stat) + // - Parse Time (gauge) + // - Scheduler Heartbeat (graph) + + // Row 2: Executor Status + // - Open Slots (stat) + // - Queued Tasks (stat) + // - Running Tasks (stat) + // - Tasks distribution (graph) + + // Row 3: Task Performance + // - Task Duration by DAG (heatmap или graph) + // - Task Failures Rate (graph) + // - Task Success Rate (graph) + + // Row 4: Per-DAG Details + // - Top DAGs by runtime (table) + // - Failed tasks by DAG (table) + ] + } +} +``` + +--- + +## 4. Алерты Grafana + +### 4.1. Создать файл: `configs/grafana/provisioning/alerting/airflow-alert-rules.yml` + +```yaml +apiVersion: 1 +groups: + - orgId: 1 + name: airflow_alerts + folder: Airflow + interval: 30s + rules: + # Alert: Scheduler не отправляет heartbeats + - uid: airflow_scheduler_down + title: Airflow Scheduler Down + condition: B + data: + - refId: A + relativeTimeRange: + from: 300 + to: 0 + datasourceUid: prometheus_uid + model: + expr: rate(airflow_scheduler_heartbeat[5m]) + instant: true + - refId: B + relativeTimeRange: + from: 0 + to: 0 + datasourceUid: __expr__ + model: + type: threshold + expression: A + conditions: + - evaluator: + type: lt + params: [0.1] + noDataState: NoData + execErrState: Error + for: 2m + annotations: + summary: "Airflow scheduler не отправляет heartbeats" + description: "Scheduler possible down — rate(airflow_scheduler_heartbeat) < 0.1 в течение 2 минут" + labels: + severity: critical + + # Alert: В очереди слишком много задач + - uid: airflow_queue_high + title: Airflow Queue Backlog + condition: B + data: + - refId: A + datasourceUid: prometheus_uid + model: + expr: airflow_executor_queued_tasks + instant: true + - refId: B + datasourceUid: __expr__ + model: + type: threshold + expression: A + conditions: + - evaluator: + type: gt + params: [50] + noDataState: OK + execErrState: Error + for: 5m + annotations: + summary: "В очереди Airflow > 50 задач" + description: "airflow_executor_queued_tasks превышает 50 в течение 5 минут — возможна перегрузка" + labels: + severity: warning + + # Alert: Много падений тасков + - uid: airflow_task_failures_high + title: High Task Failure Rate + condition: B + data: + - refId: A + datasourceUid: prometheus_uid + model: + expr: rate(airflow_task_failures[5m]) + instant: true + - refId: B + datasourceUid: __expr__ + model: + type: threshold + expression: A + conditions: + - evaluator: + type: gt + params: [0.1] + noDataState: OK + execErrState: Error + for: 3m + annotations: + summary: "Высокий rate падений тасков" + description: "rate(airflow_task_failures) > 0.1 в течение 3 минут — проверьте логи DAG" + labels: + severity: warning + + # Alert: DAG parsing занимает слишком много времени + - uid: airflow_parse_time_high + title: High DAG Parse Time + condition: B + data: + - refId: A + datasourceUid: prometheus_uid + model: + expr: airflow_dag_processing_total_parse_time + instant: true + - refId: B + datasourceUid: __expr__ + model: + type: threshold + expression: A + conditions: + - evaluator: + type: gt + params: [30] + noDataState: OK + execErrState: Error + for: 5m + annotations: + summary: "Парсинг DAG занимает > 30 секунд" + description: "Возможно, есть тяжелые DAG — оптимизируйте или разбейте на под-DAG" + labels: + severity: info +``` + +--- + +## 5. Проверка после деплоя + +```bash +# 1. Перезапуск стека с новым сервисом +make reload-monitoring + +# 2. Проверка, что statsd-exporter поднялся +curl -s http://localhost:9102/metrics | grep airflow + +# 3. Проверка таргета в Prometheus +curl -s http://localhost:9090/api/v1/targets | grep -A5 airflow + +# 4. Проверка дашборда в Grafana +open http://localhost:3000/d/airflow-overview + +# 5. Проверка алертов +curl -s -u admin:admin http://localhost:3000/api/v1/provisioning/alert-rules | grep airflow +``` + +--- + +## 6. Troubleshooting + +### Нет метрик airflow_* в Prometheus + +1. Проверить env vars в Airflow: + ```bash + docker compose exec airflow-webserver env | grep STATSD + ``` + Ожидается: `AIRFLOW__METRICS__STATSD_ON=True` + +2. Проверить доступность statsd-exporter: + ```bash + docker compose exec airflow-webserver nc -zv statsd-exporter 8125 + ``` + +3. Проверить метрики в statsd-exporter напрямую: + ```bash + curl -s http://localhost:9102/metrics + ``` + +### StatsD метрики не мапятся + +Проверить формат mapping-файла: +```bash +docker compose logs statsd-exporter | grep -i "mapping\|error" +``` + +### Airflow не отправляет метрики + +Проверить конфигурацию внутри Airflow: +```bash +docker compose exec airflow-webserver airflow config get-value metrics statsd_on +docker compose exec airflow-webserver airflow config get-value metrics statsd_host +``` + +--- + +## 7. Зависимости и порядок применения + +1. Создать `configs/statsd_mapping.yml` +2. Добавить `statsd-exporter` в `docker-compose.yml` +3. Добавить StatsD env vars в `airflow-default-env` +4. Обновить `configs/prometheus.yml` (добавить job airflow) +5. Создать дашборд `configs/grafana/provisioning/dashboards/airflow-overview.json` +6. Создать алерты `configs/grafana/provisioning/alerting/airflow-alert-rules.yml` +7. Запустить `make reload-monitoring`