81 lines
2.7 KiB
Python
81 lines
2.7 KiB
Python
"""Negodata 인증 서버 부하 테스트.
|
|
|
|
실행:
|
|
pip install locust
|
|
locust -f loadtest/locustfile.py --host http://localhost:9400
|
|
# 웹 UI: http://localhost:8089 에서 사용자 수/spawn rate 입력
|
|
|
|
# 헤드리스(CI) 예시 - 100 VU, 10/s 증가, 2분:
|
|
locust -f loadtest/locustfile.py --host http://localhost:9400 \
|
|
--headless -u 100 -r 10 -t 2m
|
|
|
|
주의:
|
|
- /login 은 bcrypt 검증(CPU 바운드)이 들어가 가장 무겁다. RPS 가 낮으면
|
|
거의 확실히 bcrypt cost 가 병목이다 (DB 아님).
|
|
- 부하를 올리며 서버 측에서 PostgreSQL 커넥션 수를 함께 모니터링하라:
|
|
SELECT count(*) FROM pg_stat_activity;
|
|
(pool_size + max_overflow) x 2(R/W) x 워커수 가 max_connections 를 넘으면
|
|
max_connections 초과 시 실패한다.
|
|
"""
|
|
|
|
import random
|
|
|
|
from locust import HttpUser, between, events, task
|
|
|
|
|
|
# 각 시뮬레이션 유저는 고유 계정을 만들어 로그인 흐름을 반복한다.
|
|
class AuthUser(HttpUser):
|
|
wait_time = between(0.5, 2.0)
|
|
|
|
def on_start(self):
|
|
# 유저별 고유 계정 생성 후 1회 로그인하여 토큰 확보.
|
|
self.user_id = f"load_{random.randint(0, 1_000_000_000)}"
|
|
self.password = "pw1234"
|
|
self.token = None
|
|
|
|
self.client.post(
|
|
"/v1/auth/create",
|
|
json={"id": self.user_id, "pw": self.password, "nickname": "load"},
|
|
name="POST /v1/auth/create",
|
|
)
|
|
self._login()
|
|
|
|
def _login(self):
|
|
with self.client.post(
|
|
"/v1/auth/login",
|
|
json={"id": self.user_id, "pw": self.password},
|
|
name="POST /v1/auth/login",
|
|
catch_response=True,
|
|
) as resp:
|
|
if resp.status_code == 200 and resp.json().get("result", {}).get("success"):
|
|
self.token = resp.json().get("access_token")
|
|
resp.success()
|
|
else:
|
|
resp.failure(f"login failed: {resp.status_code} {resp.text[:120]}")
|
|
|
|
@task(5)
|
|
def me(self):
|
|
# JWT 검증만 하는 경량 경로 (DB 無). bcrypt 경로와 처리량 비교용.
|
|
if not self.token:
|
|
return
|
|
self.client.get(
|
|
"/v1/auth/me",
|
|
headers={"Authorization": f"Bearer {self.token}"},
|
|
name="GET /v1/auth/me",
|
|
)
|
|
|
|
@task(2)
|
|
def login(self):
|
|
# bcrypt + DB write 가 포함된 무거운 경로.
|
|
self._login()
|
|
|
|
@task(1)
|
|
def healthz(self):
|
|
# 베이스라인 (앱 오버헤드 측정).
|
|
self.client.get("/healthz", name="GET /healthz")
|
|
|
|
|
|
@events.test_start.add_listener
|
|
def _on_start(environment, **kwargs):
|
|
print("부하 테스트 시작 - PostgreSQL 커넥션 수 모니터링 권장 (pg_stat_activity)")
|