Files

643 lines
28 KiB
Python
Raw Permalink Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""Сторож собственных прогонов: журнал `job_runs` наконец кто-то читает.
Планировщик gateway пишет строку на каждый прогон джоба (`job_runs`: имя,
статус, хвост трейсбека в `error`). Запись была всегда, читателя не было
ни одного: канал к соседнему агенту лежал 38 часов подряд, планировщик
честно записал это 231 раз, и до человека не доехало ничего. Этот модуль -
вторая половина прибора: показание из таблицы превращается в тревогу.
Что здесь считается показанием:
- тревога берётся только с ХВОСТА: три и более `failed` подряд, считая от
последнего прогона джоба. Одиночный упавший прогон - сетевой чих, а не
болезнь, и будить на него человека значит выучить его не смотреть;
- окно счёта - неделя, а не сутки: иначе суточный джоб не наберёт трёх
падений подряд никогда, и `отметка`, `закрытие`, `голосовой`, `память`
не охранялись бы вовсе;
- отсутствие записей - тоже показание. Джоб с регулярным кроном, у которого
последний прогон старше `FACTOR` его периодов, «не запускался» - это
отдельная тревога, не «лежит». Период выводится из крона тем же
`every_from_cron`, что у сторожа соседа, и расписание, которое не
сводится к «раз в N часов», в давности не участвует и называется в логе
неохраняемым: подставить ему число значило бы соврать. Джобы без крона
(вебхуки, события) давностью не меряются вовсе, а джоб, которого в
журнале нет ни разу, молча ждёт своей первой записи;
- пустая выдача - это показание о запросе, а не о здоровье джобов. Ноль
строк за сутки (кроме своих) означает, что планировщик вообще не
работал, и это отдельная тревога с отдельным текстом, а не тишина;
- упавший запрос к БД - не показание вовсе: решает джоб, здесь такого
входа просто нет.
Свои собственные прогоны из счёта выкинуты (`me`): джоб, который лежит,
о себе не доложит - наблюдатель внутри объекта не сообщает о смерти
объекта. Это дыра прибора, и она названа в README, а не заклеена.
Текст тревоги проходит `mask`: из `error` берётся первая строка, режется
по длине и чистится - общим `redact` gateway плюс два своих правила на то,
что `redact` по конструкции не ловит внутри строки (`ИМЯ=значение` не с
начала строки и хвост запроса у URL). Тревога уезжает в чат к диспетчеру,
логи гейтвея видны в чатах - секрету тут негде быть.
"""
from __future__ import annotations
import itertools
import json
import re
from dataclasses import dataclass, field
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 Mapping, Sequence
from datetime import tzinfo
__all__ = [
"DAY",
"ERROR_CHARS",
"FACTOR",
"FAILED",
"FORGET",
"MAX_LINES",
"REPEAT",
"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(days=7)
"""Окно журнала: неделя. Суточный джоб набирает в нём три падения подряд за
трое суток, недельный (`память`) - нет: у него три падения это три недели,
и его охраняет давность, а не серия."""
DAY = timedelta(hours=24)
"""Окно тревоги о пустом журнале: сутки без единого чужого прогона."""
FACTOR = 2
"""Давность = столько периодов джоба без записи. Два - это «пропущено два
срабатывания подряд»: один пропуск случается честно (передеплой пришёлся на
минуту крона, а догонять пропущенное планировщик не умеет), два подряд -
уже не случайность."""
GRAIN = timedelta(minutes=1)
"""Шаг, в котором период крона должен сводиться к целому: крон минутный."""
LAG = timedelta(minutes=10)
"""Запись в журнал ложится по КОНЦУ прогона: срабатывание моложе этого
могло ещё не дописаться, и пропуском его не считаем."""
STREAK = 3
"""Сколько `failed` подряд считаем «джоб лежит»."""
REPEAT = timedelta(hours=24)
"""Чаще этого об одном и том же джобе не говорим."""
FORGET = timedelta(days=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,}))")
"""`ИМЯ=значение` где угодно в строке: общий `redact` ловит такие пары только
с начала строки, а трейсбек кладёт их в середину. Просмотр нулевой длины -
чтобы `RuntimeError: MSOS_MCP=…` не съедало секрет первой же парой: обычный
`sub` откусил бы `RuntimeError:` вместе со значением и до вложенной пары
не дошёл."""
_QUERY = re.compile(r"(https?://[^\s?#]+)\?[^\s#]*")
"""Хвост запроса у URL сносим целиком: имя параметра с ключом может быть любым."""
def _iso(when: datetime | None) -> str | None:
if when is None:
return None
return when.astimezone(UTC).strftime("%Y-%m-%dT%H:%M:%SZ")
def _parse_iso(raw: str | None) -> datetime | None:
if not raw:
return None
try:
when = datetime.fromisoformat(raw)
except ValueError:
return None
return when if when.tzinfo else when.replace(tzinfo=UTC)
def _plural(count: int, forms: tuple[str, str, str]) -> str:
if count % 10 == 1 and count % 100 != 11:
return forms[0]
if count % 10 in (2, 3, 4) and count % 100 not in (12, 13, 14):
return forms[1]
return forms[2]
def _age(delta: timedelta) -> str:
"""Возраст человеку: «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
for match in _PAIR.finditer(text):
if match.start(3) < pos or not is_secret_name(match[1]):
continue
out.append(text[pos : match.start(3)])
out.append(MASK)
pos = match.end(3)
out.append(text[pos:])
return "".join(out)
def mask(text: str) -> str:
"""Убрать из строки всё, что похоже на секрет, прежде чем её показывать."""
return _QUERY.sub(r"\1?" + MASK, _mask_pairs(redact(text)))
def first_line(error: str | None) -> str:
"""Первая строка ошибки - чистая и короткая. Трейсбек человеку не нужен."""
for line in mask(error or "").splitlines():
head = line.strip()
if not head:
continue
if len(head) > ERROR_CHARS:
head = head[: ERROR_CHARS - 1].rstrip() + "…"
return head
return "без текста ошибки"
@dataclass(frozen=True, slots=True)
class Run:
"""Одна строка журнала: ровно то, на что мы смотрим, без ORM."""
id: int
job: str
status: str
started_at: datetime
error: str | None = None
@property
def failed(self) -> bool:
return self.status == FAILED
@dataclass(frozen=True, slots=True)
class Streak:
"""Хвост подряд упавших прогонов одного джоба."""
job: str
count: int
since: datetime
"""Начало серии - первое падение из тех, что видно в окне."""
age: timedelta
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)
class Clear:
"""Джоб, о котором мы тревожились, больше не лежит."""
job: str
status: str
"""Статус последнего прогона."""
count: int
"""Сколько подряд упавших осталось в хвосте: 0 - чисто."""
@dataclass(slots=True)
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
def load(cls, path: Path) -> Runs:
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)
return cls(
path=path,
alerted=_stamps(raw.get("alerted")),
quiet=_stamps(raw.get("quiet")),
empty_alerted=_parse_iso(raw.get("empty_alerted")),
)
def save(self) -> None:
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)
tmp = self.path.with_suffix(".tmp")
tmp.write_text(json.dumps(body, ensure_ascii=False, indent=1), encoding="utf-8")
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:
"""Что решил один прогон сторожа журнала."""
alarms: tuple[Streak, ...]
clears: tuple[Clear, ...]
empty: bool
"""Прогонов за сутки не было вовсе - тревога о планировщике, не о джобе."""
seen: int
"""Сколько чужих прогонов было в окне."""
mine: int
"""Сколько своих - их в счёт не берём, но в тексте называем."""
note: str
quiet: tuple[Quiet, ...] = ()
resumed: tuple[Resumed, ...] = ()
unguarded: tuple[str, ...] = ()
"""Джобы с кроном, у которого нет регулярного периода: давностью не меряем."""
def streak(runs: Sequence[Run], *, now: datetime) -> Streak | None:
"""Хвост падений одного джоба, считая от последнего прогона.
`runs` - прогоны одного джоба, новейший первым. Серия, упёршаяся в край
окна, помечается `capped`: её настоящую длину мы не знаем.
"""
failures = list(itertools.takewhile(lambda run: run.failed, runs))
if not failures:
return None
oldest = failures[-1]
return Streak(
job=failures[0].job,
count=len(failures),
since=oldest.started_at,
age=now - oldest.started_at,
error=first_line(failures[0].error),
capped=len(failures) == len(runs),
)
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 = [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 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=len(others),
mine=len(mine),
note=(
f"чужих прогонов за сутки нет вовсе (своих {len(mine)})"
+ ("; о пустоте уже говорили меньше суток назад" if quiet else "")
),
)
by_job: dict[str, list[Run]] = {}
for run in others:
by_job.setdefault(run.job, []).append(run)
alarms: list[Streak] = []
clears: list[Clear] = []
held: list[str] = []
for job, rows in by_job.items():
tail = streak(rows, now=now)
if tail is not None and tail.count >= STREAK:
told = state.alerted.get(job)
if told is None or now - told >= REPEAT:
alarms.append(tail)
else:
held.append(job)
elif job in state.alerted:
clears.append(
Clear(
job=job,
status=rows[0].status,
count=0 if tail is None else tail.count,
)
)
forgotten = [
job
for job, told in state.alerted.items()
if job not in by_job and now - told >= FORGET
]
for job in forgotten:
del state.alerted[job]
note = (
f"прогонов за неделю {len(others)} по {len(by_job)} джобам (своих {len(mine)})"
)
if alarms:
note += "; лежат: " + ", ".join(sorted(tail.job for tail in alarms))
if held:
note += "; уже говорили меньше суток назад: " + ", ".join(sorted(held))
if clears:
note += "; починились: " + ", ".join(sorted(clear.job for clear in clears))
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=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,
)
def streak_line(tail: Streak) -> str:
"""«джоб `вайбграм` лежит 4 часа, 24 прогона подряд с 12.09 18:00 UTC: …»."""
runs = f"{tail.count} {_plural(tail.count, ('прогон', 'прогона', 'прогонов'))}"
return (
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 "")
+ f": {tail.error}"
)
def alarm_text(report: Report) -> str:
"""Тревога человеку: что лежит, сколько и с какой ошибкой."""
lines = [f"- {streak_line(tail)}" for tail in report.alarms[:MAX_LINES]]
rest = len(report.alarms) - len(lines)
if rest > 0:
lines.append(f"- и ещё {rest} {_plural(rest, ('джоб', 'джоба', 'джобов'))}")
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})"
return (
f"отбой: джоб `{clear.job}` больше не лежит подряд "
f"(упавших в хвосте {clear.count}, порог {STREAK})"
)
def clear_text(report: Report) -> str:
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:
"""Отдельная тревога: за сутки не отработал вообще никто."""
return (
f"За последние сутки в журнале прогонов (`job_runs`) нет ни одной записи, "
f"кроме моих собственных (их {report.mine}): не отработал ни один джоб. "
"Это показание о планировщике, а не о каком-то одном джобе. Смотреть "
"стоит три вещи: поднялся ли планировщик вообще (в логе gateway "
"`scheduler: no postgres driver - cron and webhooks are off`), живы ли "
"кроны в pgqueuer, и не заведена ли база заново - на пустой базе эта же "
"тревога выглядит так же."
)