ElebobaandClaude Opus 5.5
79ea5ce2a5
CI / ruff · mypy · pytest (pull_request) Successful in 24m47s
Найдено смоуком на живых контейнерах (examples/, #19): - контейнер с restart: on-failure поднимается за секунду — между раундами его не видно в exited; краш теперь распознаётся и по росту RestartCount (сравнение со снапшотами до текущего раунда), повторы — счётчик - алерт закрывается, только когда условия нет resolve_after (60 с): иначе флапающий сервис получал бы «✅ восстановлено» после каждого раунда - правки сообщений со счётчиком — не чаще раза в минуту на алерт Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
132 lines
4.8 KiB
Python
132 lines
4.8 KiB
Python
"""Every 15 s: restart count, CPU and RAM of each running managed container → `container_snapshots`.
|
|
|
|
The history answers «are restarts growing over the last 5 minutes» (S-3) and feeds alerts (#15).
|
|
Old rows are removed by `app.db.maintenance`.
|
|
"""
|
|
|
|
import asyncio
|
|
import logging
|
|
from collections.abc import Awaitable, Callable
|
|
from dataclasses import dataclass, replace
|
|
from datetime import datetime
|
|
|
|
from sqlalchemy import select
|
|
|
|
from app.db.models import ContainerSnapshot, utcnow
|
|
from app.db.session import SessionMaker
|
|
from app.runtime.docker import Details, DockerClient, DockerError, Stats
|
|
|
|
log = logging.getLogger(__name__)
|
|
|
|
|
|
class SnapshotStore:
|
|
def __init__(self, sessions: SessionMaker) -> None:
|
|
self._sessions = sessions
|
|
|
|
async def add(self, rows: list[ContainerSnapshot]) -> None:
|
|
if rows:
|
|
async with self._sessions.begin() as session:
|
|
session.add_all(rows)
|
|
|
|
async def oldest_since(self, container: str, since: datetime) -> ContainerSnapshot | None:
|
|
async with self._sessions() as session:
|
|
return await session.scalar(
|
|
select(ContainerSnapshot)
|
|
.where(ContainerSnapshot.container == container, ContainerSnapshot.at >= since)
|
|
.order_by(ContainerSnapshot.at)
|
|
.limit(1)
|
|
)
|
|
|
|
async def since(self, container: str, since: datetime) -> list[ContainerSnapshot]:
|
|
async with self._sessions() as session:
|
|
rows = await session.scalars(
|
|
select(ContainerSnapshot)
|
|
.where(ContainerSnapshot.container == container, ContainerSnapshot.at >= since)
|
|
.order_by(ContainerSnapshot.at)
|
|
)
|
|
return list(rows)
|
|
|
|
async def latest(self, container: str) -> ContainerSnapshot | None:
|
|
async with self._sessions() as session:
|
|
return await session.scalar(
|
|
select(ContainerSnapshot)
|
|
.where(ContainerSnapshot.container == container)
|
|
.order_by(ContainerSnapshot.at.desc())
|
|
.limit(1)
|
|
)
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class Observation:
|
|
"""One container in one round: what the alert detectors look at (#15)."""
|
|
|
|
details: Details
|
|
stats: Stats | None # None when not running or stats failed
|
|
at: datetime | None = None # the round's time; its own snapshot row carries the same `at`
|
|
|
|
|
|
async def _observe(docker: DockerClient, container_id: str) -> Observation | None:
|
|
try:
|
|
details = await docker.inspect(container_id)
|
|
except DockerError as e:
|
|
log.warning("snapshot %s: %s", container_id[:12], e)
|
|
return None
|
|
stats = None
|
|
if details.state == "running":
|
|
try:
|
|
stats = await docker.stats(container_id)
|
|
except DockerError as e:
|
|
log.warning("stats %s: %s", container_id[:12], e)
|
|
return Observation(details, stats) # `at` is set by collect_once
|
|
|
|
|
|
def _row(o: Observation, at: datetime) -> ContainerSnapshot | None:
|
|
if o.stats is None:
|
|
return None
|
|
return ContainerSnapshot(
|
|
container=o.details.name,
|
|
at=at,
|
|
restart_count=o.details.restart_count,
|
|
cpu_pct=o.stats.cpu_pct,
|
|
mem_pct=round(o.stats.mem_pct, 1),
|
|
health=o.details.health,
|
|
)
|
|
|
|
|
|
async def collect_once(docker: DockerClient, store: SnapshotStore, now: datetime | None = None) -> list[Observation]:
|
|
"""Inspect every managed container, stats for the running ones; a snapshot row per running container."""
|
|
at = now or utcnow()
|
|
found = await asyncio.gather(*(_observe(docker, c.id) for c in await docker.containers()))
|
|
observations = [replace(o, at=at) for o in found if o is not None]
|
|
await store.add([row for o in observations if (row := _row(o, at)) is not None])
|
|
return observations
|
|
|
|
|
|
# Called after every round: the observations, or None when Docker itself was unreachable
|
|
RoundHook = Callable[[list[Observation] | None], Awaitable[None]]
|
|
|
|
|
|
async def snapshot_loop(
|
|
docker: DockerClient, store: SnapshotStore, interval: float, on_round: RoundHook | None = None
|
|
) -> None:
|
|
failing = False
|
|
while True:
|
|
observations: list[Observation] | None = None
|
|
try:
|
|
observations = await collect_once(docker, store)
|
|
if failing:
|
|
log.info("snapshots: docker reachable again")
|
|
failing = False
|
|
except DockerError as e:
|
|
if not failing: # one line per outage, not one every 15 s
|
|
log.warning("snapshots: %s", e)
|
|
failing = True
|
|
except Exception: # a broken round must not kill the bot
|
|
log.exception("snapshots failed")
|
|
if on_round is not None:
|
|
try:
|
|
await on_round(observations)
|
|
except Exception:
|
|
log.exception("snapshot hook failed")
|
|
await asyncio.sleep(interval)
|