From ca7f057e419b838b0b18a317ec18aabbb562ab8d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=EB=AF=BC=ED=97=8C?= Date: Mon, 13 Jul 2026 17:29:08 +0900 Subject: [PATCH] =?UTF-8?q?feat(lps):=20=EC=95=8C=EB=A6=BC=20=EB=A3=B0=204?= =?UTF-8?q?=EC=A2=85=20=EC=B6=94=EA=B0=80=20=E2=80=94=20=EB=8D=B0=EB=93=9C?= =?UTF-8?q?=EB=9D=BC=EC=9D=B8=C2=B7=EA=B2=80=EC=83=89=EC=9B=90=EA=B0=80?= =?UTF-8?q?=C2=B7=ED=8F=AC=ED=8A=B8=20=EA=B3=A0=EA=B0=88=C2=B7=EC=98=88?= =?UTF-8?q?=EC=82=B0=20=EB=88=84=EC=88=98?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 협의로 선정한 조기 신호 4종을 AlertManager 에 추가한다. - deadline: 최근 1h JobDeadlineExceeded 수 ≥ LPS_ALERT_DEADLINE_1H(5). 재시도로 살아나면 dead 룰엔 안 잡히는 크롤 행 반복 신호를 별도 집계. - cost: 최근 1h 완료 잡 검색원가 합 ≥ LPS_ALERT_COST_1H_USD(1.0). 비용의 87%가 프록시 대역폭 — 리소스차단 풀림·재시도 루프의 조용한 비용 폭주를 감시. job.result 의 metrics.cost.total_usd JSONB 합산. - proxy_ports_low: 가용 포트 비율 ≤ LPS_ALERT_PORTS_LOW_PCT(30%). 쿨다운 격리 누적 — blocks_1h(80건)보다 먼저 우는 대규모 차단 조기 신호. 워커별 프록시 중 가장 소진된 것 기준(min). - budget_leak: 최근 6h end_reason=block 세션 ≥ LPS_ALERT_BLOCK_SESSIONS_6H(1). 요청 예산(3회)을 지켰는데도 차단됨 = 예산 하향 검토 신호. - deadline_1h·cost_1h_usd 는 queue.ops() 에 편입 → /v1/lps/ops 로도 노출. 포트·세션 지표는 워커 웹훅 스냅샷에 포함(프록시 상태는 워커에만 있음). - 테스트 4건 추가(ops 집계 2·포트 스냅샷 2), 전체 145 passed. Co-Authored-By: Claude Fable 5 --- lps/crud/job_crud.py | 15 ++++++++++++--- lps/docs/operations.md | 7 ++++++- lps/tests/test_alerts.py | 28 +++++++++++++++++++++++++++ lps/tests/test_job_queue.py | 23 ++++++++++++++++++++++ lps/tests/test_lps_api.py | 4 +++- lps/worker_main.py | 38 ++++++++++++++++++++++++++++++++++++- 6 files changed, 109 insertions(+), 6 deletions(-) diff --git a/lps/crud/job_crud.py b/lps/crud/job_crud.py index c221d24..ee48f4d 100644 --- a/lps/crud/job_crud.py +++ b/lps/crud/job_crud.py @@ -217,13 +217,22 @@ class JobQueue: (lease_until IS NOT NULL AND lease_until < now()) OR run_started_at < now() - interval '10 minutes' )) AS stuck_running, - COALESCE(EXTRACT(EPOCH FROM (now() - min(created_at) FILTER (WHERE status = 1)))::int, 0) AS oldest_pending_sec + COALESCE(EXTRACT(EPOCH FROM (now() - min(created_at) FILTER (WHERE status = 1)))::int, 0) AS oldest_pending_sec, + -- 데드라인 강제종료(크롤 행 신호). 재시도로 살아나면 dead 엔 안 잡혀 별도 집계. + count(*) FILTER (WHERE last_error LIKE 'JobDeadlineExceeded%' + AND updated_at > now() - interval '1 hour') AS deadline_1h, + -- 최근 1h 완료 잡의 검색원가 합($) — 비용 폭주(리소스차단 풀림·재시도 루프) 감시. + COALESCE(sum((result #>> '{metrics,cost,total_usd}')::float) + FILTER (WHERE status = 3 AND updated_at > now() - interval '1 hour'), 0) AS cost_1h_usd FROM job """) s = await DB_SESSION_MNG.start_session(self.DB, DBWRType.DB_READ.value) try: - row = (await s.execute(sql)).mappings().first() - return {k: int(v) for k, v in dict(row).items()} + snap = dict((await s.execute(sql)).mappings().first()) + cost = float(snap.pop("cost_1h_usd") or 0) + snap = {k: int(v) for k, v in snap.items()} + snap["cost_1h_usd"] = round(cost, 4) + return snap finally: await DB_SESSION_MNG.end_session(self.DB, DBWRType.DB_READ.value) diff --git a/lps/docs/operations.md b/lps/docs/operations.md index 2419bc9..2285b0e 100644 --- a/lps/docs/operations.md +++ b/lps/docs/operations.md @@ -152,12 +152,17 @@ SELECT key, until, reason FROM search_negative ORDER BY created_at DESC; | `stuck` | lease 만료 RUNNING 잔존 | (0 초과 시) | | `db_pool` | DB 커넥션 풀 포화율(%) — 워커·API 각자 자기 풀 감시 | `LPS_ALERT_POOL_PCT`(90) | | `source_fail:` | 소스별 최근 30분 시도 N회 이상 & 성공 0건(쿼터 소진·셀렉터 드리프트·전면 차단 신호) | `LPS_ALERT_SOURCE_FAIL_30M`(5) | +| `deadline` | 최근 1h 잡 데드라인 강제종료 수(크롤 행 반복 신호 — 재시도로 살아나면 dead 엔 안 잡힘) | `LPS_ALERT_DEADLINE_1H`(5) | +| `cost` | 최근 1h 완료 잡 검색원가 합($) — 비용 폭주(리소스차단 풀림·재시도 루프) 감시 | `LPS_ALERT_COST_1H_USD`(1.0) | +| `proxy_ports_low` | 가용 프록시 포트 비율(%) — 쿨다운 격리 누적, blocks 보다 먼저 우는 대규모 차단 조기 신호 | `LPS_ALERT_PORTS_LOW_PCT`(30) | +| `budget_leak` | 최근 6h '예산 회전에도 차단된' IP 세션 수 — 현재 요청 예산이 안전하지 않다는 신호(예산 하향 검토) | `LPS_ALERT_BLOCK_SESSIONS_6H`(1) | ``` LPS_ALERT_WEBHOOK=https://hooks.slack.com/... # 있으면 웹훅 알림 전송(워커·API 공통) LPS_ALERT_COOLDOWN_MIN=30 # 같은 룰 재발송 억제 시간 ``` -풀 사용률은 `GET /v1/lps/ops` 의 `pool_pct`(API 프로세스 기준)로도 노출된다 — 외부 모니터 스크랩용. +지표는 알림 없이도 `GET /v1/lps/ops` 로 노출된다(`pool_pct`·`deadline_1h`·`cost_1h_usd` 포함) — 외부 모니터 스크랩용. +(`proxy_ports_avail`·`block_sessions_6h` 는 워커 웹훅 스냅샷에만 포함 — 프록시 상태는 워커 프로세스에만 있음) ## 5. 테스트 diff --git a/lps/tests/test_alerts.py b/lps/tests/test_alerts.py index 0197875..7ab0b9d 100644 --- a/lps/tests/test_alerts.py +++ b/lps/tests/test_alerts.py @@ -97,3 +97,31 @@ def test_pool_status_shape(): assert set(st) == {"checked_out", "capacity", "pct"} assert st["capacity"] > 0 # R/W 2엔진 × (pool+overflow) assert 0 <= st["pct"] <= 100 + + +# ---- 가용 프록시 포트 현황(포트 고갈 룰의 데이터) -------------------------- + +def test_proxy_ports_snapshot_min_across_workers(): + from config.config_models import DecodoConfig + from services.search.proxy import DecodoProxy + from worker_main import _proxy_ports_snapshot + + def _proxy(): + return DecodoProxy(DecodoConfig(host="h", username="u", password="p", + port_start=10001, port_end=10010, session_minutes=10)) + + class _Ad: + def __init__(self, proxy): + self._proxy = proxy + + p1, p2 = _proxy(), _proxy() + p2.mark_burned(10001) + p2.mark_burned(10002) + avail, total = _proxy_ports_snapshot([_Ad(p1), _Ad(p2), object()]) # 프록시 없는 어댑터 혼재 OK + assert (avail, total) == (8, 10) # 가장 소진된 워커(p2) 기준 + + +def test_proxy_ports_snapshot_none_without_proxy(): + from worker_main import _proxy_ports_snapshot + assert _proxy_ports_snapshot([object()]) is None + assert _proxy_ports_snapshot(None) is None diff --git a/lps/tests/test_job_queue.py b/lps/tests/test_job_queue.py index 488bc22..b211dc6 100644 --- a/lps/tests/test_job_queue.py +++ b/lps/tests/test_job_queue.py @@ -91,3 +91,26 @@ def test_backoff_is_exponential_capped(): assert compute_backoff(2, base=5) == 10 assert compute_backoff(3, base=5) == 20 assert compute_backoff(100, base=5, cap=600) == 600 + + +async def test_ops_counts_deadline_and_cost(q): + # 완료 잡 2건의 검색원가 합산 — claim 반환 잡을 완료(순서 의존 제거) + for code, cost in (("b", 0.01), ("c", 0.02)): + await q.enqueue(JobType.SEARCH.value, {"q": code}) + job = await q.claim("w1") + await q.complete(job["job_id"], "w1", {"metrics": {"cost": {"total_usd": cost}}}) + # 데드라인 강제종료 — 재큐(PENDING)로 살아나도 deadline_1h 에 잡혀야 한다(dead 와 별개 축) + j1 = await q.enqueue(JobType.SEARCH.value, {"q": "a"}) + await q.claim("w1") + await q.fail(j1, "w1", "JobDeadlineExceeded: 300s", backoff_sec=0) + snap = await q.ops() + assert snap["deadline_1h"] == 1 + assert abs(snap["cost_1h_usd"] - 0.03) < 1e-9 + + +async def test_ops_cost_ignores_jobs_without_metrics(q): + jid = await q.enqueue(JobType.SEARCH.value, {"q": "x"}) + await q.claim("w1") + await q.complete(jid, "w1", {"count": 3}) # metrics 없는 결과 — 합산에서 무시(0) + snap = await q.ops() + assert snap["cost_1h_usd"] == 0 diff --git a/lps/tests/test_lps_api.py b/lps/tests/test_lps_api.py index 3e659a3..faf8206 100644 --- a/lps/tests/test_lps_api.py +++ b/lps/tests/test_lps_api.py @@ -89,6 +89,8 @@ async def test_ops_snapshot(client, clean_jobs): r = await client.get("/v1/lps/ops") assert r.status_code == 200 j = r.json() - for k in ("pending", "running", "done", "dead", "dead_1h", "stuck_running", "oldest_pending_sec", "blocks_1h"): + for k in ("pending", "running", "done", "dead", "dead_1h", "stuck_running", "oldest_pending_sec", + "blocks_1h", "deadline_1h", "pool_pct"): assert k in j and isinstance(j[k], int) + assert isinstance(j["cost_1h_usd"], (int, float)) assert j["pending"] == 1 diff --git a/lps/worker_main.py b/lps/worker_main.py index d748ada..4a59bb2 100644 --- a/lps/worker_main.py +++ b/lps/worker_main.py @@ -114,7 +114,22 @@ async def _warmup_worker(worker_adapters, tries: int = 3, attempt_timeout: float LOG.w(f"[warmup:{ad.source}] {tries}회 실패(첫 잡에서 재시도): {type(ex).__name__}") -async def run_ops_monitor(queue, bot_log, stop, interval: float = 30.0, adapters=None, alerts=None): +def _proxy_ports_snapshot(adapters) -> tuple[int, int] | None: + """워커 프록시들의 가용 포트 현황 — (최소 가용 수, 전체 포트 수). 프록시 미사용이면 None. + 쿨다운 맵은 프록시 인스턴스(워커)별이라 가장 소진된 워커 기준(min)으로 본다.""" + proxies = {} + for ad in (adapters or []): + p = getattr(ad, "_proxy", None) + if p is not None and p.enabled: + proxies[id(p)] = p + if not proxies: + return None + any_p = next(iter(proxies.values())) + total = any_p.port_end - any_p.port_start + 1 + return min(p.available_ports() for p in proxies.values()), total + + +async def run_ops_monitor(queue, bot_log, stop, interval: float = 30.0, adapters=None, alerts=None, ip_log=None): """워커 헬스 하트비트 + 임계 알림. 주기적으로 (1) 하트비트 파일 갱신(Docker HEALTHCHECK 가 행/좀비 워커 감지) (2) 큐/차단/DB풀/소스별 실패 지표 점검 → AlertManager 로 발화 (룰별 쿨다운으로 스팸 방지, 조건 해소 시 회복 알림).""" @@ -124,7 +139,12 @@ async def run_ops_monitor(queue, bot_log, stop, interval: float = 30.0, adapters th_lag = int(os.environ.get("LPS_ALERT_QUEUE_LAG_SEC", "300")) th_pool = int(os.environ.get("LPS_ALERT_POOL_PCT", "90")) th_srcfail = int(os.environ.get("LPS_ALERT_SOURCE_FAIL_30M", "5")) + th_deadline = int(os.environ.get("LPS_ALERT_DEADLINE_1H", "5")) + th_cost = float(os.environ.get("LPS_ALERT_COST_1H_USD", "1.0")) + th_ports = int(os.environ.get("LPS_ALERT_PORTS_LOW_PCT", "30")) + th_leak = int(os.environ.get("LPS_ALERT_BLOCK_SESSIONS_6H", "1")) alerts = alerts or AlertManager(origin="worker") + ip_log = ip_log or IpSessionLog() while not stop.is_set(): try: with open(hb_path, "w") as f: @@ -142,6 +162,22 @@ async def run_ops_monitor(queue, bot_log, stop, interval: float = 30.0, adapters await alerts.check("stuck", snap["stuck_running"] > 0, f"stuck={snap['stuck_running']}", snap) await alerts.check("db_pool", pool["pct"] >= th_pool, f"DB 풀 포화 {pool['pct']}% (checked_out {pool['checked_out']}/{pool['capacity']})", snap) + await alerts.check("deadline", snap["deadline_1h"] >= th_deadline, + f"잡 데드라인 강제종료 1h={snap['deadline_1h']} — 크롤 행 반복 신호", snap) + await alerts.check("cost", snap["cost_1h_usd"] >= th_cost, + f"검색원가 1h=${snap['cost_1h_usd']} — 비용 폭주(리소스차단 풀림·재시도 루프) 점검", snap) + # 가용 프록시 포트 고갈 — 쿨다운 격리 누적. blocks_1h 보다 먼저 우는 대규모 차단 조기 신호. + ports = _proxy_ports_snapshot(adapters) + if ports: + avail, total = ports + snap["proxy_ports_avail"], snap["proxy_ports_total"] = avail, total + await alerts.check("proxy_ports_low", avail * 100 <= total * th_ports, + f"가용 프록시 포트 {avail}/{total} — 대규모 차단 진행 신호", snap) + # 예산 누수 — 요청 예산을 지켰는데도 차단된 IP 세션 발생 = 현재 예산이 안전하지 않다는 신호. + block_sessions = (await ip_log.recent_stats(360)).get("block", 0) + snap["block_sessions_6h"] = block_sessions + await alerts.check("budget_leak", block_sessions >= th_leak, + f"예산 회전에도 차단된 IP 세션 6h={block_sessions} — LPS_IP_REQUEST_BUDGET 하향 검토", snap) # 소스별 장기 실패 — 최근 30분간 시도는 있는데 성공이 0건(쿼터 소진·셀렉터 드리프트·전면 차단 신호) per_source: dict[str, list[int]] = {} for ad in (adapters or []):