135 lines
5.5 KiB
Python
135 lines
5.5 KiB
Python
"""Раз в 3 часа: хвосты логов своих стеков; в мастер уходит только новое.
|
||
|
||
Мандат Бобра (20.09.2026): «можешь и логи периодами посматривать на предмет
|
||
ошибок». Держит это механизм, а не память модели: код ходит по стекам своей
|
||
области, а мастеру достаётся инжект только тогда, когда в хвостах появилось
|
||
то, чего в прошлом проходе не было. Джоб ничего не чинит - только читает.
|
||
|
||
Стеки берутся честным листанием (`Komodo.listing`), а не первой страницей
|
||
`ListStacks`, и длина сверяется с `GetStacksSummary.total`: расхождение едет
|
||
и в лог, и в инжект - неполный список хуже пустого.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import asyncio
|
||
import logging
|
||
from datetime import UTC, datetime
|
||
from typing import TYPE_CHECKING, Any
|
||
|
||
from beaver_agent.hands import KOMODO
|
||
from beaver_agent.hands.komodo_logs import (
|
||
SCOPE,
|
||
TAIL,
|
||
Seen,
|
||
StackScan,
|
||
Sweep,
|
||
brief,
|
||
daily,
|
||
digest,
|
||
read_stack,
|
||
)
|
||
from beaver_agent.vault import STATE_DIR, ZONE
|
||
|
||
if TYPE_CHECKING:
|
||
from beaver_gateway.jobs.scheduler import JobRun
|
||
|
||
from beaver_agent.hands.komodo import Komodo
|
||
|
||
_log = logging.getLogger(__name__)
|
||
|
||
STATE_FILE = STATE_DIR / "логи-флота.json"
|
||
"""Состояние прохода: что уже видели и о чём уже сказали."""
|
||
PARALLEL = 4
|
||
"""Логи тянутся с periphery: четырёх параллельных хватает, больше - грубо."""
|
||
|
||
|
||
async def _one(komodo: Komodo, stack: str, tail: int, gate: asyncio.Semaphore) -> Any:
|
||
async with gate:
|
||
try:
|
||
log = await komodo.read(
|
||
"GetStackLog",
|
||
{"stack": stack, "services": [], "tail": tail, "timestamps": True},
|
||
)
|
||
except Exception as exc: # noqa: BLE001 - стек не прочитан, но проход идёт
|
||
_log.warning("логи: %s не прочитан: %s", stack, exc)
|
||
return StackScan(stack=stack, error=f"{type(exc).__name__}: {exc}"[:200])
|
||
return read_stack(stack, log)
|
||
|
||
|
||
async def sweep(
|
||
komodo: Komodo,
|
||
*,
|
||
scope: tuple[str, ...] = SCOPE,
|
||
tail: int = TAIL,
|
||
now: datetime | None = None,
|
||
) -> Sweep:
|
||
"""Обойти стеки области и снять хвосты; Komodo молчит - так и скажем."""
|
||
at = now or datetime.now(UTC)
|
||
try:
|
||
stacks = await komodo.listing("ListStacks")
|
||
except Exception as exc: # noqa: BLE001 - недоступный Komodo это не «чисто»
|
||
_log.exception("логи: Komodo не отдал список стеков")
|
||
return Sweep(
|
||
at=at,
|
||
unreachable=f"{type(exc).__name__}: {exc}"[:200],
|
||
tail=tail,
|
||
scope=scope,
|
||
)
|
||
summary: int | None = None
|
||
try:
|
||
summary = int((await komodo.read("GetStacksSummary", {}))["total"])
|
||
except Exception: # noqa: BLE001 - сверка необязательна, проход без неё идёт
|
||
_log.warning("логи: GetStacksSummary не прочитан, сверить длину списка нечем")
|
||
if summary is not None and summary != len(stacks):
|
||
_log.error(
|
||
"логи: листание дало %s стеков, GetStacksSummary - %s", len(stacks), summary
|
||
)
|
||
mine = sorted(s["name"] for s in stacks if s["info"].get("server_name") in scope)
|
||
gate = asyncio.Semaphore(PARALLEL)
|
||
scans = await asyncio.gather(*(_one(komodo, name, tail, gate) for name in mine))
|
||
return Sweep(
|
||
at=at,
|
||
scans=tuple(scans),
|
||
in_scope=len(mine),
|
||
fleet=len(stacks),
|
||
summary=summary,
|
||
tail=tail,
|
||
scope=scope,
|
||
)
|
||
|
||
|
||
async def logs_watch(run: JobRun) -> None:
|
||
"""Проход по логам: состояние на диске, инжект - на новом и раз в сутки.
|
||
|
||
Инжекта два и они разные: обычный - про новое, всплески и отбой; суточный -
|
||
про длящееся, которое в проходах молчит. Первый проход суток у нас в 01:40,
|
||
он же и несёт сводку: в 00:00 ротация мастера, лезть туда незачем.
|
||
"""
|
||
if KOMODO is None:
|
||
_log.info("логи: руки komodo нет, проход пропущен")
|
||
return
|
||
seen = Seen.load(STATE_FILE)
|
||
report = digest(seen, await sweep(KOMODO), ZONE)
|
||
if report.worth_telling and await run.inject_master(
|
||
brief(report, ZONE), urgency="normal", origin="логи"
|
||
):
|
||
seen.mark_told(report)
|
||
if report.summary and await run.inject_master(
|
||
daily(report, ZONE), urgency="normal", origin="логи"
|
||
):
|
||
seen.mark_daily(report)
|
||
seen.save()
|
||
_log.info(
|
||
"логи: новое %s, всплеск %s, стихло %s, длится %s (сводка %s), "
|
||
"не прочитано %s, стеков %s из %s",
|
||
len(report.new),
|
||
len(report.surge),
|
||
len(report.quiet),
|
||
len(report.ongoing),
|
||
report.summary,
|
||
len(report.trouble),
|
||
len(report.sweep.read),
|
||
report.sweep.in_scope,
|
||
)
|