beaver-agent/beaver_agent/jobs/runs.py

"""Сторож собственных прогонов: раз в час читает `job_runs` и будит человека.

Подключение к базе - то же самое, которым пишет сам планировщик
(`run.conversations.db.session()` и модель `JobRunRecord`): своей строки
подключения у джоба нет, и второй правды о том, где журнал, тоже.

Кроны и зона, в которой они читаются, берутся у самого планировщика
(`run.scheduler.jobs`, `run.scheduler.tz`): давность меряется по тому
расписанию, что реально зарегистрировано, а не по копии списка.

Устроен как сторож соседа (`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, func, select

from beaver_agent.hands.heartbeat import BOOT
from beaver_agent.hands.runs import (
    STREAK,
    WINDOW,
    Report,
    Run,
    Runs,
    alarm_text,
    clear_text,
    empty_text,
    look,
    quiet_text,
)
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 = 20000
"""Потолок строк за окно. Неделя нормальной работы - это около 1700 строк
(`вайбграм` раз в 10 минут даёт 1008 из них); упереться в потолок можно
только при шторме, и тогда серии считаются по новейшим, а недостающий хвост
честно виден как `capped` в тексте тревоги."""

BRIEF = (
    "Сторож собственных прогонов (таблица `job_runs`, окно неделя) нашёл "
    "лежащие джобы: у каждого {streak} и больше падений подряд, считая от "
    "последнего прогона.\n"
    "{body}\n"
    'Будить владельца - inject(conversation="master", urgency="wake", '
    "text=до 3 строк: какой джоб лежит, сколько прогонов подряд и что в "
    "ошибке). Это авария у НАС, а не у соседа: джоб пишется в журнал каждый "
    "прогон, и раз он падает подряд, сам он не починится. Подробности - "
    "в логах gateway. Если inject вернул ошибку - это сбой доставки, а не "
    "«не срочно»: повтори один раз с тем же текстом."
)

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"
    'Владельца НЕ будить: 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(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:
    """Прогон сторожа: журнал → решение → тревога, «отбой» или молчание."""
    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)
    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()


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.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:
    await run.spawn_job(agent="beaver-triage", title=TITLE, text=text)