tgit/app/runtime/checks.py

"""«🩺 Запустить проверки» (SPEC S-3): independent checks per container, each with its own timeout.

A check that raises or hangs becomes ❌ with the reason; the others still run and the report is always complete.
Details are short and language-neutral (numbers, codes); the titles live in `config/texts.yaml`.
"""

import asyncio
from collections.abc import Awaitable, Callable
from dataclasses import dataclass
from datetime import datetime, timedelta
from enum import StrEnum
from pathlib import Path

import httpx
import regex
import yaml
from pydantic import BaseModel, ConfigDict, Field, field_validator

from app.db.models import utcnow
from app.runtime.docker import Details, DockerClient, Stats
from app.runtime.snapshots import SnapshotStore

REGEX_TIMEOUT = 1.0  # seconds per log line: a pathological pattern can't stall the bot


class ChecksConfig(BaseModel):
    model_config = ConfigDict(extra="forbid", frozen=True)

    window: int = Field(default=300, gt=0)
    cpu_pct: float = Field(default=90, gt=0)
    mem_pct: float = Field(default=90, gt=0, le=100)
    error_pattern: str
    http_timeout: float = Field(default=5, gt=0)
    check_timeout: float = Field(default=15, gt=0)

    @field_validator("error_pattern")
    @classmethod
    def _compiles(cls, value: str) -> str:
        try:
            regex.compile(value)
        except regex.error as e:  # not a ValueError: pydantic wouldn't turn it into a validation error
            raise ValueError(f"error_pattern: {e}") from e
        return value


def load_checks(config_dir: Path) -> ChecksConfig:
    with (config_dir / "checks.yaml").open(encoding="utf-8") as f:
        return ChecksConfig.model_validate(yaml.safe_load(f))


class CheckName(StrEnum):
    RUNNING = "running"
    HEALTH = "health"
    RESTARTS = "restarts"
    CPU = "cpu"
    MEM = "mem"
    LOG_ERRORS = "log_errors"
    HTTP = "http"


class Verdict(StrEnum):
    OK = "✅"
    FAIL = "❌"
    SKIP = "➖"  # not applicable: no healthcheck, not running, no health_url


@dataclass(frozen=True)
class CheckResult:
    name: CheckName
    verdict: Verdict
    detail: str = ""


@dataclass(frozen=True)
class ContainerReport:
    details: Details
    results: list[CheckResult]

    @property
    def ok(self) -> bool:
        return all(r.verdict is not Verdict.FAIL for r in self.results)


@dataclass(frozen=True)
class Context:
    details: Details
    docker: DockerClient
    store: SnapshotStore
    cfg: ChecksConfig
    http: httpx.AsyncClient
    now: datetime
    stats: Stats | None  # None when not running

    @property
    def since(self) -> datetime:
        return self.now - timedelta(seconds=self.cfg.window)


async def check_running(ctx: Context) -> CheckResult:
    d = ctx.details
    if d.state == "running":
        return CheckResult(CheckName.RUNNING, Verdict.OK, d.state)
    detail = f"{d.state}, exit {d.exit_code}" + (", OOM" if d.oom_killed else "")
    return CheckResult(CheckName.RUNNING, Verdict.FAIL, detail)


async def check_health(ctx: Context) -> CheckResult:
    health = ctx.details.health
    if health is None:
        return CheckResult(CheckName.HEALTH, Verdict.SKIP, "no healthcheck")
    return CheckResult(CheckName.HEALTH, Verdict.OK if health == "healthy" else Verdict.FAIL, health)


async def check_restarts(ctx: Context) -> CheckResult:
    now_count = ctx.details.restart_count
    before = await ctx.store.oldest_since(ctx.details.name, ctx.since)
    if before is None:
        return CheckResult(CheckName.RESTARTS, Verdict.SKIP, f"{now_count}, no history")
    grown = now_count - before.restart_count
    if grown > 0:
        return CheckResult(CheckName.RESTARTS, Verdict.FAIL, f"{before.restart_count} → {now_count}")
    return CheckResult(CheckName.RESTARTS, Verdict.OK, str(now_count))


async def check_cpu(ctx: Context) -> CheckResult:
    if ctx.stats is None:
        return CheckResult(CheckName.CPU, Verdict.SKIP)
    pct = ctx.stats.cpu_pct
    verdict = Verdict.OK if pct < ctx.cfg.cpu_pct else Verdict.FAIL
    return CheckResult(CheckName.CPU, verdict, f"{pct:.1f}% / {ctx.cfg.cpu_pct:g}%")


async def check_mem(ctx: Context) -> CheckResult:
    if ctx.stats is None:
        return CheckResult(CheckName.MEM, Verdict.SKIP)
    pct = ctx.stats.mem_pct
    verdict = Verdict.OK if pct < ctx.cfg.mem_pct else Verdict.FAIL
    return CheckResult(CheckName.MEM, verdict, f"{pct:.1f}% / {ctx.cfg.mem_pct:g}%")


def count_errors(text: str, pattern: str) -> tuple[int, str]:
    """Matching lines and the first of them (trimmed)."""
    compiled = regex.compile(pattern)
    hits, first = 0, ""
    for line in text.splitlines():
        try:
            matched = compiled.search(line, timeout=REGEX_TIMEOUT) is not None
        except TimeoutError:
            matched = False
        if matched:
            hits += 1
            first = first or line.strip()[:120]
    return hits, first


async def check_log_errors(ctx: Context) -> CheckResult:
    text = await ctx.docker.logs(ctx.details.id, since=ctx.since)
    hits, first = await asyncio.to_thread(count_errors, text, ctx.cfg.error_pattern)
    if hits:
        return CheckResult(CheckName.LOG_ERRORS, Verdict.FAIL, f"{hits}: {first}")
    return CheckResult(CheckName.LOG_ERRORS, Verdict.OK, "0")


async def check_http(ctx: Context) -> CheckResult:
    url = ctx.details.health_url
    if url is None:
        return CheckResult(CheckName.HTTP, Verdict.SKIP, "no health_url")
    try:
        response = await ctx.http.get(url, timeout=ctx.cfg.http_timeout)
    except httpx.HTTPError as e:
        return CheckResult(CheckName.HTTP, Verdict.FAIL, type(e).__name__)
    ok = response.status_code < 400
    return CheckResult(CheckName.HTTP, Verdict.OK if ok else Verdict.FAIL, f"HTTP {response.status_code}")


Check = Callable[[Context], Awaitable[CheckResult]]
CHECKS: list[tuple[CheckName, Check]] = [
    (CheckName.RUNNING, check_running),
    (CheckName.HEALTH, check_health),
    (CheckName.RESTARTS, check_restarts),
    (CheckName.CPU, check_cpu),
    (CheckName.MEM, check_mem),
    (CheckName.LOG_ERRORS, check_log_errors),
    (CheckName.HTTP, check_http),
]


async def _guarded(name: CheckName, check: Check, ctx: Context) -> CheckResult:
    try:
        return await asyncio.wait_for(check(ctx), timeout=ctx.cfg.check_timeout)
    except TimeoutError:
        return CheckResult(name, Verdict.FAIL, "timeout")
    except Exception as e:  # one broken check must not spoil the report
        return CheckResult(name, Verdict.FAIL, f"{type(e).__name__}: {str(e)[:80]}")


class Checker:
    def __init__(
        self, docker: DockerClient, store: SnapshotStore, cfg: ChecksConfig, http: httpx.AsyncClient | None = None
    ) -> None:
        self._docker = docker
        self._store = store
        self._cfg = cfg
        self._http = http or httpx.AsyncClient(follow_redirects=False)

    async def check(self, container_id: str, now: datetime | None = None) -> ContainerReport:
        details = await self._docker.inspect(container_id)
        stats = None
        if details.state == "running":
            try:
                stats = await asyncio.wait_for(self._docker.stats(container_id), timeout=self._cfg.check_timeout)
            except Exception:  # cpu/mem become ➖, the rest still runs
                stats = None
        ctx = Context(details, self._docker, self._store, self._cfg, self._http, now or utcnow(), stats)
        results = await asyncio.gather(*(_guarded(name, fn, ctx) for name, fn in CHECKS))
        return ContainerReport(details, list(results))

    async def check_all(self, now: datetime | None = None) -> list[ContainerReport]:
        containers = await self._docker.containers()
        return list(await asyncio.gather(*(self.check(c.id, now) for c in containers)))

    async def aclose(self) -> None:
        await self._http.aclose()