fix(tradein/scrapers): honest run status — стоп 'done' поверх провала и нуля
Три прод-факта, где status='done' врал о реальном исходе прогона:
- avito_detail_backfill 15.08: {"attempted":64,"failed":57,"enriched":6,"blocked":1}
-> 'done'. mark_backfill_finished звал mark_done, потому что produced=6 (>0);
ни _sweep_run_did_nothing (нет anchors_total/errors_count у backfill'ов), ни
_phase_totally_failed (голые "attempted"/"failed" без фазового префикса) эту
форму counters не ловили. Новый _failed_ratio_too_high внутри mark_done:
failed/attempted >= 0.5 -> 'failed', >= 0.15 -> тоже 'failed' (другая
формулировка причины в error-тексте) — 'partial' статусом не заведён: это
потребовало бы DROP+ADD CHECK constraint (051_scrape_runs_extend.sql) и
дообучения ещё 4 мест (Literal-фильтр admin API, статусы фронта, оба
IN-списка сторожей) — тот же класс проводки, что и у ban_kind (#2686/#2764),
который сознательно не стал новым статусом.
- yandex_newbuilding_sweep 26.07-10.08: десять прогонов подряд 'done' при
processed=5 succeeded=0 rows_inserted=0 failed_resolve=4-5 — сторож нулевого
результата (_alert_if_consecutive_zero_results) не видел ни один результатный
ключ этого sweep'а и молчал навсегда. _RESULT_COUNTER_KEYS дополнен
rows_inserted/processed (именно в этом порядке — rows_inserted это результат,
processed это попытки; иначе "5 обработано, 0 записано" замаскировалось бы
под measured-5).
- admin-витрина показывала new_count=0 у трёх подряд cian_full_load при реально
сохранённых saved_inserted=482/214/239 — full-load'ы не пишут ни 'new_count',
ни 'lots_inserted'. _column_counts дополнен saved_inserted/rows_inserted.
Правки продублированы в scraper_kit/orchestration/runs.py (byte-эквивалент
app.services.scrape_runs, см. докстринг модуля) для параллели: единственный
текущий писатель "attempted"/"failed" (mark_backfill_finished) живёт только в
app-копии, но приоритет ключей/константы держим синхронными на будущее.
Не тронуто: сознательно пустые sweep'ы (errors_count=0, honest empty) и малые
батчи (attempted < 3) — доля отказов на них не считается диагнозом.
Tests: tests/test_honest_run_status_failed_ratio.py (41 кейс, оба модуля,
включая точные прод-числа из трёх фактов выше) + regression-прогон 609 тестов
по всем файлам, трогающим scrape_runs/orchestration.runs — 0 регрессий.
This commit is contained in:
bot-backend
parent
e8fe9faa13
commit
885031420e
3 changed files with 468 additions and 8 deletions
|
|
@ -143,6 +143,18 @@ def _pick_int(counters: Mapping[str, Any], *keys: str) -> int | None:
|
|||
# unique_fetched — full-load'ы avito/cian/yandex (4 источника, 133 прогона) — раньше
|
||||
# сторож их не видел, хотя у cian_full_load 6 из 38 успешных прогонов
|
||||
# реально дали ноль.
|
||||
# rows_inserted — yandex_newbuilding_sweep (единственный писатель ключа с таким
|
||||
# именем на верхнем уровне counters): проверено на проде 26.07-10.08 —
|
||||
# десять прогонов подряд, все 'done', processed=5 succeeded=0
|
||||
# rows_inserted=0 failed_resolve=4-5. Ни total_seen/lots_fetched/
|
||||
# unique_fetched у него нет, поэтому раньше _run_result_count всегда
|
||||
# возвращал None ("не измерено") и стрик у сторожа не копился никогда
|
||||
# (honest-run-status).
|
||||
# processed — тот же sweep: сколько домов взял в работу. НАМЕРЕННО стоит ПОСЛЕ
|
||||
# rows_inserted в кортеже — processed это счётчик ПОПЫТОК (аналог
|
||||
# attempted), а не результата: у него ненулевое значение (=limit) даже
|
||||
# когда rows_inserted=0, и если бы он читался первым, «5 обработано,
|
||||
# 0 записано» замаскировалось бы под measured-5, а не measured-0.
|
||||
# Сводить сюда счётчики ОСТАЛЬНЫХ задач бессмысленно: на проде 28 источников (2650
|
||||
# прогонов) не имеют общего результатного ключа вовсе — у каждого свой словарь
|
||||
# (deactivated / rows_written / poi_loaded / snapshotted / upserted / listings_matched
|
||||
|
|
@ -150,7 +162,13 @@ def _pick_int(counters: Mapping[str, Any], *keys: str) -> int | None:
|
|||
# трёх мониторов результата нет по смыслу. Ноль у них — часто ЗДОРОВЫЙ ответ
|
||||
# (deactivate_stale_* без протухших объявлений). Поэтому сторож не угадывает их
|
||||
# словарь, а честно признаёт, что мерить нечем — см. _run_result_count.
|
||||
_RESULT_COUNTER_KEYS = ("total_seen", "lots_fetched", "unique_fetched")
|
||||
_RESULT_COUNTER_KEYS = (
|
||||
"total_seen",
|
||||
"lots_fetched",
|
||||
"unique_fetched",
|
||||
"rows_inserted",
|
||||
"processed",
|
||||
)
|
||||
|
||||
|
||||
def _run_result_count(counters: Mapping[str, Any] | None) -> int | None:
|
||||
|
|
@ -282,6 +300,63 @@ def _phase_totally_failed(counters: Mapping[str, Any]) -> str | None:
|
|||
return None
|
||||
|
||||
|
||||
# honest-run-status (2026-08-15): доля отказов, которая обесценивает формально ненулевой
|
||||
# сбор. Прод-факт avito_detail_backfill 15.08: {"attempted":64,"failed":57,"enriched":6,
|
||||
# "blocked":1} — 89% попыток отказали, а mark_backfill_finished всё равно звал mark_done,
|
||||
# потому что "produced != 0" (6 обогащено). Ни _sweep_run_did_nothing (нужны
|
||||
# anchors_total/errors_count, у backfill'ов их нет), ни _phase_totally_failed (нужна пара
|
||||
# "<phase>_attempted"/"<phase>_failed" — здесь голые "attempted"/"failed" без фазового
|
||||
# префикса, `"attempted".endswith("_attempted")` не матчит) эту форму counters не ловят —
|
||||
# обе проверки написаны под СВОИ формы, а не под backfill'овскую.
|
||||
#
|
||||
# Порог 'failed' — половина и больше отказов: сбор для практических целей провалился,
|
||||
# даже если несколько записей всё же обогатились. Порог 'partial' НЕ заведён отдельным
|
||||
# статусом scrape_runs.status — это потребовало бы миграции (DROP+ADD CHECK constraint,
|
||||
# 051_scrape_runs_extend.sql) и обучило бы новому значению ещё 4 места (Literal-фильтр
|
||||
# admin API, хардкод статусов фронта, оба IN-списка сторожей) — тот же класс "оборванной
|
||||
# проводки", из-за которого заведён #2686/ban_kind. Вместо статуса — тот же диагноз, что и
|
||||
# у ban_kind: causa в тексте `error`, терминальный статус один ('failed'). 0.15..0.5 —
|
||||
# та же 'failed', но с другой формулировкой причины ("деградировал", не "провалился"), чтобы
|
||||
# оператор видел разницу читая error, не только status.
|
||||
FAILED_RATIO_FAILED_THRESHOLD = 0.5
|
||||
FAILED_RATIO_DEGRADED_THRESHOLD = 0.15
|
||||
# Минимум попыток, при котором доля вообще что-то значит — иначе 1 отказ из 2 (=0.5)
|
||||
# палит статус на шуме единичного случая. То же рассуждение и то же число, что у
|
||||
# _PHASE_MIN_ATTEMPTS (см. выше).
|
||||
_FAILED_RATIO_MIN_ATTEMPTS = _PHASE_MIN_ATTEMPTS
|
||||
|
||||
|
||||
def _failed_ratio_too_high(counters: Mapping[str, Any]) -> str | None:
|
||||
"""Прогон, у которого доля отказов слишком велика, даже если что-то собрано.
|
||||
|
||||
Возвращает текст причины (для error) либо None. Читает ГОЛЫЕ ключи "attempted"/
|
||||
"failed" (без фазового префикса) — сейчас это словарь только у четырёх
|
||||
detail-backfill'ов (avito/yandex/domclick/newbuilding_enrich), все идут через
|
||||
mark_backfill_finished → mark_done. `attempted < _FAILED_RATIO_MIN_ATTEMPTS` или
|
||||
отсутствие любого из ключей → None (нечем/не о чём судить — счётчики либо не
|
||||
заполнены, либо принадлежат другому источнику со своим словарём).
|
||||
|
||||
Что признак НЕ доказывает: КТО виноват (площадка, наш прокси, наш парсер) — поэтому
|
||||
'failed' без диагноза, как и у #2625/#2700/#2764.
|
||||
"""
|
||||
attempted = _pick_int(counters, "attempted")
|
||||
failed = _pick_int(counters, "failed")
|
||||
if attempted is None or failed is None or attempted < _FAILED_RATIO_MIN_ATTEMPTS:
|
||||
return None
|
||||
ratio = failed / max(attempted, 1)
|
||||
if ratio >= FAILED_RATIO_FAILED_THRESHOLD:
|
||||
verb = "провалился"
|
||||
elif ratio >= FAILED_RATIO_DEGRADED_THRESHOLD:
|
||||
verb = "деградировал"
|
||||
else:
|
||||
return None
|
||||
return (
|
||||
f"failed-ratio-honest-status: сбор {verb} — {failed} из {attempted} попыток "
|
||||
f"отказали (доля {ratio:.0%}); формально ненулевой результат этого не искупает. "
|
||||
f"Причина НЕ установлена — статус 'failed' без диагноза"
|
||||
)
|
||||
|
||||
|
||||
def _column_counts(counters: dict[str, int]) -> tuple[int | None, int | None]:
|
||||
"""Извлечь значения для dedicated-колонок total_seen / new_count из jsonb-counters.
|
||||
|
||||
|
|
@ -292,13 +367,22 @@ def _column_counts(counters: dict[str, int]) -> tuple[int | None, int | None]:
|
|||
показывала total_seen=0 при реально сохранённых строках (audit #1871/#1926).
|
||||
|
||||
Приоритет ключей:
|
||||
- total_seen ← _RESULT_COUNTER_KEYS (total_seen / lots_fetched / unique_fetched)
|
||||
- new_count ← 'new_count' (если уже есть) иначе 'lots_inserted'
|
||||
- total_seen ← _RESULT_COUNTER_KEYS (total_seen / lots_fetched / unique_fetched /
|
||||
rows_inserted / processed)
|
||||
- new_count ← 'new_count' / 'lots_inserted' / 'saved_inserted' / 'rows_inserted'
|
||||
(первый присутствующий). 'saved_inserted' — full-load'ы (cian/avito/yandex,
|
||||
CianFullLoadCounters и аналоги в pipeline.py): на проде витрина показывала
|
||||
new_count=0 у трёх подряд cian_full_load при реально сохранённых
|
||||
saved_inserted=482/214/239 (honest-run-status) — ключ 'new_count'/'lots_inserted'
|
||||
у full-load'ов в counters не пишется вовсе. 'rows_inserted' — тот же ключ,
|
||||
которым yandex_newbuilding_sweep сообщает число upsert'ов.
|
||||
|
||||
Возвращает (total_seen, new_count); None для ключа, которого нет в counters —
|
||||
тогда соответствующая колонка не перезаписывается (COALESCE-семантика в UPDATE).
|
||||
"""
|
||||
return _run_result_count(counters), _pick_int(counters, "new_count", "lots_inserted")
|
||||
return _run_result_count(counters), _pick_int(
|
||||
counters, "new_count", "lots_inserted", "saved_inserted", "rows_inserted"
|
||||
)
|
||||
|
||||
|
||||
def _alert_if_consecutive_failures(db: Session, source: str) -> None:
|
||||
|
|
@ -558,6 +642,11 @@ def mark_done(db: Session, run_id: int, counters: dict[str, int]) -> None:
|
|||
#2700: там же — отказ называть успехом прогон, у которого отказала КАЖДАЯ попытка
|
||||
целой фазы (см. _phase_totally_failed). Отличие от #2625: тот случай про «не сделано
|
||||
ничего», этот — про «одно направление работы мертво, а суммарный сбор это прячет».
|
||||
|
||||
honest-run-status: там же — отказ называть успехом прогон с высокой долей отказов,
|
||||
даже если собрано > 0 (см. _failed_ratio_too_high). Отличие от #2625/#2700: те два
|
||||
смотрят на «всё или ничего» (все якоря / вся фаза), этот — на ДОЛЮ отказов у
|
||||
detail-backfill'ов, где ни один из первых двух признаков не матчит форму counters.
|
||||
"""
|
||||
did_nothing = _sweep_run_did_nothing(counters)
|
||||
if did_nothing is not None:
|
||||
|
|
@ -569,6 +658,11 @@ def mark_done(db: Session, run_id: int, counters: dict[str, int]) -> None:
|
|||
logger.error("%s run_id=%d", phase_dead, run_id)
|
||||
mark_failed(db, run_id, phase_dead, counters)
|
||||
return
|
||||
ratio_bad = _failed_ratio_too_high(counters)
|
||||
if ratio_bad is not None:
|
||||
logger.error("%s run_id=%d", ratio_bad, run_id)
|
||||
mark_failed(db, run_id, ratio_bad, counters)
|
||||
return
|
||||
total_seen, new_count = _column_counts(counters)
|
||||
row = db.execute(
|
||||
text(
|
||||
|
|
|
|||
267
tradein-mvp/backend/tests/test_honest_run_status_failed_ratio.py
Normal file
267
tradein-mvp/backend/tests/test_honest_run_status_failed_ratio.py
Normal file
|
|
@ -0,0 +1,267 @@
|
|||
"""honest-run-status (2026-08-15): статус прогона не должен рапортовать 'done' поверх
|
||||
провала или нуля. Три прод-факта закрыты этой правкой:
|
||||
|
||||
(a) avito_detail_backfill 15.08: {"attempted":64,"failed":57,"enriched":6,"blocked":1}
|
||||
-> status='done' — 89% отказов, статус зелёный. mark_backfill_finished звал
|
||||
mark_done, потому что produced=6 (>0); ни _sweep_run_did_nothing (нет
|
||||
anchors_total/errors_count у backfill'ов), ни _phase_totally_failed (ключи
|
||||
"attempted"/"failed" без фазового префикса) эту форму counters не ловили.
|
||||
Фикс: _failed_ratio_too_high внутри mark_done.
|
||||
|
||||
(b) yandex_newbuilding_sweep 26.07-10.08: десять прогонов подряд 'done' при
|
||||
processed=5, succeeded=0, rows_inserted=0, failed_resolve=4-5 — сторож нулевого
|
||||
результата (_alert_if_consecutive_zero_results) слеп, т.к. _RESULT_COUNTER_KEYS
|
||||
не знал ни одного ключа этого sweep'а (total_seen/lots_fetched/unique_fetched).
|
||||
Фикс: _RESULT_COUNTER_KEYS дополнен rows_inserted/processed (в этом порядке —
|
||||
rows_inserted это РЕЗУЛЬТАТ, processed это ПОПЫТКИ).
|
||||
|
||||
(c) admin-витрина показывала new_count=0 у трёх подряд cian_full_load, хотя реально
|
||||
сохранено saved_inserted=482/214/239 — full-load'ы не пишут ни 'new_count', ни
|
||||
'lots_inserted'. Фикс: _column_counts дополнен saved_inserted/rows_inserted.
|
||||
|
||||
Проверяем на обоих модулях (kit-копия и app-копия — байт-эквивалентны по докстрингу
|
||||
runs.py), тем же паттерном, что test_2625_run_that_did_nothing.py.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
from typing import Any
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
import pytest
|
||||
|
||||
os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")
|
||||
|
||||
from scraper_kit.orchestration import runs as kit_runs
|
||||
|
||||
from app.services import scrape_runs as app_runs
|
||||
|
||||
_MODULES = {"kit": kit_runs, "app": app_runs}
|
||||
|
||||
|
||||
def _capture_status(mod: Any, counters: dict[str, Any]) -> list[str]:
|
||||
"""Прогнать mark_done на фейковой сессии, вернуть статусы всех UPDATE'ов.
|
||||
|
||||
Тот же helper, что в test_2625_run_that_did_nothing.py — читаем СТАТУС В SQL, а не
|
||||
имя вызванной функции.
|
||||
"""
|
||||
statuses: list[str] = []
|
||||
|
||||
def _execute(stmt: Any, *args: Any, **kwargs: Any) -> MagicMock:
|
||||
sql = str(stmt)
|
||||
for status in ("done", "failed", "banned"):
|
||||
if f"status = '{status}'" in sql:
|
||||
statuses.append(status)
|
||||
return MagicMock()
|
||||
|
||||
db = MagicMock()
|
||||
db.execute.side_effect = _execute
|
||||
with patch.object(mod, "sentry_sdk", MagicMock()):
|
||||
mod.mark_done(db, 1, dict(counters))
|
||||
return statuses
|
||||
|
||||
|
||||
def _capture_backfill_status(
|
||||
counters: dict[str, Any], *, source: str = "avito_detail_backfill", aborted: bool = False
|
||||
) -> list[str]:
|
||||
"""Прогнать app_runs.mark_backfill_finished на фейковой сессии (mark_done НЕ мокан —
|
||||
в отличие от test_backfill_honest_status.py, здесь важно именно его РЕАЛЬНОЕ
|
||||
поведение: mark_backfill_finished решает вызвать mark_done, а решает ли mark_done
|
||||
остаться 'done' или сам себя переквалифицировать в 'failed' — предмет этого теста).
|
||||
|
||||
mark_backfill_finished есть только в app_runs (kit-копия его не держит — см.
|
||||
docstring модуля runs.py, "mark_skipped есть только здесь" — тот же принцип
|
||||
относится к продуктовым финализаторам detail-backfill'ов).
|
||||
"""
|
||||
statuses: list[str] = []
|
||||
|
||||
def _execute(stmt: Any, *args: Any, **kwargs: Any) -> MagicMock:
|
||||
sql = str(stmt)
|
||||
for status in ("done", "failed", "banned"):
|
||||
if f"status = '{status}'" in sql:
|
||||
statuses.append(status)
|
||||
return MagicMock()
|
||||
|
||||
db = MagicMock()
|
||||
db.execute.side_effect = _execute
|
||||
with patch.object(app_runs, "sentry_sdk", MagicMock()):
|
||||
app_runs.mark_backfill_finished(
|
||||
db, 1, dict(counters), source=source, aborted_by_blocks=aborted
|
||||
)
|
||||
return statuses
|
||||
|
||||
|
||||
# ── (a) failed_ratio: прод-факт avito_detail_backfill 15.08 ─────────────────────────
|
||||
|
||||
|
||||
def test_prod_fact_avito_15_08_no_longer_done() -> None:
|
||||
"""{"attempted":64,"failed":57,"enriched":6,"blocked":1} — 89% отказов — 'failed',
|
||||
НЕ 'done'. Красный на старом коде (produced=6 != 0 -> mark_done -> 'done')."""
|
||||
counters = {"attempted": 64, "failed": 57, "enriched": 6, "blocked": 1}
|
||||
assert _capture_backfill_status(counters) == ["failed"]
|
||||
|
||||
|
||||
def test_prod_fact_avito_reason_names_the_ratio() -> None:
|
||||
reason = app_runs._failed_ratio_too_high(
|
||||
{"attempted": 64, "failed": 57, "enriched": 6, "blocked": 1}
|
||||
)
|
||||
assert reason is not None
|
||||
assert "failed-ratio-honest-status" in reason
|
||||
assert "57 из 64" in reason
|
||||
assert "89%" in reason
|
||||
|
||||
|
||||
@pytest.mark.parametrize("name", list(_MODULES))
|
||||
@pytest.mark.parametrize(
|
||||
("counters", "flagged", "why"),
|
||||
[
|
||||
({"attempted": 64, "failed": 57}, True, "прод-факт: 89% отказов"),
|
||||
({"attempted": 10, "failed": 5}, True, "ровно порог failed (0.5)"),
|
||||
({"attempted": 20, "failed": 3}, True, "ровно порог degraded (0.15)"),
|
||||
({"attempted": 20, "failed": 2}, False, "ниже порога degraded (0.10)"),
|
||||
({"attempted": 2, "failed": 2}, False, "ratio=1.0, но < _FAILED_RATIO_MIN_ATTEMPTS"),
|
||||
({"attempted": 0, "failed": 0}, False, "нет попыток вовсе"),
|
||||
({"failed": 5}, False, "нет attempted — чужой словарь"),
|
||||
({"attempted": 50}, False, "нет failed — чужой словарь"),
|
||||
({}, False, "пустые counters"),
|
||||
(
|
||||
{"anchors_total": 5, "errors_count": 5, "lots_fetched": 0},
|
||||
False,
|
||||
"sweep-словарь (anchors_total), не detail-backfill",
|
||||
),
|
||||
],
|
||||
)
|
||||
def test_failed_ratio_classifier_boundaries(
|
||||
name: str, counters: dict[str, Any], flagged: bool, why: str
|
||||
) -> None:
|
||||
reason = _MODULES[name]._failed_ratio_too_high(counters)
|
||||
assert (reason is not None) is flagged, why
|
||||
|
||||
|
||||
# ── (5) не должен палить прогоны с малой/умеренной долей отказов ────────────────────
|
||||
|
||||
|
||||
@pytest.mark.parametrize("name", list(_MODULES))
|
||||
def test_low_failure_ratio_stays_done(name: str) -> None:
|
||||
"""Штатный шум (10% отказов) не становится 'failed' — не каждый отказ диагноз."""
|
||||
counters = {"attempted": 50, "enriched": 45, "failed": 5}
|
||||
assert _capture_status(_MODULES[name], counters) == ["done"]
|
||||
|
||||
|
||||
def test_tiny_batch_zero_produced_fails_via_old_rule_not_ratio() -> None:
|
||||
"""2 попытки, обе отказали, produced=0 — доля тут не при чём (attempted < floor
|
||||
_FAILED_RATIO_MIN_ATTEMPTS, _failed_ratio_too_high вернул бы None); статус всё
|
||||
равно 'failed', но по СТАРОМУ правилу #2674 (produced==0), внутри
|
||||
mark_backfill_finished — mark_done/_failed_ratio_too_high тут не вызываются вовсе.
|
||||
Показывает, что новая проверка не дублирует и не подменяет старую."""
|
||||
counters = {"attempted": 2, "enriched": 0, "failed": 2}
|
||||
assert _capture_backfill_status(counters) == ["failed"]
|
||||
|
||||
|
||||
def test_tiny_batch_with_partial_success_stays_done() -> None:
|
||||
"""2 попытки, 1 успех, 1 отказ (ratio=0.5, но attempted < floor=3) — стрик слишком
|
||||
короткий, чтобы доля что-то значила -> остаётся 'done'."""
|
||||
counters = {"attempted": 2, "enriched": 1, "failed": 1}
|
||||
assert _capture_backfill_status(counters) == ["done"]
|
||||
|
||||
|
||||
@pytest.mark.parametrize("name", list(_MODULES))
|
||||
def test_honest_empty_sweep_unaffected_by_failed_ratio(name: str) -> None:
|
||||
"""Сознательно спящее расписание (город без новостроек): sweep-словарь без
|
||||
attempted/failed вовсе -> failed_ratio не о чем судить, честная пустота остаётся
|
||||
'done' (см. также test_2625_run_that_did_nothing.py::test_honest_empty_stays_done)."""
|
||||
counters = {"anchors_total": 1, "errors_count": 0, "lots_fetched": 0}
|
||||
assert _capture_status(_MODULES[name], counters) == ["done"]
|
||||
|
||||
|
||||
# ── (b) _RESULT_COUNTER_KEYS: прод-факт yandex_newbuilding_sweep 26.07-10.08 ─────────
|
||||
|
||||
|
||||
def test_prod_fact_yandex_newbuilding_sweep_measured_as_zero() -> None:
|
||||
"""processed=5, succeeded=0, rows_inserted=0, failed_resolve=4 — раньше
|
||||
_run_result_count возвращал None ("не измерено"); теперь — измеренный 0."""
|
||||
counters = {
|
||||
"total": 309,
|
||||
"fetchable": 200,
|
||||
"pending": 50,
|
||||
"processed": 5,
|
||||
"skipped_already_enriched": 0,
|
||||
"succeeded": 0,
|
||||
"resolved_slug": 1,
|
||||
"failed_resolve": 4,
|
||||
"failed_fetch": 0,
|
||||
"rows_inserted": 0,
|
||||
"duration_sec": 42.0,
|
||||
}
|
||||
assert app_runs._run_result_count(counters) == 0
|
||||
assert kit_runs._run_result_count(counters) == 0
|
||||
|
||||
|
||||
def test_rows_inserted_takes_priority_over_processed() -> None:
|
||||
"""rows_inserted (результат) читается ПЕРЕД processed (попытки) — иначе "5
|
||||
обработано, 0 записано" замаскировалось бы под measured-5."""
|
||||
counters = {"processed": 5, "rows_inserted": 0}
|
||||
assert app_runs._run_result_count(counters) == 0
|
||||
|
||||
|
||||
def test_processed_is_fallback_when_rows_inserted_absent() -> None:
|
||||
counters = {"processed": 3}
|
||||
assert app_runs._run_result_count(counters) == 3
|
||||
|
||||
|
||||
@pytest.mark.parametrize("name", list(_MODULES))
|
||||
def test_zero_result_watchdog_now_fires_for_newbuilding_sweep_streak(name: str) -> None:
|
||||
"""(b) integration: 3 подряд yandex_newbuilding_sweep-подобных 'done' с
|
||||
rows_inserted=0 -> алерт срабатывает. До фикса _RESULT_COUNTER_KEYS сторож считал
|
||||
результат "не измеренным" и молчал бы вечно (см. #2703 в docstring модуля)."""
|
||||
mod = _MODULES[name]
|
||||
row = MagicMock()
|
||||
row.status = "done"
|
||||
row.counters = {"processed": 5, "succeeded": 0, "rows_inserted": 0, "failed_resolve": 4}
|
||||
db = MagicMock()
|
||||
result = MagicMock()
|
||||
result.fetchall.return_value = [row, row, row]
|
||||
db.execute.return_value = result
|
||||
with patch.object(mod, "sentry_sdk") as mock_sentry:
|
||||
mod._alert_if_consecutive_zero_results(db, "yandex_newbuilding_sweep")
|
||||
mock_sentry.capture_message.assert_called_once()
|
||||
|
||||
|
||||
# ── (c) _column_counts: прод-факт cian_full_load new_count=0 при saved_inserted>0 ───
|
||||
|
||||
|
||||
@pytest.mark.parametrize("name", list(_MODULES))
|
||||
def test_prod_fact_cian_full_load_saved_inserted_surfaces_as_new_count(name: str) -> None:
|
||||
"""saved_inserted=482 (прод-факт: три подряд прогона 482/214/239) — new_count
|
||||
больше не 0, хотя ключей 'new_count'/'lots_inserted' в counters нет вовсе."""
|
||||
counters = {"unique_fetched": 1200, "saved_inserted": 482, "saved_updated": 30}
|
||||
total_seen, new_count = _MODULES[name]._column_counts(counters)
|
||||
assert total_seen == 1200
|
||||
assert new_count == 482
|
||||
|
||||
|
||||
@pytest.mark.parametrize("name", list(_MODULES))
|
||||
def test_yandex_newbuilding_rows_inserted_surfaces_as_new_count(name: str) -> None:
|
||||
counters = {"rows_inserted": 7}
|
||||
_, new_count = _MODULES[name]._column_counts(counters)
|
||||
assert new_count == 7
|
||||
|
||||
|
||||
@pytest.mark.parametrize("name", list(_MODULES))
|
||||
def test_new_count_priority_unchanged_by_new_keys(name: str) -> None:
|
||||
"""'new_count' явный ключ всё ещё побеждает 'lots_inserted'/'saved_inserted' —
|
||||
расширение списка не меняет приоритет уже существующих ключей."""
|
||||
counters = {"new_count": 5, "lots_inserted": 99, "saved_inserted": 1}
|
||||
_, new_count = _MODULES[name]._column_counts(counters)
|
||||
assert new_count == 5
|
||||
|
||||
|
||||
@pytest.mark.parametrize("name", list(_MODULES))
|
||||
def test_lots_inserted_still_beats_saved_inserted(name: str) -> None:
|
||||
"""Порядок пикулярно НЕ переставлен для уже существующей пары — 'lots_inserted'
|
||||
(city/newbuilding-sweep'ы) проверяется раньше 'saved_inserted' (full-load'ы),
|
||||
т.к. это разные, непересекающиеся семейства источников."""
|
||||
counters = {"lots_inserted": 12, "saved_inserted": 999}
|
||||
_, new_count = _MODULES[name]._column_counts(counters)
|
||||
assert new_count == 12
|
||||
|
|
@ -138,6 +138,18 @@ def _pick_int(counters: Mapping[str, Any], *keys: str) -> int | None:
|
|||
# unique_fetched — full-load'ы avito/cian/yandex (4 источника, 133 прогона) — раньше
|
||||
# сторож их не видел, хотя у cian_full_load 6 из 38 успешных прогонов
|
||||
# реально дали ноль.
|
||||
# rows_inserted — yandex_newbuilding_sweep (единственный писатель ключа с таким
|
||||
# именем на верхнем уровне counters): проверено на проде 26.07-10.08 —
|
||||
# десять прогонов подряд, все 'done', processed=5 succeeded=0
|
||||
# rows_inserted=0 failed_resolve=4-5. Ни total_seen/lots_fetched/
|
||||
# unique_fetched у него нет, поэтому раньше _run_result_count всегда
|
||||
# возвращал None ("не измерено") и стрик у сторожа не копился никогда
|
||||
# (honest-run-status).
|
||||
# processed — тот же sweep: сколько домов взял в работу. НАМЕРЕННО стоит ПОСЛЕ
|
||||
# rows_inserted в кортеже — processed это счётчик ПОПЫТОК (аналог
|
||||
# attempted), а не результата: у него ненулевое значение (=limit) даже
|
||||
# когда rows_inserted=0, и если бы он читался первым, «5 обработано,
|
||||
# 0 записано» замаскировалось бы под measured-5, а не measured-0.
|
||||
# Сводить сюда счётчики ОСТАЛЬНЫХ задач бессмысленно: на проде 28 источников (2650
|
||||
# прогонов) не имеют общего результатного ключа вовсе — у каждого свой словарь
|
||||
# (deactivated / rows_written / poi_loaded / snapshotted / upserted / listings_matched
|
||||
|
|
@ -145,7 +157,13 @@ def _pick_int(counters: Mapping[str, Any], *keys: str) -> int | None:
|
|||
# трёх мониторов результата нет по смыслу. Ноль у них — часто ЗДОРОВЫЙ ответ
|
||||
# (deactivate_stale_* без протухших объявлений). Поэтому сторож не угадывает их
|
||||
# словарь, а честно признаёт, что мерить нечем — см. _run_result_count.
|
||||
_RESULT_COUNTER_KEYS = ("total_seen", "lots_fetched", "unique_fetched")
|
||||
_RESULT_COUNTER_KEYS = (
|
||||
"total_seen",
|
||||
"lots_fetched",
|
||||
"unique_fetched",
|
||||
"rows_inserted",
|
||||
"processed",
|
||||
)
|
||||
|
||||
|
||||
def _run_result_count(counters: Mapping[str, Any] | None) -> int | None:
|
||||
|
|
@ -277,6 +295,68 @@ def _phase_totally_failed(counters: Mapping[str, Any]) -> str | None:
|
|||
return None
|
||||
|
||||
|
||||
# honest-run-status (2026-08-15): доля отказов, которая обесценивает формально ненулевой
|
||||
# сбор. Прод-факт avito_detail_backfill 15.08: {"attempted":64,"failed":57,"enriched":6,
|
||||
# "blocked":1} — 89% попыток отказали, а mark_backfill_finished всё равно звал mark_done,
|
||||
# потому что "produced != 0" (6 обогащено). Ни _sweep_run_did_nothing (нужны
|
||||
# anchors_total/errors_count, у backfill'ов их нет), ни _phase_totally_failed (нужна пара
|
||||
# "<phase>_attempted"/"<phase>_failed" — здесь голые "attempted"/"failed" без фазового
|
||||
# префикса, `"attempted".endswith("_attempted")` не матчит) эту форму counters не ловят —
|
||||
# обе проверки написаны под СВОИ формы, а не под backfill'овскую.
|
||||
#
|
||||
# Порог 'failed' — половина и больше отказов: сбор для практических целей провалился,
|
||||
# даже если несколько записей всё же обогатились. Порог 'partial' НЕ заведён отдельным
|
||||
# статусом scrape_runs.status — это потребовало бы миграции (DROP+ADD CHECK constraint,
|
||||
# 051_scrape_runs_extend.sql) и обучило бы новому значению ещё 4 места (Literal-фильтр
|
||||
# admin API, хардкод статусов фронта, оба IN-списка сторожей) — тот же класс "оборванной
|
||||
# проводки", из-за которого заведён #2686/ban_kind. Вместо статуса — тот же диагноз, что и
|
||||
# у ban_kind: causa в тексте `error`, терминальный статус один ('failed'). 0.15..0.5 —
|
||||
# та же 'failed', но с другой формулировкой причины ("деградировал", не "провалился"), чтобы
|
||||
# оператор видел разницу читая error, не только status.
|
||||
#
|
||||
# mark_backfill_finished (единственный писатель "attempted"/"failed" на верхнем уровне
|
||||
# counters) живёт только в app.services.scrape_runs — здесь эта проверка сейчас неактивна
|
||||
# ни для одного реального вызывающего, но kit-копия держится байт-эквивалентной app-копии
|
||||
# (см. docstring модуля), и будущий kit-native job с тем же словарём получит её даром.
|
||||
FAILED_RATIO_FAILED_THRESHOLD = 0.5
|
||||
FAILED_RATIO_DEGRADED_THRESHOLD = 0.15
|
||||
# Минимум попыток, при котором доля вообще что-то значит — иначе 1 отказ из 2 (=0.5)
|
||||
# палит статус на шуме единичного случая. То же рассуждение и то же число, что у
|
||||
# _PHASE_MIN_ATTEMPTS (см. выше).
|
||||
_FAILED_RATIO_MIN_ATTEMPTS = _PHASE_MIN_ATTEMPTS
|
||||
|
||||
|
||||
def _failed_ratio_too_high(counters: Mapping[str, Any]) -> str | None:
|
||||
"""Прогон, у которого доля отказов слишком велика, даже если что-то собрано.
|
||||
|
||||
Возвращает текст причины (для error) либо None. Читает ГОЛЫЕ ключи "attempted"/
|
||||
"failed" (без фазового префикса) — сейчас это словарь только у четырёх
|
||||
detail-backfill'ов (avito/yandex/domclick/newbuilding_enrich), все идут через
|
||||
mark_backfill_finished → mark_done. `attempted < _FAILED_RATIO_MIN_ATTEMPTS` или
|
||||
отсутствие любого из ключей → None (нечем/не о чём судить — счётчики либо не
|
||||
заполнены, либо принадлежат другому источнику со своим словарём).
|
||||
|
||||
Что признак НЕ доказывает: КТО виноват (площадка, наш прокси, наш парсер) — поэтому
|
||||
'failed' без диагноза, как и у #2625/#2700/#2764.
|
||||
"""
|
||||
attempted = _pick_int(counters, "attempted")
|
||||
failed = _pick_int(counters, "failed")
|
||||
if attempted is None or failed is None or attempted < _FAILED_RATIO_MIN_ATTEMPTS:
|
||||
return None
|
||||
ratio = failed / max(attempted, 1)
|
||||
if ratio >= FAILED_RATIO_FAILED_THRESHOLD:
|
||||
verb = "провалился"
|
||||
elif ratio >= FAILED_RATIO_DEGRADED_THRESHOLD:
|
||||
verb = "деградировал"
|
||||
else:
|
||||
return None
|
||||
return (
|
||||
f"failed-ratio-honest-status: сбор {verb} — {failed} из {attempted} попыток "
|
||||
f"отказали (доля {ratio:.0%}); формально ненулевой результат этого не искупает. "
|
||||
f"Причина НЕ установлена — статус 'failed' без диагноза"
|
||||
)
|
||||
|
||||
|
||||
def _column_counts(counters: dict[str, int]) -> tuple[int | None, int | None]:
|
||||
"""Извлечь значения для dedicated-колонок total_seen / new_count из jsonb-counters.
|
||||
|
||||
|
|
@ -287,13 +367,22 @@ def _column_counts(counters: dict[str, int]) -> tuple[int | None, int | None]:
|
|||
показывала total_seen=0 при реально сохранённых строках (audit #1871/#1926).
|
||||
|
||||
Приоритет ключей:
|
||||
- total_seen ← _RESULT_COUNTER_KEYS (total_seen / lots_fetched / unique_fetched)
|
||||
- new_count ← 'new_count' (если уже есть) иначе 'lots_inserted'
|
||||
- total_seen ← _RESULT_COUNTER_KEYS (total_seen / lots_fetched / unique_fetched /
|
||||
rows_inserted / processed)
|
||||
- new_count ← 'new_count' / 'lots_inserted' / 'saved_inserted' / 'rows_inserted'
|
||||
(первый присутствующий). 'saved_inserted' — full-load'ы (cian/avito/yandex,
|
||||
CianFullLoadCounters и аналоги в pipeline.py): на проде витрина показывала
|
||||
new_count=0 у трёх подряд cian_full_load при реально сохранённых
|
||||
saved_inserted=482/214/239 (honest-run-status) — ключ 'new_count'/'lots_inserted'
|
||||
у full-load'ов в counters не пишется вовсе. 'rows_inserted' — тот же ключ,
|
||||
которым yandex_newbuilding_sweep сообщает число upsert'ов.
|
||||
|
||||
Возвращает (total_seen, new_count); None для ключа, которого нет в counters —
|
||||
тогда соответствующая колонка не перезаписывается (COALESCE-семантика в UPDATE).
|
||||
"""
|
||||
return _run_result_count(counters), _pick_int(counters, "new_count", "lots_inserted")
|
||||
return _run_result_count(counters), _pick_int(
|
||||
counters, "new_count", "lots_inserted", "saved_inserted", "rows_inserted"
|
||||
)
|
||||
|
||||
|
||||
def _alert_if_consecutive_failures(db: Session, source: str) -> None:
|
||||
|
|
@ -632,6 +721,11 @@ def mark_done(db: Session, run_id: int, counters: dict[str, int]) -> None:
|
|||
#2700: там же — отказ называть успехом прогон, у которого отказала КАЖДАЯ попытка
|
||||
целой фазы (см. _phase_totally_failed). Отличие от #2625: тот случай про «не сделано
|
||||
ничего», этот — про «одно направление работы мертво, а суммарный сбор это прячет».
|
||||
|
||||
honest-run-status: там же — отказ называть успехом прогон с высокой долей отказов,
|
||||
даже если собрано > 0 (см. _failed_ratio_too_high). Отличие от #2625/#2700: те два
|
||||
смотрят на «всё или ничего» (все якоря / вся фаза), этот — на ДОЛЮ отказов у
|
||||
detail-backfill'ов, где ни один из первых двух признаков не матчит форму counters.
|
||||
"""
|
||||
did_nothing = _sweep_run_did_nothing(counters)
|
||||
if did_nothing is not None:
|
||||
|
|
@ -643,6 +737,11 @@ def mark_done(db: Session, run_id: int, counters: dict[str, int]) -> None:
|
|||
logger.error("%s run_id=%d", phase_dead, run_id)
|
||||
mark_failed(db, run_id, phase_dead, counters)
|
||||
return
|
||||
ratio_bad = _failed_ratio_too_high(counters)
|
||||
if ratio_bad is not None:
|
||||
logger.error("%s run_id=%d", ratio_bad, run_id)
|
||||
mark_failed(db, run_id, ratio_bad, counters)
|
||||
return
|
||||
total_seen, new_count = _column_counts(counters)
|
||||
row = db.execute(
|
||||
text(
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue