Merge pull request 'fix(grafana): добавлено Reduce перед порогами алертов' (#29) from fix/28-grafana-alert-reduce into main
Reviewed-on: #29
This commit was merged in pull request #29.
This commit is contained in:
ddmitry committed
commit
a39f613a01
5 files changed
+114
-43
No files matched your search
@@ -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
|
||||
@@ -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:
|
||||
|
||||
@@ -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** снова растёт;
|
||||
|
||||
@@ -6,6 +6,7 @@ prometheus-client==0.21.1
|
||||
|
||||
# Тестирование
|
||||
pytest==8.3.4
|
||||
PyYAML==6.0.2
|
||||
|
||||
# Утилиты
|
||||
python-json-logger==2.0.7
|
||||
@@ -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
|
||||
Reference in new issue
Block a user