a63793daf0
API(enqueue/조회, asyncpg I/O 바운드) 부하 테스트로 멀티코어 활용·한계 측정. - loadtest/locustfile.py: enqueue(고유코드 write)+조회 가중 부하. - loadtest/bench_multicore.sh: PROCESS_COUNT 1→N 자동 비교(--processes 로 부하생성기도 멀티프로세스). - server_configs: PROCESS_COUNT / DB_POOL_SIZE / DB_MAX_OVERFLOW env override(코드·toml 수정 없이 튜닝). - loadtest/README.md: 발견 문서화. 발견(11코어·1500users): 기본 풀(10/20)로 워커 늘리면 실패 폭증(1w=0 → 4w=10336) — (pool+overflow)×2엔진×workers=240 > PG max_connections=100 커넥션 고갈(SQLAlchemy pool checkout 실패). 증명: 풀 8/4(96<100)로 PC=4 재실행 → RPS 1336→2705(2배), 실패 0. 코드는 멀티코어 활용 가능, 막는 건 풀 오버서브스크립션. 규칙: (pool+overflow)×2×process_count ≤ max_connections. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
64 lines
2.9 KiB
Python
64 lines
2.9 KiB
Python
"""LPS **API 서버** 부하 테스트 (enqueue/조회 경로).
|
||
|
||
부하 대상은 API(web_main) 다 — 워커(브라우저 크롤)는 프록시/브라우저에 묶여 처리량이 결정되므로
|
||
Locust 대상이 아니다. API 는 요청을 받아 job 테이블에 적재만 하고 즉시 응답한다(asyncpg I/O 바운드).
|
||
|
||
측정 목적
|
||
1) **멀티코어 활용**: API 는 asyncio(스레드 1개)라 단일 프로세스=단일 코어. uvicorn workers(=process_count)
|
||
를 1→N 으로 올리며 RPS 가 스케일하는지 본다. (bench_multicore.sh 가 자동 비교)
|
||
2) **부하 한계**: enqueue 는 job write(+dedupe unique index). 한계는 대개 DB(커넥션 풀·write 경합).
|
||
(pool_size+max_overflow)×2엔진×workers 가 PG max_connections 를 넘으면 거기서 막힌다.
|
||
|
||
⚠️ **워커는 끄고** 실행하라(부하 중 실제 크롤=프록시/AI 비용). 잡은 PENDING 으로 쌓였다가 끝나면 정리:
|
||
psql -h 127.0.0.1 -U postgres -d lps_db -c "TRUNCATE job;"
|
||
-- 부하 중 커넥션 관측: SELECT count(*) FROM pg_stat_activity WHERE datname='lps_db';
|
||
|
||
실행
|
||
locust -f loadtest/locustfile.py --host http://localhost:9600 # 웹 UI(:8089)
|
||
locust -f loadtest/locustfile.py --host http://localhost:9600 --headless -u 200 -r 20 -t 2m
|
||
"""
|
||
|
||
import random
|
||
|
||
from locust import HttpUser, between, task
|
||
|
||
|
||
class LpsApiUser(HttpUser):
|
||
# 실제 클라이언트처럼 짧게 쉬며 반복(과도한 wait 없이 API 한계를 본다)
|
||
wait_time = between(0.05, 0.3)
|
||
|
||
def on_start(self):
|
||
self.last_job = None
|
||
|
||
@task(6)
|
||
def enqueue_search(self):
|
||
# 고유 product_code → 실제 INSERT(활성 중복 dedupe 회피). 부하의 핵심 write 경로.
|
||
code = f"LOAD-{random.randint(0, 2_000_000_000)}"
|
||
body = {"data": [{"product_code": code, "product_name": "부하테스트 상품",
|
||
"specification": "1박스", "job_type": "batch"}]}
|
||
with self.client.post("/v1/lps/search", json=body, name="POST /search", catch_response=True) as r:
|
||
if r.status_code == 200 and r.json().get("accepted", 0) == 1:
|
||
self.last_job = (r.json().get("items") or [{}])[0].get("job_id")
|
||
r.success()
|
||
else:
|
||
r.failure(f"{r.status_code} {r.text[:120]}")
|
||
|
||
@task(3)
|
||
def poll_job(self): # 접수 후 상태 폴링(DB read)
|
||
if not self.last_job:
|
||
return
|
||
with self.client.get(f"/v1/lps/jobs/{self.last_job}", name="GET /jobs/{id}", catch_response=True) as r:
|
||
r.success() if r.status_code == 200 else r.failure(f"{r.status_code}")
|
||
|
||
@task(1)
|
||
def queue_stats(self):
|
||
self.client.get("/v1/lps/queue/stats", name="GET /queue/stats")
|
||
|
||
@task(1)
|
||
def ops(self):
|
||
self.client.get("/v1/lps/ops", name="GET /ops")
|
||
|
||
@task(1)
|
||
def readyz(self):
|
||
self.client.get("/readyz", name="GET /readyz")
|