o2o-negosium-original/lps/loadtest.py

"""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())