configs/grafana/provisioning/alerting/airflow-alert-rules.yml

@@ -33,9 +33,8 @@ groups:
             model:
               conditions:
                 - evaluator:
-                    params:
+                    params: []
-                      - 0.1
+                    type: gt
-                    type: lt
                   operator:
                     type: and
                   query:
@@ -51,16 +50,17 @@ groups:
               expression: A
               intervalMs: 1000
               maxDataPoints: 43200
+              reducer: last
               refId: B
-              type: threshold
+              type: reduce
           - refId: C
             datasourceUid: __expr__
             model:
               conditions:
                 - evaluator:
                     params:
-                      - 0
+                      - 0.1
-                    type: gt
+                    type: lt
                   operator:
                     type: and
                   query:
@@ -114,8 +114,7 @@ groups:
             model:
               conditions:
                 - evaluator:
-                    params:
+                    params: []
-                      - 50
                     type: gt
                   operator:
                     type: and
@@ -132,15 +131,16 @@ groups:
               expression: A
               intervalMs: 1000
               maxDataPoints: 43200
+              reducer: last
               refId: B
-              type: threshold
+              type: reduce
           - refId: C
             datasourceUid: __expr__
             model:
               conditions:
                 - evaluator:
                     params:
-                      - 0
+                      - 50
                     type: gt
                   operator:
                     type: and
@@ -164,7 +164,7 @@ groups:
         for: 5m
         annotations:
           summary: "В очереди Airflow > 50 задач"
-          description: "airflow_executor_queued_tasks = {{ $values.A }} (превышает 50 в течение 5 минут) — возможна перегрузка"
+          description: "airflow_executor_queued_tasks = {{ $values.B.Value }} (превышает 50 в течение 5 минут) — возможна перегрузка"
         labels:
           severity: warning
 
@@ -195,8 +195,7 @@ groups:
             model:
               conditions:
                 - evaluator:
-                    params:
+                    params: []
-                      - 0.1
                     type: gt
                   operator:
                     type: and
@@ -213,15 +212,16 @@ groups:
               expression: A
               intervalMs: 1000
               maxDataPoints: 43200
+              reducer: last
               refId: B
-              type: threshold
+              type: reduce
           - refId: C
             datasourceUid: __expr__
             model:
               conditions:
                 - evaluator:
                     params:
-                      - 0
+                      - 0.1
                     type: gt
                   operator:
                     type: and
@@ -245,7 +245,7 @@ groups:
         for: 3m
         annotations:
           summary: "Высокий rate падений тасков"
-          description: "rate(airflow_task_failures_total) = {{ $values.A }} (превышает 0.1 в течение 3 минут) — проверьте логи DAG"
+          description: "rate(airflow_task_failures_total) = {{ $values.B.Value }} (превышает 0.1 в течение 3 минут) — проверьте логи DAG"
         labels:
           severity: warning
 
@@ -276,8 +276,7 @@ groups:
             model:
               conditions:
                 - evaluator:
-                    params:
+                    params: []
-                      - 30
                     type: gt
                   operator:
                     type: and
@@ -294,15 +293,16 @@ groups:
               expression: A
               intervalMs: 1000
               maxDataPoints: 43200
+              reducer: last
               refId: B
-              type: threshold
+              type: reduce
           - refId: C
             datasourceUid: __expr__
             model:
               conditions:
                 - evaluator:
                     params:
-                      - 0
+                      - 30
                     type: gt
                   operator:
                     type: and
@@ -326,6 +326,6 @@ groups:
         for: 5m
         annotations:
           summary: "Парсинг DAG занимает > 30 секунд"
-          description: "airflow_dag_processing_total_parse_time = {{ $values.A }}s — возможно, есть тяжелые DAG"
+          description: "airflow_dag_processing_total_parse_time = {{ $values.B.Value }}s — возможно, есть тяжелые DAG"
         labels:
           severity: info

configs/grafana/provisioning/alerting/kafka-alert-rules.yml

@@ -33,9 +33,8 @@ groups:
             model:
               conditions:
                 - evaluator:
-                    params:
+                    params: []
-                      - 1
+                    type: gt
-                    type: lt
                   operator:
                     type: and
                   query:
@@ -51,16 +50,17 @@ groups:
               expression: A
               intervalMs: 1000
               maxDataPoints: 43200
+              reducer: last
               refId: B
-              type: threshold
+              type: reduce
           - refId: C
             datasourceUid: __expr__
             model:
               conditions:
                 - evaluator:
                     params:
-                      - 0
+                      - 1
-                    type: gt
+                    type: lt
                   operator:
                     type: and
                   query:
@@ -83,7 +83,7 @@ groups:
         for: 1m
         annotations:
           summary: "Kafka broker недоступен"
-          description: "Количество доступных брокеров: {{ $values.A }} (ожидается ≥ 1)"
+          description: "Количество доступных брокеров: {{ $values.B.Value }} (ожидается ≥ 1)"
         labels:
           severity: critical
 
@@ -114,8 +114,7 @@ groups:
             model:
               conditions:
                 - evaluator:
-                    params:
+                    params: []
-                      - 10000
                     type: gt
                   operator:
                     type: and
@@ -132,15 +131,16 @@ groups:
               expression: A
               intervalMs: 1000
               maxDataPoints: 43200
+              reducer: last
               refId: B
-              type: threshold
+              type: reduce
           - refId: C
             datasourceUid: __expr__
             model:
               conditions:
                 - evaluator:
                     params:
-                      - 0
+                      - 10000
                     type: gt
                   operator:
                     type: and
