feat(lps): 다중 상품 병렬 검색(워커별 브라우저 세트) + 부하테스트 스크립트

WORKER_CONCURRENCY 를 늘려도 공유 브라우저 lock 때문에 직렬화되던 문제를 고쳐 진짜 병렬 검색. - worker_main: _build_worker(i) 로 워커마다 자립 세트(브라우저 어댑터·AI·핸들러) 생성. 프로필 분리(user_data_dir_w{i}, ProcessSingleton 충돌 회피) + 워커별 다른 프록시 포트 (proxy.seed_offset 로 100포트를 균등 분할=다른 IP). naver/judge/keyword 도 워커별(공유상태 경합 제거). 프리플라이트는 대표 프록시로 게이트 1회 확인. - proxy.seed_offset(k): 워커 시작 포트 분산. - loadtest.py: N개 상품 제출→폴링→처리량·지연(p50/p95)·AI/DECODO/총비용 집계. 실측(동시성2, 4상품): 순차합 323s→벽시계 181s(~1.8x), 상품당 $0.0071, 1000건 ~$7. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
2026-07-09 16:28:08 +09:00 · 2026-07-09 16:28:08 +09:00 · ad26e6ce6f
commit ad26e6ce6f
parent bf19d4a8fd
3 changed files with 160 additions and 31 deletions
--- a/lps/loadtest.py
+++ b/lps/loadtest.py
@ -0,0 +1,113 @@
+"""LPS 부하 테스트 — 여러 상품을 한 번에 제출하고 처리량·지연·비용을 집계한다.
+
+여러 상품 동시 검색을 재현한다. 워커 동시성(WORKER_CONCURRENCY)만큼 상품이 병렬 처리된다.
+  python loadtest.py                 # 기본 상품셋
+  N=8 python loadtest.py             # 상품 8개(기본셋에서 반복 확장)
+  BASE=http://localhost:9600 python loadtest.py
+
+측정: 벽시계 총시간, 처리량(상품/분), 상품별·집계 지연(p50/p95), AI·DECODO·총비용.
+워커가 떠 있어야 하고, 워커 동시성이 낮으면 상품들이 큐에서 순차 대기한다(그게 부하의 핵심 관측).
+"""
+
+import asyncio
+import os
+import time
+
+import httpx
+
+BASE = os.environ.get("BASE", "http://localhost:9600")
+N = int(os.environ.get("N", "6"))
+
+# 실제 매칭 잘 되는 상품셋(반복해서 N개로 확장). product_code 는 부하테스트 전용 프리픽스.
+_CATALOG = [
+    {"product_name": "코카콜라 제로", "specification": "355ml 24캔"},
+    {"product_name": "농심 신라면", "specification": "1박스 40개입"},
+    {"product_name": "오리온 초코파이", "specification": "12개입"},
+    {"product_name": "제주 삼다수", "specification": "2L 12개입"},
+    {"product_name": "맥심 모카골드 커피믹스", "specification": "1박스 160개입"},
+    {"product_name": "스탠리 텀블러", "specification": "887ml"},
+]
+
+
+def _products(n):
+    out = []
+    for i in range(n):
+        base = _CATALOG[i % len(_CATALOG)]
+        out.append({"product_code": f"LT{i:03d}", "job_type": "batch", **base})
+    return out
+
+
+def _pct(values, p):
+    if not values:
+        return 0
+    s = sorted(values)
+    k = min(len(s) - 1, int(round((p / 100) * (len(s) - 1))))
+    return s[k]
+
+
+async def main():
+    products = _products(N)
+    print(f"■ 부하 테스트: 상품 {N}개 → {BASE}  (워커 동시성만큼 병렬 처리)\n")
+
+    async with httpx.AsyncClient(timeout=30) as c:
+        # 큐/헬스 사전 확인
+        try:
+            stats = (await c.get(f"{BASE}/v1/lps/queue/stats")).json()["counts"]
+            print(f"시작 큐 상태: {stats}")
+        except Exception as e:
+            print(f"API 접속 실패({e}). 서버가 떠 있나요?"); return
+
+        t0 = time.monotonic()
+        r = (await c.post(f"{BASE}/v1/lps/search", json={"data": products})).json()
+        jobs = [it["job_id"] for it in r.get("items", []) if it.get("job_id")]
+        print(f"접수: {len(jobs)}건 (중복 스킵 {N - len(jobs)})\n폴링 중…\n")
+
+        results = {}
+        while len(results) < len(jobs):
+            await asyncio.sleep(2)
+            pend = [j for j in jobs if j not in results]
+            got = await asyncio.gather(*[c.get(f"{BASE}/v1/lps/jobs/{j}") for j in pend])
+            for j, resp in zip(pend, got):
+                d = resp.json()
+                if d.get("status") in ("DONE", "DEAD"):
+                    results[j] = d
+            done = len(results)
+            print(f"\r  진행 {done}/{len(jobs)}  ({time.monotonic()-t0:.0f}s 경과)", end="", flush=True)
+
+        wall = time.monotonic() - t0
+        print("\n")
+
+        # 집계
+        durs, ai_costs, proxy_costs, totals = [], [], [], []
+        found = dead = 0
+        print(f"{'상품코드':<8} {'상태':<5} {'결과':<10} {'소요':>6} {'AI$':>9} {'DECODO$':>9} {'총$':>9}")
+        print("-" * 66)
+        for j in jobs:
+            d = results[j]
+            o = d.get("output") or {}
+            m = o.get("metrics") or {}
+            cost = m.get("cost") or {}
+            code = (o.get("query") or "")[:8]
+            st = d.get("status", "?")
+            dur = (m.get("duration_ms") or 0) / 1000
+            if st == "DONE":
+                found += 1 if o.get("outcome") == "found" else 0
+                durs.append(dur); ai_costs.append(cost.get("ai_usd", 0))
+                proxy_costs.append(cost.get("proxy_usd", 0)); totals.append(cost.get("total_usd", 0))
+            else:
+                dead += 1
+            print(f"{code:<8} {st:<5} {o.get('outcome','-'):<10} {dur:>5.1f}s "
+                  f"{cost.get('ai_usd',0):>9.6f} {cost.get('proxy_usd',0):>9.6f} {cost.get('total_usd',0):>9.6f}")
+
+        print("-" * 66)
+        print(f"\n■ 집계 ({len(jobs)}건, DEAD {dead})")
+        print(f"  벽시계 총시간   : {wall:.1f}s   처리량: {len(jobs)/wall*60:.1f} 상품/분")
+        print(f"  상품 지연        : p50 {_pct(durs,50):.1f}s · p95 {_pct(durs,95):.1f}s · max {max(durs) if durs else 0:.1f}s")
+        print(f"  ※ 순차합 대비 병렬: 상품별 소요 합 {sum(durs):.0f}s → 벽시계 {wall:.0f}s (동시성 효과)")
+        print(f"  비용 합계        : AI ${sum(ai_costs):.5f} + DECODO ${sum(proxy_costs):.5f} = ${sum(totals):.5f}")
+        print(f"  상품당 평균 비용 : ${(sum(totals)/len(totals)) if totals else 0:.6f}")
+        print(f"  1,000건 추정 비용: ${(sum(totals)/len(totals)*1000) if totals else 0:.2f}")
+
+
+if __name__ == "__main__":
+    asyncio.run(main())
--- a/lps/services/search/proxy.py
+++ b/lps/services/search/proxy.py
@ -37,6 +37,10 @@ class DecodoProxy:
        """시간창과 무관하게 즉시 다음 포트(=새 IP)로 회전. 봇 감지 시 호출."""
        self._rotate_offset += 1

