feat(lps): P4 관측·알림·워커 헬스 — readyz/ops + 하트비트/HEALTHCHECK + 임계 알림
프로덕션 운영 가시성. 행/좀비 워커 감지 + 큐/차단 지표 노출 + 임계 알림. - 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 <noreply@anthropic.com>
This commit is contained in:
민헌
parent
3a8d669970
commit
d6dd47e652
@ -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 은 로그를 삼킴).
|
||||
|
||||
@ -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)
|
||||
|
||||
@ -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)
|
||||
|
||||
@ -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.<domain>.<file> 를 import 후 include.
|
||||
app.include_router(router.v1.lps.search.router)
|
||||
|
||||
@ -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,
|
||||
|
||||
@ -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)
|
||||
|
||||
@ -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
|
||||
|
||||
@ -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)
|
||||
|
||||
Loading…
Reference in New Issue
Block a user