ElebobaandClaude Opus 5.5
6c6f766abc
CI / ruff · mypy · pytest (pull_request) Successful in 4m49s
- детекторы по состоянию после каждого раунда снапшотов: краш, OOM, цикл рестартов, unhealthy, перегрузка ≥ 1 мин, мало места, Docker недоступен; пороги в config/alerts.yaml - AlertService: ключ kind+service, повтор в 10 мин — счётчик в исходном сообщении, «✅ восстановлено» ответом; разовые события (🛡) со счётчиком - рассылка всем активным (🛡 — только админам), темп ~25/с, RetryAfter, заблокировавший бота пропускается; alert_deliveries для правок и ответов - «🚨 Алерты»: активные + 7 дней по 10, карточка; кнопки «📜 Логи», «🩺 Статус» под алертом открывают новое сообщение, а не затирают алерт - 🛡 атаки через агента идут через общий сервис алертов - убраны заглушки «Раздел в разработке» — у всех разделов свои роутеры Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
97 lines
3.5 KiB
Python
97 lines
3.5 KiB
Python
from dataclasses import replace
|
|
from datetime import UTC, datetime, timedelta
|
|
|
|
from app.alerts.detectors import (
|
|
AlertsConfig,
|
|
container_conditions,
|
|
disk_condition,
|
|
docker_condition,
|
|
load_alerts_config,
|
|
)
|
|
from app.db.models import AlertKind, ContainerSnapshot
|
|
from app.runtime.docker import Details, Stats
|
|
from app.runtime.snapshots import Observation
|
|
from tests.conftest import ROOT
|
|
|
|
CFG = load_alerts_config(ROOT / "config")
|
|
NOW = datetime(2026, 10, 17, 12, 0, tzinfo=UTC)
|
|
RUNNING = Details(
|
|
id="a" * 64,
|
|
name="svc",
|
|
service="svc",
|
|
image="svc:latest",
|
|
state="running",
|
|
health="healthy",
|
|
exit_code=0,
|
|
restart_count=5,
|
|
started_at=NOW - timedelta(hours=1),
|
|
finished_at=None,
|
|
oom_killed=False,
|
|
labels={},
|
|
)
|
|
CALM = Stats(cpu_pct=5, mem_bytes=10, mem_limit=100)
|
|
|
|
|
|
def snap(seconds_ago: int, restarts: int = 5, cpu: float = 5, mem: float = 10) -> ContainerSnapshot:
|
|
return ContainerSnapshot(
|
|
container="svc", at=NOW - timedelta(seconds=seconds_ago), restart_count=restarts, cpu_pct=cpu, mem_pct=mem
|
|
)
|
|
|
|
|
|
def kinds(o: Observation, history: list[ContainerSnapshot] | None = None) -> set[AlertKind]:
|
|
return {c.kind for c in container_conditions(o, history or [], CFG, NOW)}
|
|
|
|
|
|
def test_config_defaults() -> None:
|
|
assert AlertsConfig() == CFG
|
|
|
|
|
|
def test_healthy_container_is_quiet() -> None:
|
|
assert kinds(Observation(RUNNING, CALM), [snap(300), snap(15)]) == set()
|
|
|
|
|
|
def test_crash_oom_and_clean_stop() -> None:
|
|
crashed = replace(RUNNING, state="exited", exit_code=1, health=None)
|
|
oom = replace(crashed, exit_code=137, oom_killed=True)
|
|
stopped = replace(crashed, exit_code=0)
|
|
|
|
assert kinds(Observation(crashed, None)) == {AlertKind.CRASH}
|
|
assert kinds(Observation(oom, None)) == {AlertKind.OOM}
|
|
assert kinds(Observation(stopped, None)) == set() # docker stop is not an incident
|
|
|
|
|
|
def test_restart_loop_counts_within_window() -> None:
|
|
history = [snap(900, restarts=0), snap(500, restarts=2), snap(60, restarts=4)]
|
|
assert kinds(Observation(RUNNING, CALM), history) == {AlertKind.RESTART_LOOP} # 5 - 2 ≥ 3 within 10 min
|
|
assert kinds(Observation(replace(RUNNING, restart_count=4), CALM), history) == set()
|
|
|
|
|
|
def test_unhealthy() -> None:
|
|
assert kinds(Observation(replace(RUNNING, health="unhealthy"), CALM)) == {AlertKind.UNHEALTHY}
|
|
assert kinds(Observation(replace(RUNNING, health="starting"), CALM)) == set()
|
|
|
|
|
|
def test_overload_needs_a_full_minute() -> None:
|
|
hot = Stats(cpu_pct=150, mem_bytes=10, mem_limit=100)
|
|
held = [snap(60, cpu=120), snap(45, cpu=130), snap(30, cpu=140), snap(15, cpu=150)]
|
|
blip = [snap(15, cpu=150)]
|
|
dipped = [snap(60, cpu=120), snap(30, cpu=40), snap(15, cpu=150)]
|
|
|
|
assert kinds(Observation(RUNNING, hot), held) == {AlertKind.OVERLOAD}
|
|
assert kinds(Observation(RUNNING, hot), blip) == set()
|
|
assert kinds(Observation(RUNNING, hot), dipped) == set()
|
|
mem_hot = Stats(cpu_pct=1, mem_bytes=95, mem_limit=100)
|
|
assert kinds(Observation(RUNNING, mem_hot), [snap(60, mem=95), snap(15, mem=96)]) == {AlertKind.OVERLOAD}
|
|
|
|
|
|
def test_disk_and_docker() -> None:
|
|
assert disk_condition(None, CFG) == []
|
|
assert disk_condition(50, CFG) == []
|
|
[low] = disk_condition(4.2, CFG)
|
|
assert low.kind is AlertKind.DISK_LOW and low.service is None
|
|
|
|
assert docker_condition(None, CFG, NOW) == []
|
|
assert docker_condition(NOW - timedelta(minutes=2), CFG, NOW) == []
|
|
[down] = docker_condition(NOW - timedelta(minutes=6), CFG, NOW)
|
|
assert down.key == (AlertKind.DEPENDENCY_DOWN, "docker")
|