From 335e5e76122f6a689bcc0876b54ced4078f120c9 Mon Sep 17 00:00:00 2001 From: Dmitry Dementev Date: Tue, 1 Sep 2026 00:51:03 +0300 Subject: [PATCH] =?UTF-8?q?fix(grafana):=20=D0=B4=D0=BE=D0=B1=D0=B0=D0=B2?= =?UTF-8?q?=D0=BB=D0=B5=D0=BD=D0=BE=20Reduce=20=D0=BF=D0=B5=D1=80=D0=B5?= =?UTF-8?q?=D0=B4=20=D0=BF=D0=BE=D1=80=D0=BE=D0=B3=D0=B0=D0=BC=D0=B8=20?= =?UTF-8?q?=D0=B0=D0=BB=D0=B5=D1=80=D1=82=D0=BE=D0=B2?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Зачем: - диапазонные запросы Prometheus возвращали временные ряды, из-за чего Grafana не вычисляла алерты Airflow и Kafka. - Что: - между запросом Prometheus и Threshold добавлен Reduce(last) в семи правилах, а аннотации переведены на итоговое число. - добавлены регрессионные тесты конфигурации и зависимость PyYAML для их запуска. - в уроке 5 уточнено время срабатывания и восстановления алерта scheduler. - Проверка: - uv run --with-requirements generator/requirements.txt pytest tests/ -q -p no:cacheprovider: 33 passed. - uv run --with-requirements requirements.txt pytest tests/ -q -p no:cacheprovider из generator: 218 passed. - Docker Compose config и вычисление правил в Grafana 11.5.2 прошли успешно. --- .../alerting/airflow-alert-rules.yml | 42 ++++++------ .../alerting/kafka-alert-rules.yml | 36 +++++----- docs/course/lessons/05_monitoring.md | 10 +-- generator/requirements.txt | 1 + tests/test_grafana_alert_rules.py | 68 +++++++++++++++++++ 5 files changed, 114 insertions(+), 43 deletions(-) create mode 100644 tests/test_grafana_alert_rules.py diff --git a/configs/grafana/provisioning/alerting/airflow-alert-rules.yml b/configs/grafana/provisioning/alerting/airflow-alert-rules.yml index 406642b..8554c75 100644 --- a/configs/grafana/provisioning/alerting/airflow-alert-rules.yml +++ b/configs/grafana/provisioning/alerting/airflow-alert-rules.yml @@ -33,9 +33,8 @@ groups: model: conditions: - evaluator: - params: - - 0.1 - type: lt + params: [] + type: gt 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 - type: gt + - 0.1 + type: lt operator: type: and query: @@ -114,8 +114,7 @@ groups: model: conditions: - evaluator: - params: - - 50 + params: [] 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: - - 0.1 + params: [] 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: - - 30 + params: [] 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 diff --git a/configs/grafana/provisioning/alerting/kafka-alert-rules.yml b/configs/grafana/provisioning/alerting/kafka-alert-rules.yml index 4546f35..24c53d8 100644 --- a/configs/grafana/provisioning/alerting/kafka-alert-rules.yml +++ b/configs/grafana/provisioning/alerting/kafka-alert-rules.yml @@ -33,9 +33,8 @@ groups: model: conditions: - evaluator: - params: - - 1 - type: lt + params: [] + type: gt 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 - type: gt + - 1 + 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: - - 10000 + params: [] 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: - - 0.1 - type: lt + params: [] + type: gt 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 - type: gt + - 0.1 + type: lt operator: type: and query: diff --git a/docs/course/lessons/05_monitoring.md b/docs/course/lessons/05_monitoring.md index 01bd03f..a7d2d83 100644 --- a/docs/course/lessons/05_monitoring.md +++ b/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** снова растёт; diff --git a/generator/requirements.txt b/generator/requirements.txt index badfad0..924e8f5 100644 --- a/generator/requirements.txt +++ b/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 diff --git a/tests/test_grafana_alert_rules.py b/tests/test_grafana_alert_rules.py new file mode 100644 index 0000000..7ded511 --- /dev/null +++ b/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