diff --git a/lps/loadtest.py b/lps/loadtest.py new file mode 100644 index 0000000..ab53050 --- /dev/null +++ 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()) diff --git a/lps/services/search/proxy.py b/lps/services/search/proxy.py index bbad8f3..977c1c4 100644 --- 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 diff --git a/lps/worker_main.py b/lps/worker_main.py index 1632aa0..ebcbb39 100644 --- 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()