+    def seed_offset(self, k: int):
+        """워커별 시작 포트 분산용 — 동시 워커가 같은 포트(=같은 IP)를 쓰지 않도록 시작점을 벌린다."""
+        self._rotate_offset = k
+
    def _port(self) -> int:
        """시간창 + 수동 오프셋 기반 포트 선택. 창 안에선 동일 IP, rotate()나 창 변화 시 다음 IP."""
        n = self.port_end - self.port_start + 1
--- a/lps/worker_main.py
+++ b/lps/worker_main.py
@ -1,9 +1,9 @@
 # LPS 워커 프로세스 진입점 (API 와 분리 실행 — 코드베이스 공유, 독립 스케일).
 #   python worker_main.py
-#   WORKER_CONCURRENCY=2 python worker_main.py
+#   WORKER_CONCURRENCY=3 python worker_main.py   # 상품 3개 동시 검색(권장 2~3, 로컬)
 #
-# 쿠팡 검색은 브라우저(Chrome)라 무겁고 컨텍스트당 직렬이므로 기본 동시성은 1.
-# 여러 브라우저로 늘리려면 CoupangAdapter 인스턴스를 워커마다 따로 준다.
+# 브라우저 어댑터는 컨텍스트당 직렬(lock)이라, 진짜 병렬을 위해 **워커마다 자기 브라우저 세트**를 준다:
+# 프로필 분리(user_data_dir_w{i}) + 워커별 다른 프록시 포트(=다른 IP). 동시성 N → 최대 4×N Chrome.

 import asyncio
 import os
