52 lines
1.6 KiB
Python
52 lines
1.6 KiB
Python
|
"""Сопоставление входящего алерта с командой по правилам лейблов (как Team в Grafana IRM)."""
|
|||
|
|
|
|||
|
|
from __future__ import annotations
|
|||
|
|
|
|||
|
|
from typing import Any, Sequence
|
|||
|
|
from uuid import UUID
|
|||
|
|
|
|||
|
|
import asyncpg
|
|||
|
|
|
|||
|
|
|
|||
|
|
def match_team_for_labels(
|
|||
|
|
labels: dict[str, Any],
|
|||
|
|
rules: Sequence[asyncpg.Record | tuple[UUID, str, str]],
|
|||
|
|
) -> UUID | None:
|
|||
|
|
"""
|
|||
|
|
rules — упорядочены по приоритету (выше priority — раньше проверка).
|
|||
|
|
Первое совпадение label_key == label_value возвращает team_id.
|
|||
|
|
"""
|
|||
|
|
if not labels or not rules:
|
|||
|
|
return None
|
|||
|
|
flat: dict[str, str] = {
|
|||
|
|
str(k): "" if v is None else str(v) for k, v in labels.items()
|
|||
|
|
}
|
|||
|
|
for row in rules:
|
|||
|
|
if isinstance(row, tuple):
|
|||
|
|
tid, key, val = row[0], row[1], row[2]
|
|||
|
|
else:
|
|||
|
|
tid = row["team_id"]
|
|||
|
|
key = row["label_key"]
|
|||
|
|
val = row["label_value"]
|
|||
|
|
if flat.get(str(key)) == str(val):
|
|||
|
|
return tid if isinstance(tid, UUID) else UUID(str(tid))
|
|||
|
|
return None
|
|||
|
|
|
|||
|
|
|
|||
|
|
async def fetch_team_rules(conn: asyncpg.Connection) -> list[asyncpg.Record]:
|
|||
|
|
return await conn.fetch(
|
|||
|
|
"""
|
|||
|
|
SELECT team_id, label_key, label_value
|
|||
|
|
FROM team_label_rules
|
|||
|
|
ORDER BY priority DESC, id ASC
|
|||
|
|
"""
|
|||
|
|
)
|
|||
|
|
|
|||
|
|
|
|||
|
|
async def resolve_team_id_for_labels(
|
|||
|
|
conn: asyncpg.Connection,
|
|||
|
|
labels: dict[str, Any],
|
|||
|
|
) -> UUID | None:
|
|||
|
|
rules = await fetch_team_rules(conn)
|
|||
|
|
return match_team_for_labels(labels, list(rules))
|