perf(cadastre): freshness skip + grid-walk heartbeat + reduce WMS log noise
Three operational improvements after ekb_full v3 completed at 99.1%: Fix H — per-quarter freshness skip (scrape_cadastre.py): - enqueue_cadastre_harvest queries cad_quarter_stats.fetched_at per quarter - Skip quarters where fetched_at >= NOW() - skip_fresh_hours (default 1h) - extra_config.skip_fresh_hours=0 disables skip (full re-fetch) - Saves ~30k NSPD calls on repeated runs of same scope Fix I — heartbeat in grid-walk loop (bulk_harvest.py:_grid_walk_category): - 225 cells x 70-300ms = 15-75s heartbeat silence - Pass update_progress callback to _grid_walk_category - Call every heartbeat_every (50) cells with phase progress info - Prevents cleanup_zombies false-positive cancellation Fix J — NSPD WMS HTTP 500 noise to debug (bulk_harvest.py:_grid_walk_category): - NSPD WMS endpoint frequently returns HTTP 500 (server-side issue) - Was: logger.warning each, 200+ warnings per quarter - Now: logger.debug (visible via --log-level=DEBUG) 40 tests passing.
This commit is contained in:
lekss361
parent
d104222643
commit
ae04dcb4b9
2 changed files with 71 additions and 5 deletions
|
|
@ -218,6 +218,7 @@ async def harvest_quarter(
|
|||
client=client,
|
||||
quarter=quarter,
|
||||
layer_id=cat_id,
|
||||
update_progress=update_progress,
|
||||
)
|
||||
result.grid_walk_requests += n_requests
|
||||
|
||||
|
|
@ -276,12 +277,19 @@ async def _grid_walk_category(
|
|||
layer_id: int,
|
||||
grid_size: int = 15,
|
||||
tile_size: int = 512,
|
||||
update_progress: Callable[[dict[str, Any]], None] | None = None,
|
||||
heartbeat_every: int = 50,
|
||||
) -> tuple[int, int]:
|
||||
"""Grid-walk layer_id в bbox квартала.
|
||||
|
||||
Генерирует grid_size×grid_size ячеек, для каждой делает wms_feature_info.
|
||||
Дедуплицирует по cad_num, upsert'ит новые объекты.
|
||||
|
||||
Args:
|
||||
update_progress: Fix I — callback для heartbeat внутри grid-walk loop.
|
||||
225 cells × ~70ms = ~15-75s без heartbeat, zombie-cleanup ложно
|
||||
помечал worker как мёртвого. Вызываем каждые heartbeat_every cells.
|
||||
|
||||
Returns:
|
||||
(upserted_count, requests_count)
|
||||
"""
|
||||
|
|
@ -300,7 +308,19 @@ async def _grid_walk_category(
|
|||
upserted = 0
|
||||
requests = 0
|
||||
|
||||
for cell_bbox, click_xy in grid_points:
|
||||
for idx, (cell_bbox, click_xy) in enumerate(grid_points):
|
||||
# Fix I: heartbeat каждые N cells — обновляет cadastre_jobs.heartbeat_at
|
||||
# чтобы cleanup_zombies (5min stale check) не cancel'нул живой worker.
|
||||
if update_progress is not None and idx > 0 and idx % heartbeat_every == 0:
|
||||
update_progress(
|
||||
{
|
||||
"phase": f"grid_walk_{layer_id}_cell_{idx}",
|
||||
"quarter": quarter,
|
||||
"grid_cells_done": idx,
|
||||
"grid_cells_total": len(grid_points),
|
||||
}
|
||||
)
|
||||
|
||||
try:
|
||||
features = await client.wms_feature_info(
|
||||
layer_id=layer_id,
|
||||
|
|
@ -311,7 +331,10 @@ async def _grid_walk_category(
|
|||
)
|
||||
requests += 1
|
||||
except Exception as e:
|
||||
logger.warning(
|
||||
# Fix J: NSPD WMS возвращает HTTP 500 на десятки cells per quarter —
|
||||
# это server-side noise, не наш bug. logger.debug чтобы не засорять
|
||||
# prod логи (раньше было warning → 200+ warnings per quarter).
|
||||
logger.debug(
|
||||
"_grid_walk_category: wms_feature_info error layer=%d quarter=%s: %s",
|
||||
layer_id,
|
||||
quarter,
|
||||
|
|
|
|||
|
|
@ -357,22 +357,65 @@ def enqueue_cadastre_harvest(self: Any, job_id: int) -> dict[str, Any]:
|
|||
logger.warning("enqueue_cadastre_harvest: job=%d имеет пустой список quarters", job_id)
|
||||
return {"job_id": job_id, "enqueued": 0}
|
||||
|
||||
# Fix H: per-quarter freshness skip — не переобрабатывать quarters с свежим
|
||||
# cad_quarter_stats.fetched_at (последний успешный обход <FRESHNESS_HOURS).
|
||||
# extra_config.skip_fresh_hours=N управляет порогом; 0 → отключает skip.
|
||||
# Это сэкономит ~30k NSPD calls при повторных запусках того же scope.
|
||||
extra_config = job["extra_config"] or {}
|
||||
skip_fresh_hours = extra_config.get("skip_fresh_hours", 1)
|
||||
skipped_fresh: set[str] = set()
|
||||
if skip_fresh_hours > 0:
|
||||
threshold = _dt.datetime.now(_dt.UTC) - _dt.timedelta(hours=skip_fresh_hours)
|
||||
rows = (
|
||||
db.execute(
|
||||
text(
|
||||
"SELECT cad_number FROM cad_quarter_stats "
|
||||
"WHERE cad_number = ANY(CAST(:qs AS text[])) "
|
||||
"AND fetched_at >= :t"
|
||||
),
|
||||
{"qs": quarters, "t": threshold},
|
||||
)
|
||||
.scalars()
|
||||
.all()
|
||||
)
|
||||
skipped_fresh = set(rows)
|
||||
if skipped_fresh:
|
||||
# Корректируем targets_total — skipped quarters не входят в прогресс
|
||||
db.execute(
|
||||
text(
|
||||
"UPDATE cadastre_jobs SET targets_total = :new_total, "
|
||||
"targets_skipped = COALESCE(targets_skipped, 0) + :sk "
|
||||
"WHERE job_id = :id"
|
||||
),
|
||||
{
|
||||
"new_total": len(quarters) - len(skipped_fresh),
|
||||
"sk": len(skipped_fresh),
|
||||
"id": job_id,
|
||||
},
|
||||
)
|
||||
|
||||
# Обновить статус на running
|
||||
_update_job(db, job_id, status="running", started_at=_dt.datetime.now(_dt.UTC))
|
||||
db.commit()
|
||||
|
||||
enqueued = 0
|
||||
for quarter in quarters:
|
||||
if quarter in skipped_fresh:
|
||||
continue
|
||||
bulk_harvest_quarter_task.apply_async(
|
||||
args=[quarter, job_id],
|
||||
queue=CADASTRE_QUEUE,
|
||||
)
|
||||
enqueued += 1
|
||||
|
||||
logger.info(
|
||||
"enqueue_cadastre_harvest: job=%d enqueued %d quarters",
|
||||
"enqueue_cadastre_harvest: job=%d enqueued %d (skipped_fresh=%d, threshold=%dh)",
|
||||
job_id,
|
||||
len(quarters),
|
||||
enqueued,
|
||||
len(skipped_fresh),
|
||||
skip_fresh_hours,
|
||||
)
|
||||
return {"job_id": job_id, "enqueued": len(quarters)}
|
||||
return {"job_id": job_id, "enqueued": enqueued, "skipped_fresh": len(skipped_fresh)}
|
||||
|
||||
finally:
|
||||
db.close()
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue