From 91bd50a66d1f931e13d834e954fba1219da53646 Mon Sep 17 00:00:00 2001 From: h Date: Tue, 22 Sep 2026 20:15:15 +0200 Subject: [PATCH] feat(hands,jobs): job-run watchdog looks a week back and alarms on jobs that stopped running --- README.md | 48 ++++-- beaver_agent/hands/heartbeat.py | 14 +- beaver_agent/hands/runs.py | 291 +++++++++++++++++++++++++++++--- beaver_agent/jobs/__init__.py | 7 +- beaver_agent/jobs/runs.py | 97 ++++++++--- tests/test_runs.py | 234 ++++++++++++++++++++++++- 6 files changed, 620 insertions(+), 71 deletions(-) diff --git a/README.md b/README.md index b241da8..7b0a7e3 100644 --- a/README.md +++ b/README.md @@ -277,7 +277,7 @@ that 231 times, and nothing reached a human: the outage surfaced ten days later, by hand, out of curiosity. The record existed, the reader did not. This job is the reader. -- `прогоны` (`:27` every hour) reads `job_runs` for the last 24 hours through +- `прогоны` (`:27` every hour) reads `job_runs` for the last 7 days through the same connection the scheduler itself writes with (`run.conversations.db.session()` and the `JobRunRecord` model) - no second truth about where the journal lives. @@ -290,8 +290,31 @@ the reader. over the real outage it reads: «джоб `вайбграм` лежит 23 часа и дольше, 144 прогона подряд с 12.09 22:00 UTC: HubError: vibegram 530 bad_json». It repeats at most once a day per job, and when the job recovers one short `отбой` line goes out and the counter resets. + The window is a week, not a day, so a daily job (`отметка`, `закрытие`, + `голосовой`) gets its three failures in a row in three days. +- **A job that stopped running is an alarm too, with its own text**: «джоб + `закрытие` не запускался 2 суток при периоде 24 часа». For every job with a + cron, the period comes from the registered cron (`run.scheduler.jobs`) via + the same `every_from_cron` the neighbour watchdog uses, and the last run is + taken from the whole journal, not the window. The threshold is **two + periods**: one missed fire happens honestly (a redeploy landed on the cron + minute, and the scheduler does not catch up), two in a row do not. It is + never finer than two of the watchdog's own periods (2 hours), because an + hourly watcher cannot tell a shorter gap from its own blindness between + runs. Time when the scheduler was not running non-critical jobs at all is + not counted as a miss: before the process started, and while the + watchdog's own hourly chain is broken (the gateway was down, or the budget + was throttled and non-critical crons were deferred without a record). Same + discipline as failures: once a day per job, and an `отбой` when a run shows + up. +- What silence does not cover, on purpose: jobs without a cron (webhooks, + events) are not measured at all; a job that has never appeared in the + journal waits silently for its first row; and a cron that does not reduce to + a regular period (`куратор`, `0 2,8,12,16,20 * * *`) is skipped and named in + the log as unguarded every run, because any number put in its place would + be made up. `куратор` still falls under the failure count. - **An empty answer is a statement about the query, not about health.** Zero - rows in 24 hours (other than this job's own) means no job ran at all, and + rows in the last 24 hours (other than this job's own) means no job ran at all, and that is its own alarm with its own text - the scheduler is what to look at, not any one job. A failed query is no reading whatsoever: the run logs it and stays silent, because a database that did not answer is not evidence of @@ -303,22 +326,15 @@ the reader. carrying the key can be named anything. State between runs is `мета/бобер/.состояние/прогоны.json` (who was told and -when). A lost file means a lying job is reported again and an `отбой` for +when, separately for failures and for silence). A lost file means a lying job is reported again and an `отбой` for something reported before the loss never comes - the evidence for it is gone. -Two blind spots, both deliberate: - -- **The watchdog cannot report its own death.** Its own runs are excluded from - the count (`run.job.name`), because a job that is down will not file a report - about itself - an observer inside the object does not report the object's - death. If `прогоны` itself starts failing, nothing here will say so; that - needs an outside eye, like the watchdog ring above. -- **The window is 24 hours, so a job that runs less than three times a day can - never accumulate three consecutive failures inside it.** `отметка`, - `закрытие`, `голосовой` (daily) and `память` (weekly) are therefore - unwatched by this job in practice; `вайбграм`, `логи`, `куратор`, `досуг`, - `ротация` and `сторож` are inside it. Widening the counting window past the - query window is the fix, when it is worth it. +One blind spot, deliberate: **the watchdog cannot report its own death.** Its +own runs are excluded from both the failure count and the silence check +(`run.job.name`), because a job that is down will not file a report about +itself - an observer inside the object does not report the object's death. If +`прогоны` itself starts failing or stops running, nothing here will say so; +that needs an outside eye, like the watchdog ring above. ## Policy (`beaver_agent/policy.py`) diff --git a/beaver_agent/hands/heartbeat.py b/beaver_agent/hands/heartbeat.py index a3a40a7..9904475 100644 --- a/beaver_agent/hands/heartbeat.py +++ b/beaver_agent/hands/heartbeat.py @@ -151,13 +151,19 @@ def parse_beat(text: str, kind: str = "hb") -> Beat | None: def every_from_cron( - expression: str | None, tz: tzinfo, *, now: datetime | None = None + expression: str | None, + tz: tzinfo, + *, + now: datetime | None = None, + grain: timedelta = timedelta(hours=1), ) -> timedelta | None: - """Шаг крона, если он сводится к «раз в N часов»; иначе None. + """Шаг крона, если он сводится к «раз в N `grain`»; иначе None. Считаем по стенным часам зоны, в которой планировщик читает кроны: суточный крон - это 24 часа и в сутки перевода стрелок тоже, а вот - `0 2,8,12,16,20 * * *` не сводится ни к какому N и даёт None. + `0 2,8,12,16,20 * * *` не сводится ни к какому N и даёт None. `grain` + по умолчанию час - такой период печатается в отметке соседу; сторожу + прогонов нужен и `*/10`, он просит минуту. """ if not expression: return None @@ -171,7 +177,7 @@ def every_from_cron( if len(steps) != 1: return None step = steps.pop() - return step if step.total_seconds() % 3600 == 0 else None + return step if step % grain == timedelta(0) else None def beat_line( diff --git a/beaver_agent/hands/runs.py b/beaver_agent/hands/runs.py index 37eafc1..049e428 100644 --- a/beaver_agent/hands/runs.py +++ b/beaver_agent/hands/runs.py @@ -11,6 +11,17 @@ - тревога берётся только с ХВОСТА: три и более `failed` подряд, считая от последнего прогона джоба. Одиночный упавший прогон - сетевой чих, а не болезнь, и будить на него человека значит выучить его не смотреть; +- окно счёта - неделя, а не сутки: иначе суточный джоб не наберёт трёх + падений подряд никогда, и `отметка`, `закрытие`, `голосовой`, `память` + не охранялись бы вовсе; +- отсутствие записей - тоже показание. Джоб с регулярным кроном, у которого + последний прогон старше `FACTOR` его периодов, «не запускался» - это + отдельная тревога, не «лежит». Период выводится из крона тем же + `every_from_cron`, что у сторожа соседа, и расписание, которое не + сводится к «раз в N часов», в давности не участвует и называется в логе + неохраняемым: подставить ему число значило бы соврать. Джобы без крона + (вебхуки, события) давностью не меряются вовсе, а джоб, которого в + журнале нет ни разу, молча ждёт своей первой записи; - пустая выдача - это показание о запросе, а не о здоровье джобов. Ноль строк за сутки (кроме своих) означает, что планировщик вообще не работал, и это отдельная тревога с отдельным текстом, а не тишина; @@ -38,13 +49,19 @@ from datetime import UTC, datetime, timedelta from pathlib import Path # noqa: TC003 - в аннотации dataclass from typing import TYPE_CHECKING +from beaver_gateway.jobs.scheduler import next_run from beaver_gateway.security.redact import MASK, is_secret_name, redact +from beaver_agent.hands.heartbeat import every_from_cron + if TYPE_CHECKING: - from collections.abc import Sequence + from collections.abc import Mapping, Sequence + from datetime import tzinfo __all__ = [ + "DAY", "ERROR_CHARS", + "FACTOR", "FAILED", "FORGET", "MAX_LINES", @@ -52,22 +69,39 @@ __all__ = [ "STREAK", "WINDOW", "Clear", + "Quiet", "Report", + "Resumed", "Run", "Runs", "Streak", "alarm_text", + "awake", "clear_text", "empty_text", "first_line", "look", "mask", + "quiet_text", "streak", ] -WINDOW = timedelta(hours=24) -"""Окно журнала: сутки. Джоб реже трёх раз в сутки за него не наберёт серии -из трёх падений - это слепая зона, названная в README.""" +WINDOW = timedelta(days=7) +"""Окно журнала: неделя. Суточный джоб набирает в нём три падения подряд за +трое суток, недельный (`память`) - нет: у него три падения это три недели, +и его охраняет давность, а не серия.""" +DAY = timedelta(hours=24) +"""Окно тревоги о пустом журнале: сутки без единого чужого прогона.""" +FACTOR = 2 +"""Давность = столько периодов джоба без записи. Два - это «пропущено два +срабатывания подряд»: один пропуск случается честно (передеплой пришёлся на +минуту крона, а догонять пропущенное планировщик не умеет), два подряд - +уже не случайность.""" +GRAIN = timedelta(minutes=1) +"""Шаг, в котором период крона должен сводиться к целому: крон минутный.""" +LAG = timedelta(minutes=10) +"""Запись в журнал ложится по КОНЦУ прогона: срабатывание моложе этого +могло ещё не дописаться, и пропуском его не считаем.""" STREAK = 3 """Сколько `failed` подряд считаем «джоб лежит».""" REPEAT = timedelta(hours=24) @@ -80,6 +114,7 @@ ERROR_CHARS = 160 MAX_LINES = 10 """Сколько джобов перечисляем в одной тревоге: сразу всё не читают.""" FAILED = "failed" +_DAYS_FROM = timedelta(hours=48) STATE_VERSION = 1 _PAIR = re.compile(r"(?=([A-Za-z][\w.-]{2,})\s*([:=])\s*([^\s,;)\]}\"']{6,}))") @@ -117,14 +152,29 @@ def _plural(count: int, forms: tuple[str, str, str]) -> str: def _age(delta: timedelta) -> str: - """Возраст человеку: «4 часа», «45 минут». Секунды тут не нужны никому.""" + """Возраст человеку: «4 часа», «45 минут», «15 суток». Секунды не нужны.""" minutes = max(int(delta.total_seconds()) // 60, 0) if minutes < 60: return f"{minutes} {_plural(minutes, ('минуту', 'минуты', 'минут'))}" hours = minutes // 60 + if delta >= _DAYS_FROM: + days = hours // 24 + return f"{days} {_plural(days, ('сутки', 'суток', 'суток'))}" return f"{hours} {_plural(hours, ('час', 'часа', 'часов'))}" +def _period(delta: timedelta) -> str: + """Период крона: «10 минут», «24 часа», «7 суток» - без округления.""" + minutes = int(delta.total_seconds()) // 60 + if minutes % (24 * 60) == 0 and delta >= _DAYS_FROM: + days = minutes // (24 * 60) + return f"{days} {_plural(days, ('сутки', 'суток', 'суток'))}" + if minutes % 60 == 0: + hours = minutes // 60 + return f"{hours} {_plural(hours, ('час', 'часа', 'часов'))}" + return f"{minutes} {_plural(minutes, ('минута', 'минуты', 'минут'))}" + + def _mask_pairs(text: str) -> str: out: list[str] = [] pos = 0 @@ -182,7 +232,25 @@ class Streak: error: str capped: bool """Серия упёрлась в край окна: падений было НЕ МЕНЬШЕ `count`, а сколько - на самом деле - за сутками не видно, и врать точным числом нельзя.""" + на самом деле - за неделей не видно, и врать точным числом нельзя.""" + + +@dataclass(frozen=True, slots=True) +class Quiet: + """Джоб с кроном, который перестал запускаться.""" + + job: str + age: timedelta + """Сколько прошло с начала последнего прогона - любого, хоть упавшего.""" + period: timedelta + + +@dataclass(frozen=True, slots=True) +class Resumed: + """Джоб, о молчании которого говорили, снова запустился.""" + + job: str + status: str @dataclass(frozen=True, slots=True) @@ -203,6 +271,8 @@ class Runs: path: Path alerted: dict[str, datetime] = field(default_factory=dict) """Джоб → когда о нём ушла тревога. Он же признак «сейчас в тревоге».""" + quiet: dict[str, datetime] = field(default_factory=dict) + """То же для «не запускался»: отдельно, у тревог разные отбои.""" empty_alerted: datetime | None = None @classmethod @@ -213,14 +283,10 @@ class Runs: return cls(path=path) if raw.get("version") != STATE_VERSION: return cls(path=path) - alerted = {} - for job, when in dict(raw.get("alerted") or {}).items(): - parsed = _parse_iso(str(when)) - if parsed is not None: - alerted[str(job)] = parsed return cls( path=path, - alerted=alerted, + alerted=_stamps(raw.get("alerted")), + quiet=_stamps(raw.get("quiet")), empty_alerted=_parse_iso(raw.get("empty_alerted")), ) @@ -228,6 +294,7 @@ class Runs: body = { "version": STATE_VERSION, "alerted": {job: _iso(when) for job, when in sorted(self.alerted.items())}, + "quiet": {job: _iso(when) for job, when in sorted(self.quiet.items())}, "empty_alerted": _iso(self.empty_alerted), } self.path.parent.mkdir(parents=True, exist_ok=True) @@ -236,6 +303,15 @@ class Runs: tmp.replace(self.path) +def _stamps(raw: dict[str, str] | None) -> dict[str, datetime]: + out = {} + for job, when in (raw or {}).items(): + parsed = _parse_iso(str(when)) + if parsed is not None: + out[str(job)] = parsed + return out + + @dataclass(frozen=True, slots=True) class Report: """Что решил один прогон сторожа журнала.""" @@ -249,6 +325,10 @@ class Report: mine: int """Сколько своих - их в счёт не берём, но в тексте называем.""" note: str + quiet: tuple[Quiet, ...] = () + resumed: tuple[Resumed, ...] = () + unguarded: tuple[str, ...] = () + """Джобы с кроном, у которого нет регулярного периода: давностью не меряем.""" def streak(runs: Sequence[Run], *, now: datetime) -> Streak | None: @@ -271,27 +351,41 @@ def streak(runs: Sequence[Run], *, now: datetime) -> Streak | None: ) -def look(state: Runs, runs: Sequence[Run], *, now: datetime, me: str) -> Report: +def look( + state: Runs, + runs: Sequence[Run], + *, + now: datetime, + me: str, + crons: Mapping[str, str] | None = None, + latest: Mapping[str, Run] | None = None, + tz: tzinfo = UTC, + boot: datetime | None = None, +) -> Report: """Разобрать окно журнала и решить, о чём говорить человеку. + `crons` - джоб → крон, только джобы с кроном; `latest` - последний прогон + каждого джоба за ВСЮ историю журнала, а не за окно; `boot` - старт + процесса. Без `crons` давность не проверяется вовсе. + Состояние здесь только чистится от забытых джобов; отметки «сказали» и «сказали про пустоту» ставит джоб - и только когда тревога действительно ушла: несостоявшаяся доставка не должна заглушать следующий прогон. """ - mine = sum(1 for run in runs if run.job == me) + mine = [run for run in runs if run.job == me] others = sorted( (run for run in runs if run.job != me), key=lambda run: run.id, reverse=True ) - if not others: + if not any(now - run.started_at < DAY for run in others): quiet = state.empty_alerted is not None and now - state.empty_alerted < REPEAT return Report( alarms=(), clears=(), empty=not quiet, - seen=0, - mine=mine, + seen=len(others), + mine=len(mine), note=( - f"чужих прогонов за сутки нет вовсе (своих {mine})" + f"чужих прогонов за сутки нет вовсе (своих {len(mine)})" + ("; о пустоте уже говорили меньше суток назад" if quiet else "") ), ) @@ -328,7 +422,9 @@ def look(state: Runs, runs: Sequence[Run], *, now: datetime, me: str) -> Report: for job in forgotten: del state.alerted[job] - note = f"прогонов за сутки {len(others)} по {len(by_job)} джобам (своих {mine})" + note = ( + f"прогонов за неделю {len(others)} по {len(by_job)} джобам (своих {len(mine)})" + ) if alarms: note += "; лежат: " + ", ".join(sorted(tail.job for tail in alarms)) if held: @@ -338,12 +434,140 @@ def look(state: Runs, runs: Sequence[Run], *, now: datetime, me: str) -> Report: if forgotten: note += "; забыты (пропали из окна): " + ", ".join(sorted(forgotten)) + silence = _silence( + state, + mine, + now=now, + me=me, + crons=crons or {}, + latest=latest or {}, + tz=tz, + boot=boot, + ) + return Report( alarms=tuple(sorted(alarms, key=lambda tail: tail.count, reverse=True)), clears=tuple(sorted(clears, key=lambda clear: clear.job)), empty=False, seen=len(others), - mine=mine, + mine=len(mine), + note=note + silence.note, + quiet=silence.quiet, + resumed=silence.resumed, + unguarded=silence.unguarded, + ) + + +@dataclass(frozen=True, slots=True) +class _Silence: + quiet: tuple[Quiet, ...] + resumed: tuple[Resumed, ...] + unguarded: tuple[str, ...] + note: str + + +def awake( + mine: Sequence[Run], *, now: datetime, step: timedelta, boot: datetime | None +) -> datetime: + """С какого момента планировщик заведомо крутил некритичные джобы. + + Показание - собственные прогоны сторожа: он сам некритичный, и когда + гейтвей лежит или бюджет зажат (`throttled` откладывает некритичный крон + без записи в журнал), его прогонов нет так же, как и чужих. Цепочка своих + прогонов без разрыва длиннее `step`, считая от сейчас, плюс старт + процесса - это время, за которое чужой пропуск действительно пропуск, + а не общий простой. + """ + anchor = now + for run in sorted(mine, key=lambda run: run.started_at, reverse=True): + if anchor - run.started_at > step: + break + anchor = min(anchor, run.started_at) + return anchor if boot is None else max(anchor, boot) + + +def _silence( + state: Runs, + mine: Sequence[Run], + *, + now: datetime, + me: str, + crons: Mapping[str, str], + latest: Mapping[str, Run], + tz: tzinfo, + boot: datetime | None, +) -> _Silence: + """Давность: какие джобы с кроном перестали запускаться.""" + own = every_from_cron(crons.get(me), tz, now=now, grain=GRAIN) + guarded: dict[str, timedelta] = {} + unguarded: list[str] = [] + for job, cron in crons.items(): + if job == me: + continue + period = every_from_cron(cron, tz, now=now, grain=GRAIN) + if period is None: + unguarded.append(job) + else: + guarded[job] = period + + note = "" + if unguarded: + note += "; давностью не охраняются (крон без регулярного периода): " + ( + ", ".join(sorted(unguarded)) + ) + dropped = [job for job in state.quiet if job not in guarded] + for job in dropped: + del state.quiet[job] + if dropped: + note += "; сняты с давности (нет в расписании): " + ", ".join(sorted(dropped)) + if not guarded: + return _Silence((), (), tuple(sorted(unguarded)), note) + if own is None: + # Свой период не выведен - нет и опоры, когда планировщик заведомо + # работал. Любое число тут было бы выдумано. + note += "; свой крон без регулярного периода - давность не проверяется" + return _Silence((), (), tuple(sorted(unguarded)), note) + + since = awake(mine, now=now, step=own * FACTOR, boot=boot) + quiet: list[Quiet] = [] + resumed: list[Resumed] = [] + held: list[str] = [] + fresh: list[str] = [] + for job, period in guarded.items(): + last = latest.get(job) + if last is None: + fresh.append(job) + continue + age = now - last.started_at + # Порог не тоньше собственного: чаще раза в час сторож не смотрит, и + # пропуск короче двух своих периодов неотличим от его же слепоты. + overdue = ( + age > max(period, own) * FACTOR + and next_run(crons[job], tz, max(last.started_at, since)) <= now - LAG + ) + told = state.quiet.get(job) + if overdue: + if told is None or now - told >= REPEAT: + quiet.append(Quiet(job=job, age=age, period=period)) + else: + held.append(job) + elif told is not None and last.started_at > told: + resumed.append(Resumed(job=job, status=last.status)) + + if fresh: + note += "; ещё ни разу не запускались: " + ", ".join(sorted(fresh)) + if quiet: + note += "; не запускаются: " + ", ".join(sorted(q.job for q in quiet)) + if held: + note += "; о молчании уже говорили меньше суток назад: " + ", ".join( + sorted(held) + ) + if resumed: + note += "; снова запускаются: " + ", ".join(sorted(r.job for r in resumed)) + return _Silence( + quiet=tuple(sorted(quiet, key=lambda q: q.age, reverse=True)), + resumed=tuple(sorted(resumed, key=lambda r: r.job)), + unguarded=tuple(sorted(unguarded)), note=note, ) @@ -355,7 +579,7 @@ def streak_line(tail: Streak) -> str: f"джоб `{tail.job}` лежит {_age(tail.age)}" f"{' и дольше' if tail.capped else ''}, {runs} подряд с " f"{tail.since.astimezone(UTC).strftime('%d.%m %H:%M UTC')}" - + (" (это всё окно суток, серия началась раньше)" if tail.capped else "") + + (" (это всё окно недели, серия началась раньше)" if tail.capped else "") + f": {tail.error}" ) @@ -369,6 +593,27 @@ def alarm_text(report: Report) -> str: return "\n".join(lines) +def quiet_line(quiet: Quiet) -> str: + """«джоб `закрытие` не запускался 50 часов при периоде 24 часа».""" + return ( + f"джоб `{quiet.job}` не запускался {_age(quiet.age)} " + f"при периоде {_period(quiet.period)}" + ) + + +def quiet_text(report: Report) -> str: + """Тревога о молчании: какие джобы с кроном перестали запускаться.""" + lines = [f"- {quiet_line(quiet)}" for quiet in report.quiet[:MAX_LINES]] + rest = len(report.quiet) - len(lines) + if rest > 0: + lines.append(f"- и ещё {rest} {_plural(rest, ('джоб', 'джоба', 'джобов'))}") + return "\n".join(lines) + + +def resumed_line(resumed: Resumed) -> str: + return f"отбой: джоб `{resumed.job}` снова запускается ({resumed.status})" + + def clear_line(clear: Clear) -> str: if clear.count == 0: return f"отбой: джоб `{clear.job}` снова проходит ({clear.status})" @@ -379,7 +624,9 @@ def clear_line(clear: Clear) -> str: def clear_text(report: Report) -> str: - return "\n".join(f"- {clear_line(clear)}" for clear in report.clears) + lines = [clear_line(clear) for clear in report.clears] + lines += [resumed_line(resumed) for resumed in report.resumed] + return "\n".join(f"- {line}" for line in lines) def empty_text(report: Report) -> str: diff --git a/beaver_agent/jobs/__init__.py b/beaver_agent/jobs/__init__.py index fcb7ab9..3b13ac0 100644 --- a/beaver_agent/jobs/__init__.py +++ b/beaver_agent/jobs/__init__.py @@ -38,9 +38,10 @@ jobs = [ # ради чего заведён; чтение 25 событий стоит копейки. Job("отметка", beat, cron="12 17 * * *", critical=False), Job("сторож", watch, cron="42 */6 * * *", critical=False), - # Сторож собственного журнала прогонов. Час - это компромисс: окно счёта - # сутки, и джоб, упавший три раза подряд, лежит уже заметно дольше часа, - # так что чаще смысла нет, а реже - тревога стареет на свой же период. + # Сторож собственного журнала прогонов. Час - это компромисс: джоб, + # упавший три раза подряд, лежит уже заметно дольше часа, так что чаще + # смысла нет, а реже - тревога стареет на свой же период. Этот же час - + # разрешение давности: молчание короче двух часов он не различает. # ":27" выбрана пустой: на ":00" ротация и куратор, на ":40" логи, на # ":42" сторож соседа, а собственный прогон этого джоба не должен ждать # очереди за чужим спавном. diff --git a/beaver_agent/jobs/runs.py b/beaver_agent/jobs/runs.py index 5ed231f..89706fc 100644 --- a/beaver_agent/jobs/runs.py +++ b/beaver_agent/jobs/runs.py @@ -4,6 +4,10 @@ (`run.conversations.db.session()` и модель `JobRunRecord`): своей строки подключения у джоба нет, и второй правды о том, где журнал, тоже. +Кроны и зона, в которой они читаются, берутся у самого планировщика +(`run.scheduler.jobs`, `run.scheduler.tz`): давность меряется по тому +расписанию, что реально зарегистрировано, а не по копии списка. + Устроен как сторож соседа (`beaver_agent/jobs/heartbeat.py`): состояние файлом в `мета/бобер/.состояние/`, тревога - спавном джоба к триажу, и запись «сказали» ставится только после того, как тревога ушла. @@ -16,8 +20,9 @@ from datetime import UTC, datetime from typing import TYPE_CHECKING from beaver_gateway.storage.models import JobRunRecord -from sqlmodel import col, select +from sqlmodel import col, func, select +from beaver_agent.hands.heartbeat import BOOT from beaver_agent.hands.runs import ( STREAK, WINDOW, @@ -28,6 +33,7 @@ from beaver_agent.hands.runs import ( clear_text, empty_text, look, + quiet_text, ) from beaver_agent.vault import STATE_DIR @@ -42,13 +48,14 @@ STATE_FILE = STATE_DIR / "прогоны.json" TITLE = "прогоны" -LIMIT = 5000 -"""Потолок строк за окно. Сутки нормальной работы - это сотни строк; упереться -в потолок можно только при шторме, и тогда серии считаются по новейшим, а -недостающий хвост честно виден как `capped` в тексте тревоги.""" +LIMIT = 20000 +"""Потолок строк за окно. Неделя нормальной работы - это около 1700 строк +(`вайбграм` раз в 10 минут даёт 1008 из них); упереться в потолок можно +только при шторме, и тогда серии считаются по новейшим, а недостающий хвост +честно виден как `capped` в тексте тревоги.""" BRIEF = ( - "Сторож собственных прогонов (таблица `job_runs`, окно сутки) нашёл " + "Сторож собственных прогонов (таблица `job_runs`, окно неделя) нашёл " "лежащие джобы: у каждого {streak} и больше падений подряд, считая от " "последнего прогона.\n" "{body}\n" @@ -60,6 +67,20 @@ BRIEF = ( "«не срочно»: повтори один раз с тем же текстом." ) +QUIET_BRIEF = ( + "Сторож собственных прогонов нашёл джобы, которые ПЕРЕСТАЛИ ЗАПУСКАТЬСЯ: " + "у каждого есть крон, а последней записи в журнале `job_runs` больше " + "двух периодов.\n" + "{body}\n" + 'Будить владельца - inject(conversation="master", urgency="wake", ' + "text=до 3 строк: какой джоб молчит, сколько и при каком периоде). Это " + "не падение: джоб не падает, его просто никто не зовёт. Смотреть стоит " + "расписание (не снят ли джоб из `beaver_agent/jobs/__init__.py`), кроны " + "в pgqueuer и лог gateway на старте - джоб, упавший на импорте, не " + "регистрируется вовсе. Если inject вернул ошибку - это сбой доставки, " + "а не «не срочно»: повтори один раз с тем же текстом." +) + CLEAR_BRIEF = ( "Сторож собственных прогонов: то, о чём он тревожился, починилось.\n" "{body}\n" @@ -95,20 +116,35 @@ async def recent(run: JobRun, *, since: datetime) -> list[Run]: ) async with run.conversations.db.session() as session: rows = (await session.exec(stmt)).all() - return [ - Run( - id=row.id or 0, - job=row.job, - status=row.status, - started_at=( - row.started_at - if row.started_at.tzinfo - else row.started_at.replace(tzinfo=UTC) - ), - error=row.error, - ) - for row in rows - ] + return [_run(row) for row in rows] + + +def _run(row: JobRunRecord) -> Run: + return Run( + id=row.id or 0, + job=row.job, + status=row.status, + started_at=( + row.started_at + if row.started_at.tzinfo + else row.started_at.replace(tzinfo=UTC) + ), + error=row.error, + ) + + +async def latest(run: JobRun) -> dict[str, Run]: + """Последний прогон каждого джоба за всю историю журнала. + + Не за окно: джоб, молчащий дольше недели, иначе выглядел бы «ни разу не + запускавшимся» и уходил бы в тихий пропуск ровно тогда, когда молчит + дольше всего. Тем же запросом gateway строит свою сводку по джобам. + """ + newest = select(func.max(JobRunRecord.id)).group_by(col(JobRunRecord.job)) + stmt = select(JobRunRecord).where(col(JobRunRecord.id).in_(newest)) + async with run.conversations.db.session() as session: + rows = (await session.exec(stmt)).all() + return {row.job: _run(row) for row in rows} async def watch_runs(run: JobRun) -> None: @@ -116,13 +152,24 @@ async def watch_runs(run: JobRun) -> None: now = datetime.now(UTC) try: rows = await recent(run, since=now - WINDOW) + last = await latest(run) except Exception: # noqa: BLE001 - отказ запроса это отсутствие показания # Журнал не прочитан - сказать о здоровье джобов нечего, и молчание # базы выдавать за здоровье нельзя ровно так же, как за болезнь. _log.exception("прогоны: журнал не прочитан, прогон без показания") return state = Runs.load(STATE_FILE) - report = look(state, rows, now=now, me=run.job.name) + crons = {job.name: job.cron for job in run.scheduler.jobs if job.cron} + report = look( + state, + rows, + now=now, + me=run.job.name, + crons=crons, + latest=last, + tz=run.scheduler.tz, + boot=BOOT, + ) _log.info("прогоны: %s", report.note) await _tell(run, report, state, now=now) state.save() @@ -136,10 +183,16 @@ async def _tell(run: JobRun, report: Report, state: Runs, *, now: datetime) -> N await _spawn(run, BRIEF.format(body=alarm_text(report), streak=STREAK)) for tail in report.alarms: state.alerted[tail.job] = now - if report.clears: + if report.quiet: + await _spawn(run, QUIET_BRIEF.format(body=quiet_text(report))) + for quiet in report.quiet: + state.quiet[quiet.job] = now + if report.clears or report.resumed: await _spawn(run, CLEAR_BRIEF.format(body=clear_text(report))) for clear in report.clears: state.alerted.pop(clear.job, None) + for resumed in report.resumed: + state.quiet.pop(resumed.job, None) async def _spawn(run: JobRun, text: str) -> None: diff --git a/tests/test_runs.py b/tests/test_runs.py index 0085c56..2655c2c 100644 --- a/tests/test_runs.py +++ b/tests/test_runs.py @@ -9,11 +9,13 @@ from beaver_agent.hands.runs import ( Run, Runs, alarm_text, + awake, clear_text, empty_text, first_line, look, mask, + quiet_text, ) NOW = datetime(2026, 9, 22, 15, 42, tzinfo=UTC) @@ -126,7 +128,7 @@ def test_streak_running_past_the_window_says_at_least(tmp_path): assert tail.capped and tail.count == 144 text = alarm_text(report) assert "лежит 24 часа и дольше, 144 прогона подряд" in text - assert "это всё окно суток, серия началась раньше" in text + assert "это всё окно недели, серия началась раньше" in text def test_empty_window_is_its_own_alarm(tmp_path): @@ -210,9 +212,12 @@ def test_state_survives_a_round_trip(tmp_path): class FakeRun: - def __init__(self, database=None, *, name="прогоны"): + def __init__(self, database=None, *, name="прогоны", jobs=()): self.job = SimpleNamespace(name=name, cron="27 * * * *") self.conversations = SimpleNamespace(db=database) + self.scheduler = SimpleNamespace( + jobs=[self.job, *(SimpleNamespace(name=n, cron=c) for n, c in jobs)], tz=UTC + ) self.spawned = [] async def spawn_job(self, *, agent, text, title=None): @@ -259,7 +264,7 @@ async def test_job_reads_the_real_journal_and_spawns(tmp_path, monkeypatch): record("вайбграм", "failed", ago=10), record("досуг", "done", ago=5), record("прогоны", "done", ago=3), - # Старше суток - в окно не попадает + # Одно падение - не серия, сколько бы окно ни было record("логи", "failed", ago=60 * 30), ], ) @@ -271,7 +276,7 @@ async def test_job_reads_the_real_journal_and_spawns(tmp_path, monkeypatch): agent, title, text = run_.spawned[0] assert agent == "beaver-triage" and title == "прогоны" assert "джоб `вайбграм` лежит" in text and "3 прогона подряд" in text - assert "логи" not in text # за окном + assert "логи" not in text # одно падение assert "прогоны" not in text.split("\n")[1] # сам себя не считает # Тревога записана - следующий прогон в те же сутки молчит assert Runs.load(tmp_path / "прогоны.json").alerted["вайбграм"] is not None @@ -364,3 +369,224 @@ async def test_the_watcher_never_reports_itself(tmp_path, monkeypatch, name): assert not run_.spawned await database.dispose() + + +# --- Окно неделя: суточные джобы тоже набирают серию --------------------------- + + +def test_daily_job_failing_three_days_in_a_row_is_an_alarm(tmp_path): + day = 24 * 60 + rows = [ + run(1, "закрытие", "done", ago=4 * day - 60), + run(2, "закрытие", "failed", ago=3 * day - 60), + run(3, "закрытие", "failed", ago=2 * day - 60), + run(4, "закрытие", "failed", ago=day - 60), + ] + report = look(state(tmp_path), rows, now=NOW, me="прогоны") + (tail,) = report.alarms + assert tail.job == "закрытие" and tail.count == 3 and not tail.capped + assert "джоб `закрытие` лежит 2 суток, 3 прогона подряд" in alarm_text(report) + + +async def test_job_reads_a_week_back(tmp_path, monkeypatch): + from beaver_agent.jobs import runs as job + + monkeypatch.setattr(job, "STATE_FILE", tmp_path / "прогоны.json") + day = 24 * 60 + database = await make_db( + tmp_path, + [ + record("голосовой", "failed", ago=3 * day - 30), + record("голосовой", "failed", ago=2 * day - 30), + record("голосовой", "failed", ago=day - 30), + # Старше недели - в окно не попадает + record("память", "failed", ago=9 * day), + ], + ) + run_ = FakeRun(database) + + await job.watch_runs(cast("Any", run_)) + + (spawned,) = run_.spawned + assert "джоб `голосовой` лежит" in spawned[2] + assert "память" not in spawned[2] + await database.dispose() + + +# --- Давность: джоб перестал запускаться --------------------------------------- + +CRONS = {"прогоны": "27 * * * *", "закрытие": "20 4 * * *", "вайбграм": "*/10 * * * *"} + + +def hourly_me(hours=72): + """Непрерывная цепочка своих прогонов: планировщик жив всё это время.""" + return [ + run(5000 + i, "прогоны", "done", ago=60 * (hours - i) - 15) + for i in range(hours) + ] + + +def alive(ago=5): + return run(9000, "вайбграм", "done", ago=ago) + + +def silent(tmp_path, *, closing_ago_h, mine=None, boot=None, **kw): + last = run(1, "закрытие", "done", ago=closing_ago_h * 60) + return look( + kw.pop("runs", None) or state(tmp_path), + [alive(), *(hourly_me() if mine is None else mine)], + now=NOW, + me="прогоны", + crons=kw.pop("crons", CRONS), + latest=kw.pop("latest", {"закрытие": last, "вайбграм": alive()}), + tz=UTC, + boot=boot, + ) + + +def test_job_silent_longer_than_two_periods_is_its_own_alarm(tmp_path): + report = silent(tmp_path, closing_ago_h=50) + (quiet,) = report.quiet + assert quiet.job == "закрытие" and quiet.period == timedelta(hours=24) + assert not report.alarms # это не «лежит» + text = quiet_text(report) + assert "джоб `закрытие` не запускался 2 суток при периоде 24 часа" in text + assert "лежит" not in text + + +def test_one_missed_fire_is_not_silence(tmp_path): + # 40 часов: пропущено одно срабатывание, не два + assert not silent(tmp_path, closing_ago_h=40).quiet + + +def test_silence_is_told_once_a_day_and_cleared_by_a_run(tmp_path): + told = state(tmp_path, quiet={"закрытие": NOW - timedelta(hours=5)}) + held = silent(tmp_path, closing_ago_h=50, runs=told) + assert not held.quiet + assert "о молчании уже говорили меньше суток назад: закрытие" in held.note + older = state(tmp_path, quiet={"закрытие": NOW - timedelta(hours=25)}) + assert silent(tmp_path, closing_ago_h=50, runs=older).quiet + # Запуск появился после тревоги - «отбой», а не молчание + back = silent(tmp_path, closing_ago_h=1, runs=told) + (resumed,) = back.resumed + assert resumed.job == "закрытие" and not back.clears + assert "отбой: джоб `закрытие` снова запускается (done)" in clear_text(back) + + +def test_downtime_of_the_scheduler_is_not_a_missed_fire(tmp_path): + # Гейтвей только что поднялся: после старта срабатывания ещё не было + fresh = silent(tmp_path, closing_ago_h=50, boot=NOW - timedelta(minutes=30)) + assert not fresh.quiet + # Сторож сам не бегал сутки (бюджет зажат, некритичное откладывается): + # пропуски за это время - общий простой, а не молчание одного джоба + gap = [run(1, "прогоны", "done", ago=60 * 30)] + assert not silent(tmp_path, closing_ago_h=50, mine=gap).quiet + assert awake(gap, now=NOW, step=timedelta(hours=2), boot=None) == NOW + + +def test_short_period_job_uses_the_watchers_resolution(tmp_path): + latest = {"закрытие": run(1, "закрытие", "done", ago=60)} + # Вайбграм раз в 10 минут, молчит 70 минут: сторож часовой, и такой + # пропуск неотличим от его же слепоты между прогонами + latest["вайбграм"] = run(2, "вайбграм", "done", ago=70) + assert not silent(tmp_path, closing_ago_h=1, latest=latest).quiet + latest["вайбграм"] = run(2, "вайбграм", "done", ago=150) + (quiet,) = silent(tmp_path, closing_ago_h=1, latest=latest).quiet + assert quiet.job == "вайбграм" + assert "не запускался 2 часа при периоде 10 минут" in quiet_text( + silent(tmp_path, closing_ago_h=1, latest=latest) + ) + + +def test_job_without_cron_is_not_checked(tmp_path): + # deploy - вебхук: крона у него нет, в `crons` он не попадает вовсе + latest = { + "deploy": run(1, "deploy", "done", ago=60 * 24 * 30), + "закрытие": run(2, "закрытие", "done", ago=60), + } + report = silent(tmp_path, closing_ago_h=1, latest=latest) + assert not report.quiet and "deploy" not in report.note + + +def test_job_never_seen_in_the_journal_waits_silently(tmp_path): + report = silent(tmp_path, closing_ago_h=0, latest={"вайбграм": alive()}) + assert not report.quiet and not report.resumed + assert "ещё ни разу не запускались: закрытие" in report.note + + +def test_irregular_schedule_is_skipped_and_named(tmp_path): + crons = {**CRONS, "куратор": "0 2,8,12,16,20 * * *"} + report = silent(tmp_path, closing_ago_h=1, crons=crons, latest={}) + assert report.unguarded == ("куратор",) + assert "давностью не охраняются" in report.note and "куратор" in report.note + assert not report.quiet + + +def test_the_watcher_does_not_measure_its_own_silence(tmp_path): + latest = { + "прогоны": run(1, "прогоны", "done", ago=60 * 24 * 5), + "закрытие": run(2, "закрытие", "done", ago=60), + } + report = silent(tmp_path, closing_ago_h=1, latest=latest) + assert not report.quiet + + +def test_job_removed_from_the_schedule_leaves_silence_state(tmp_path): + told = state(tmp_path, quiet={"старый": NOW - timedelta(hours=3)}) + report = silent(tmp_path, closing_ago_h=1, runs=told) + assert not report.resumed and "старый" not in told.quiet + assert "сняты с давности (нет в расписании): старый" in report.note + + +def test_real_schedule_names_only_the_curator_unguarded(tmp_path): + from beaver_agent.jobs import jobs + + crons = {job.name: job.cron for job in jobs if job.cron} + report = silent(tmp_path, closing_ago_h=1, crons=crons, latest={}) + assert report.unguarded == ("куратор",) + # Вебхуки в расписание давности не попадают + assert {"deploy", "komodo", "t3code"}.isdisjoint(crons) + + +async def test_job_alarms_on_a_silent_job_and_logs_the_unguarded( + tmp_path, monkeypatch, caplog +): + from beaver_agent.jobs import runs as job + + monkeypatch.setattr(job, "STATE_FILE", tmp_path / "прогоны.json") + monkeypatch.setattr(job, "BOOT", datetime.now(UTC) - timedelta(days=10)) + database = await make_db( + tmp_path, + [ + record("закрытие", "done", ago=60 * 50), + *(record("прогоны", "done", ago=60 * h - 15) for h in range(1, 73)), + record("вайбграм", "done", ago=5), + record("deploy", "done", ago=60 * 24 * 20), + ], + ) + run_ = FakeRun( + database, + jobs=[ + ("закрытие", "20 4 * * *"), + ("вайбграм", "*/10 * * * *"), + ("куратор", "0 2,8,12,16,20 * * *"), + ("deploy", None), + ("новый", "0 3 * * *"), + ], + ) + + with caplog.at_level("INFO", logger=job.__name__): + await job.watch_runs(cast("Any", run_)) + + (spawned,) = run_.spawned + text = spawned[2] + assert "ПЕРЕСТАЛИ ЗАПУСКАТЬСЯ" in text + assert "джоб `закрытие` не запускался 2 суток при периоде 24 часа" in text + assert "deploy" not in text and "новый" not in text + assert "давностью не охраняются" in caplog.text and "куратор" in caplog.text + assert "ещё ни разу не запускались: новый" in caplog.text + assert "закрытие" in Runs.load(tmp_path / "прогоны.json").quiet + # В те же сутки не повторяется + await job.watch_runs(cast("Any", run_)) + assert len(run_.spawned) == 1 + await database.dispose()