From d6dd47e65245a025857cd8753f28ec40437120bb Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=EB=AF=BC=ED=97=8C?= Date: Thu, 9 Jul 2026 23:21:43 +0900 Subject: [PATCH] =?UTF-8?q?feat(lps):=20P4=20=EA=B4=80=EC=B8=A1=C2=B7?= =?UTF-8?q?=EC=95=8C=EB=A6=BC=C2=B7=EC=9B=8C=EC=BB=A4=20=ED=97=AC=EC=8A=A4?= =?UTF-8?q?=20=E2=80=94=20readyz/ops=20+=20=ED=95=98=ED=8A=B8=EB=B9=84?= =?UTF-8?q?=ED=8A=B8/HEALTHCHECK=20+=20=EC=9E=84=EA=B3=84=20=EC=95=8C?= =?UTF-8?q?=EB=A6=BC?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 프로덕션 운영 가시성. 행/좀비 워커 감지 + 큐/차단 지표 노출 + 임계 알림. - API: /readyz(DB 도달성=readiness, 실패 503; /healthz=liveness와 구분). /v1/lps/ops(플랫 JSON): 큐 카운트 + oldest_pending_sec(큐지연) + dead_1h + stuck_running + blocks_1h. - crud: JobQueue.ops()/ping(), BotDetectionLog.recent_count(). - worker: run_ops_monitor — 하트비트 파일 주기 갱신(Docker HEALTHCHECK 가 신선도로 행 워커 감지) + 임계(DEAD/차단/큐지연/stuck) 초과 시 WARN 로그 + (LPS_ALERT_WEBHOOK 있으면) Slack 호환 웹훅. - Dockerfile.worker: HEALTHCHECK(하트비트 <120s). 임계·웹훅은 env(LPS_ALERT_*). - 테스트: readyz/ops 2종. 검증: 컨테이너 healthy 판정, 하트비트 갱신, ops 스냅샷 정상. 91 tests. Co-Authored-By: Claude Fable 5 --- lps/Dockerfile.worker | 7 +++++- lps/crud/bot_detection.py | 9 +++++++ lps/crud/job_crud.py | 30 ++++++++++++++++++++++ lps/router/router.py | 20 +++++++++++++-- lps/router/v1/lps/search.py | 10 ++++++++ lps/services/lps_service.py | 11 +++++++- lps/tests/test_lps_api.py | 15 +++++++++++ lps/worker_main.py | 50 ++++++++++++++++++++++++++++++++++++- 8 files changed, 147 insertions(+), 5 deletions(-) diff --git a/lps/Dockerfile.worker b/lps/Dockerfile.worker index 8bc6edc..7082d4a 100644 --- a/lps/Dockerfile.worker +++ b/lps/Dockerfile.worker @@ -22,7 +22,12 @@ COPY . . ENV APP_ENV=local \ PYTHONUNBUFFERED=1 \ LPS_CHROME_EXECUTABLE=/usr/bin/chromium \ - DISPLAY=:99 + DISPLAY=:99 \ + LPS_HEARTBEAT_FILE=/tmp/lps_worker_heartbeat + +# 하트비트 신선도(<120s)로 행/좀비 워커 감지. start-period 는 웜업(브라우저 기동) 여유. +HEALTHCHECK --interval=30s --timeout=8s --start-period=120s --retries=3 \ + CMD python -c "import os,time,sys; p=os.environ['LPS_HEARTBEAT_FILE']; sys.exit(0 if os.path.exists(p) and time.time()-os.path.getmtime(p)<120 else 1)" # Xvfb(가상 디스플레이)를 백그라운드로 띄우고 python 을 exec 로 승계 실행. # → 헤드풀 Chromium 이 :99 에 뜨고, 워커 로그는 그대로 docker logs 로 나온다(xvfb-run 은 로그를 삼킴). diff --git a/lps/crud/bot_detection.py b/lps/crud/bot_detection.py index 8d0618f..e1692f1 100644 --- a/lps/crud/bot_detection.py +++ b/lps/crud/bot_detection.py @@ -26,3 +26,12 @@ class BotDetectionLog: raise finally: await DB_SESSION_MNG.end_session(self.DB, DBWRType.DB_WRITE.value) + + async def recent_count(self, minutes: int = 60) -> int: + """최근 N분간 봇 감지(차단) 건수 — 차단율 급증 알림·모니터링용.""" + sql = text("SELECT count(*) FROM bot_detection WHERE created_at > now() - make_interval(mins => :m)") + s = await DB_SESSION_MNG.start_session(self.DB, DBWRType.DB_READ.value) + try: + return int((await s.execute(sql, {"m": minutes})).scalar() or 0) + finally: + await DB_SESSION_MNG.end_session(self.DB, DBWRType.DB_READ.value) diff --git a/lps/crud/job_crud.py b/lps/crud/job_crud.py index 44633ad..0c7c31c 100644 --- a/lps/crud/job_crud.py +++ b/lps/crud/job_crud.py @@ -200,3 +200,33 @@ class JobQueue: return {js.name: by_val.get(js.value, 0) for js in JobStatus} return await self._tx(run) + + async def ops(self) -> dict: + """운영 스냅샷(모니터링·알림용): 상태별 카운트 + 큐 지연(가장 오래된 PENDING 나이) + + 최근 1시간 DEAD + lease 만료 stuck(reaper 가 회수 못한 좀비 신호).""" + sql = text(""" + SELECT + count(*) FILTER (WHERE status = 1) AS pending, + count(*) FILTER (WHERE status = 2) AS running, + count(*) FILTER (WHERE status = 3) AS done, + count(*) FILTER (WHERE status = 4) AS dead, + count(*) FILTER (WHERE status = 4 AND updated_at > now() - interval '1 hour') AS dead_1h, + count(*) FILTER (WHERE status = 2 AND lease_until IS NOT NULL AND lease_until < now()) AS stuck_running, + COALESCE(EXTRACT(EPOCH FROM (now() - min(created_at) FILTER (WHERE status = 1)))::int, 0) AS oldest_pending_sec + 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()} + finally: + await DB_SESSION_MNG.end_session(self.DB, DBWRType.DB_READ.value) + + async def ping(self) -> bool: + """DB 도달성 확인(readiness). 실패 시 예외.""" + s = await DB_SESSION_MNG.start_session(self.DB, DBWRType.DB_READ.value) + try: + await s.execute(text("SELECT 1")) + return True + finally: + await DB_SESSION_MNG.end_session(self.DB, DBWRType.DB_READ.value) diff --git a/lps/router/router.py b/lps/router/router.py index 0db0df1..93518e6 100644 --- a/lps/router/router.py +++ b/lps/router/router.py @@ -50,13 +50,29 @@ async def log_time(request: Request, call_next): @app.get( path="/healthz", - summary="헬스체크", - description="서버 기동 시각(API_SERVER_START_TIME)을 반환하는 헬스체크 엔드포인트.", + summary="헬스체크(liveness)", + description="서버 기동 시각을 반환. 프로세스가 살아있는지만 확인(DB 무관).", responses={404: {"description": "Not found"}}, ) async def healthz(): return API_SERVER_START_TIME +@app.get( + path="/readyz", + summary="레디니스(readiness)", + description="DB 도달성까지 확인. 오케스트레이터/LB 가 트래픽 라우팅 여부 판단에 사용. 실패 시 503.", +) +async def readyz(): + from fastapi import Response + from crud.job_crud import JobQueue + try: + await JobQueue().ping() + return {"ready": True} + except Exception as ex: + return Response(content=f'{{"ready": false, "error": "{type(ex).__name__}"}}', + media_type="application/json", status_code=503) + + # 각 도메인 라우터를 등록한다. 새 기능 추가 시 router.v1.. 를 import 후 include. app.include_router(router.v1.lps.search.router) diff --git a/lps/router/v1/lps/search.py b/lps/router/v1/lps/search.py index 52a6e78..d7425d9 100644 --- a/lps/router/v1/lps/search.py +++ b/lps/router/v1/lps/search.py @@ -41,6 +41,16 @@ async def queue_stats(service: LpsService = Depends()): return RemoveNoneResponse(await service.stats()) +@router.get( + path="/ops", + summary="운영 스냅샷(모니터링)", + description="큐 카운트 + 큐 지연(oldest_pending_sec) + 최근1h DEAD(dead_1h) + lease만료 stuck + " + "최근1h 차단(blocks_1h). 외부 모니터가 스크랩·임계 알림하기 좋은 플랫 JSON.", +) +async def ops(service: LpsService = Depends()): + return await service.ops() + + @router.get( path="/products/{product_code}/history", response_model=Res_PriceHistory, diff --git a/lps/services/lps_service.py b/lps/services/lps_service.py index 30f29f6..71876fb 100644 --- a/lps/services/lps_service.py +++ b/lps/services/lps_service.py @@ -11,6 +11,7 @@ from fastapi import Depends from common.enums import ErrorType, JobStatus, JobType from crud.job_crud import JobQueue from crud.price_history import PriceHistory +from crud.bot_detection import BotDetectionLog from router.v1.lps.protocol import ( EnqueuedItem, PricePoint, @@ -26,9 +27,11 @@ _PRIORITY = {"new_product": 1, "manual": 2, "partner": 3, "batch": 4} class LpsService: - def __init__(self, queue: JobQueue = Depends(JobQueue), history: PriceHistory = Depends(PriceHistory)): + def __init__(self, queue: JobQueue = Depends(JobQueue), history: PriceHistory = Depends(PriceHistory), + bot_log: BotDetectionLog = Depends(BotDetectionLog)): self.queue = queue self.history = history + self.bot_log = bot_log async def submit_search(self, items: list[SearchItem]) -> Res_Search: res = Res_Search() @@ -70,6 +73,12 @@ class LpsService: res.counts = await self.queue.counts() return res + async def ops(self) -> dict: + """운영 스냅샷(모니터링·알림용, 플랫 JSON): 큐 카운트·지연·최근 DEAD·최근 차단 수.""" + snap = await self.queue.ops() + snap["blocks_1h"] = await self.bot_log.recent_count(60) + return snap + async def price_history(self, product_code: str, limit: int = 100) -> Res_PriceHistory: res = Res_PriceHistory(product_code=product_code) rows = await self.history.list_by_product(product_code, limit) diff --git a/lps/tests/test_lps_api.py b/lps/tests/test_lps_api.py index 0b08579..3e659a3 100644 --- a/lps/tests/test_lps_api.py +++ b/lps/tests/test_lps_api.py @@ -77,3 +77,18 @@ async def test_queue_stats(client, clean_jobs): r = await client.get("/v1/lps/queue/stats") counts = r.json()["counts"] assert counts["PENDING"] == 2 and counts["DONE"] == 0 and counts["DEAD"] == 0 + + +async def test_readyz(client): + r = await client.get("/readyz") # DB 도달 → ready + assert r.status_code == 200 and r.json()["ready"] is True + + +async def test_ops_snapshot(client, clean_jobs): + await client.post("/v1/lps/search", json={"data": [{"product_code": "A", "product_name": "x"}]}) + 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"): + assert k in j and isinstance(j[k], int) + assert j["pending"] == 1 diff --git a/lps/worker_main.py b/lps/worker_main.py index 5b0b990..eb09852 100644 --- a/lps/worker_main.py +++ b/lps/worker_main.py @@ -7,6 +7,9 @@ import asyncio import os +import time + +import httpx from common.logger import LOG from config.server_configs import web_server_config, openai_config, decodo_config @@ -86,6 +89,50 @@ async def _warmup_worker(worker_adapters, tries: int = 3): LOG.w(f"[warmup:{ad.source}] {tries}회 실패(첫 잡에서 재시도): {type(ex).__name__}") +async def _post_webhook(url: str, text: str, snap: dict): + """Slack 호환 웹훅으로 알림 전송(있을 때만). 실패는 무시.""" + try: + async with httpx.AsyncClient(timeout=5) as c: + await c.post(url, json={"text": f":rotating_light: LPS {text}\n```{snap}```"}) + except Exception: + pass + + +async def run_ops_monitor(queue, bot_log, stop, interval: float = 30.0): + """워커 헬스 하트비트 + 임계 알림. 주기적으로 (1) 하트비트 파일 갱신(Docker HEALTHCHECK 가 + 행/좀비 워커 감지) (2) 큐/차단 지표 점검 → 임계 초과 시 WARN 로그 + (env 있으면) 웹훅 알림.""" + hb_path = os.environ.get("LPS_HEARTBEAT_FILE", "/tmp/lps_worker_heartbeat") + webhook = os.environ.get("LPS_ALERT_WEBHOOK") + th_dead = int(os.environ.get("LPS_ALERT_DEAD_1H", "20")) + th_blocks = int(os.environ.get("LPS_ALERT_BLOCKS_1H", "80")) + th_lag = int(os.environ.get("LPS_ALERT_QUEUE_LAG_SEC", "300")) + while not stop.is_set(): + try: + with open(hb_path, "w") as f: + f.write(str(int(time.time()))) # 하트비트(mtime) — HEALTHCHECK 가 신선도 확인 + except Exception: + pass + try: + snap = await queue.ops() + snap["blocks_1h"] = await bot_log.recent_count(60) + alerts = [] + if snap["dead_1h"] >= th_dead: alerts.append(f"DEAD 1h={snap['dead_1h']}") + if snap["blocks_1h"] >= th_blocks: alerts.append(f"차단 1h={snap['blocks_1h']}") + if snap["oldest_pending_sec"] >= th_lag: alerts.append(f"큐지연={snap['oldest_pending_sec']}s") + if snap["stuck_running"] > 0: alerts.append(f"stuck={snap['stuck_running']}") + if alerts: + msg = "[ops-alert] " + " · ".join(alerts) + LOG.w(msg) + if webhook: + await _post_webhook(webhook, msg, snap) + except Exception as ex: + LOG.e_no_callstack(f"[ops-monitor] {type(ex).__name__}: {ex}") + try: + await asyncio.wait_for(stop.wait(), timeout=interval) + except asyncio.TimeoutError: + pass + + async def run_browser_reaper(adapters, stop, idle_sec: float = 120.0, interval: float = 30.0): """유휴 브라우저 정리 루프 — 일정 시간 검색 없는 어댑터의 Chrome 을 닫아 메모리를 회수한다. 쿠키는 user_data_dir 에 남아, 다음 검색 때 재기동해도 (같은 IP면) 웜 유지.""" @@ -133,7 +180,8 @@ async def main(concurrency: int = 1): tasks.append(asyncio.create_task(run_reaper(queue, stop))) tasks.append(asyncio.create_task(run_browser_reaper(all_adapters, stop))) # 유휴 브라우저 정리 - LOG.i(f"LPS 워커 {concurrency}개 + reaper + 브라우저정리(유휴 120s) 기동 (워커별 세트 · 상품 {concurrency}개 동시)") + tasks.append(asyncio.create_task(run_ops_monitor(queue, BotDetectionLog(), stop))) # 하트비트 + 임계 알림 + LOG.i(f"LPS 워커 {concurrency}개 + reaper + 브라우저정리 + ops모니터(하트비트/알림) 기동 (워커별 세트 · 상품 {concurrency}개 동시)") try: await asyncio.gather(*tasks)