add vps check
This commit is contained in:
@@ -108,6 +108,31 @@ def _read_last_db_progress_ts(redis_client: Redis) -> int | None:
|
||||
return max_ts or None
|
||||
|
||||
|
||||
def _has_active_recent_progress(redis_client: Redis, now_ts: int, stall_seconds: int) -> bool:
|
||||
"""Return True when a task is still reporting non-DB progress.
|
||||
|
||||
Hourly only_new runs can legitimately skip every listed vehicle as existing.
|
||||
Those runs may not emit fast_db_progress for a long time, but they still emit
|
||||
regular listing/segment progress. Treating old DB timestamps as fatal caused
|
||||
the watchdog to kill healthy production workers during such scans.
|
||||
"""
|
||||
for key in redis_client.scan_iter(match="iaai:state:task_progress:*"):
|
||||
try:
|
||||
payload = redis_client.get(key)
|
||||
if not payload:
|
||||
continue
|
||||
data = json.loads(payload)
|
||||
stage = str(data.get("stage") or "")
|
||||
if stage in {"failed", "sync_done", "segment_task_failed", "segment_task_soft_timeout"}:
|
||||
continue
|
||||
ts = _safe_int(data.get("ts"), 0)
|
||||
if ts > 0 and now_ts - ts <= max(60, int(stall_seconds)):
|
||||
return True
|
||||
except Exception:
|
||||
continue
|
||||
return False
|
||||
|
||||
|
||||
def _reset_bootstrap_checkpoint_for_db_idle(redis_client: Redis) -> None:
|
||||
pipe = redis_client.pipeline()
|
||||
pipe.delete(SYNC_LISTING_CHECKPOINT_KEY)
|
||||
@@ -219,6 +244,12 @@ def main() -> None:
|
||||
restart_reason = f"progress_age={age}s > {stall_seconds}s"
|
||||
db_idle_restart = False
|
||||
if age <= stall_seconds:
|
||||
# If the task is actively reporting progress, do not require DB writes.
|
||||
# Hourly only_new listing scans often skip already-known cars and can
|
||||
# legitimately have no DB writes while still moving through segments.
|
||||
if _has_active_recent_progress(redis_client, now_ts, stall_seconds):
|
||||
time.sleep(check_interval)
|
||||
continue
|
||||
last_db_ts = _read_last_db_progress_ts(redis_client)
|
||||
db_age = None if last_db_ts is None else max(0, now_ts - int(last_db_ts))
|
||||
if db_age is None:
|
||||
|
||||
@@ -578,9 +578,14 @@ def _should_restart_for_db_idle(progress: dict, db_idle_restart_seconds: int) ->
|
||||
stage = str(progress.get("stage") or "")
|
||||
if stage in TERMINAL_PROGRESS_STAGES:
|
||||
return False
|
||||
# Listing/segment progress means the task is alive even if it writes no new
|
||||
# cars. This is normal for hourly only_new runs when all vehicles are already
|
||||
# present in DB. Do not kill healthy scans just because DB progress is idle.
|
||||
progress_ts = _safe_int(progress.get("ts")) or 0
|
||||
if progress_ts > 0 and int(time.time()) - progress_ts < int(db_idle_restart_seconds):
|
||||
return False
|
||||
if stage in STALL_WATCHDOG_LONG_RUNNING_STAGES:
|
||||
timeout = _stall_timeout_for_progress(stage, db_idle_restart_seconds)
|
||||
progress_ts = _safe_int(progress.get("ts")) or 0
|
||||
return progress_ts > 0 and int(time.time()) - progress_ts >= timeout
|
||||
|
||||
segments_total = _safe_int(progress.get("segments_total"))
|
||||
@@ -591,7 +596,6 @@ def _should_restart_for_db_idle(progress: dict, db_idle_restart_seconds: int) ->
|
||||
return False
|
||||
|
||||
now_ts = int(time.time())
|
||||
progress_ts = _safe_int(progress.get("ts")) or 0
|
||||
if progress_ts <= 0:
|
||||
return False
|
||||
|
||||
@@ -1051,10 +1055,17 @@ def _build_listing_segments(settings: Settings) -> list[dict[str, str | int | No
|
||||
|
||||
runtime_config = RuntimeConfig.from_file(settings.runtime_config_file)
|
||||
brands = runtime_config.filters.include.brands
|
||||
if settings.scraping_profile.http_first and settings.listing.fast_segment_year_splits:
|
||||
|
||||
# Fast HTTP-first режим не использует UI year-фильтры IAAI.
|
||||
# Значит runtime-сегменты должны быть "один бренд = один сегмент"
|
||||
# без доп. year split, иначе получаем много лишних долгих сегментов,
|
||||
# которые fast-профиль всё равно игнорирует.
|
||||
if settings.scraping_profile.http_first:
|
||||
segments = build_fast_listing_segments_for_makes(list(brands))
|
||||
else:
|
||||
elif settings.listing.fast_segment_year_splits:
|
||||
segments = build_listing_segments_for_makes(list(brands))
|
||||
else:
|
||||
segments = build_fast_listing_segments_for_makes(list(brands))
|
||||
if not segments:
|
||||
logger.warning(
|
||||
"IAAI_LISTING_SEGMENTS=runtime, but runtime_config filters.brands is empty; segmented listing disabled"
|
||||
|
||||
@@ -1,5 +1,6 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import unittest
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
@@ -50,6 +51,27 @@ class TestSelfHeal(unittest.TestCase):
|
||||
|
||||
self.assertIsNone(ts)
|
||||
|
||||
def test_active_recent_progress_prevents_db_idle_restart(self) -> None:
|
||||
redis_client = MagicMock()
|
||||
redis_client.scan_iter.return_value = ["iaai:state:task_progress:hourly"]
|
||||
redis_client.get.return_value = json.dumps(
|
||||
{
|
||||
"task_id": "hourly",
|
||||
"stage": "fast_listing_collected",
|
||||
"ts": 1000,
|
||||
"last_db_progress_ts": 1,
|
||||
"skipped_existing": 2417,
|
||||
}
|
||||
)
|
||||
|
||||
active = self_heal._has_active_recent_progress(
|
||||
redis_client,
|
||||
now_ts=1010,
|
||||
stall_seconds=900,
|
||||
)
|
||||
|
||||
self.assertTrue(active)
|
||||
|
||||
@patch("iaai_scraper.worker.self_heal.time.sleep", return_value=None)
|
||||
@patch("iaai_scraper.worker.self_heal.os.kill")
|
||||
@patch("builtins.open")
|
||||
|
||||
@@ -19,6 +19,8 @@ class TestWorkerTaskLockHelpers(unittest.TestCase):
|
||||
|
||||
settings = Settings(runtime_config_file=runtime_config_file)
|
||||
settings.listing.listing_segments_json = "runtime"
|
||||
settings.scraping_profile.http_first = False
|
||||
settings.listing.fast_segment_year_splits = True
|
||||
|
||||
segs = tasks._build_listing_segments(settings)
|
||||
|
||||
|
||||
18
vps_check.sh
Normal file
18
vps_check.sh
Normal file
@@ -0,0 +1,18 @@
|
||||
#!/usr/bin/env bash
|
||||
set -euo pipefail
|
||||
cd /root/iaai-parser
|
||||
|
||||
echo '--- services ---'
|
||||
docker compose ps
|
||||
|
||||
echo '--- env speed ---'
|
||||
grep -nE 'IAAI_PROFILE|IAAI_SCRAPING_PROFILE|IAAI_HTTP_FIRST|IAAI_LISTING_SEGMENTS|IAAI_FAST_SEGMENT_YEAR_SPLITS|CELERY_WORKER_CONCURRENCY|IAAI_FETCH_CONCURRENCY|CELERY_BATCH_SIZE' .env
|
||||
|
||||
echo '--- db count ---'
|
||||
docker compose exec -T postgres psql -U iaai -d iaai_scraper -t -c "select count(*) as cars_count, max(last_seen_at) as last_seen from iaai_cars;"
|
||||
|
||||
echo '--- redis progress ---'
|
||||
docker compose exec -T redis redis-cli MGET iaai:state:sync_listing_checkpoint iaai:state:sync_segments_done iaai:state:sync_segments_total iaai:state:sync_full_scan_done
|
||||
|
||||
echo '--- recent worker progress ---'
|
||||
docker compose logs --tail=80 worker | grep -E 'Segment [0-9]+/[0-9]+|Fast HTTP-first listing started|Fast HTTP-first listing collected|Fast HTTP-first DB batch|ERROR|WARNING' | tail -40
|
||||
Reference in New Issue
Block a user