147 lines
6.6 KiB
Python
147 lines
6.6 KiB
Python
"""Сторож собственных прогонов: раз в час читает `job_runs` и будит человека.
|
|
|
|
Подключение к базе - то же самое, которым пишет сам планировщик
|
|
(`run.conversations.db.session()` и модель `JobRunRecord`): своей строки
|
|
подключения у джоба нет, и второй правды о том, где журнал, тоже.
|
|
|
|
Устроен как сторож соседа (`beaver_agent/jobs/heartbeat.py`): состояние
|
|
файлом в `мета/бобер/.состояние/`, тревога - спавном джоба к триажу, и
|
|
запись «сказали» ставится только после того, как тревога ушла.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
from datetime import UTC, datetime
|
|
from typing import TYPE_CHECKING
|
|
|
|
from beaver_gateway.storage.models import JobRunRecord
|
|
from sqlmodel import col, select
|
|
|
|
from beaver_agent.hands.runs import (
|
|
STREAK,
|
|
WINDOW,
|
|
Report,
|
|
Run,
|
|
Runs,
|
|
alarm_text,
|
|
clear_text,
|
|
empty_text,
|
|
look,
|
|
)
|
|
from beaver_agent.vault import STATE_DIR
|
|
|
|
if TYPE_CHECKING:
|
|
from beaver_gateway.jobs.scheduler import JobRun
|
|
|
|
_log = logging.getLogger(__name__)
|
|
|
|
STATE_FILE = STATE_DIR / "прогоны.json"
|
|
"""Кому уже сказали. Без файла тревога о лежащем джобе повторится, а «отбой»
|
|
по тому, о ком говорили до потери файла, не придёт - показания не осталось."""
|
|
|
|
TITLE = "прогоны"
|
|
|
|
LIMIT = 5000
|
|
"""Потолок строк за окно. Сутки нормальной работы - это сотни строк; упереться
|
|
в потолок можно только при шторме, и тогда серии считаются по новейшим, а
|
|
недостающий хвост честно виден как `capped` в тексте тревоги."""
|
|
|
|
BRIEF = (
|
|
"Сторож собственных прогонов (таблица `job_runs`, окно сутки) нашёл "
|
|
"лежащие джобы: у каждого {streak} и больше падений подряд, считая от "
|
|
"последнего прогона.\n"
|
|
"{body}\n"
|
|
'Будить владельца - inject(conversation="master", urgency="wake", '
|
|
"text=до 3 строк: какой джоб лежит, сколько прогонов подряд и что в "
|
|
"ошибке). Это авария у НАС, а не у соседа: джоб пишется в журнал каждый "
|
|
"прогон, и раз он падает подряд, сам он не починится. Подробности - "
|
|
"в логах gateway. Если inject вернул ошибку - это сбой доставки, а не "
|
|
"«не срочно»: повтори один раз с тем же текстом."
|
|
)
|
|
|
|
CLEAR_BRIEF = (
|
|
"Сторож собственных прогонов: то, о чём он тревожился, починилось.\n"
|
|
"{body}\n"
|
|
'Владельца НЕ будить: inject(conversation="master", urgency="normal", '
|
|
"text=одна строка «отбой»). Человек знает про тревогу, ему нужно знать, "
|
|
"что она кончилась."
|
|
)
|
|
|
|
EMPTY_BRIEF = (
|
|
"Сторож собственных прогонов не нашёл в журнале НИЧЕГО.\n"
|
|
"{body}\n"
|
|
'Будить владельца - inject(conversation="master", urgency="wake", '
|
|
"text=до 3 строк: что за сутки не отработал ни один джоб и куда смотреть). "
|
|
"Молчащий планировщик - это молчание всех джобов разом, включая тех, что "
|
|
"носят человеку почту."
|
|
)
|
|
|
|
|
|
async def recent(run: JobRun, *, since: datetime) -> list[Run]:
|
|
"""Прогоны всех джобов за окно, новейшие первыми.
|
|
|
|
`started_at` в базе лежит наивным UTC (так его кладёт планировщик), и
|
|
сравнение в `where` тоже наивное - как это делает gateway в своих
|
|
запросах по времени.
|
|
"""
|
|
stmt = (
|
|
select(JobRunRecord)
|
|
.where(
|
|
col(JobRunRecord.started_at) >= since.astimezone(UTC).replace(tzinfo=None)
|
|
)
|
|
.order_by(col(JobRunRecord.id).desc())
|
|
.limit(LIMIT)
|
|
)
|
|
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
|
|
]
|
|
|
|
|
|
async def watch_runs(run: JobRun) -> None:
|
|
"""Прогон сторожа: журнал → решение → тревога, «отбой» или молчание."""
|
|
now = datetime.now(UTC)
|
|
try:
|
|
rows = await recent(run, since=now - WINDOW)
|
|
except Exception: # noqa: BLE001 - отказ запроса это отсутствие показания
|
|
# Журнал не прочитан - сказать о здоровье джобов нечего, и молчание
|
|
# базы выдавать за здоровье нельзя ровно так же, как за болезнь.
|
|
_log.exception("прогоны: журнал не прочитан, прогон без показания")
|
|
return
|
|
state = Runs.load(STATE_FILE)
|
|
report = look(state, rows, now=now, me=run.job.name)
|
|
_log.info("прогоны: %s", report.note)
|
|
await _tell(run, report, state, now=now)
|
|
state.save()
|
|
|
|
|
|
async def _tell(run: JobRun, report: Report, state: Runs, *, now: datetime) -> None:
|
|
if report.empty:
|
|
await _spawn(run, EMPTY_BRIEF.format(body=empty_text(report)))
|
|
state.empty_alerted = now
|
|
if report.alarms:
|
|
await _spawn(run, BRIEF.format(body=alarm_text(report), streak=STREAK))
|
|
for tail in report.alarms:
|
|
state.alerted[tail.job] = now
|
|
if report.clears:
|
|
await _spawn(run, CLEAR_BRIEF.format(body=clear_text(report)))
|
|
for clear in report.clears:
|
|
state.alerted.pop(clear.job, None)
|
|
|
|
|
|
async def _spawn(run: JobRun, text: str) -> None:
|
|
await run.spawn_job(agent="beaver-triage", title=TITLE, text=text)
|