From 9737167c761fa20634d7107accda5160db51b20b Mon Sep 17 00:00:00 2001 From: h Date: Sun, 20 Sep 2026 18:26:05 +0000 Subject: [PATCH] =?UTF-8?q?feat(jobs,hands):=20=D1=81=D0=BA=D0=B0=D0=BD?= =?UTF-8?q?=D0=B5=D1=80=20=D0=BB=D0=BE=D0=B3=D0=BE=D0=B2=20=D1=84=D0=BB?= =?UTF-8?q?=D0=BE=D1=82=D0=B0=20=D0=B1=D1=83=D0=B4=D0=B8=D1=82=20=D1=82?= =?UTF-8?q?=D0=BE=D0=BB=D1=8C=D0=BA=D0=BE=20=D0=BD=D0=BE=D0=B2=D1=8B=D0=BC?= =?UTF-8?q?=20=D0=B8=20=D0=BD=D0=B5=D1=81=D1=91=D1=82=20=D1=81=D0=B2=D0=BE?= =?UTF-8?q?=D1=8E=20=D0=BE=D0=B1=D0=BB=D0=B0=D1=81=D1=82=D1=8C?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- beaver_agent/hands/komodo_logs.py | 541 ++++++++++++++++++++++++++++++ beaver_agent/jobs/__init__.py | 4 + beaver_agent/jobs/logs.py | 124 +++++++ beaver_agent/vault.py | 3 + tests/test_komodo_logs.py | 465 +++++++++++++++++++++++++ 5 files changed, 1137 insertions(+) create mode 100644 beaver_agent/hands/komodo_logs.py create mode 100644 beaver_agent/jobs/logs.py create mode 100644 tests/test_komodo_logs.py diff --git a/beaver_agent/hands/komodo_logs.py b/beaver_agent/hands/komodo_logs.py new file mode 100644 index 0000000..b4fcd64 --- /dev/null +++ b/beaver_agent/hands/komodo_logs.py @@ -0,0 +1,541 @@ +"""Логи флота периодами: что в хвостах похоже на беду и чего там раньше не было. + +Ходит и читает джоб `jobs/logs.py`, судит этот модуль. Правило одно: мастера +дёргает только НОВОЕ. То, что молотит в логе третьи сутки, уже сказано один +раз и больше не показывается; показывается то, чего в прошлом проходе не было, +и обратный переход «было → нет» - с временем последнего события, не с временем +прохода (логи берутся с `--timestamps`, время в строке настоящее). + +Показание всегда несёт свою область: сколько стеков в области, сколько из них +прочиталось, какой отрезок логов попал в хвост. Без этого «всё чисто» +неотличимо от «половину не посмотрел», поэтому чего не смог - тоже в инжекте. +""" + +from __future__ import annotations + +import hashlib +import json +import re +from dataclasses import dataclass, field +from datetime import UTC, datetime, timedelta, tzinfo +from typing import TYPE_CHECKING, Any + +from beaver_agent.hands.komodo import ANSI + +if TYPE_CHECKING: + from pathlib import Path + +__all__ = [ + "SCOPE", + "SIGNALS", + "TAIL", + "Entry", + "Report", + "Seen", + "StackScan", + "Sweep", + "brief", + "digest", + "read_stack", +] + +SCOPE = ("dell-home", "personal-main") +"""Свои серверы. Машины заказчиков (md-cars-*, veris-*, ms-agents-*) не наши: +там чужой прод и чужие секреты, и читать их логи в свой контекст - нельзя.""" + +TAIL = 500 +"""Строк на стек. Болтливый стек столько строк проживает за минуты - об этом +в инжекте говорится отдельно, а не выдаётся за просмотренный промежуток.""" + +SIGNALS: tuple[tuple[str, str], ...] = ( + ("критично", r"\bCRITICAL\b|\bFATAL\b|\bEMERG\b"), + ("паника", r"\bpanic:|\bpanicked at\b|\bfatal error:"), + ("oom", r"\bOOM\b|OOMKilled|[Oo]ut of memory|oom[-_]kill"), + ("traceback", r"Traceback \(most recent call last\)|\bUnhandled exception\b"), + ("падение", r"exited with code (?!0\b)\d+|\bexit status [1-9]|Exited \([1-9]\d*\)"), + ("рестарт-цикл", r"[Rr]estarting \(\d+\)|back-?off.{0,20}restart|CrashLoopBackOff"), + ("healthcheck", r"health-?\s?check (?:failed|error)|\bunhealthy\b"), + ("ошибка", r"\bERROR\b|\bERR!\b"), +) +"""Что считается бедой: метка и выражение. Правится одним местом; порядок - +приоритет, строка получает первую подошедшую метку.""" + +MATCHERS = tuple((label, re.compile(pattern)) for label, pattern in SIGNALS) + +SELF_MARK = "Логи флота" +"""Свой же инжект видно в stdout gateway: ловить себя по кругу не надо.""" + +LINE_RE = re.compile(r"^(?P[^\s|]+)\s*\|\s?(?P.*)$") +STAMP_RE = re.compile( + r"^(?P\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}(?:\.\d+)?Z)(?:\s(?P.*))?$" +) +"""Пустая строка тоже со штампом (`сервис | время` и ничего) - это не мусор.""" +DIGITS_RE = re.compile(r"\b[0-9a-f]{7,}\b|\d+") +"""Числа и хеши в строке - шум: без них «то же самое» узнаётся между проходами.""" +SECRET_RE = re.compile( + r"(?i)\b(?:token|secret|api[-_ ]?key|password|passwd|authorization)\b\s*[=:]\s*\S+" + r"|\b[A-Za-z0-9_\-]{32,}\b" +) +"""Инжект уезжает в чат и в vault: похожее на ключ в показании не нужно.""" + +SAMPLE_CHARS = 180 +MAX_NEW = 25 +MAX_QUIET = 10 +MAX_TROUBLE = 10 +GAP_NAMES = 3 +UNPARSED_LOUD = 10 +"""Столько неразобранных строк на проход - уже не обрезанная голова, а дыра.""" +KEEP = timedelta(days=14) +"""Сколько помнить беду, которой больше не видно: дальше пусть будет новой.""" +KEY_CHARS = 12 +STATE_VERSION = 1 +EPOCH = datetime(1970, 1, 1, tzinfo=UTC) + + +@dataclass(frozen=True, slots=True) +class Hit: + """Одна беда в хвосте одного стека, уже слитая по повторам.""" + + key: str + stack: str + service: str + signal: str + sample: str + at: datetime | None + count: int + + +@dataclass(frozen=True, slots=True) +class StackScan: + """Что дал один стек: находки, отрезок его хвоста или причина отказа.""" + + stack: str + hits: tuple[Hit, ...] = () + first: datetime | None = None + last: datetime | None = None + lines: int = 0 + unparsed: int = 0 + """Строки без `сервис | время`: обрезанная голова хвоста и прочий мусор.""" + error: str | None = None + + @property + def ok(self) -> bool: + return self.error is None + + +@dataclass(frozen=True, slots=True) +class Sweep: + """Проход целиком - вместе с тем, какую область он покрыл.""" + + at: datetime + scans: tuple[StackScan, ...] = () + in_scope: int = 0 + fleet: int = 0 + summary: int | None = None + tail: int = TAIL + scope: tuple[str, ...] = SCOPE + unreachable: str | None = None + """Komodo не ответил вовсе - ни одного стека, и это не «чисто».""" + + @property + def read(self) -> tuple[StackScan, ...]: + return tuple(s for s in self.scans if s.ok) + + @property + def trouble(self) -> tuple[str, ...]: + if self.unreachable: + return (f"Komodo не ответил: {self.unreachable}",) + out = [f"{s.stack}: {s.error}" for s in self.scans if not s.ok] + if self.summary is not None and self.summary != self.fleet: + out.append( + f"листание дало {self.fleet} стеков, а GetStacksSummary - " + f"{self.summary}: список неполон" + ) + return tuple(out) + + @property + def span(self) -> tuple[datetime | None, datetime | None]: + """Отрезок, который реально попал в хвосты всех прочитанных стеков.""" + firsts = [s.first for s in self.read if s.first] + lasts = [s.last for s in self.read if s.last] + return (min(firsts) if firsts else None, max(lasts) if lasts else None) + + @property + def shortest(self) -> StackScan | None: + """Стек с самым коротким хвостом: он и ограничивает, что мы видели.""" + stamped = [s for s in self.read if s.first] + if not stamped: + return None + return max(stamped, key=lambda s: s.first or EPOCH) + + def gaps(self, since: datetime | None) -> tuple[str, ...]: + """Стеки, чей хвост начался позже прошлого прохода: там дыра.""" + if since is None: + return () + return tuple(s.stack for s in self.read if s.first and s.first > since) + + +@dataclass(slots=True) +class Entry: + """Беда, которую уже видели: когда впервые, когда последний раз, сколько раз.""" + + key: str + stack: str + service: str + signal: str + sample: str + first: datetime | None = None + last: datetime | None = None + """Время события из строки лога, а не время прохода.""" + looked: datetime | None = None + """Когда её последний раз видели в хвосте: по этому и забываем.""" + count: int = 0 + told: bool = False + """Сказали ли о ней мастеру. Инжект не доехал - не сказали, повторим.""" + + @property + def where(self) -> str: + return f"{self.stack}/{self.service}" + + +@dataclass(frozen=True, slots=True) +class Report: + """Что из прохода стоит мастера.""" + + sweep: Sweep + new: tuple[Entry, ...] = () + quiet: tuple[Entry, ...] = () + trouble: tuple[str, ...] = () + gaps: tuple[str, ...] = () + cold: bool = False + """Первый проход: состояния не было, и вся область легла в базовую линию.""" + baseline: int = 0 + """Сколько бед легло в базовую линию в этом проходе.""" + baselined: tuple[str, ...] = () + """Стеки, впервые прочитанные в этом проходе: их прошлое не показывается.""" + + @property + def worth_telling(self) -> bool: + return bool(self.new or self.quiet or self.trouble or self.cold) + + +@dataclass(slots=True) +class Seen: + """Состояние между проходами: файл на диске, переживает передеплой.""" + + path: Path + entries: dict[str, Entry] = field(default_factory=dict) + baselined: set[str] = field(default_factory=set) + """Стеки, чьё прошлое уже принято за данность: их старое не «новое».""" + trouble: tuple[str, ...] = () + """О каких отказах уже сказали: Komodo лежит - скажем один раз, не каждые 3 ч.""" + last_pass: datetime | None = None + fresh: bool = True + """Файла не было: базовую линию только снимаем, стену старых бед не шлём.""" + + @classmethod + def load(cls, path: Path) -> Seen: + try: + raw = json.loads(path.read_text(encoding="utf-8")) + except (OSError, ValueError): + return cls(path=path) + if raw.get("version") != STATE_VERSION: + return cls(path=path) + entries = { + key: Entry( + key=key, + stack=row["stack"], + service=row["service"], + signal=row["signal"], + sample=row["sample"], + first=_parse(row.get("first")), + last=_parse(row.get("last")), + looked=_parse(row.get("looked")), + count=int(row.get("count", 0)), + told=bool(row.get("told")), + ) + for key, row in raw.get("entries", {}).items() + } + return cls( + path=path, + entries=entries, + baselined=set(raw.get("baselined", ())), + trouble=tuple(raw.get("trouble", ())), + last_pass=_parse(raw.get("last_pass")), + fresh=False, + ) + + def save(self) -> None: + body = { + "version": STATE_VERSION, + "last_pass": _iso(self.last_pass), + "baselined": sorted(self.baselined), + "trouble": list(self.trouble), + "entries": { + key: { + "stack": e.stack, + "service": e.service, + "signal": e.signal, + "sample": e.sample, + "first": _iso(e.first), + "last": _iso(e.last), + "looked": _iso(e.looked), + "count": e.count, + "told": e.told, + } + for key, e in self.entries.items() + }, + } + self.path.parent.mkdir(parents=True, exist_ok=True) + tmp = self.path.with_suffix(".tmp") + tmp.write_text(json.dumps(body, ensure_ascii=False), encoding="utf-8") + tmp.replace(self.path) + + def mark_told(self, report: Report) -> None: + """Инжект доехал: новое больше не новое, стихшее забывается.""" + for entry in report.new: + entry.told = True + for entry in report.quiet: + self.entries.pop(entry.key, None) + self.trouble = report.trouble + + +def _iso(at: datetime | None) -> str | None: + return at.isoformat() if at else None + + +def _parse(raw: Any) -> datetime | None: + if not isinstance(raw, str): + return None + try: + at = datetime.fromisoformat(raw) + except ValueError: + return None + return at if at.tzinfo else at.replace(tzinfo=UTC) + + +def _mask(text: str) -> str: + """Замазать похожее на секрет: строка лога едет мастеру как есть.""" + return SECRET_RE.sub("<скрыто>", text) + + +def _normal(text: str) -> str: + """Строка без чисел, хешей и адресов: по ней узнаётся «та же самая беда».""" + return DIGITS_RE.sub("#", ANSI.sub("", text)).strip() + + +def _key(stack: str, service: str, text: str) -> str: + raw = f"{stack}\0{service}\0{_normal(text)}".encode() + return hashlib.blake2s(raw, digest_size=KEY_CHARS // 2).hexdigest() + + +def signal_of(text: str) -> str | None: + for label, matcher in MATCHERS: + if matcher.search(text): + return label + return None + + +def read_stack(stack: str, log: dict[str, Any]) -> StackScan: + """Ответ GetStackLog → находки стека; ошибка Komodo остаётся ошибкой.""" + if not isinstance(log, dict) or "error" in log: + error = log.get("error") if isinstance(log, dict) else "ответ не разобран" + return StackScan(stack=stack, error=str(error)[:200]) + text = (log.get("stdout") or "") + "\n" + (log.get("stderr") or "") + found: dict[str, Hit] = {} + stamps: list[datetime] = [] + lines = 0 + unparsed = 0 + for raw in text.splitlines(): + line = ANSI.sub("", raw).strip() + if not line: + continue + match = LINE_RE.match(line) + stamped = STAMP_RE.match(match.group("rest")) if match else None + if match is None or stamped is None: + # Голову хвоста Komodo обрезает по месту, и от прохода к проходу она + # разная: недоразобранная строка - не находка, иначе каждый раз «ново». + unparsed += 1 + continue + lines += 1 + service = match.group("service") + at = datetime.fromisoformat(stamped.group("ts")).astimezone(UTC) + stamps.append(at) + rest = stamped.group("text") or "" + if SELF_MARK in rest: + continue + signal = signal_of(rest) + if signal is None: + continue + key = _key(stack, service, rest) + before = found.get(key) + found[key] = Hit( + key=key, + stack=stack, + service=service, + signal=signal, + sample=(before.sample if before else _mask(rest)[:SAMPLE_CHARS]), + at=at or (before.at if before else None), + count=(before.count if before else 0) + 1, + ) + return StackScan( + stack=stack, + hits=tuple(found.values()), + first=min(stamps) if stamps else None, + last=max(stamps) if stamps else None, + lines=lines, + unparsed=unparsed, + ) + + +def _forget(seen: Seen, now: datetime) -> None: + """Беду, которой давно не видно в хвостах, забываем: вернётся - будет новой. + + Считаем по времени прохода, в котором её видели, а не по времени события: + строка от прошлого месяца может спокойно лежать в хвосте тихого стека, и + забывать её - значит каждый проход объявлять её новой. + """ + for key, entry in list(seen.entries.items()): + if entry.told and entry.looked is not None and now - entry.looked > KEEP: + del seen.entries[key] + + +def digest(seen: Seen, sweep: Sweep) -> Report: + """Сравнить проход с прошлым: что новое, что стихло, о чём молчать.""" + gaps = sweep.gaps(seen.last_pass) + _forget(seen, sweep.at) + looked = {scan.stack for scan in sweep.read} + newcomers = looked - seen.baselined + hits = {hit.key: hit for scan in sweep.read for hit in scan.hits} + baseline = 0 + for key, hit in hits.items(): + entry = seen.entries.get(key) + if entry is None: + entry = Entry( + key=key, + stack=hit.stack, + service=hit.service, + signal=hit.signal, + sample=hit.sample, + first=hit.at or sweep.at, + looked=sweep.at, + ) + seen.entries[key] = entry + if hit.stack in newcomers: + # Стек читаем впервые: в хвосте лежит всё его прошлое, и вываливать + # его как «новое» - врать. Принимаем за данность, дальше - только рост. + entry.told = True + baseline += 1 + entry.last = hit.at or entry.last or sweep.at + entry.looked = sweep.at + entry.count += hit.count + seen.baselined |= looked + # Стихло - только у стеков, которые прочитались: чего не видели, того не судим. + quiet = tuple( + e + for e in seen.entries.values() + if e.told and e.stack in looked and e.key not in hits + ) + seen.last_pass = sweep.at + # Холодный старт без единого прочитанного стека базой не считается: иначе + # первый удачный проход вывалит всю область как «новое». + cold = seen.fresh and bool(sweep.read) + seen.fresh = seen.fresh and not sweep.read + fresh = tuple(e for e in seen.entries.values() if not e.told) + trouble = sweep.trouble if sweep.trouble != seen.trouble else () + return Report( + sweep=sweep, + new=tuple(sorted(fresh, key=lambda e: (e.stack, e.service))), + quiet=tuple(sorted(quiet, key=lambda e: (e.stack, e.service))), + trouble=trouble, + gaps=gaps, + cold=cold, + baseline=baseline, + baselined=tuple(sorted(newcomers)), + ) + + +def _when(at: datetime | None, tz: tzinfo) -> str: + return at.astimezone(tz).strftime("%d.%m %H:%M") if at else "время не в строке" + + +def _scope_line(sweep: Sweep, gaps: tuple[str, ...], tz: tzinfo) -> str: + since, until = sweep.span + where = ", ".join(sweep.scope) + if sweep.unreachable: + return f"Область: {where} - не прочитано ничего, списка стеков нет." + out = [ + ( + f"Область: {where} - прочитано {len(sweep.read)} стеков из " + f"{sweep.in_scope} в области (во флоте {sweep.fleet})" + ), + ( + f"хвост {sweep.tail} строк на стек: строки с {_when(since, tz)} " + f"по {_when(until, tz)}" + ), + ] + short = sweep.shortest + if short is not None: + out.append( + f"самый короткий хвост у {short.stack} - только с {_when(short.first, tz)}" + ) + blind = sum(s.unparsed for s in sweep.read) + if blind > UNPARSED_LOUD: + out.append(f"строк без метки времени пропущено {blind} - их не смотрел") + if gaps: + names = ", ".join(gaps[:GAP_NAMES]) + more = f" и ещё {len(gaps) - GAP_NAMES}" if len(gaps) > GAP_NAMES else "" + out.append( + f"у {len(gaps)} стеков хвост начался позже прошлого прохода " + f"({names}{more}) - между проходами часть строк прошла мимо" + ) + return "; ".join(out) + "." + + +def _entry_line(entry: Entry, tz: tzinfo, *, new: bool) -> str: + when = _when(entry.last, tz) + times = f" ×{entry.count}" if entry.count > 1 else "" + head = f"- {entry.where} {when}{times} {entry.signal}: " + if not new: + head = f"- {entry.where} {entry.signal}, последнее в {when}: " + return head + entry.sample + + +def brief(report: Report, tz: tzinfo = UTC) -> str: + """Инжект: что нового, где, сколько раз - и всегда область просмотра.""" + sweep = report.sweep + lines = [f"{SELF_MARK}: проход {sweep.at.astimezone(tz):%d.%m %H:%M}."] + lines.append(_scope_line(sweep, report.gaps, tz)) + if report.cold: + lines.append( + f"Первый проход: снял базовую линию, в хвостах {report.baseline} " + "таких строк - дальше дёргаю только тем, чего в прошлом проходе не было." + ) + elif report.baselined: + names = ", ".join(report.baselined[:GAP_NAMES]) + more = ( + f" и ещё {len(report.baselined) - GAP_NAMES}" + if len(report.baselined) > GAP_NAMES + else "" + ) + lines.append( + f"Впервые прочитаны ({report.baseline} строк в базу): {names}{more} - " + "их прошлое принял за данность, показываю только то, что появится дальше." + ) + if report.new: + lines.append(f"Новое ({len(report.new)}):") + lines.extend(_entry_line(e, tz, new=True) for e in report.new[:MAX_NEW]) + if len(report.new) > MAX_NEW: + lines.append(f"…и ещё {len(report.new) - MAX_NEW} - смотри komodo(logs).") + elif not report.cold and sweep.read: + lines.append("Нового в этой области нет.") + if report.quiet: + lines.append(f"Стихло ({len(report.quiet)}), время - события, не прохода:") + lines.extend(_entry_line(e, tz, new=False) for e in report.quiet[:MAX_QUIET]) + if len(report.quiet) > MAX_QUIET: + lines.append(f"…и ещё {len(report.quiet) - MAX_QUIET}.") + if report.trouble: + lines.append(f"Не смог посмотреть ({len(report.trouble)}):") + lines.extend(f"- {t}" for t in report.trouble[:MAX_TROUBLE]) + lines.append("Это только чтение: чинить или забить - решаешь ты.") + return "\n".join(lines) diff --git a/beaver_agent/jobs/__init__.py b/beaver_agent/jobs/__init__.py index f8d69ad..9e3342d 100644 --- a/beaver_agent/jobs/__init__.py +++ b/beaver_agent/jobs/__init__.py @@ -10,6 +10,7 @@ from beaver_agent.jobs.closing import close_idle from beaver_agent.jobs.deploy import deploy from beaver_agent.jobs.komodo import komodo_alert from beaver_agent.jobs.leisure import tick +from beaver_agent.jobs.logs import logs_watch from beaver_agent.jobs.memory import curate, memory from beaver_agent.jobs.rotation import rotate from beaver_agent.jobs.t3code import t3code_event @@ -29,6 +30,9 @@ jobs = [ Job("память", memory, cron="30 4 * * 0", critical=False), Job("голосовой", rotate_voice, cron="5 4 * * *", critical=False), Job("куратор", curate, cron="0 2,8,12,16,20 * * *", critical=False), + # Логи - раз в 3 часа и на ":40": на ":00" ротация и куратор. Чаще смысла + # мало (36 стеков = 36 чтений за проход), реже - хвост уезжает совсем. + Job("логи", logs_watch, cron="40 1,4,7,10,13,16,19,22 * * *", critical=False), Job("deploy", deploy, webhook=True), Job("komodo", komodo_alert, webhook=True, dedupe=False), Job("t3code", t3code_event, webhook=True, dedupe=False), diff --git a/beaver_agent/jobs/logs.py b/beaver_agent/jobs/logs.py new file mode 100644 index 0000000..4285e7e --- /dev/null +++ b/beaver_agent/jobs/logs.py @@ -0,0 +1,124 @@ +"""Раз в 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, + 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: + """Проход по логам: состояние на диске, инжект - только на новом.""" + if KOMODO is None: + _log.info("логи: руки komodo нет, проход пропущен") + return + seen = Seen.load(STATE_FILE) + report = digest(seen, await sweep(KOMODO)) + if not report.worth_telling: + seen.save() + _log.info( + "логи: нового нет, стеков прочитано %s из %s", + len(report.sweep.read), + report.sweep.in_scope, + ) + return + if await run.inject_master(brief(report, ZONE), urgency="normal", origin="логи"): + seen.mark_told(report) + seen.save() + _log.info( + "логи: новое %s, стихло %s, не прочитано %s", + len(report.new), + len(report.quiet), + len(report.trouble), + ) diff --git a/beaver_agent/vault.py b/beaver_agent/vault.py index 25c4f1a..8bb5abf 100644 --- a/beaver_agent/vault.py +++ b/beaver_agent/vault.py @@ -33,6 +33,9 @@ REPLIES = BEAVER / "реплики" CURATOR_JOURNAL = BEAVER / "куратор.md" LEISURE = BEAVER / "досуг.md" """Раскладка суток по тикам досуга: пишет её модель, механизм только читает.""" +STATE_DIR = Path(os.environ.get("BEAVER_STATE_DIR", VAULT / ".состояние")) +"""Машинное состояние джобов (не для чтения моделью): точка в начале - такой +каталог Obsidian не синкает и не показывает, а том `vault` переживает деплой.""" STATE = BEAVER / "состояние.md" STATE_MAX_LINES = 60 ATTACHMENTS = BEAVER / "вложения" diff --git a/tests/test_komodo_logs.py b/tests/test_komodo_logs.py new file mode 100644 index 0000000..04a73e9 --- /dev/null +++ b/tests/test_komodo_logs.py @@ -0,0 +1,465 @@ +from datetime import UTC, datetime, timedelta + +from beaver_agent.hands.komodo import Komodo +from beaver_agent.hands.komodo_logs import ( + KEEP, + SELF_MARK, + Seen, + StackScan, + Sweep, + brief, + digest, + read_stack, +) +from beaver_agent.jobs.logs import sweep as sweep_fleet + +NOW = datetime(2026, 9, 20, 18, 0, tzinfo=UTC) + + +def log(*lines): + return {"stdout": "\n".join(lines), "stderr": ""} + + +def line(service, at, text): + return f"{service} | {at}Z {text}" + + +def a_stack(name, server="dell-home"): + return { + "id": name, + "name": name, + "info": {"state": "running", "server_name": server}, + } + + +def fleet(answers, calls=None): + """Komodo, отвечающий заготовками; calls копит запросы к нему.""" + + class Fake(Komodo): + async def read(self, request, params): + if calls is not None: + calls.append((request, params)) + answer = answers.get(request, []) + return answer(params) if callable(answer) else answer + + return Fake(url="u", key="k", secret="s") + + +def swept(scans, **kw): + kw.setdefault("in_scope", len(scans)) + kw.setdefault("fleet", len(scans)) + return Sweep(at=kw.pop("at", NOW), scans=tuple(scans), **kw) + + +KNOWN = ("beaver-agent", "gitea", "democalc", "mix", "healthbot") + + +def seen(tmp_path, *, fresh=False, known=KNOWN): + """Состояние с уже снятой базовой линией: проход не первый.""" + state = Seen.load(tmp_path / "логи.json") + if not fresh: + state.fresh = False + state.baselined = set(known) + return state + + +def test_a_line_gives_its_service_its_time_and_its_signal(): + scan = read_stack( + "beaver-agent", + log( + line("beaver-gateway", "2026-09-20T17:40:01.123456789", "ERROR boom"), + line("beaver-gateway", "2026-09-20T17:41:02.000000000", "ERROR boom"), + line("beaver-obsidian", "2026-09-20T17:42:03.000000000", "Fully synced"), + ), + ) + (hit,) = scan.hits + assert hit.service == "beaver-gateway" + assert hit.signal == "ошибка" + assert hit.count == 2 + assert hit.at == datetime(2026, 9, 20, 17, 41, 2, tzinfo=UTC) + assert scan.first == datetime(2026, 9, 20, 17, 40, 1, 123456, tzinfo=UTC) + assert scan.last == datetime(2026, 9, 20, 17, 42, 3, tzinfo=UTC) + + +def test_the_same_trouble_with_other_numbers_is_the_same_trouble(): + scan = read_stack( + "gitea", + log( + line("gitea", "2026-09-20T17:40:01.0", "ERROR request 8871 failed"), + line("gitea", "2026-09-20T17:40:09.0", "ERROR request 9134 failed"), + ), + ) + (hit,) = scan.hits + assert hit.count == 2 + + +def test_every_named_signal_is_caught(): + texts = ( + "CRITICAL disk is gone", + "panic: runtime error", + "container OOMKilled", + "Traceback (most recent call last):", + "service exited with code 137", + "beaver-gateway Restarting (1) 3 seconds ago", + "container is unhealthy", + "ERROR plain", + ) + scan = read_stack( + "mix", + log(*(line("svc", f"2026-09-20T17:4{i}:00.0", t) for i, t in enumerate(texts))), + ) + assert {h.signal for h in scan.hits} == { + "критично", + "паника", + "oom", + "traceback", + "падение", + "рестарт-цикл", + "healthcheck", + "ошибка", + } + assert "exited with code 0" not in {h.sample for h in scan.hits} + + +def test_a_clean_exit_and_our_own_inject_are_not_trouble(): + scan = read_stack( + "beaver-agent", + log( + line("svc", "2026-09-20T17:40:00.0", "worker exited with code 0"), + line("beaver-gateway", "2026-09-20T17:41:00.0", f"{SELF_MARK}: ERROR boom"), + ), + ) + assert scan.hits == () + + +def test_komodos_own_error_is_not_a_clean_stack(): + scan = read_stack("gitea", {"error": "periphery unreachable"}) + assert not scan.ok + assert scan.error == "periphery unreachable" + assert Sweep(at=NOW, scans=(scan,), in_scope=1, fleet=1).trouble == ( + "gitea: periphery unreachable", + ) + + +def test_the_first_pass_only_takes_the_baseline(tmp_path): + state = seen(tmp_path, fresh=True) + scan = read_stack( + "beaver-agent", log(line("svc", "2026-09-20T17:40:00.0", "ERROR старая беда")) + ) + report = digest(state, swept([scan])) + assert report.cold and report.baseline == 1 + assert report.new == () + text = brief(report) + assert "базовую линию" in text + assert "прочитано 1 стеков из 1" in text + + +def test_only_the_new_line_wakes_the_master(tmp_path): + old = line("svc", "2026-09-20T17:40:00.0", "ERROR молотит третьи сутки") + state = seen(tmp_path, fresh=True) + first = digest(state, swept([read_stack("beaver-agent", log(old))])) + state.mark_told(first) + state.save() + + fresh_line = line("svc", "2026-09-20T18:50:00.0", "CRITICAL диск кончился") + state = Seen.load(tmp_path / "логи.json") + report = digest( + state, + swept( + [read_stack("beaver-agent", log(old, old, fresh_line))], + at=NOW + timedelta(hours=3), + ), + ) + (new,) = report.new + assert new.signal == "критично" + text = brief(report) + assert "диск кончился" in text + assert "молотит третьи сутки" not in text + + +def test_a_repeat_carries_how_many_times(tmp_path): + noisy = [ + line("svc", f"2026-09-20T17:4{i}:00.0", "ERROR раз за разом") for i in range(5) + ] + report = digest(seen(tmp_path), swept([read_stack("beaver-agent", log(*noisy))])) + assert "×5" in brief(report) + + +def test_a_trouble_that_stopped_is_dated_by_the_event_not_by_the_look(tmp_path): + was = line("svc", "2026-09-20T12:03:00.0", "ERROR было и прошло") + state = seen(tmp_path) + state.mark_told(digest(state, swept([read_stack("beaver-agent", log(was))]))) + + later = NOW + timedelta(hours=6) + report = digest(state, swept([read_stack("beaver-agent", log())], at=later)) + (quiet,) = report.quiet + assert quiet.last == datetime(2026, 9, 20, 12, 3, tzinfo=UTC) + text = brief(report) + assert "последнее в 20.09 12:03" in text + assert f"{later:%H:%M}" not in text.split("\n", 1)[1] + + +def test_a_stack_that_did_not_answer_is_not_a_stack_that_went_quiet(tmp_path): + was = line("svc", "2026-09-20T12:03:00.0", "ERROR было") + state = seen(tmp_path) + state.mark_told(digest(state, swept([read_stack("beaver-agent", log(was))]))) + + report = digest( + state, + swept( + [StackScan(stack="beaver-agent", error="timeout")], + at=NOW + timedelta(hours=3), + ), + ) + assert report.quiet == () + assert report.trouble == ("beaver-agent: timeout",) + assert "beaver-agent: timeout" in brief(report) + + +def test_an_undelivered_inject_leaves_the_trouble_new(tmp_path): + scan = read_stack( + "beaver-agent", log(line("svc", "2026-09-20T17:40:00.0", "ERROR боль")) + ) + state = seen(tmp_path) + first = digest(state, swept([scan])) + assert len(first.new) == 1 + state.save() # инжект не доехал: mark_told не звали + + state = Seen.load(state.path) + again = digest( + state, swept([read_stack("beaver-agent", log())], at=NOW + timedelta(hours=3)) + ) + assert [e.sample for e in again.new] == [e.sample for e in first.new] + assert again.quiet == () + + +def test_the_state_survives_a_restart(tmp_path): + scan = read_stack( + "beaver-agent", log(line("svc", "2026-09-20T17:40:00.0", "ERROR боль")) + ) + state = seen(tmp_path) + state.mark_told(digest(state, swept([scan]))) + state.save() + + state = Seen.load(state.path) + assert not state.fresh + report = digest(state, swept([scan], at=NOW + timedelta(hours=3))) + assert report.new == () and report.quiet == () + + +def test_an_old_line_still_lying_in_the_tail_is_not_new_again(tmp_path): + # Тихий стек держит августовскую строку в хвосте месяцами: она не новая. + scan = read_stack( + "democalc", log(line("pg-1", "2026-08-22T15:49:51.0", "ERROR: relation")) + ) + state = seen(tmp_path) + state.mark_told(digest(state, swept([scan]))) + state.save() + + state = Seen.load(state.path) + assert digest(state, swept([scan], at=NOW + timedelta(hours=3))).new == () + + +def test_a_trouble_forgotten_by_time_comes_back_as_new(tmp_path): + scan = read_stack( + "beaver-agent", log(line("svc", "2026-09-20T17:40:00.0", "ERROR боль")) + ) + state = seen(tmp_path) + state.mark_told(digest(state, swept([scan]))) + late = NOW + KEEP + timedelta(days=1) + report = digest(state, swept([scan], at=late)) + assert len(report.new) == 1 + + +def test_a_silent_pass_is_not_worth_the_master(tmp_path): + state = seen(tmp_path) + assert not digest(state, swept([read_stack("beaver-agent", log())])).worth_telling + + +def test_the_scope_says_how_much_was_looked_at(tmp_path): + scans = [ + read_stack( + "beaver-agent", log(line("svc", "2026-09-20T17:40:00.0", "ERROR боль")) + ), + StackScan(stack="gitea", error="timeout"), + ] + report = digest(seen(tmp_path), swept(scans, in_scope=2, fleet=53)) + head = brief(report).splitlines()[1] + assert head.startswith( + "Область: dell-home, personal-main - прочитано 1 стеков из 2" + ) + assert "во флоте 53" in head + assert "хвост 500 строк на стек: строки с 20.09 17:40 по 20.09 17:40" in head + assert "самый короткий хвост у beaver-agent - только с 20.09 17:40" in head + + +def test_a_tail_shorter_than_the_pause_between_passes_is_said_out_loud(tmp_path): + state = seen(tmp_path) + state.last_pass = NOW - timedelta(hours=3) + scan = read_stack( + "beaver-agent", + log(line("svc", "2026-09-20T17:59:00.0", "ERROR из последней минуты")), + ) + report = digest(state, swept([scan])) + assert report.gaps == ("beaver-agent",) + assert "часть строк прошла мимо" in brief(report) + + +def test_a_silent_komodo_says_so_instead_of_reporting_clean(tmp_path): + down = Sweep(at=NOW, unreachable="ClientError: connection refused") + report = digest(seen(tmp_path), down) + assert report.worth_telling + text = brief(report) + assert "Komodo не ответил" in text + assert "Нового в этой области нет" not in text + + +def test_a_lasting_outage_is_told_once(tmp_path): + down = Sweep(at=NOW, unreachable="ClientError: connection refused") + state = seen(tmp_path) + state.mark_told(digest(state, down)) + assert digest(state, down).trouble == () + + +async def test_the_sweep_lists_every_page_and_skips_other_peoples_machines(): + pages = [ + [a_stack(f"s{i}") for i in range(500)], + [a_stack("gitea", "personal-main"), a_stack("veris-runner", "veris-main")], + ] + calls = [] + komodo = fleet( + { + "ListStacks": lambda p: pages[p["page"]], + "GetStacksSummary": {"total": 502}, + "GetStackLog": log(), + }, + calls, + ) + out = await sweep_fleet(komodo, now=NOW) + asked = [p["stack"] for r, p in calls if r == "GetStackLog"] + assert "veris-runner" not in asked + assert "gitea" in asked and "s0" in asked + assert out.fleet == 502 and out.in_scope == 501 + assert out.trouble == () + assert all(p.get("timestamps") for r, p in calls if r == "GetStackLog") + + +async def test_the_sweep_shouts_when_the_list_and_the_summary_disagree(): + komodo = fleet( + { + "ListStacks": [a_stack("beaver-agent")], + "GetStacksSummary": {"total": 53}, + "GetStackLog": log(), + } + ) + out = await sweep_fleet(komodo, now=NOW) + assert out.trouble == ( + "листание дало 1 стеков, а GetStacksSummary - 53: список неполон", + ) + + +async def test_a_stack_that_throws_does_not_end_the_sweep(): + def logs(params): + if params["stack"] == "gitea": + msg = "connection reset" + raise RuntimeError(msg) + return log(line("svc", "2026-09-20T17:40:00.0", "ERROR боль")) + + komodo = fleet( + { + "ListStacks": [a_stack("beaver-agent"), a_stack("gitea", "personal-main")], + "GetStacksSummary": {"total": 2}, + "GetStackLog": logs, + } + ) + out = await sweep_fleet(komodo, now=NOW) + assert len(out.read) == 1 + assert out.trouble == ("gitea: RuntimeError: connection reset",) + + +async def test_a_dead_komodo_makes_a_sweep_that_admits_it(): + class Dead(Komodo): + async def read(self, request, params): + msg = "connection refused" + raise OSError(msg) + + out = await sweep_fleet(Dead(url="u", key="k", secret="s"), now=NOW) + assert out.read == () + assert out.unreachable == "OSError: connection refused" + assert out.trouble == ("Komodo не ответил: OSError: connection refused",) + + +def test_a_line_without_a_stamp_is_not_a_find(): + # Голову хвоста Komodo обрезает по месту: полстроки - не новая беда. + scan = read_stack( + "democalc", + log( + "49.260 UTC [8772] ERROR: relation does not exist", + line("postgres-1", "2026-09-20T17:40:00.0", "ERROR: настоящая"), + ), + ) + assert [h.sample for h in scan.hits] == ["ERROR: настоящая"] + assert scan.unparsed == 1 and scan.lines == 1 + + +def test_many_unreadable_lines_are_said_out_loud(tmp_path): + scan = StackScan(stack="beaver-agent", lines=5, unparsed=40) + head = brief(digest(seen(tmp_path), swept([scan]))).splitlines()[1] + assert "строк без метки времени пропущено 40" in head + + +def test_an_empty_line_is_a_line_not_a_blind_spot(): + scan = read_stack("dell-home-dawarich", log("app-1 | 2026-09-17T14:37:38.510Z")) + assert scan.unparsed == 0 and scan.lines == 1 and scan.hits == () + + +def test_a_stack_read_for_the_first_time_does_not_dump_its_past(tmp_path): + state = seen(tmp_path) + scan = read_stack( + "lodge", log(line("svc", "2026-09-19T10:00:00.0", "ERROR старое, до нас")) + ) + report = digest(state, swept([scan])) + assert report.new == () + assert report.baselined == ("lodge",) and report.baseline == 1 + assert "Впервые прочитаны" in brief(report) + + later = read_stack( + "lodge", + log( + line("svc", "2026-09-19T10:00:00.0", "ERROR старое, до нас"), + line("svc", "2026-09-20T19:00:00.0", "ERROR уже при нас"), + ), + ) + after = digest(state, swept([later], at=NOW + timedelta(hours=3))) + assert [e.sample for e in after.new] == ["ERROR уже при нас"] + + +def test_a_first_pass_into_a_dead_komodo_does_not_burn_the_baseline(tmp_path): + state = seen(tmp_path, fresh=True) + down = Sweep(at=NOW, unreachable="OSError: connection refused") + report = digest(state, down) + assert not report.cold and state.fresh + assert report.trouble == ("Komodo не ответил: OSError: connection refused",) + + scan = read_stack( + "beaver-agent", log(line("svc", "2026-09-20T17:40:00.0", "ERROR боль")) + ) + later = digest(state, swept([scan], at=NOW + timedelta(hours=3))) + assert later.cold and later.new == () + + +def test_a_secret_in_the_line_does_not_ride_into_the_chat(): + scan = read_stack( + "gitea", + log( + line( + "svc", + "2026-09-20T17:40:00.0", + "ERROR auth failed token=ghp_ZZ123 for kJ8sdfLKJ23kjhsdf98234hkjsdf7823x", + ) + ), + ) + (hit,) = scan.hits + assert "ghp_ZZ123" not in hit.sample + assert "kJ8sdfLKJ23kjhsdf98234hkjsdf7823x" not in hit.sample + assert hit.sample.startswith("ERROR auth failed")