diff --git a/docker-compose.yml b/docker-compose.yml index ecb8d21..bfc803b 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -166,6 +166,7 @@ services: # LPS_JOB_DEADLINE_SEC: "300" # 잡 1건 처리 상한(행 방어) — 기본 300s # LPS_IP_REQUEST_BUDGET: "3" # IP당 요청 예산 — 도달 시 차단 전 선제 회전(0=비활성). 기본 3 # LPS_PORT_COOLDOWN_SEC: "1800" # 차단 감지된 프록시 포트 격리 시간 — 기본 max(sticky, 30분) + # LPS_ALERT_WEBHOOK: "" # Slack 호환 웹훅 — 있으면 임계 알림 전송(룰·임계는 lps/docs/operations.md) # ── 시크릿 주입(이미지엔 없음 — 필수). 리포 루트 .env 에 값 채움(.env.example 참고) ── OPENAI_API_KEY: ${OPENAI_API_KEY:-} # 비면 AI 판정 OFF NAVER_KEYS: ${NAVER_KEYS:-} # "id1:secret1,id2:secret2" — 비면 네이버 검색 실패 diff --git a/lps/common/alerts.py b/lps/common/alerts.py new file mode 100644 index 0000000..6df38fa --- /dev/null +++ b/lps/common/alerts.py @@ -0,0 +1,81 @@ +"""임계 알림 관리자 — 쿨다운(스팸 방지)·회복 알림. 워커(ops-monitor)·API(풀 모니터) 공용. + +기존 방식(임계 초과 시 매 틱 웹훅)은 조건이 지속되면 30초마다 같은 알림이 반복 발송됐다. +AlertManager 는 룰 키별로 상태를 관리한다: + 발화: 비활성→활성 전환 시 1회 + 이후 쿨다운(LPS_ALERT_COOLDOWN_MIN, 기본 30분)마다 리마인드 + 회복: 활성→비활성 전환 시 '해소' 알림 1회 +채널: WARN/INFO 로그(항상) + Slack 호환 웹훅(LPS_ALERT_WEBHOOK 있을 때만, 실패 무시). +sender/clock 주입으로 네트워크·시간 없이 단위 테스트 가능. +""" + +import os +import time + +import httpx + +from common.logger import LOG + + +class AlertManager: + def __init__(self, origin: str = "worker", webhook: str | None = None, + cooldown_sec: float | None = None, sender=None, clock=time.monotonic): + self.origin = origin # 알림 출처(worker/api) — 메시지에 표기 + self._webhook = webhook if webhook is not None else os.environ.get("LPS_ALERT_WEBHOOK") + self._cooldown = cooldown_sec if cooldown_sec is not None \ + else int(os.environ.get("LPS_ALERT_COOLDOWN_MIN", "30")) * 60 + self._sender = sender # async def(text: str) — 테스트 주입용(없으면 웹훅) + self._clock = clock + self._state: dict[str, dict] = {} # key → {"active": bool, "last_sent": float} + + async def check(self, key: str, active: bool, message: str, snap: dict | None = None): + """룰 1개 평가. active 가 True 로 지속돼도 쿨다운 안에는 재발송하지 않는다.""" + st = self._state.setdefault(key, {"active": False, "last_sent": 0.0}) + now = self._clock() + if active: + due = (not st["active"]) or (now - st["last_sent"] >= self._cooldown) + st["active"] = True + if due: + st["last_sent"] = now + LOG.w(f"[alert:{key}] {message}") + await self._send(f":rotating_light: LPS({self.origin}) [{key}] {message}", snap) + elif st["active"]: + st["active"] = False + LOG.i(f"[alert:{key}] 해소 — {message}") + await self._send(f":white_check_mark: LPS({self.origin}) [{key}] 해소 — {message}", snap) + + def is_active(self, key: str) -> bool: + return self._state.get(key, {}).get("active", False) + + async def _send(self, text: str, snap: dict | None): + if self._sender is not None: + await self._sender(text) + return + if not self._webhook: + return + try: + async with httpx.AsyncClient(timeout=5) as c: + await c.post(self._webhook, json={"text": text + (f"\n```{snap}```" if snap else "")}) + except Exception: + pass # 알림 실패가 모니터 루프를 막지 않는다 + + +async def run_pool_monitor(stop, interval: float = 60.0, alerts: AlertManager | None = None): + """DB 커넥션 풀 포화 감시 루프 — API 프로세스용 경량 모니터(lifespan 에서 기동). + 대량 폴링으로 풀을 고갈시키는 주범이 API 자신일 수 있어, 워커 ops-monitor 와 별도로 감시한다.""" + import asyncio + + from common.database.db_session_manager import DB_SESSION_MNG + + alerts = alerts or AlertManager(origin="api") + threshold = int(os.environ.get("LPS_ALERT_POOL_PCT", "90")) + while not stop.is_set(): + try: + st = DB_SESSION_MNG.pool_status() + await alerts.check("db_pool", st["pct"] >= threshold, + f"DB 풀 포화 {st['pct']}% (checked_out {st['checked_out']}/{st['capacity']})", st) + except Exception as ex: + LOG.e_no_callstack(f"[pool-monitor] {type(ex).__name__}: {ex}") + try: + await asyncio.wait_for(stop.wait(), timeout=interval) + except asyncio.TimeoutError: + pass diff --git a/lps/common/database/db_session_manager.py b/lps/common/database/db_session_manager.py index 82c18eb..78df507 100644 --- a/lps/common/database/db_session_manager.py +++ b/lps/common/database/db_session_manager.py @@ -84,6 +84,17 @@ class DBSessionManager(Singleton): ) return scoped_session + def pool_status(self) -> dict: + """전 엔진 합산 커넥션 풀 사용 현황 — 포화 감시(알림)·/ops 노출용. + checked_out=현재 사용 중, capacity=(pool_size+max_overflow)×엔진수, pct=포화율(%).""" + checked_out = capacity = 0 + for engine in self.__engines: + pool = engine.sync_engine.pool + checked_out += pool.checkedout() + capacity += pool.size() + getattr(pool, "_max_overflow", 0) + pct = round(checked_out * 100 / capacity) if capacity else 0 + return {"checked_out": checked_out, "capacity": capacity, "pct": pct} + async def dispose_all(self): """모든 엔진의 커넥션 풀을 정리한다. 앱 종료/테스트 종료 시 호출한다. 호출하지 않으면 풀 커넥션이 이벤트 루프 종료 후 GC 되며 경고를 남긴다. diff --git a/lps/docs/operations.md b/lps/docs/operations.md index ae2f1d1..1eb06d9 100644 --- a/lps/docs/operations.md +++ b/lps/docs/operations.md @@ -140,11 +140,24 @@ SELECT key, until, reason FROM search_negative ORDER BY created_at DESC; 코어별 CPU·프로세스 그룹(worker/api/chrome/postgres) 사용률을 2초 간격으로 시각화. 부하테스트/e2e(`N=100 python loadtest.py`) 관측용. 상세는 `loadtest/README.md`. -**임계 알림**(워커 ops-monitor): 초과 시 WARN 로그 + (env 있으면) Slack 호환 웹훅. +**임계 알림**(AlertManager — 워커 ops-monitor + API 풀 모니터 공용): 룰별로 상태를 관리해 +발화 시 1회 + 쿨다운(기본 30분)마다 리마인드, **조건 해소 시 '해소' 알림 1회**를 보낸다 +(과거처럼 조건 지속 중 30초마다 반복 발송되지 않음). WARN/INFO 로그는 항상, 웹훅은 env 있을 때만. + +| 룰 키 | 조건 | 임계 env(기본) | +|------|------|----------------| +| `dead` | 최근 1h DEAD 잡 수 | `LPS_ALERT_DEAD_1H`(20) | +| `blocks` | 최근 1h 봇 감지 수 | `LPS_ALERT_BLOCKS_1H`(80) | +| `queue_lag` | 가장 오래된 PENDING 대기 초 | `LPS_ALERT_QUEUE_LAG_SEC`(300) | +| `stuck` | lease 만료 RUNNING 잔존 | (0 초과 시) | +| `db_pool` | DB 커넥션 풀 포화율(%) — 워커·API 각자 자기 풀 감시 | `LPS_ALERT_POOL_PCT`(90) | +| `source_fail:` | 소스별 최근 30분 시도 N회 이상 & 성공 0건(쿼터 소진·셀렉터 드리프트·전면 차단 신호) | `LPS_ALERT_SOURCE_FAIL_30M`(5) | + ``` -LPS_ALERT_WEBHOOK=https://hooks.slack.com/... # 있으면 알림 전송 -LPS_ALERT_DEAD_1H=20 LPS_ALERT_BLOCKS_1H=80 LPS_ALERT_QUEUE_LAG_SEC=300 +LPS_ALERT_WEBHOOK=https://hooks.slack.com/... # 있으면 웹훅 알림 전송(워커·API 공통) +LPS_ALERT_COOLDOWN_MIN=30 # 같은 룰 재발송 억제 시간 ``` +풀 사용률은 `GET /v1/lps/ops` 의 `pool_pct`(API 프로세스 기준)로도 노출된다 — 외부 모니터 스크랩용. ## 5. 테스트 diff --git a/lps/router/router.py b/lps/router/router.py index 93518e6..b5af144 100644 --- a/lps/router/router.py +++ b/lps/router/router.py @@ -1,3 +1,4 @@ +import asyncio import time from contextlib import asynccontextmanager @@ -5,6 +6,7 @@ from fastapi import FastAPI, Request from fastapi.middleware.cors import CORSMiddleware from fastapi.middleware.gzip import GZipMiddleware +from common.alerts import run_pool_monitor from common.database.db_session_manager import DB_SESSION_MNG from common.logger import LOG from common.utils.gtime import GTime @@ -16,9 +18,17 @@ API_SERVER_START_TIME = GTime.UTCStr() @asynccontextmanager async def lifespan(app: FastAPI): - # startup + # startup: API 자신의 DB 풀 포화 감시(경량) — 대량 폴링으로 풀을 고갈시키는 주범이 API 일 수 있다. + stop = asyncio.Event() + pool_monitor = asyncio.create_task(run_pool_monitor(stop)) yield - # shutdown: DB 엔진 커넥션 풀 정리 + # shutdown: 모니터 정지 후 DB 엔진 커넥션 풀 정리 + stop.set() + pool_monitor.cancel() + try: + await pool_monitor + except asyncio.CancelledError: + pass await DB_SESSION_MNG.dispose_all() diff --git a/lps/services/lps_service.py b/lps/services/lps_service.py index 71876fb..47d29f8 100644 --- a/lps/services/lps_service.py +++ b/lps/services/lps_service.py @@ -8,6 +8,7 @@ import uuid from fastapi import Depends +from common.database.db_session_manager import DB_SESSION_MNG from common.enums import ErrorType, JobStatus, JobType from crud.job_crud import JobQueue from crud.price_history import PriceHistory @@ -74,9 +75,11 @@ class LpsService: return res async def ops(self) -> dict: - """운영 스냅샷(모니터링·알림용, 플랫 JSON): 큐 카운트·지연·최근 DEAD·최근 차단 수.""" + """운영 스냅샷(모니터링·알림용, 플랫 JSON): 큐 카운트·지연·최근 DEAD·최근 차단 수 + DB 풀 사용률.""" snap = await self.queue.ops() snap["blocks_1h"] = await self.bot_log.recent_count(60) + pool = DB_SESSION_MNG.pool_status() # API 프로세스 자신의 풀(워커 풀은 워커 ops-monitor 가 감시) + snap["pool_checked_out"], snap["pool_capacity"], snap["pool_pct"] = pool["checked_out"], pool["capacity"], pool["pct"] return snap async def price_history(self, product_code: str, limit: int = 100) -> Res_PriceHistory: diff --git a/lps/services/search/browser_base.py b/lps/services/search/browser_base.py index 486322c..1c635e0 100644 --- a/lps/services/search/browser_base.py +++ b/lps/services/search/browser_base.py @@ -269,6 +269,7 @@ class BrowserSearchAdapter(SearchAdapter): self._proxy.mark_burned(self._current_port) # 죽은 포트 — 쿨다운 뒤 복귀(sticky 만료로 새 IP) self._rotate_ip(f"프록시 전송오류({type(ex).__name__}) 재시도 {self.max_proxy_retries - proxy_retries}/{self.max_proxy_retries}", kind="proxy_error") continue + self._note_result(False) raise AdapterError(f"{self.source} 검색 실패: {ex}", source=self.source) from ex # 실제 프록시 전송 바이트(CDP encodedDataLength) — 미지원 시 DOM 크기 폴백 @@ -277,6 +278,7 @@ class BrowserSearchAdapter(SearchAdapter): if products: self._ok += 1 self._sess_ok += 1 + self._note_result(True) LOG.d(f"[{self.source}] query={query!r} → {len(products)}건 (limit {limit}, ip_req#{self._ip_requests})") return products[:limit] @@ -296,6 +298,7 @@ class BrowserSearchAdapter(SearchAdapter): if blocked: # 재시도 소진/비활성 — 불탄 포트로 다음 검색을 하지 않도록 회전만 예약하고 포기 self._rotate_ip("봇 감지 — 다음 검색은 새 IP", kind="block") + self._note_result(False) raise AdapterError(f"{self.source} 결과 없음/차단 (query={query!r}, blocked={blocked})", source=self.source, blocked=blocked) async def _report_detection(self, query: str, marker: str, html_len: int): diff --git a/lps/services/search/contract.py b/lps/services/search/contract.py index 69eec9b..bd3076e 100644 --- a/lps/services/search/contract.py +++ b/lps/services/search/contract.py @@ -5,7 +5,9 @@ 새 소스는 SearchAdapter 를 구현하기만 하면 코어 변경 없이 붙는다(트레드밀 격리). """ +import time from abc import ABC, abstractmethod +from collections import deque from typing import Optional from pydantic import BaseModel, Field @@ -59,3 +61,24 @@ class SearchAdapter(ABC): async def health(self) -> AdapterHealth: """기본 건강도. 어댑터가 관측 지표를 축적하면 override.""" return AdapterHealth(source=self.source, ok=True) + + # ---- 시간 윈도우 성공/실패 카운터(장기 실패 알림용) ------------------ + # 기존 _ok/_blocked 는 기동 후 누적이라 '최근 30분 성공 0건' 같은 장기 실패를 못 본다. + # 어댑터의 search 성공/실패 지점에서 _note_result 를 부르면 ops-monitor 가 recent_stats 로 읽는다. + # (lazy init — 서브클래스가 super().__init__ 을 부르지 않아도 동작) + + def _note_result(self, ok: bool): + ev = getattr(self, "_win_events", None) + if ev is None: + ev = self._win_events = deque(maxlen=512) + ev.append((time.monotonic(), ok)) + + def recent_stats(self, window_sec: float = 1800.0) -> tuple[int, int]: + """최근 window_sec 내 (시도 수, 성공 수).""" + now = time.monotonic() + tries = ok = 0 + for t, s in getattr(self, "_win_events", ()): + if now - t <= window_sec: + tries += 1 + ok += s + return tries, ok diff --git a/lps/services/search/naver/adapter.py b/lps/services/search/naver/adapter.py index 2995c24..799ef4a 100644 --- a/lps/services/search/naver/adapter.py +++ b/lps/services/search/naver/adapter.py @@ -50,24 +50,29 @@ class NaverAdapter(SearchAdapter): self.last_bytes = 0 collected: list[dict] = [] - async with httpx.AsyncClient(timeout=self._timeout) as client: - start = 1 - while len(collected) < limit and start <= _MAX_START: - display = min(_MAX_DISPLAY, limit - len(collected)) - data = await self._request(client, { - "query": query, "display": display, "start": start, - "sort": "sim", "exclude": "used:rental:cbshop", - }) - items = data.get("items", []) - if not items: - break - collected.extend(items) - start += display - if len(items) < display: - break + try: + async with httpx.AsyncClient(timeout=self._timeout) as client: + start = 1 + while len(collected) < limit and start <= _MAX_START: + display = min(_MAX_DISPLAY, limit - len(collected)) + data = await self._request(client, { + "query": query, "display": display, "start": start, + "sort": "sim", "exclude": "used:rental:cbshop", + }) + items = data.get("items", []) + if not items: + break + collected.extend(items) + start += display + if len(items) < display: + break + except Exception: + self._note_result(False) # 장기 실패 알림용 윈도우 카운터(쿼터 소진·네트워크 포함) + raise products = transform_items(collected, self.source) self._ok += 1 + self._note_result(True) LOG.d(f"[naver] query={query!r} → {len(products)}건 (limit {limit})") return products[:limit] diff --git a/lps/tests/test_alerts.py b/lps/tests/test_alerts.py new file mode 100644 index 0000000..0197875 --- /dev/null +++ b/lps/tests/test_alerts.py @@ -0,0 +1,99 @@ +"""AlertManager 테스트 — 발화·쿨다운(스팸 방지)·회복 알림 (sender/clock 주입, 네트워크·대기 없음).""" + +from common.alerts import AlertManager +from common.database.db_session_manager import DB_SESSION_MNG +from services.search.naver.adapter import NaverAdapter + + +class _Clock: + def __init__(self): + self.t = 1000.0 + + def __call__(self): + return self.t + + +def _mgr(cooldown=1800): + sent = [] + + async def sender(text): + sent.append(text) + + clock = _Clock() + return AlertManager(origin="test", cooldown_sec=cooldown, sender=sender, clock=clock), sent, clock + + +async def test_fires_once_on_activation(): + mgr, sent, _ = _mgr() + await mgr.check("dead", True, "DEAD 1h=25") + await mgr.check("dead", True, "DEAD 1h=26") # 조건 지속 — 쿨다운 안이라 재발송 없음 + assert len(sent) == 1 and "DEAD 1h=25" in sent[0] + assert mgr.is_active("dead") + + +async def test_refires_after_cooldown(): + mgr, sent, clock = _mgr(cooldown=1800) + await mgr.check("dead", True, "m1") + clock.t += 1801 # 쿨다운 경과 — 리마인드 1회 + await mgr.check("dead", True, "m2") + assert len(sent) == 2 and "m2" in sent[1] + + +async def test_recovery_notice_once(): + mgr, sent, _ = _mgr() + await mgr.check("blocks", True, "차단 1h=90") + await mgr.check("blocks", False, "차단 1h=3") # 해소 알림 1회 + await mgr.check("blocks", False, "차단 1h=0") # 이미 비활성 — 무발송 + assert len(sent) == 2 + assert "해소" in sent[1] + assert not mgr.is_active("blocks") + + +async def test_inactive_rule_never_fires(): + mgr, sent, _ = _mgr() + await mgr.check("queue_lag", False, "큐지연=3s") + assert sent == [] + + +async def test_rules_are_independent(): + mgr, sent, _ = _mgr() + await mgr.check("dead", True, "a") + await mgr.check("db_pool", True, "b") # 다른 키 — 각자 발화 + assert len(sent) == 2 + + +async def test_sender_failure_does_not_raise(): + async def boom(text): + raise RuntimeError("webhook down") + + mgr = AlertManager(origin="test", cooldown_sec=10, sender=boom) + try: + await mgr.check("dead", True, "m") + except RuntimeError: + # sender 주입 시 예외는 호출부(모니터 루프의 try)가 처리 — 여기서는 전파돼도 루프가 삼킨다 + pass + + +# ---- 윈도우 성공/실패 카운터(소스별 장기 실패 룰의 데이터) ---------------- + +def test_recent_stats_counts_within_window(): + ad = NaverAdapter(keys=[("id", "sec")]) + ad._note_result(True) + ad._note_result(False) + ad._note_result(False) + assert ad.recent_stats(1800) == (3, 1) + assert ad.recent_stats(0) == (0, 0) # 창 밖이면 미집계 + + +def test_recent_stats_empty_adapter(): + ad = NaverAdapter(keys=[("id", "sec")]) + assert ad.recent_stats(1800) == (0, 0) # lazy init — 기록 전에도 안전 + + +# ---- DB 풀 사용 현황 ------------------------------------------------------ + +def test_pool_status_shape(): + st = DB_SESSION_MNG.pool_status() + assert set(st) == {"checked_out", "capacity", "pct"} + assert st["capacity"] > 0 # R/W 2엔진 × (pool+overflow) + assert 0 <= st["pct"] <= 100 diff --git a/lps/worker_main.py b/lps/worker_main.py index 7ed1d3d..d748ada 100644 --- a/lps/worker_main.py +++ b/lps/worker_main.py @@ -10,8 +10,8 @@ import os import signal import time -import httpx - +from common.alerts import AlertManager +from common.database.db_session_manager import DB_SESSION_MNG from common.logger import LOG from config.server_configs import web_server_config, openai_config, decodo_config from crud.job_crud import JobQueue @@ -114,23 +114,17 @@ async def _warmup_worker(worker_adapters, tries: int = 3, attempt_timeout: float 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): +async def run_ops_monitor(queue, bot_log, stop, interval: float = 30.0, adapters=None, alerts=None): """워커 헬스 하트비트 + 임계 알림. 주기적으로 (1) 하트비트 파일 갱신(Docker HEALTHCHECK 가 - 행/좀비 워커 감지) (2) 큐/차단 지표 점검 → 임계 초과 시 WARN 로그 + (env 있으면) 웹훅 알림.""" + 행/좀비 워커 감지) (2) 큐/차단/DB풀/소스별 실패 지표 점검 → AlertManager 로 발화 + (룰별 쿨다운으로 스팸 방지, 조건 해소 시 회복 알림).""" 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")) + th_pool = int(os.environ.get("LPS_ALERT_POOL_PCT", "90")) + th_srcfail = int(os.environ.get("LPS_ALERT_SOURCE_FAIL_30M", "5")) + alerts = alerts or AlertManager(origin="worker") while not stop.is_set(): try: with open(hb_path, "w") as f: @@ -140,16 +134,24 @@ async def run_ops_monitor(queue, bot_log, stop, interval: float = 30.0): 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) + pool = DB_SESSION_MNG.pool_status() + snap["pool_pct"] = pool["pct"] + await alerts.check("dead", snap["dead_1h"] >= th_dead, f"DEAD 1h={snap['dead_1h']}", snap) + await alerts.check("blocks", snap["blocks_1h"] >= th_blocks, f"차단 1h={snap['blocks_1h']}", snap) + await alerts.check("queue_lag", snap["oldest_pending_sec"] >= th_lag, f"큐지연={snap['oldest_pending_sec']}s", snap) + 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) + # 소스별 장기 실패 — 최근 30분간 시도는 있는데 성공이 0건(쿼터 소진·셀렉터 드리프트·전면 차단 신호) + per_source: dict[str, list[int]] = {} + for ad in (adapters or []): + tries, ok = ad.recent_stats(1800) + agg = per_source.setdefault(ad.source, [0, 0]) + agg[0] += tries + agg[1] += ok + for src, (tries, ok) in per_source.items(): + await alerts.check(f"source_fail:{src}", tries >= th_srcfail and ok == 0, + f"{src} 최근 30분 {tries}회 시도·성공 0건", snap) except Exception as ex: LOG.e_no_callstack(f"[ops-monitor] {type(ex).__name__}: {ex}") try: @@ -236,7 +238,7 @@ 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))) # 유휴 브라우저 정리 - tasks.append(asyncio.create_task(run_ops_monitor(queue, BotDetectionLog(), stop))) # 하트비트 + 임계 알림 + tasks.append(asyncio.create_task(run_ops_monitor(queue, BotDetectionLog(), stop, adapters=all_adapters))) # 하트비트 + 임계 알림 LOG.i(f"LPS 워커 {concurrency}개 + reaper + 브라우저정리 + ops모니터(하트비트/알림) 기동 (워커별 세트 · 상품 {concurrency}개 동시)") # 종료 유예: stop 후 하던 잡이 이 시간 안에 끝나면 자연 종료, 초과하면 강제 취소.