feat(workers): data-freshness monitor (Celery beat → GlitchTip) #1904
6 changed files with 287 additions and 15 deletions
|
|
@ -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 ───────────────────────────────────────────────────
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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"
|
||||
|
|
|
|||
107
backend/app/workers/tasks/scrape_freshness_check.py
Normal file
107
backend/app/workers/tasks/scrape_freshness_check.py
Normal file
|
|
@ -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': '<msg>'} — исключение НЕ пробрасывается,
|
||||
чтобы не сорвать 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
|
||||
145
backend/tests/workers/test_scrape_freshness_check.py
Normal file
145
backend/tests/workers/test_scrape_freshness_check.py
Normal file
|
|
@ -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
|
||||
|
|
@ -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;
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue