beaver/beaver-agent
2
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(hands,jobs): журнал прогонов дописывается, шторм новых видов глушится громко

Browse Source
This commit is contained in:
hh committed 2026-09-20 21:36:08 +00:00
1 parent 06b4119ad3
commit 4b24795e9b
3 files changed
+391 -45
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

beaver_agent/hands/komodo_logs.py
+294 -43
View File
@@ -55,7 +55,9 @@ __all__ = [
"daily",
"digest",
"grouped",
"note",
"read_stack",
"record",
]
SCOPE = ("dell-home", "personal-main")
@@ -148,6 +150,30 @@ NORM_PASSES = 3
свою норму за собой, и порог не срабатывает никогда - варёная лягушка."""
SURGE_HOLD = timedelta(hours=12)
"""Одна беда - не чаще раза в полсуток, даже если растёт ступенями."""
MUTE_KINDS = 15
"""Столько новых ВИДОВ строк от одного стека за проход - это не пятнадцать
бед, а беда, печатающая каждый раз неповторимый текст (нормализация чисел и
хешей такую не схлопывает). Дальше стек глушится - но громко, строкой в
инжекте, а не молча.
Отсчитано от живых чисел (20.09.2026): самый шумный на новые виды стек -
`komodo`, 67 видов за сутки, то есть ~8 за проход; `personal-main-caddy` -
12 за сутки; у остальных 0-1. 15 за проход - вдвое выше худшего известного
фона. Цифра из рассуждения, не из замера шторма: пересмотреть после недели."""
MUTE_SHOW = 3
"""Сколько видов из шторма всё-таки показать - чтобы было видно, что глушится."""
MUTE_CLEAR = 3
"""Поток спал до стольких видов за проход - глушение снимается."""
MUTE_MIN = timedelta(hours=24)
"""Но не раньше суток: шторм идёт рывками, и без выдержки глушение мигало бы
через проход, а каждое включение - строка в инжекте."""
JOURNAL_MAX = 2_000_000
"""Байт на файл журнала, дальше - поворот в `.1`. При восьми проходах в сутки
это месяцы истории в двух файлах: пороги пересматривать будет по чему."""
JOURNAL_TROUBLES = 50
"""Сколько активных бед писать в строку прогона - самые громкие."""
TREND_UP, TREND_DOWN = 1.5, 0.67
"""Полтора раза в сутках - уже тренд, между ними - то же самое."""
@@ -288,6 +314,38 @@ class Entry:
return f"{self.stack}/{self.service}"
@dataclass(frozen=True, slots=True)
class Active:
"""Беда, которая прирастала в этом проходе: строка журнала прогонов."""
entry: Entry
gain: int
@dataclass(slots=True)
class Mute:
"""Приглушённый стек: сколько видов проглочено и с каких пор."""
stack: str
since: datetime
kinds: int = 0
"""Всего видов, проглоченных с начала глушения."""
day_kinds: int = 0
day_before: int = 0
told: bool = False
"""Сказали ли мастеру, что глушим. Не сказали - повторим в следующий проход:
молчаливое глушение - это ровно тот отказ, против которого оно и сделано."""
@dataclass(frozen=True, slots=True)
class Storm:
"""Шторм новых видов от одного стека - одной строкой вместо сорока."""
stack: str
kinds: int
shown: int
@dataclass(frozen=True, slots=True)
class Ongoing:
"""Длящееся одним рядом: стек, сервис и сигнал, а не отдельная строка.
@@ -334,6 +392,14 @@ class Report:
quiet: tuple[Entry, ...] = ()
ongoing: tuple[Entry, ...] = ()
"""Длящееся: молчит в обычном проходе, идёт в суточную сводку."""
active: tuple[Active, ...] = ()
"""Кто прирастал в этом проходе - это и пишется в журнал прогонов."""
storms: tuple[Storm, ...] = ()
"""Стеки, чей поток новых видов пришлось заглушить в этом проходе."""
unmuted: tuple[str, ...] = ()
"""Стеки, у которых поток спал и глушение снято."""
muted: tuple[Mute, ...] = ()
"""Кто приглушён сейчас - напоминание раз в сутки, пока не спадёт."""
summary: bool = False
"""Сутки закрылись и сводку за них ещё не отдавали. Она уходит всегда -
и когда длящегося нет: молчание должно быть отличимо от мёртвого джоба."""
@@ -359,7 +425,15 @@ class Report:
@property
def worth_telling(self) -> bool:
"""Суточная сводка сюда не входит: она едет своим инжектом."""
return bool(self.new or self.surge or self.quiet or self.trouble or self.cold)
return bool(
self.new
or self.surge
or self.quiet
or self.trouble
or self.cold
or self.storms
or self.unmuted
)
@dataclass(slots=True)
@@ -370,6 +444,8 @@ class Seen:
entries: dict[str, Entry] = field(default_factory=dict)
baselined: set[str] = field(default_factory=set)
"""Стеки, чьё прошлое уже принято за данность: их старое не «новое»."""
mutes: dict[str, Mute] = field(default_factory=dict)
"""Стеки в приглушённом режиме: их новые виды глотаются, но считаются."""
trouble: tuple[str, ...] = ()
"""О каких отказах уже сказали: Komodo лежит - скажем один раз, не каждые 3 ч."""
last_pass: datetime | None = None
@@ -427,6 +503,17 @@ class Seen:
path=path,
entries=entries,
baselined=set(raw.get("baselined", ())),
mutes={
stack: Mute(
stack=stack,
since=_parse(row.get("since")) or EPOCH,
kinds=int(row.get("kinds", 0)),
day_kinds=int(row.get("day_kinds", 0)),
day_before=int(row.get("day_before", 0)),
told=bool(row.get("told")),
)
for stack, row in raw.get("mutes", {}).items()
},
trouble=tuple(raw.get("trouble", ())),
last_pass=_parse(raw.get("last_pass")),
day=raw.get("day"),
@@ -455,6 +542,16 @@ class Seen:
"blind_before": self.blind_before,
"summary_day": self.summary_day,
"baselined": sorted(self.baselined),
"mutes": {
stack: {
"since": _iso(m.since),
"kinds": m.kinds,
"day_kinds": m.day_kinds,
"day_before": m.day_before,
"told": m.told,
}
for stack, m in self.mutes.items()
},
"trouble": list(self.trouble),
"entries": {
key: {
@@ -497,6 +594,10 @@ class Seen:
item.entry.surged = report.sweep.at
for entry in report.quiet:
entry.quieted = report.sweep.at
for storm in report.storms:
mute = self.mutes.get(storm.stack)
if mute is not None:
mute.told = True
self.trouble = report.trouble
def mark_daily(self, report: Report) -> None:
@@ -609,6 +710,8 @@ def _roll(seen: Seen, day: str) -> None:
entry.day_earlier = entry.day_before
entry.day_before = entry.day_count
entry.day_count = 0
for mute in seen.mutes.values():
mute.day_before, mute.day_kinds = mute.day_kinds, 0
seen.closed = seen.day
seen.passes_before, seen.passes_today = seen.passes_today, 0
seen.misses_before, seen.misses_today = seen.misses_today, 0
@@ -616,6 +719,34 @@ def _roll(seen: Seen, day: str) -> None:
seen.day = day
def _mutes(
seen: Seen, kinds: dict[str, int], at: datetime
) -> tuple[tuple[Storm, ...], tuple[str, ...], dict[str, int]]:
"""Кого глушим, с кого глушение снимаем и сколько видов всё же показать.
Глушение никогда не бывает тихим: пока про него не сказали, оно едет в
инжект каждый проход, а пока держится - напоминает о себе в суточной сводке.
"""
storms: list[Storm] = []
unmuted: list[str] = []
budget: dict[str, int] = {}
for stack, mute in list(seen.mutes.items()):
calm = kinds.get(stack, 0) <= MUTE_CLEAR
if calm and mute.told and at - mute.since >= MUTE_MIN:
del seen.mutes[stack]
unmuted.append(stack)
for stack, count in kinds.items():
if stack in seen.mutes:
if not seen.mutes[stack].told:
storms.append(Storm(stack=stack, kinds=count, shown=MUTE_SHOW))
continue
if count > MUTE_KINDS:
seen.mutes[stack] = Mute(stack=stack, since=at)
budget[stack] = MUTE_SHOW
storms.append(Storm(stack=stack, kinds=count, shown=MUTE_SHOW))
return tuple(storms), tuple(sorted(unmuted)), budget
def _surging(entry: Entry, gain: int, at: datetime) -> bool:
"""Стал ли знакомый фон снова новостью.
@@ -625,6 +756,11 @@ def _surging(entry: Entry, gain: int, at: datetime) -> bool:
"""
if not entry.told or entry.quieted is not None:
return False
if not entry.passes:
# Первая встреча: в хвосте лежит суточный запас строк, и «прирост» тут
# говорит про длину хвоста, а не про скорость беды. Ни кратность, ни
# потолок к нему не применимы - это база, а не всплеск.
return False
if entry.surged is not None and at - entry.surged <= SURGE_HOLD:
return False
if gain >= SURGE_CEILING:
@@ -636,6 +772,76 @@ def _surging(entry: Entry, gain: int, at: datetime) -> bool:
)
@dataclass(slots=True)
class _Take:
"""Что проход сделал с записями состояния: для сборки отчёта."""
baseline: int = 0
born: set[str] = field(default_factory=set)
surge: list[Surge] = field(default_factory=list)
active: list[Active] = field(default_factory=list)
def _absorb(
seen: Seen,
sweep: Sweep,
hits: dict[str, Hit],
newcomers: set[str],
budget: dict[str, int],
) -> _Take:
"""Разложить находки прохода по состоянию: прирост, норма, глушение."""
take = _Take()
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.times[0] if hit.times else sweep.at,
looked=sweep.at,
)
seen.entries[key] = entry
take.born.add(key)
mute = seen.mutes.get(hit.stack)
if mute is not None:
# Шторм: показываем первые несколько, остальное глотаем - но в
# счёт, и стек идёт в суточную сводку как приглушённый.
if budget.get(hit.stack, 0) > 0:
budget[hit.stack] -= 1
else:
entry.told = True
mute.kinds += 1
mute.day_kinds += 1
if hit.stack in newcomers:
# Стек читаем впервые: в хвосте лежит всё его прошлое, и вываливать
# его как «новое» - врать. Принимаем за данность, дальше - только рост.
entry.told = True
take.baseline += 1
entry.looked = sweep.at
gain = hit.since(entry.last)
if not gain:
# Та же строка, что и в прошлый проход: хвост перечитан, беда не росла.
continue
if _surging(entry, gain, sweep.at):
take.surge.append(Surge(entry=entry, gain=gain, norm=entry.norm))
if entry.quieted is not None:
# Стихшая беда вернулась: это снова новость, а не продолжение.
entry.quieted = None
entry.told = False
entry.count += gain
entry.passes += 1
entry.day_count += gain
if entry.passes <= NORM_PASSES:
# Норма берётся по молодости беды и дальше не трогается.
entry.norm = entry.count / entry.passes
entry.last = hit.at or entry.last or sweep.at
take.active.append(Active(entry=entry, gain=gain))
return take
def digest(seen: Seen, sweep: Sweep, tz: tzinfo = UTC) -> Report:
"""Сравнить проход с прошлым: что новое, что всплеск, что стихло, что длится."""
gaps = sweep.gaps(seen.last_pass)
@@ -648,46 +854,12 @@ def digest(seen: Seen, sweep: Sweep, tz: tzinfo = UTC) -> Report:
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
born: set[str] = set()
surge: list[Surge] = []
kinds: dict[str, int] = {}
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.times[0] if hit.times else sweep.at,
looked=sweep.at,
)
seen.entries[key] = entry
born.add(key)
if hit.stack in newcomers:
# Стек читаем впервые: в хвосте лежит всё его прошлое, и вываливать
# его как «новое» - врать. Принимаем за данность, дальше - только рост.
entry.told = True
baseline += 1
entry.looked = sweep.at
gain = hit.since(entry.last)
if not gain:
# Та же строка, что и в прошлый проход: хвост перечитан, беда не росла.
continue
if _surging(entry, gain, sweep.at):
surge.append(Surge(entry=entry, gain=gain, norm=entry.norm))
if entry.quieted is not None:
# Стихшая беда вернулась: это снова новость, а не продолжение.
entry.quieted = None
entry.told = False
entry.count += gain
entry.passes += 1
entry.day_count += gain
if entry.passes <= NORM_PASSES:
# Норма берётся по молодости беды и дальше не трогается.
entry.norm = entry.count / entry.passes
entry.last = hit.at or entry.last or sweep.at
if key not in seen.entries and hit.stack not in newcomers:
kinds[hit.stack] = kinds.get(hit.stack, 0) + 1
storms, unmuted, budget = _mutes(seen, kinds, sweep.at)
take = _absorb(seen, sweep, hits, newcomers, budget)
seen.baselined |= looked
# Стихло - только про то, о чём говорили, и только у прочитанных стеков:
# чего не видели, того не судим, а про фон мастер и не слышал.
@@ -697,7 +869,7 @@ def digest(seen: Seen, sweep: Sweep, tz: tzinfo = UTC) -> Report:
if e.announced
and e.quieted is None
and e.stack in looked
and e.key not in born
and e.key not in take.born
and (
e.key not in hits
or (e.last is not None and sweep.at - e.last > QUIET_AFTER)
@@ -722,9 +894,13 @@ def digest(seen: Seen, sweep: Sweep, tz: tzinfo = UTC) -> Report:
return Report(
sweep=sweep,
new=tuple(sorted(fresh, key=lambda e: (e.stack, e.service))),
surge=tuple(sorted(surge, key=lambda s: -s.gain)),
surge=tuple(sorted(take.surge, key=lambda s: -s.gain)),
quiet=tuple(sorted(quiet, key=lambda e: (e.stack, e.service))),
ongoing=tuple(sorted(ongoing, key=lambda e: -e.day_before)),
active=tuple(sorted(take.active, key=lambda a: -a.gain)),
storms=storms,
unmuted=unmuted,
muted=tuple(seen.mutes.values()),
summary=seen.closed is not None and seen.summary_day != day,
day=seen.closed or "",
today=day,
@@ -734,11 +910,61 @@ def digest(seen: Seen, sweep: Sweep, tz: tzinfo = UTC) -> Report:
trouble=trouble,
gaps=gaps,
cold=cold,
baseline=baseline,
baseline=take.baseline,
baselined=tuple(sorted(newcomers)),
)
def record(report: Report, tz: tzinfo = UTC) -> dict[str, Any]:
"""Строка журнала прогонов: по ней через неделю пересматривают пороги."""
sweep = report.sweep
return {
"at": sweep.at.astimezone(tz).isoformat(timespec="seconds"),
"day": report.today,
"pass_of_day": None,
"stacks": len(sweep.read),
"in_scope": sweep.in_scope,
"fleet": sweep.fleet,
"misses": max(sweep.in_scope - len(sweep.read), 0),
"blind": 1 if sweep.unreachable else 0,
"tail": sweep.tail,
"lines": sum(s.lines for s in sweep.read),
"unparsed": sum(s.unparsed for s in sweep.read),
"new": len(report.new),
"surge": len(report.surge),
"quiet": len(report.quiet),
"ongoing": len(report.ongoing),
"active": len(report.active),
"muted": {m.stack: m.kinds for m in report.muted},
"troubles": [
{
"key": a.entry.key,
"where": a.entry.where,
"signal": a.entry.signal,
"gain": a.gain,
"total": a.entry.count,
"norm": round(a.entry.norm, 2),
"day": a.entry.day_count,
}
for a in report.active[:JOURNAL_TROUBLES]
],
}
def note(path: Path, row: dict[str, Any], *, max_bytes: int = JOURNAL_MAX) -> None:
"""Дописать строку прогона в jsonl; файл дорос - поворот в `.1`.
Состояние - снимок и перезаписывается, по нему истории не собрать. Журнал
только дописывается: пороги брались из рассуждения, и пересматривать их
через неделю можно будет по числам, а не по памяти.
"""
path.parent.mkdir(parents=True, exist_ok=True)
if path.exists() and path.stat().st_size > max_bytes:
path.replace(path.with_suffix(path.suffix + ".1"))
with path.open("a", encoding="utf-8") as fh:
fh.write(json.dumps(row, ensure_ascii=False) + "\n")
def _when(at: datetime | None, tz: tzinfo) -> str:
return at.astimezone(tz).strftime("%d.%m %H:%M") if at else "время не в строке"
@@ -813,6 +1039,22 @@ def grouped(entries: tuple[Entry, ...]) -> tuple[Ongoing, ...]:
return tuple(sorted(out, key=lambda o: -o.day_before))
def _storm_line(storm: Storm) -> str:
return (
f"Стек {storm.stack}: {storm.kinds} новых ВИДОВ строк за проход - это не "
f"{storm.kinds} бед, а строки с неповторимым текстом. Показываю "
f"{storm.shown}, остальное глушу и напоминаю раз в сутки, пока не спадёт."
)
def _mute_line(mute: Mute, tz: tzinfo) -> str:
return (
f"Приглушён {mute.stack}: за те сутки проглочено {mute.day_before} новых "
f"видов строк (всего {mute.kinds}), глушу с {_when(mute.since, tz)} - "
"поток пока не спал."
)
def _daily_line(row: Ongoing, tz: tzinfo) -> str:
kinds = f", строк {row.kinds} видов" if row.kinds > 1 else ""
return (
@@ -843,6 +1085,10 @@ def daily(report: Report, tz: tzinfo = UTC) -> str:
f"Отказов доступа за те сутки: чтений стеков {report.misses}, "
f"проходов без ответа Komodo {report.blind}."
)
lines.extend(
_mute_line(mute, tz)
for mute in sorted(report.muted, key=lambda m: -m.day_before)
)
if not rows:
lines.append("Длящегося нет: всё, о чём я говорил, стихло или не повторялось.")
return "\n".join(lines)
@@ -894,6 +1140,11 @@ def brief(report: Report, tz: tzinfo = UTC) -> str:
"но её стало в разы больше своей нормы:"
)
lines.extend(_surge_line(item, tz) for item in report.surge[:MAX_NEW])
lines.extend(_storm_line(storm) for storm in report.storms)
if report.unmuted:
lines.append(
f"Поток новых видов спал, глушение снял: {', '.join(report.unmuted)}."
)
if report.quiet:
lines.append(f"Стихло ({len(report.quiet)}), время - события, не прохода:")
lines.extend(_entry_line(e, tz, new=False) for e in report.quiet[:MAX_QUIET])