@ -28,52 +28,64 @@ from worker.runner import Worker, run_reaper
 LOG.SetPrefix(f"{web_server_config.server_name}-worker")


-async def main(concurrency: int = 1):
-    queue = JobQueue()
-    # 쿠팡(브라우저, 무거움) + 네이버(오픈API, 가벼움) 동시 검색 → 병합 최저가
-    # DECODO 프록시: .env 에 값 있으면 쿠팡만 sticky+주기적 회전으로 경유(없으면 직접 연결)
+def _build_worker(i: int, concurrency: int, has_openai: bool, neg_cache, history):
+    """워커 1개의 자립 세트(브라우저 어댑터·AI·핸들러)를 만든다.
+    프로필 분리(user_data_dir_w{i}) + 워커별 다른 프록시 포트(=다른 IP)로 진짜 병렬을 보장한다."""
+    # 워커별 프록시(다른 포트=다른 IP). 100포트를 워커 수로 균등 분할해 시작점을 벌린다.
    proxy = DecodoProxy()
-    LOG.i(f"DECODO 프록시: {'ON(sticky ' + str(proxy.session_minutes) + '분 회전)' if proxy.enabled else 'OFF(미설정)'}")
-    # 시작 프리플라이트: 살아있는 프록시 포트를 선점하고 egress IP 를 로그로 남긴다(빠른 실패·가시성).
-    # residential IP 는 실행 중에도 죽으므로 실제 회복은 런타임 IP 회전(전송오류·봇감지)이 담당.
-    if proxy.enabled:
-        egress_ip, egress_port = await proxy.healthcheck()
-        if egress_ip:
-            LOG.i(f"DECODO 프리플라이트 OK — egress IP {egress_ip} (port {egress_port})")
-        else:
-            LOG.w("DECODO 프리플라이트 실패 — 살아있는 포트를 못 찾음(런타임 회전으로 재시도)")
-    # 봇 감지 시: 감지 기록(DB) + 새 IP 로 회전 후 재시도
+    if proxy.enabled and concurrency > 1:
+        n = proxy.port_end - proxy.port_start + 1
+        proxy.seed_offset(i * max(1, n // concurrency))
    bot_log = BotDetectionLog()
+    suffix = f"_w{i}" if concurrency > 1 else ""
+
+    def _pf(source):  # 워커별 Chrome 프로필 경로(중복 실행 시 ProcessSingleton 충돌 방지)
+        return f"/tmp/lps_{source}{suffix}"
+
    adapters = {
-        "coupang": CoupangAdapter(headless=False, proxy=proxy, on_detect=bot_log.record),
-        "naver": NaverAdapter(),
+        "coupang": CoupangAdapter(headless=False, user_data_dir=_pf("coupang"), proxy=proxy, on_detect=bot_log.record),
+        "naver": NaverAdapter(),   # httpx 직접(프록시 미경유) — 워커별 인스턴스(last_bytes 경합 회피)
    }
-    # 오픈마켓 폴백 크롤러: 네이버가 그 몰을 커버 못 했을 때만 lazy 하게 실사이트 크롤(브라우저는 첫 사용 시 기동).
-    #   G마켓·옥션(ESM '잠시만' 챌린지) + 11번가(PC). 리소스차단 OFF(렌더/챌린지 보호)는 어댑터 기본값.
    fallback_adapters = {
-        "gmarket": EsmAdapter("gmarket", headless=False, proxy=proxy, on_detect=bot_log.record),
-        "auction": EsmAdapter("auction", headless=False, proxy=proxy, on_detect=bot_log.record),
-        "st11": ElevenStAdapter(headless=False, proxy=proxy, on_detect=bot_log.record),
+        "gmarket": EsmAdapter("gmarket", headless=False, user_data_dir=_pf("gmarket"), proxy=proxy, on_detect=bot_log.record),
+        "auction": EsmAdapter("auction", headless=False, user_data_dir=_pf("auction"), proxy=proxy, on_detect=bot_log.record),
+        "st11": ElevenStAdapter(headless=False, user_data_dir=_pf("st11"), proxy=proxy, on_detect=bot_log.record),
    }
-    LOG.i(f"오픈마켓 폴백 크롤: {', '.join(fallback_adapters)} (네이버 미커버 몰만)")
-    # OpenAI 키 있으면 '같은 상품' AI 판정 + 재검색어 생성 활성화
-    has_openai = bool(openai_config.api_key)
+    # AI 도 워커별 인스턴스 — 공유 상태(last_usage) 경합 원천 제거
    judge = SimilarityJudge() if has_openai else None
    keyword_gen = KeywordGenerator() if has_openai else None
-    LOG.i(f"AI(판정+검색어생성): {'ON' if has_openai else 'OFF(키 없음)'}")
    handler = build_search_handler(
        adapters, judge=judge, keyword_gen=keyword_gen,
-        neg_cache=NegativeCache(), history=PriceHistory(),
+        neg_cache=neg_cache, history=history,
        fallback_adapters=fallback_adapters,
        ai_model=openai_config.model,
        proxy_cost_per_gb=decodo_config.cost_per_gb,
    )
+    return handler, list(adapters.values()) + list(fallback_adapters.values())
+
+
+async def main(concurrency: int = 1):
+    queue = JobQueue()
+    neg_cache, history = NegativeCache(), PriceHistory()   # DB 기반 — 워커 공유 안전
+    has_openai = bool(openai_config.api_key)
+
+    # 시작 프리플라이트: DECODO 게이트가 살아있는지(인증) 대표 프록시로 1회 확인. 포트는 워커별로 각자 잡음.
+    probe = DecodoProxy()
+    LOG.i(f"DECODO 프록시: {'ON(sticky ' + str(probe.session_minutes) + '분 회전)' if probe.enabled else 'OFF(미설정)'}")
+    if probe.enabled:
+        egress_ip, egress_port = await probe.healthcheck()
+        LOG.i(f"DECODO 프리플라이트 OK — egress IP {egress_ip} (port {egress_port})") if egress_ip \
+            else LOG.w("DECODO 프리플라이트 실패 — 살아있는 포트를 못 찾음(런타임 회전으로 재시도)")
+    LOG.i(f"AI(판정+검색어생성): {'ON' if has_openai else 'OFF(키 없음)'} · 오픈마켓 폴백: gmarket, auction, st11")

    stop = asyncio.Event()
    listeners: list[JobListener] = []
    tasks: list[asyncio.Task] = []
+    all_adapters = []

    for i in range(concurrency):
+        handler, worker_adapters = _build_worker(i, concurrency, has_openai, neg_cache, history)
+        all_adapters += worker_adapters
        listener = JobListener()
        await listener.start()
        listeners.append(listener)
@ -81,7 +93,7 @@ async def main(concurrency: int = 1):
        tasks.append(asyncio.create_task(worker.run(listener, stop)))

    tasks.append(asyncio.create_task(run_reaper(queue, stop)))
-    LOG.i(f"LPS 워커 {concurrency}개 + reaper 기동")
+    LOG.i(f"LPS 워커 {concurrency}개 + reaper 기동 (워커별 브라우저 세트 — 상품 {concurrency}개 동시 검색)")

    try:
        await asyncio.gather(*tasks)
@ -89,7 +101,7 @@ async def main(concurrency: int = 1):
        stop.set()
        for listener in listeners:
            await listener.close()
-        for adapter in list(adapters.values()) + list(fallback_adapters.values()):
+        for adapter in all_adapters:
            await adapter.close()