From 6d6abf30b18aca2fb6f970cb5647f9291eb09d00 Mon Sep 17 00:00:00 2001 From: Light1YT Date: Thu, 25 Jun 2026 16:02:13 +0500 Subject: [PATCH 1/2] feat(workers): Celery-beat data-freshness monitor with GlitchTip alerts MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Extract compute_freshness(db) from the /admin/scrape/freshness handler (single source of truth, endpoint response unchanged) and add a daily beat task scrape_freshness_check that alerts via sentry_sdk.capture_message when a source is stale/failed — level=error if a critical source fully failed, else warning; also logs at matching severity so a no-DSN worker still surfaces it. Registered in celery include + beat (09:00 MSK daily, after the nightly scraper cluster). Closes #freshness-monitor (rank 7 follow-up) --- backend/app/api/v1/admin_scrape.py | 22 ++- backend/app/workers/beat_schedule.py | 12 ++ backend/app/workers/celery_app.py | 1 + .../workers/tasks/scrape_freshness_check.py | 107 +++++++++++++ .../workers/test_scrape_freshness_check.py | 145 ++++++++++++++++++ 5 files changed, 283 insertions(+), 4 deletions(-) create mode 100644 backend/app/workers/tasks/scrape_freshness_check.py create mode 100644 backend/tests/workers/test_scrape_freshness_check.py diff --git a/backend/app/api/v1/admin_scrape.py b/backend/app/api/v1/admin_scrape.py index 8023ed59..fd7c54fc 100644 --- a/backend/app/api/v1/admin_scrape.py +++ b/backend/app/api/v1/admin_scrape.py @@ -1343,12 +1343,13 @@ def _classify_freshness(age_days: float | None, fresh_days: float, stale_days: f return "critical" -@router.get("/freshness") -def scrape_freshness( - db: Annotated[Session, Depends(get_db)], -) -> dict[str, Any]: +def compute_freshness(db: Session) -> dict[str, Any]: """Свежесть данных по каждому источнику (для dashboard + alerting). + Чистая функция (один источник истины для HTTP-эндпоинта и beat-таски + scrape_freshness_check). Принимает открытую сессию, НЕ коммитит/закрывает её — + жизненным циклом сессии управляет вызывающий (FastAPI-зависимость или таска). + Для каждого источника один проход по его run-таблице агрегирует: - last_success_at — max(COALESCE(finished_at, started_at)) WHERE status='done' - last_attempt_at — max(started_at) (или created_at fallback) по всем прогонам @@ -1462,6 +1463,19 @@ def scrape_freshness( } +@router.get("/freshness") +def scrape_freshness( + db: Annotated[Session, Depends(get_db)], +) -> dict[str, Any]: + """GET-обёртка над compute_freshness — только читает, не алертит. + + Вся агрегация живёт в compute_freshness (один источник истины); эндпоинт + лишь прокидывает сессию и возвращает результат без изменений формы payload. + Sentry-алёрты по критичным источникам шлёт beat-таска scrape_freshness_check. + """ + return compute_freshness(db) + + # ── DomRF (kn) — runs list ─────────────────────────────────────────────────── diff --git a/backend/app/workers/beat_schedule.py b/backend/app/workers/beat_schedule.py index e3b56f02..cb9f8e6b 100644 --- a/backend/app/workers/beat_schedule.py +++ b/backend/app/workers/beat_schedule.py @@ -557,4 +557,16 @@ def build_beat_schedule() -> dict: "options": {"queue": "celery"}, } + # Data-freshness monitor: ежедневно 09:00 МСК (Celery conf.timezone=Europe/Moscow → + # crontab в МСК, #1233). Запускает compute_freshness по 5 run-таблицам и шлёт + # GlitchTip-алёрт, если критичный источник устарел/упал. 09:00 — после ночного + # кластера скрейперов (03:00–07:00) и в начале рабочего дня, чтобы оператор увидел + # деградацию свежести данных утром. Только читает + алертит (не пишет), дёшево. + # Техническая, не в job_settings (как refresh-analytics / mv-sales-tracker). + schedule["scrape-freshness-check-daily"] = { + "task": "tasks.scrape_freshness_check.scrape_freshness_check", + "schedule": _parse_cron("0 9 * * *"), # 09:00 MSK ежедневно + "options": {"queue": "celery"}, + } + return schedule diff --git a/backend/app/workers/celery_app.py b/backend/app/workers/celery_app.py index 5755402b..9418adc2 100644 --- a/backend/app/workers/celery_app.py +++ b/backend/app/workers/celery_app.py @@ -85,6 +85,7 @@ celery_app = Celery( "app.workers.tasks.refresh_layout_velocity", "app.workers.tasks.riasurt_sverdl_harvest", "app.workers.tasks.mv_sales_tracker_refresh", + "app.workers.tasks.scrape_freshness_check", ], ) celery_app.conf.timezone = "Europe/Moscow" diff --git a/backend/app/workers/tasks/scrape_freshness_check.py b/backend/app/workers/tasks/scrape_freshness_check.py new file mode 100644 index 00000000..63fbf11b --- /dev/null +++ b/backend/app/workers/tasks/scrape_freshness_check.py @@ -0,0 +1,107 @@ +"""Celery-beat data-freshness monitor: алёрт в GlitchTip при устаревших/упавших источниках. + +GET /api/v1/admin/scrape/freshness считает свежесть по 5 run-таблицам, но только ЧИТАЕТ +(чтобы UI-поллинг не спамил issue). Эта beat-таска периодически запускает ту же агрегацию +(compute_freshness — единый источник истины) и пушит Sentry/GlitchTip-сообщение, когда есть +проблемные источники. + +Какие источники считаем алёртабельными: + - status == "failed" (нет успехов вообще / возраст >= stale_days) — серьёзно; + - status == "stale" (возраст между fresh_days и stale_days) — предупреждение. +Severity сообщения: "error", если хотя бы один КРИТИЧНЫЙ источник (registry.critical=True) +полностью упал (failed); иначе "warning". Если sentry в worker-процессе не инициализирован +(нет glitchtip_dsn) — capture_message молча no-op, поэтому дополнительно логируем WARNING. + +Mirror conventions (zone_regulation_refresh.py / backend.md): SessionLocal()+try/finally; +broad except не валит beat-loop (любая ошибка → log + возврат summary-dict). +""" + +from __future__ import annotations + +import logging +from typing import Any, Literal + +import sentry_sdk + +from app.api.v1.admin_scrape import compute_freshness +from app.core.db import SessionLocal +from app.workers.celery_app import celery_app + +logger = logging.getLogger(__name__) + +# Статусы из compute_freshness, которые считаем поводом для алёрта. +_ALERTABLE = {"stale", "failed"} + + +def _format_age(source: dict[str, Any]) -> str: + """Человекочитаемый возраст источника для тела сообщения.""" + age = source.get("age_days") + if age is None: + return "нет успешных прогонов" + return f"{age}d" + + +@celery_app.task(name="tasks.scrape_freshness_check.scrape_freshness_check") +def scrape_freshness_check() -> dict[str, Any]: + """Проверяет свежесть источников и шлёт GlitchTip-алёрт при stale/failed. + + Returns: + Summary-dict для observability/тестов: + {'checked': N, 'alertable': M, 'stale': S, 'failed': F, + 'overall_status': str, 'alerted': bool, 'level': str|None} + При внутренней ошибке: {'error': ''} — исключение НЕ пробрасывается, + чтобы не сорвать beat-loop. + """ + db = SessionLocal() + try: + payload = compute_freshness(db) + except Exception as exc: # любая ошибка агрегации не должна валить beat + logger.exception("scrape_freshness_check: compute_freshness упал: %s", exc) + return {"error": str(exc)} + finally: + db.close() + + sources: list[dict[str, Any]] = payload.get("sources", []) + overall = payload.get("overall_status", "unknown") + + alertable = [s for s in sources if s.get("status") in _ALERTABLE] + stale = [s for s in alertable if s.get("status") == "stale"] + failed = [s for s in alertable if s.get("status") == "failed"] + + summary: dict[str, Any] = { + "checked": len(sources), + "alertable": len(alertable), + "stale": len(stale), + "failed": len(failed), + "overall_status": overall, + "alerted": False, + "level": None, + } + + if not alertable: + logger.info( + "scrape_freshness_check: все источники свежие (checked=%d, overall=%s)", + len(sources), + overall, + ) + return summary + + # Severity: критичный источник упал полностью → error, иначе warning. + critical_failed = [s for s in failed if s.get("critical")] + level: Literal["error", "warning"] = "error" if critical_failed else "warning" + + parts = [f"{s['source']} [{s['status']}, {_format_age(s)}]" for s in alertable] + message = ( + f"Data freshness alert (overall={overall}): " + f"{len(failed)} failed, {len(stale)} stale — " + ", ".join(parts) + ) + + # capture_message — no-op, если SDK не инициализирован (нет glitchtip_dsn); + # поэтому всегда дублируем в лог (WARNING/ERROR по severity). + sentry_sdk.capture_message(message, level=level) + log = logger.error if level == "error" else logger.warning + log("scrape_freshness_check: %s", message) + + summary["alerted"] = True + summary["level"] = level + return summary diff --git a/backend/tests/workers/test_scrape_freshness_check.py b/backend/tests/workers/test_scrape_freshness_check.py new file mode 100644 index 00000000..4e05af2e --- /dev/null +++ b/backend/tests/workers/test_scrape_freshness_check.py @@ -0,0 +1,145 @@ +"""Тесты beat-таски scrape_freshness_check — алёрт в GlitchTip при stale/failed источниках. + +Сеть/БД не дёргаются: мокаем SessionLocal + compute_freshness (единый источник истины) + +sentry_sdk.capture_message. Проверяем: severity-маппинг (error при failed критичного, +warning при stale), сообщение называет проблемные источники, all-ok → нет алёрта, ошибка +агрегации → graceful (исключение не пробрасывается). +""" + +from __future__ import annotations + +from typing import Any +from unittest.mock import MagicMock, patch + + +def _source( + name: str, + status: str, + *, + critical: bool = False, + age_days: float | None = 1.0, +) -> dict[str, Any]: + """Минимальный source-dict в форме compute_freshness (нужные таске поля).""" + return {"source": name, "status": status, "critical": critical, "age_days": age_days} + + +def _payload(sources: list[dict[str, Any]], overall: str) -> dict[str, Any]: + return {"generated_at": "2026-06-25T09:00:00+00:00", "overall_status": overall, + "sources": sources} + + +def _run_with(payload_or_exc: Any) -> tuple[dict[str, Any], MagicMock]: + """Запускает таску с замоканными SessionLocal/compute_freshness/capture_message. + + payload_or_exc: dict → возвращается из compute_freshness; Exception → бросается. + Возвращает (summary, capture_message_mock). + """ + from app.workers.tasks import scrape_freshness_check as task_mod + + db = MagicMock() + if isinstance(payload_or_exc, Exception): + compute = MagicMock(side_effect=payload_or_exc) + else: + compute = MagicMock(return_value=payload_or_exc) + + with ( + patch.object(task_mod, "SessionLocal", return_value=db), + patch.object(task_mod, "compute_freshness", compute), + patch.object(task_mod.sentry_sdk, "capture_message") as cap, + ): + summary = task_mod.scrape_freshness_check.run() + + # Сессию всегда закрываем (try/finally). + db.close.assert_called_once() + return summary, cap + + +def test_stale_and_failed_emit_single_alert() -> None: + """Один stale + один failed (некритичный) → ровно один capture_message, level=warning, + сообщение называет оба источника.""" + payload = _payload( + [ + _source("kn", "ok", critical=True), + _source("objective", "stale", critical=True, age_days=10.0), + _source("nspd", "failed", critical=False, age_days=None), + ], + overall="stale", + ) + summary, cap = _run_with(payload) + + cap.assert_called_once() + args, kwargs = cap.call_args + message = args[0] + assert kwargs["level"] == "warning" # упал НЕкритичный → warning + assert "objective" in message + assert "nspd" in message + assert "kn" not in message # ok-источник не упоминается + + assert summary["alerted"] is True + assert summary["level"] == "warning" + assert summary["stale"] == 1 + assert summary["failed"] == 1 + assert summary["alertable"] == 2 + assert summary["checked"] == 3 + + +def test_critical_failed_escalates_to_error() -> None: + """Критичный источник полностью упал (failed) → level=error.""" + payload = _payload( + [ + _source("kn", "failed", critical=True, age_days=None), + _source("nspd", "stale", critical=False, age_days=12.0), + ], + overall="failed", + ) + summary, cap = _run_with(payload) + + cap.assert_called_once() + _, kwargs = cap.call_args + assert kwargs["level"] == "error" + assert summary["level"] == "error" + assert summary["alerted"] is True + + +def test_all_ok_emits_no_alert() -> None: + """Все источники ok → capture_message НЕ вызывается.""" + payload = _payload( + [ + _source("kn", "ok", critical=True), + _source("objective", "ok", critical=True), + _source("nspd", "ok"), + ], + overall="ok", + ) + summary, cap = _run_with(payload) + + cap.assert_not_called() + assert summary["alerted"] is False + assert summary["level"] is None + assert summary["alertable"] == 0 + assert summary["overall_status"] == "ok" + + +def test_compute_freshness_raising_is_graceful() -> None: + """compute_freshness бросает → таска логирует и возвращает {'error': ...}, не падает.""" + summary, cap = _run_with(RuntimeError("db boom")) + + cap.assert_not_called() + assert summary == {"error": "db boom"} + + +def test_endpoint_delegates_to_compute_freshness() -> None: + """Рефактор: HTTP-эндпоинт scrape_freshness просто возвращает compute_freshness(db). + + Без живой БД: monkeypatch compute_freshness → проверяем, что handler отдаёт его + результат байт-в-байт и прокидывает ту же сессию. + """ + from app.api.v1 import admin_scrape + + sentinel = {"generated_at": "x", "overall_status": "ok", "sources": []} + db = MagicMock() + with patch.object(admin_scrape, "compute_freshness", return_value=sentinel) as comp: + result = admin_scrape.scrape_freshness(db) + + comp.assert_called_once_with(db) + assert result is sentinel -- 2.45.3 From f59d874bacdd5efea66af66f080e989d993f2370 Mon Sep 17 00:00:00 2001 From: Light1YT Date: Thu, 25 Jun 2026 16:17:23 +0500 Subject: [PATCH 2/2] chore(codegen): regen api-types after /freshness handler refactor MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The endpoint docstring (→ openapi description → api-types JSDoc) changed when scrape_freshness was split into compute_freshness + thin handler, so the committed api-types.ts was stale and openapi-codegen-check failed. Regenerated via the CI chain (only JSDoc lines differ; no shape change). Refs #73 --- frontend/src/lib/api-types.ts | 15 ++++----------- 1 file changed, 4 insertions(+), 11 deletions(-) diff --git a/frontend/src/lib/api-types.ts b/frontend/src/lib/api-types.ts index df1fc61d..7a3182c7 100644 --- a/frontend/src/lib/api-types.ts +++ b/frontend/src/lib/api-types.ts @@ -1560,18 +1560,11 @@ export interface paths { }; /** * Scrape Freshness - * @description Свежесть данных по каждому источнику (для dashboard + alerting). + * @description GET-обёртка над compute_freshness — только читает, не алертит. * - * Для каждого источника один проход по его run-таблице агрегирует: - * - last_success_at — max(COALESCE(finished_at, started_at)) WHERE status='done' - * - last_attempt_at — max(started_at) (или created_at fallback) по всем прогонам - * - last_status — status последнего прогона - * - objects_updated_24h / _7d — сумма run-счётчика по done-прогонам в окне - * - age_days — возраст last_success в днях (NULL если успехов не было) - * - freshness — fresh / stale / critical / unknown (по порогам источника) - * - status — ok / stale / failed (для UI-светофора) - * - * `overall_status` агрегирует худший статус по критичным источникам. + * Вся агрегация живёт в compute_freshness (один источник истины); эндпоинт + * лишь прокидывает сессию и возвращает результат без изменений формы payload. + * Sentry-алёрты по критичным источникам шлёт beat-таска scrape_freshness_check. */ get: operations["scrape_freshness_api_v1_admin_scrape_freshness_get"]; put?: never; -- 2.45.3