ddmitry/clickstream-ch-kafka-superset-demo
1
0 Fork 0
Code Issues 1 Pull Requests Actions Packages Projects Releases Wiki Activity

fix(grafana): добавлено Reduce перед порогами алертов #29

Merged
ddmitry merged 1 commits from fix/28-grafana-alert-reduce into main 2026-09-01 00:53:26 +03:00
Conversation 0 Commits 1 Files Changed 5
+114 -43
5 changed files with 114 additions and 43 deletions
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

configs/grafana/provisioning/alerting/airflow-alert-rules.yml
+21 -21
View File
@@ -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
configs/grafana/provisioning/alerting/kafka-alert-rules.yml
+18 -18
View File
@@ -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:
docs/course/lessons/05_monitoring.md
+6 -4
View File
@@ -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
+1
View File
@@ -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
+68
View File
@@ -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