ca7f057e41
협의로 선정한 조기 신호 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 <noreply@anthropic.com>
247 lines
11 KiB
Python
247 lines
11 KiB
Python
"""작업 큐 CRUD — PostgreSQL 을 '제대로' 큐로 쓴다.
|
|
|
|
레퍼런스의 반면교사를 전부 뒤집는다:
|
|
- 할당은 **단일 문장 원자 claim**: FOR UPDATE SKIP LOCKED 서브쿼리 + 같은 UPDATE + RETURNING.
|
|
→ 워커/디스패처가 몇이든 같은 잡 이중 할당이 원천 불가. fetch 와 claim 을 분리하지 않는다.
|
|
- 모든 전이는 **조건부 CAS**(WHERE 에 status/worker_id 가드) + RETURNING.
|
|
- 복구는 timeout 추측이 아니라 **lease 만료 소유권**(reaper 가 회수).
|
|
- 재시도/백오프/dead-letter 를 큐에 내장(스크립트 난립 제거). 외부전송도 OUTBOX 잡으로.
|
|
"""
|
|
|
|
import json
|
|
|
|
from sqlalchemy import text
|
|
|
|
from common.database.db_session_manager import DB_SESSION_MNG
|
|
from common.enums import DBType, DBWRType, JobStatus
|
|
|
|
|
|
# 잡 적재 시 워커를 즉시 깨우는 LISTEN/NOTIFY 채널(폴링 제거).
|
|
JOB_NOTIFY_CHANNEL = "lps_job"
|
|
|
|
|
|
def compute_backoff(attempts: int, base: float = 5.0, cap: float = 600.0) -> float:
|
|
"""지수 백오프(초). attempts 회 시도 후 다음 재시도까지 대기 = base * 2^(attempts-1), cap 상한."""
|
|
return min(cap, base * (2 ** max(0, attempts - 1)))
|
|
|
|
|
|
class JobQueue:
|
|
DB = DBType.MAIN.value
|
|
|
|
async def _tx(self, fn):
|
|
"""쓰기 트랜잭션(commit/rollback 은 여기서 책임)."""
|
|
s = await DB_SESSION_MNG.start_session(self.DB, DBWRType.DB_WRITE.value)
|
|
try:
|
|
res = await fn(s)
|
|
await s.commit()
|
|
return res
|
|
except Exception:
|
|
await s.rollback()
|
|
raise
|
|
finally:
|
|
await DB_SESSION_MNG.end_session(self.DB, DBWRType.DB_WRITE.value)
|
|
|
|
# ---- 적재 -----------------------------------------------------------
|
|
async def enqueue(self, job_type: int, payload: dict, priority: int = 100, dedupe_key: str | None = None, max_attempts: int = 3):
|
|
"""잡 적재. dedupe_key 가 활성(PENDING/RUNNING) 중복이면 삽입 없이 None 반환."""
|
|
sql = text("""
|
|
INSERT INTO job (job_type, priority, payload, dedupe_key, max_attempts)
|
|
VALUES (:t, :p, CAST(:payload AS jsonb), :dk, :ma)
|
|
ON CONFLICT (dedupe_key) WHERE status IN (1, 2) AND dedupe_key IS NOT NULL
|
|
DO NOTHING
|
|
RETURNING job_id
|
|
""")
|
|
|
|
async def run(s):
|
|
row = (await s.execute(sql, {
|
|
"t": job_type, "p": priority, "payload": json.dumps(payload),
|
|
"dk": dedupe_key, "ma": max_attempts,
|
|
})).first()
|
|
if row:
|
|
# 커밋 시 전달됨 → LISTEN 중인 유휴 워커를 즉시 깨운다(중복 스킵 시엔 알림 안 함).
|
|
await s.execute(text("SELECT pg_notify(:ch, '')"), {"ch": JOB_NOTIFY_CHANNEL})
|
|
return str(row[0]) if row else None
|
|
|
|
return await self._tx(run)
|
|
|
|
# ---- 원자적 claim ---------------------------------------------------
|
|
async def claim(self, worker_id: str, lease_sec: int = 120):
|
|
"""대기 잡 1건을 원자적으로 점유. 없으면 None.
|
|
FOR UPDATE SKIP LOCKED 로 잠근 행을 같은 UPDATE 에서 RUNNING 으로 전이 → 이중 할당 불가."""
|
|
sql = text("""
|
|
UPDATE job SET
|
|
status = 2,
|
|
worker_id = :wid,
|
|
lease_until = now() + make_interval(secs => :lease),
|
|
run_started_at = now(),
|
|
attempts = attempts + 1,
|
|
updated_at = now()
|
|
WHERE job_id = (
|
|
SELECT job_id FROM job
|
|
WHERE status = 1 AND run_after <= now()
|
|
ORDER BY priority ASC, created_at ASC
|
|
FOR UPDATE SKIP LOCKED
|
|
LIMIT 1
|
|
)
|
|
RETURNING job_id, job_type, payload, attempts, max_attempts
|
|
""")
|
|
|
|
async def run(s):
|
|
row = (await s.execute(sql, {"wid": worker_id, "lease": lease_sec})).mappings().first()
|
|
if not row:
|
|
return None
|
|
d = dict(row)
|
|
d["job_id"] = str(d["job_id"])
|
|
if isinstance(d.get("payload"), str):
|
|
d["payload"] = json.loads(d["payload"])
|
|
return d
|
|
|
|
return await self._tx(run)
|
|
|
|
# ---- 완료/실패 (소유권 가드) ---------------------------------------
|
|
async def complete(self, job_id: str, worker_id: str, result: dict | None = None) -> bool:
|
|
sql = text("""
|
|
UPDATE job SET status = 3, result = CAST(:result AS jsonb),
|
|
lease_until = NULL, worker_id = NULL, updated_at = now()
|
|
WHERE job_id = :id AND status = 2 AND worker_id = :wid
|
|
RETURNING job_id
|
|
""")
|
|
|
|
async def run(s):
|
|
row = (await s.execute(sql, {"id": job_id, "wid": worker_id, "result": json.dumps(result) if result is not None else None})).first()
|
|
return row is not None
|
|
|
|
return await self._tx(run)
|
|
|
|
async def fail(self, job_id: str, worker_id: str, error: str, backoff_sec: float = 5.0) -> int | None:
|
|
"""실패 처리. 시도 남으면 PENDING(run_after=백오프)으로 재큐, 소진되면 DEAD(dead-letter).
|
|
전이 후 status(JobStatus 값)를 반환. 소유 불일치면 None."""
|
|
sql = text("""
|
|
UPDATE job SET
|
|
status = CASE WHEN attempts >= max_attempts THEN 4 ELSE 1 END,
|
|
run_after = CASE WHEN attempts >= max_attempts THEN run_after
|
|
ELSE now() + make_interval(secs => :backoff) END,
|
|
last_error = :err,
|
|
lease_until = NULL,
|
|
worker_id = NULL,
|
|
updated_at = now()
|
|
WHERE job_id = :id AND status = 2 AND worker_id = :wid
|
|
RETURNING status
|
|
""")
|
|
|
|
async def run(s):
|
|
row = (await s.execute(sql, {"id": job_id, "wid": worker_id, "err": error[:2000], "backoff": backoff_sec})).first()
|
|
return int(row[0]) if row else None
|
|
|
|
return await self._tx(run)
|
|
|
|
# ---- lease 갱신(heartbeat) / 회수(reaper) ---------------------------
|
|
async def renew_lease(self, job_id: str, worker_id: str, lease_sec: int = 120) -> bool:
|
|
sql = text("""
|
|
UPDATE job SET lease_until = now() + make_interval(secs => :lease), updated_at = now()
|
|
WHERE job_id = :id AND worker_id = :wid AND status = 2
|
|
RETURNING job_id
|
|
""")
|
|
|
|
async def run(s):
|
|
row = (await s.execute(sql, {"id": job_id, "wid": worker_id, "lease": lease_sec})).first()
|
|
return row is not None
|
|
|
|
return await self._tx(run)
|
|
|
|
async def reap(self) -> list[str]:
|
|
"""만료된 lease(워커 사망 등)의 RUNNING 잡을 회수. 시도 남으면 즉시 재큐, 소진되면 DEAD.
|
|
회수된 job_id 목록 반환."""
|
|
sql = text("""
|
|
UPDATE job SET
|
|
status = CASE WHEN attempts >= max_attempts THEN 4 ELSE 1 END,
|
|
run_after = now(),
|
|
last_error = COALESCE(last_error, '') || ' [lease-expired reclaim]',
|
|
lease_until = NULL,
|
|
worker_id = NULL,
|
|
updated_at = now()
|
|
WHERE status = 2 AND lease_until IS NOT NULL AND lease_until < now()
|
|
RETURNING job_id
|
|
""")
|
|
|
|
async def run(s):
|
|
rows = (await s.execute(sql)).all()
|
|
return [str(r[0]) for r in rows]
|
|
|
|
return await self._tx(run)
|
|
|
|
# ---- 단건 조회 ------------------------------------------------------
|
|
async def get(self, job_id: str) -> dict | None:
|
|
"""잡 단건 조회(읽기). 없으면 None. status 는 정수(JobStatus 값)."""
|
|
sql = text("""
|
|
SELECT job_id, job_type, status, priority, attempts, max_attempts,
|
|
result, last_error, run_after, created_at, updated_at
|
|
FROM job WHERE job_id = :id
|
|
""")
|
|
s = await DB_SESSION_MNG.start_session(self.DB, DBWRType.DB_READ.value)
|
|
try:
|
|
row = (await s.execute(sql, {"id": job_id})).mappings().first()
|
|
if not row:
|
|
return None
|
|
d = dict(row)
|
|
d["job_id"] = str(d["job_id"])
|
|
if isinstance(d.get("result"), str):
|
|
d["result"] = json.loads(d["result"])
|
|
return d
|
|
finally:
|
|
await DB_SESSION_MNG.end_session(self.DB, DBWRType.DB_READ.value)
|
|
|
|
# ---- 관측(관리 API/메트릭용) ---------------------------------------
|
|
async def counts(self) -> dict[str, int]:
|
|
"""상태별 잡 개수(관리 API·알림용). 수동 psql 스크립트를 대체한다."""
|
|
async def run(s):
|
|
rows = (await s.execute(text("SELECT status, count(*) FROM job GROUP BY status"))).all()
|
|
by_val = {int(st): int(c) for st, c in rows}
|
|
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 + stuck(좀비 신호). stuck 은 두 축 — lease 만료(워커 사망인데 reaper
|
|
미회수) OR 실행 10분 초과(핸들러 행 — heartbeat 가 lease 를 계속 갱신해 lease 축엔 안
|
|
잡히므로 run_started_at 로 따로 본다. 잡 데드라인 300s 가 정상 작동하면 여기 안 온다)."""
|
|
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())
|
|
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,
|
|
-- 데드라인 강제종료(크롤 행 신호). 재시도로 살아나면 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:
|
|
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)
|
|
|
|
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)
|