Locust + k6로 API 부하 테스트 완전 가이드

부하 테스트란 무엇이고 왜 필요한가

서비스를 배포하고 나면 항상 이런 불안감이 생깁니다. “동시에 1,000명이 접속하면 버틸 수 있을까?” 이 질문에 답하는 가장 확실한 방법이 바로 부하 테스트(Load Testing)입니다. 실제 트래픽을 시뮬레이션해 서버의 성능 한계를 측정하고, 병목 지점을 사전에 찾아내는 과정이죠.

이번 글에서는 Python 기반의 Locust와 JavaScript 기반의 k6, 두 가지 부하 테스트 도구를 실전 FastAPI 서버에 적용하는 방법을 단계별로 정리합니다. 설치부터 시나리오 작성, 결과 분석, 그리고 병목 개선까지 실제 현장에서 활용할 수 있는 내용으로 구성했습니다.

Locust 설치 및 기본 시나리오 작성

Locust는 Python으로 테스트 시나리오를 작성할 수 있어 백엔드 개발자에게 친숙합니다. pip 한 줄로 설치가 완료되고, 웹 UI로 실시간 결과를 확인할 수 있는 것이 큰 장점입니다.

# Locust 설치
pip install locust

# locustfile.py 작성
from locust import HttpUser, task, between

class FastAPIUser(HttpUser):
    wait_time = between(0.5, 2)  # 요청 사이 대기 시간 (초)
    host = "http://localhost:8000"

    def on_start(self):
        """테스트 시작 시 로그인 처리"""
        response = self.client.post("/api/auth/login", json={
            "email": "[email protected]",
            "password": "testpassword"
        })
        if response.status_code == 200:
            token = response.json()["access_token"]
            self.client.headers.update({"Authorization": f"Bearer {token}"})

    @task(3)
    def get_items_list(self):
        """목록 조회 (가중치 3 - 가장 자주 호출)"""
        self.client.get("/api/items?page=1&limit=20")

    @task(1)
    def get_item_detail(self):
        """상세 조회 (가중치 1)"""
        self.client.get("/api/items/1")

    @task(1)
    def create_item(self):
        """생성 요청"""
        self.client.post("/api/items", json={
            "title": "테스트 항목",
            "description": "부하 테스트용 데이터"
        })

작성한 locustfile.py로 테스트를 실행하면 웹 UI(기본 포트 8089)에서 동시 사용자 수와 Spawn Rate를 조절하며 실시간으로 결과를 확인할 수 있습니다. CLI 모드로 실행하면 CI/CD 파이프라인에도 통합할 수 있습니다.

k6로 정밀한 성능 목표 기반 테스트

k6는 Go로 작성되어 성능이 뛰어나고, 시나리오를 JavaScript로 작성합니다. 특히 Thresholds(성능 목표)를 설정해 테스트 성공/실패 여부를 자동 판정할 수 있어 CI 환경에서 유용합니다. Grafana와도 공식 연동을 지원합니다.

// k6-test.js
import http from 'k6/http';
import { check, sleep } from 'k6';
import { Rate } from 'k6/metrics';

const errorRate = new Rate('errors');

// 부하 테스트 시나리오 정의
export const options = {
  stages: [
    { duration: '1m', target: 50 },   // 1분 동안 50명으로 증가
    { duration: '3m', target: 50 },   // 3분 동안 50명 유지
    { duration: '1m', target: 200 },  // 1분 동안 200명으로 급증 (스파이크)
    { duration: '2m', target: 200 },  // 2분 유지
    { duration: '1m', target: 0 },    // 종료
  ],
  thresholds: {
    http_req_duration: ['p(95)<500'],  // 95%ile 응답시간 500ms 이하
    http_req_failed: ['rate<0.01'],    // 에러율 1% 미만
    errors: ['rate<0.05'],
  },
};

const BASE_URL = 'https://api.example.com';

export default function () {
  const listRes = http.get(`${BASE_URL}/api/items?page=1&limit=20`, {
    headers: { Authorization: 'Bearer test-token' },
  });

  check(listRes, {
    '목록 조회 200': (r) => r.status === 200,
    '응답 속도 500ms 이하': (r) => r.timings.duration < 500,
  });

  errorRate.add(listRes.status !== 200);
  sleep(1);
}

FastAPI 응답 시간 로깅으로 병목 찾기

부하 테스트를 실행했다면 이제 결과를 분석해야 합니다. 핵심 지표는 RPS(초당 요청 수), 응답시간 분포(p50/p95/p99), 에러율입니다. p95가 튀기 시작하는 지점이 실질적인 서버 한계선입니다.

FastAPI 서버에서 병목을 찾을 때는 다음 순서로 접근합니다. 먼저 DB 쿼리 시간, 그 다음 외부 API 호출, 마지막으로 CPU/메모리 자원 부족 여부를 확인합니다. 미들웨어로 요청별 응답 시간을 로깅하면 슬로우 엔드포인트를 즉시 파악할 수 있습니다.

# FastAPI 응답 시간 로깅 미들웨어
import time
import logging
from fastapi import FastAPI, Request

app = FastAPI()
logger = logging.getLogger(__name__)

@app.middleware("http")
async def log_request_time(request: Request, call_next):
    start_time = time.time()
    response = await call_next(request)
    duration = (time.time() - start_time) * 1000  # ms

    logger.info(
        f"{request.method} {request.url.path} "
        f"status={response.status_code} "
        f"duration={duration:.1f}ms"
    )
    if duration > 1000:
        logger.warning(f"SLOW: {request.url.path} {duration:.1f}ms")

    response.headers["X-Process-Time"] = f"{duration:.1f}ms"
    return response

# SQLAlchemy 슬로우 쿼리 감지
from sqlalchemy import event
from sqlalchemy.engine import Engine

@event.listens_for(Engine, "before_cursor_execute")
def before_cursor_execute(conn, cursor, statement, parameters, context, executemany):
    conn.info.setdefault("query_start_time", []).append(time.time())

@event.listens_for(Engine, "after_cursor_execute")
def after_cursor_execute(conn, cursor, statement, parameters, context, executemany):
    total = time.time() - conn.info["query_start_time"].pop(-1)
    if total > 0.5:
        logger.warning(f"SLOW QUERY ({total*1000:.0f}ms): {statement[:200]}")

Locust vs k6 - 어떤 도구를 선택해야 할까

두 도구 모두 훌륭하지만 사용 목적에 따라 선택이 달라집니다. Locust는 Python 친화적이고 웹 UI가 직관적이라 초반 탐색적 테스트에 좋습니다. k6는 Go 기반이라 자원 소모가 적고, Thresholds로 합격/불합격 기준을 명확히 정의할 수 있어 CI/CD 파이프라인 통합에 적합합니다. 팀 내 언어 친숙도와 인프라 환경에 맞게 선택하거나, 두 도구를 상황에 따라 병행하는 것도 좋은 전략입니다.

코드벤터는 글로벌 협력 네트워크를 통해 다양한 규모의 서비스를 개발하며 실전 부하 테스트 경험을 쌓아왔습니다. API 성능 최적화, 인프라 설계, 부하 테스트 전략 수립이 필요하다면 코드벤터와 함께하세요. 트래픽이 늘어도 흔들리지 않는 서버를 만들어 드립니다.