Locust로 부하 테스트하기
현재 진행하고 있는 프로젝트에서 최근 부하 테스트를 담당하게 되었습니다. 테스트를 위한 도구는 여러 가지가 있지만 저희는 빠르고 간단하게 사용할 수 있는 Locust를 사용하기로 했습니다.
생각보다 Locust 사용 사례가 많지 않아서, 이 과정에서 겪었던 과정들과 문제들을 정리해 보려고 합니다.
간단한 사용법
Locust는 Python 코드를 작성할 수 있다면 간단하게 사용할 수 있습니다.
Locust에서 중요한 파일은 크게 2가지입니다.
locust.conf: 테스트 환경에 대한 설정 파일입니다.locustfile.py: 테스트 코드를 작성하는 파일입니다.
먼저 locust.conf 파일은 다음과 같은 형태로 작성합니다.
(주석은 실제 사용할 때는 지우면 됩니다.)
locustfile = locustfile.py # 실행할 파일
headless = true # 헤드리스 모드 사용 (웹 인터페이스 없음)
expect-workers = 10 # 워커 수
users = 10 # 최대 동시 사용자 수
spawn-rate = 2 # 초당 생성되는 사용자 수
run-time = 5m # 테스트 실행 시간
loglevel=INFO
# html = "output/sample.html"
# csv = "output/sample.csv"
# logfile = "output/sample.log"
host = "https://www.example.com"
# host = "http://app.sample-namespace.svc.cluster.local:8000"
다른 설정은 상황에 따라 자유롭게 변경하면 되고, 중요한 부분은 host 부분입니다.
이 부분에는 테스트를 진행할 서버의 URL을 입력합니다.
외부에서 서버에 접근할 수 있는 상황이라면, 접근할 수 있는 URL을 입력하면 됩니다.
만약 그렇지 않다면 내부에서 접근 가능한 값을 설정해야 합니다.
다음으로, locustfile.py 파일에는 테스트 코드를 작성합니다.
from locust import HttpUser, between, task
class TestApi(HttpUser):
wait_time = between(1, 2)
@task
def test_api(self):
self.client.get("/hello")
아주 간단하게 작성한 예시입니다. 지금은 단순 GET 요청이지만, 파라미터를 추가하거나 파일을 업로드하고, 추가 로직을 작성하거나 로그를 추가할 수도 있습니다.
wait_time 부분은 사용자가 요청을 보내는 시간 간격을 의미합니다.
예시에서는 1~2초 사이의 무작위 간격을 가지도록 설정한 것입니다.
부하 테스트에서 겪은 문제들
위에서 작성한 것만 보면 Locust 사용법은 간단합니다.
하지만 실제로 적용할 때는 몇 가지 문제가 있었습니다.
내부 폐쇄망에서 테스트하기
제가 테스트해야 하는 API는 격리된 Kubernetes 환경에 구성되어 있었습니다.
당연히 내부망에서만 API에 접근할 수 있었고, 또한 클러스터 외부에서는 개발 환경을 구성할 수 없었습니다.
이 조건에서 테스트를 진행하기 위해 Locust 환경 전체를 이미지로 만들고, 그 이미지를 Pod로 배포해 테스트를 진행했습니다. 컨테이너 내부에서는 Kubernetes 서비스와 포트를 통해 호출이 가능했고, 그 외에는 비슷한 방식으로 테스트를 진행했습니다.
원하는 결과를 파일로 저장하기
Locust는 기본적으로 CSV와 HTML 형식으로 결과를 저장합니다.
하지만 이 파일들은 응답 시간과 성공률 정도만 표시하고 있고, 추가로 원하는 정보가 있다면 추가 작업이 필요합니다.
또한, 이 테스트 파일을 입맛대로 수정할 수 있는 방법이 마땅치 않았습니다.
여러 방면으로 조사한 결과, Locust의 Event hooks를 사용하면 가능하다는 것을 알게 되었습니다.
다만 공식 예제와 달리 실제로는 CSV 파일을 따로 저장해야 했고,
중간값이나 분위수 등의 정보도 계산해야 했습니다.
그래서 전역으로 리스트 변수를 선언하고, 요청이 완료될 때마다 데이터를 추가한 다음
테스트 종료 이벤트에서 최종 파일을 저장하는 방식으로 구현했습니다.
공식 예제에서도 전역 변수를 사용했고, 이 정도의 로직이 부하 테스트에 큰 영향을 미치지는 않았습니다.
예시 코드는 다음과 같습니다.
from typing import Generator
from locust import HttpUser, task, between, events
from locust.runners import MasterRunner
from datetime import datetime
import logging
# import csv
# import os
# import numpy as np
API_KEY = "------"
stats = []
class TestApi(HttpUser):
wait_time = between(1, 2)
@task
def test_api(self):
data = {
# ...
}
headers = {"content-type": "application/json", "API-KEY": API_KEY}
res = self.client.post(
"/foo/bar",
json=data,
headers=headers
)
# Add the data to the stats
@events.test_stop.add_listener
def on_test_stop(environment, **kwargs):
if not isinstance(environment.runner, MasterRunner):
# CSV file processing
logging.info(f"테스트가 완료되었습니다.")
else:
logging.info("Cleaning up test data")
그 외
- Locust는 스트리밍 응답을 공식적으로 지원하지 않습니다. 이슈를 보면 클라이언트를 직접 작성하라고 되어 있는데, 저의 경우 일반 API 호출로 해결을 했습니다. 스트리밍 테스트가 필요하다면, 다른 도구를 생각해야 할 수도 있을 것 같습니다.
- 당연하지만 Locust 명령어를 통해서도 설정값을 지정해 테스트를 진행할 수 있습니다. 저의 경우에도 여러 조건에서 테스트를 진행해야 했는데, 내부망에서는 코드 작업이 쉽지 않기 때문에 미리 스크립트를 작성해 두고 명령어를 통해 테스트 조건을 지정하는 방식으로 작업을 진행했습니다.
마치며
돌아보면 지금까지 테스트다운 테스트를 해 본 적이 많지 않았던 것 같습니다. 그래서 더 제대로 하고 싶은 마음도 있었고요.
테스트 방법을 정립하기 위해서 저도 많은 시행 착오를 거쳤지만, API 자체에서 시간이나 데이터를 얻기 위한 설정도 필요했고, 팀원들이 관련 작업을 지원해 주었습니다.
또한 방법도 방법이지만, 다른 프로젝트에서도 비슷한 방법으로 부하 테스트를 진행할 수 있도록 기준을 정하고, 정리하는 것이 쉽지 않았던 것 같습니다.
테스트를 수행하는 과정이 순탄하지는 않았지만, 기한 내에 잘 마무리할 수 있었습니다.
앞으로도 테스트를 포함해 많은 작업을 진행하게 될 텐데, 이 과정에서 정돈된 프로세스를 만들어 나갈 수 있도록 하는 것이 목표입니다.