@@ -164,7 +164,7 @@ groups:
         for: 5m
         annotations:
           summary: "Высокий лаг консьюмера {{ $labels.consumergroup }}"
-          description: "Lag для {{ $labels.topic }}:{{ $labels.partition }} составляет {{ $values.A }} сообщений (порог: 10000)"
+          description: "Lag для {{ $labels.topic }}:{{ $labels.partition }} составляет {{ $values.B.Value }} сообщений (порог: 10000)"
         labels:
           severity: warning
 
@@ -195,9 +195,8 @@ groups:
             model:
               conditions:
                 - evaluator:
-                    params:
+                    params: []
-                      - 0.1
+                    type: gt
-                    type: lt
                   operator:
                     type: and
                   query:
@@ -213,16 +212,17 @@ groups:
               expression: A
               intervalMs: 1000
               maxDataPoints: 43200
+              reducer: last
               refId: B
-              type: threshold
+              type: reduce
           - refId: C
             datasourceUid: __expr__
             model:
               conditions:
                 - evaluator:
                     params:
-                      - 0
+                      - 0.1
-                    type: gt
+                    type: lt
                   operator:
                     type: and
                   query:

docs/course/lessons/05_monitoring.md

@@ -304,9 +304,10 @@ docker compose ps airflow-scheduler
 docker compose stop airflow-scheduler
 ```
 
-Подожди 2-3 минуты. У правила `Airflow Scheduler Down` стоит окно `for: 2m`, поэтому алерт не
+Подожди около 5 минут. После остановки scheduler значение `rate(...[5m])` снижается постепенно.
-обязан покраснеть мгновенно. Это нормальное поведение: мониторинг защищается от коротких
+При интервале heartbeat в 5 секунд порог будет пройден примерно через 2,5 минуты. Затем Grafana
-миганий.
+ещё 2 минуты держит правило в состоянии `Pending` из-за `for: 2m`. Точное время зависит от
+моментов сбора метрик и вычисления правила.
 
 Что ожидаем:
 
@@ -326,7 +327,8 @@ docker compose stop airflow-scheduler
 docker compose up -d airflow-scheduler
 ```
 
-Подожди ещё 2-3 минуты и проверь:
+Подожди ещё около 3 минут: пятиминутному окну `rate` нужно накопить достаточно новых сигналов
+heartbeat. Затем проверь:
 
 - `docker compose ps airflow-scheduler` показывает running/up;
 - **Scheduler Heartbeat Rate** снова растёт;

generator/requirements.txt

@@ -6,6 +6,7 @@ prometheus-client==0.21.1
 
 # Тестирование
 pytest==8.3.4
+PyYAML==6.0.2
 
 # Утилиты
 python-json-logger==2.0.7

tests/test_grafana_alert_rules.py

@@ -0,0 +1,68 @@
+import re
+from pathlib import Path
+
+import yaml
+
+
+REPO_ROOT = Path(__file__).resolve().parents[1]
+ALERTING_DIR = REPO_ROOT / "configs" / "grafana" / "provisioning" / "alerting"
+
+EXPECTED_THRESHOLDS = {
+    "airflow-alert-rules.yml": {
+        "airflow_scheduler_down": ("lt", [0.1]),
+        "airflow_queue_high": ("gt", [50]),
+        "airflow_task_failures_high": ("gt", [0.1]),
+        "airflow_parse_time_high": ("gt", [30]),
+    },
+    "kafka-alert-rules.yml": {
+        "kafka_broker_down": ("lt", [1]),
+        "kafka_consumer_lag_high": ("gt", [10000]),
+        "kafka_no_messages_produced": ("lt", [0.1]),
+    },
+}
+
+
+def _rules_by_uid(path):
+    document = yaml.safe_load(path.read_text(encoding="utf-8"))
+    return {
+        rule["uid"]: rule
+        for group in document["groups"]
+        for rule in group["rules"]
+    }
+
+
+def test_range_alerts_reduce_series_before_threshold():
+    """Диапазонный запрос сводится к числу до сравнения с порогом."""
+    for filename, expected_thresholds in EXPECTED_THRESHOLDS.items():
+        rules = _rules_by_uid(ALERTING_DIR / filename)
+
+        assert set(rules) == set(expected_thresholds)
+
+        for uid, (comparison, params) in expected_thresholds.items():
+            rule = rules[uid]
+            stages = {stage["refId"]: stage for stage in rule["data"]}
+
+            threshold = stages[rule["condition"]]["model"]
+            assert threshold["type"] == "threshold", uid
+
+            reduce_stage = stages[threshold["expression"]]["model"]
+            assert reduce_stage["type"] == "reduce", uid
+            assert reduce_stage["reducer"] == "last", uid
+
+            query = stages[reduce_stage["expression"]]["model"]
+            assert query["range"] is True, uid
+
+            evaluator = threshold["conditions"][0]["evaluator"]
+            assert evaluator == {"type": comparison, "params": params}, uid
+
+
+def test_alert_annotations_use_reduced_values():
+    """Описание алерта выводит число после Reduce, а не диапазонный ряд."""
+    for filename in EXPECTED_THRESHOLDS:
+        for uid, rule in _rules_by_uid(ALERTING_DIR / filename).items():
+            stages = {stage["refId"]: stage for stage in rule["data"]}
+
+            for annotation in rule.get("annotations", {}).values():
+                for ref_id in re.findall(r"\$values\.([A-Z])", annotation):
+                    model = stages[ref_id]["model"]
+                    assert model.get("type") == "reduce", uid