From 3ece1ba339ae25d3113162a9f2d47d8fd456c52d Mon Sep 17 00:00:00 2001 From: bot-backend Date: Sat, 30 May 2026 23:00:05 +0300 Subject: [PATCH] feat(observability): sentry alert on N consecutive scraper failures (Refs #827) --- .../backend/app/services/scrape_runs.py | 97 ++++++++- .../backend/tests/test_scrape_run_alert.py | 203 ++++++++++++++++++ 2 files changed, 294 insertions(+), 6 deletions(-) create mode 100644 tradein-mvp/backend/tests/test_scrape_run_alert.py diff --git a/tradein-mvp/backend/app/services/scrape_runs.py b/tradein-mvp/backend/app/services/scrape_runs.py index 697a6daf..66451ee0 100644 --- a/tradein-mvp/backend/app/services/scrape_runs.py +++ b/tradein-mvp/backend/app/services/scrape_runs.py @@ -10,11 +10,88 @@ import json import logging from typing import Any +import sentry_sdk from sqlalchemy import text from sqlalchemy.orm import Session logger = logging.getLogger(__name__) +# Количество последовательных неудачных запусков (failed/banned), при достижении +# которого отправляется Sentry alert. Алерт срабатывает ровно при N-й подряд ошибке +# (anti-spam: не на каждой последующей). +CONSECUTIVE_FAILURE_ALERT_THRESHOLD = 3 + + +def _alert_if_consecutive_failures(db: Session, source: str) -> None: + """Отправить Sentry alert если последние CONSECUTIVE_FAILURE_ALERT_THRESHOLD + завершённых запусков для данного source имеют статус 'failed' или 'banned'. + + Anti-spam: алерт срабатывает ТОЛЬКО когда стрик РОВНО равен порогу — т.е. запрос + возвращает ровно N последних (failed|banned) и (N+1)-й, если существует, НЕ является + failed/banned. Это предотвращает повторный алерт на каждой ошибке сверх порога. + + Best-effort: весь блок обёрнут в try/except — сбой запроса или неинициализированный + Sentry НЕ должен нарушать вызывающий mark_* путь. + """ + n = CONSECUTIVE_FAILURE_ALERT_THRESHOLD + try: + # Берём последние N+1 завершённых (non-running) запусков по source. + # Сортируем по finished_at DESC чтобы самые свежие шли первыми. + rows = db.execute( + text( + """ + SELECT status FROM scrape_runs + WHERE source = :source + AND status IN ('failed', 'banned', 'done', 'cancelled') + ORDER BY finished_at DESC NULLS LAST + LIMIT :limit + """ + ), + {"source": source, "limit": n + 1}, + ).fetchall() + + if len(rows) < n: + # Ещё не набралось N завершённых запусков вообще — алерт не нужен. + return + + # Первые N должны быть все failed/banned. + first_n = rows[:n] + if not all(r.status in ("failed", "banned") for r in first_n): + return + + # (N+1)-й запуск, если есть, тоже должен НЕ быть failed/banned — иначе мы уже + # должны были отправить алерт раньше и не стоит дублировать. + if len(rows) > n and rows[n].status in ("failed", "banned"): + return + + # Стрик ровно достиг порога — отправляем алерт. + sentry_sdk.capture_message( + f"Scraper source '{source}' has {n} consecutive failed/banned runs — " + "manual intervention may be required (expired cookies / ban / broken parser).", + level="error", + ) + logger.error( + "sentry alert sent: source=%s has %d consecutive failed/banned runs", source, n + ) + except Exception: + pass # sentry_sdk not initialised in dev, or query failed — best-effort only + + +def _alert_on_run_id(db: Session, run_id: int) -> None: + """Вспомогательная обёртка: извлекает source по run_id и вызывает + _alert_if_consecutive_failures. Best-effort — не бросает исключений. + """ + try: + row = db.execute( + text("SELECT source FROM scrape_runs WHERE id = :run_id"), + {"run_id": run_id}, + ).fetchone() + if row is None: + return + _alert_if_consecutive_failures(db, str(row.source)) + except Exception: + pass + def create_run(db: Session, *, source: str, params: dict[str, Any]) -> int: """INSERT scrape_runs(source, status='running', params, started_at=NOW()). @@ -105,6 +182,8 @@ def mark_failed(db: Session, run_id: int, error: str, counters: dict[str, int]) if row is None: logger.warning("mark_failed no-op: run_id=%d not in 'running' state", run_id) db.commit() + # Алерт после коммита — статус уже персистирован в БД; best-effort (не бросает). + _alert_on_run_id(db, run_id) def mark_banned(db: Session, run_id: int, error: str, counters: dict[str, int]) -> None: @@ -135,6 +214,8 @@ def mark_banned(db: Session, run_id: int, error: str, counters: dict[str, int]) if row is None: logger.warning("mark_banned no-op: run_id=%d not in 'running' state", run_id) db.commit() + # Алерт после коммита — статус уже персистирован в БД; best-effort (не бросает). + _alert_on_run_id(db, run_id) def mark_cancelled(db: Session, run_id: int) -> bool: @@ -156,9 +237,10 @@ def mark_cancelled(db: Session, run_id: int) -> bool: def list_recent(db: Session, *, source: str, limit: int = 10) -> list[dict[str, Any]]: """Список последних N runs по source, в порядке убывания started_at.""" - rows = db.execute( - text( - """ + rows = ( + db.execute( + text( + """ SELECT id AS run_id, source, status, params, counters, error, started_at, finished_at, heartbeat_at FROM scrape_runs @@ -166,7 +248,10 @@ def list_recent(db: Session, *, source: str, limit: int = 10) -> list[dict[str, ORDER BY started_at DESC NULLS LAST LIMIT :limit """ - ), - {"source": source, "limit": limit}, - ).mappings().all() + ), + {"source": source, "limit": limit}, + ) + .mappings() + .all() + ) return [dict(r) for r in rows] diff --git a/tradein-mvp/backend/tests/test_scrape_run_alert.py b/tradein-mvp/backend/tests/test_scrape_run_alert.py new file mode 100644 index 00000000..af8a3979 --- /dev/null +++ b/tradein-mvp/backend/tests/test_scrape_run_alert.py @@ -0,0 +1,203 @@ +"""Unit tests for consecutive-failure Sentry alert in scrape_runs. + +All tests use a fully mocked DB session — no live DB required. +The mock simulates db.execute(...).fetchall() and db.execute(...).fetchone() +to control which run statuses are returned for the query in +_alert_if_consecutive_failures / _alert_on_run_id. +""" + +from __future__ import annotations + +from types import SimpleNamespace +from unittest.mock import MagicMock, patch + +from app.services.scrape_runs import ( + CONSECUTIVE_FAILURE_ALERT_THRESHOLD, + _alert_if_consecutive_failures, + mark_banned, + mark_failed, +) + +N = CONSECUTIVE_FAILURE_ALERT_THRESHOLD # 3 + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + + +def _make_rows(*statuses: str) -> list[SimpleNamespace]: + """Build a list of fake row objects with .status attribute.""" + return [SimpleNamespace(status=s) for s in statuses] + + +def _make_db(streak_rows: list[SimpleNamespace]) -> MagicMock: + """Return a mock Session whose execute().fetchall() returns streak_rows.""" + execute_result = MagicMock() + execute_result.fetchall.return_value = streak_rows + db = MagicMock() + db.execute.return_value = execute_result + return db + + +# --------------------------------------------------------------------------- +# _alert_if_consecutive_failures — unit tests +# --------------------------------------------------------------------------- + + +class TestAlertIfConsecutiveFailures: + def test_no_alert_when_fewer_than_n_runs(self) -> None: + """< N finished runs → no alert regardless of status.""" + db = _make_db(_make_rows(*["failed"] * (N - 1))) + with patch("app.services.scrape_runs.sentry_sdk") as mock_sentry: + _alert_if_consecutive_failures(db, "avito") + mock_sentry.capture_message.assert_not_called() + + def test_no_alert_when_streak_is_n_minus_1(self) -> None: + """N-1 consecutive failures (plus one older success) → no alert.""" + # Rows ordered newest-first: N-1 failures, then one 'done' + rows = _make_rows(*["failed"] * (N - 1), "done") + db = _make_db(rows) + with patch("app.services.scrape_runs.sentry_sdk") as mock_sentry: + _alert_if_consecutive_failures(db, "avito") + mock_sentry.capture_message.assert_not_called() + + def test_alert_fires_exactly_at_n(self) -> None: + """Exactly N consecutive failures (no older run) → alert sent once.""" + rows = _make_rows(*["failed"] * N) + db = _make_db(rows) + with patch("app.services.scrape_runs.sentry_sdk") as mock_sentry: + _alert_if_consecutive_failures(db, "avito") + mock_sentry.capture_message.assert_called_once() + (msg,) = mock_sentry.capture_message.call_args.args + assert "avito" in msg + assert str(N) in msg + + def test_alert_fires_with_mixed_failed_banned(self) -> None: + """Mix of 'failed' and 'banned' within N streak → alert sent.""" + rows = _make_rows("banned", "failed", "banned") + assert len(rows) == N + db = _make_db(rows) + with patch("app.services.scrape_runs.sentry_sdk") as mock_sentry: + _alert_if_consecutive_failures(db, "cian") + mock_sentry.capture_message.assert_called_once() + + def test_alert_fires_when_n_plus_1_is_done(self) -> None: + """N failures newest + 1 older 'done' → alert (streak just reached N).""" + rows = _make_rows(*["failed"] * N, "done") + db = _make_db(rows) + with patch("app.services.scrape_runs.sentry_sdk") as mock_sentry: + _alert_if_consecutive_failures(db, "yandex") + mock_sentry.capture_message.assert_called_once() + + def test_no_alert_when_n_plus_1_is_also_failed(self) -> None: + """N failures newest + 1 older 'failed' → no alert (streak > N, already fired).""" + rows = _make_rows(*["failed"] * (N + 1)) + db = _make_db(rows) + with patch("app.services.scrape_runs.sentry_sdk") as mock_sentry: + _alert_if_consecutive_failures(db, "avito") + mock_sentry.capture_message.assert_not_called() + + def test_no_alert_when_first_n_has_success(self) -> None: + """If any of the N newest runs is 'done' → no alert.""" + # 2 failures then 1 success in newest N + rows = _make_rows("failed", "done", "failed") + db = _make_db(rows) + with patch("app.services.scrape_runs.sentry_sdk") as mock_sentry: + _alert_if_consecutive_failures(db, "avito") + mock_sentry.capture_message.assert_not_called() + + def test_sentry_exception_does_not_propagate(self) -> None: + """If sentry_sdk.capture_message raises, the function must NOT raise.""" + rows = _make_rows(*["failed"] * N) + db = _make_db(rows) + with patch("app.services.scrape_runs.sentry_sdk") as mock_sentry: + mock_sentry.capture_message.side_effect = RuntimeError("sentry down") + # Should not raise + _alert_if_consecutive_failures(db, "avito") + + def test_db_query_exception_does_not_propagate(self) -> None: + """If db.execute raises, the function must NOT raise.""" + db = MagicMock() + db.execute.side_effect = Exception("connection lost") + with patch("app.services.scrape_runs.sentry_sdk") as mock_sentry: + _alert_if_consecutive_failures(db, "avito") + mock_sentry.capture_message.assert_not_called() + + +# --------------------------------------------------------------------------- +# mark_failed / mark_banned integration — verify alert hook is called +# --------------------------------------------------------------------------- + + +def _make_db_for_mark(source: str, streak_rows: list[SimpleNamespace]) -> MagicMock: + """Mock DB for mark_failed/mark_banned. + + Sequence of execute() calls in mark_failed: + 1. UPDATE scrape_runs SET status='failed' ... RETURNING id → .first() + 2. (after rollback) SELECT source FROM scrape_runs WHERE id=... → .fetchone() + 3. SELECT status FROM scrape_runs WHERE source=... LIMIT N+1 → .fetchall() + """ + db = MagicMock() + + # Call 1 (mark_failed UPDATE): .first() returns a row with id + update_result = MagicMock() + update_result.first.return_value = SimpleNamespace(id=42) + + # Call 2 (_alert_on_run_id SELECT source): .fetchone() returns source row + source_result = MagicMock() + source_result.fetchone.return_value = SimpleNamespace(source=source) + + # Call 3 (_alert_if_consecutive_failures SELECT statuses): .fetchall() + streak_result = MagicMock() + streak_result.fetchall.return_value = streak_rows + + db.execute.side_effect = [update_result, source_result, streak_result] + return db + + +class TestMarkFailedAlertIntegration: + def test_mark_failed_triggers_alert_on_nth_failure(self) -> None: + """mark_failed integrates with alert helper: alert fires on Nth streak.""" + rows = _make_rows(*["failed"] * N) + db = _make_db_for_mark("avito", rows) + + with patch("app.services.scrape_runs.sentry_sdk") as mock_sentry: + mark_failed(db, 42, "timeout", {}) + + mock_sentry.capture_message.assert_called_once() + + def test_mark_failed_no_alert_below_threshold(self) -> None: + """mark_failed: no alert when streak < N.""" + rows = _make_rows(*["failed"] * (N - 1)) + db = _make_db_for_mark("avito", rows) + + with patch("app.services.scrape_runs.sentry_sdk") as mock_sentry: + mark_failed(db, 42, "timeout", {}) + + mock_sentry.capture_message.assert_not_called() + + +class TestMarkBannedAlertIntegration: + def test_mark_banned_triggers_alert_on_nth_failure(self) -> None: + """mark_banned integrates with alert helper: alert fires on Nth streak.""" + rows = _make_rows(*["banned"] * N) + # Reuse the same mock builder — mark_banned has same execute sequence + db = _make_db_for_mark("cian", rows) + + with patch("app.services.scrape_runs.sentry_sdk") as mock_sentry: + mark_banned(db, 42, "403 Forbidden", {}) + + mock_sentry.capture_message.assert_called_once() + (msg,) = mock_sentry.capture_message.call_args.args + assert "cian" in msg + + def test_mark_banned_no_alert_below_threshold(self) -> None: + """mark_banned: no alert when streak < N.""" + rows = _make_rows(*["banned"] * (N - 1)) + db = _make_db_for_mark("cian", rows) + + with patch("app.services.scrape_runs.sentry_sdk") as mock_sentry: + mark_banned(db, 42, "403 Forbidden", {}) + + mock_sentry.capture_message.assert_not_called()