[Locust] 웹 서버 부하 테스트 하기 (Feat. python, gcp)

Locust 란?

Locust는 웹 애플리케이션이나 API의 부하(Load)와 스트레스 테스트(Stress Test)를 수행하기 위해 만들어진 Python 기반의 오픈소스 프레임워크이다. 수천~수만 명의 사용자가 동시에 웹사이트를 사용하거나 API를 호출했을 때, 시스템이 어떻게 반응하는지, 얼마나 견디는지, 어디서 병목이 발생하는지 등을 확인할 수 있는 도구이다. 

• 사용자가 1초에 몇 명씩 접속할지, 총 몇 명이 활동할지, 어떤 행동을 반복할지 모두 프로그래밍적으로 제어 가능
• 로그인 → 게시글 조회 → 댓글 작성처럼 실제 플로우를 재현 가능
• HTTP뿐만 아니라 WebSocket, gRPC, GraphQL 등도 확장 플러그인으로 테스트 가능
• Python 코드이기 때문에 기존 코드, 설정값, 외부 API 호출과도 자연스럽게 통합 가능

 

이번 부하 테스트를 진행하게 된 배경을 간단하게 소개하자면, GCP에 새롭게 배포한 웹 서버에 동시에 약 100명의 사용자가 접속해 서비스를 이용하는 상황이다. 특히 이번 프로젝트는 한국이 아닌 타국 출장에서 서비스가 운영될 예정이었기 때문에, 상대적으로 불안정할 수 있는 네트워크 환경을 고려한 테스트가 필수적이었다.

본 테스트는 100명의 사용자가 동시에 로그인하고, 이미지를 포함한 페이지를 요청하는 상황을 가정하여 시나리오를 구성하였다.

 

 

1. Locust 설치

Mac OS 기준! 터미널에 명령어를 쳐서 Locust 를 설치해준다.

pip install locust

 

패키지가 잘 깔렸는지 확인!

locust -V

 

 

2. 로그인 → 인증 요청 흐름 테스트 코드 작성

(JWT, Bearer Token 사용)

 

코드는 실행 API에 따라 많이 달라질 것 같아, 아래에 더 자세한 설명이 있습니다! 

from locust import HttpUser, task, between
import csv
import random

class ProjectPageUser(HttpUser):
    wait_time = between(1, 2)

    users = []
    with open("[csv파일]/[경로]/mong/user.csv", newline='') as csvfile:
        reader = csv.DictReader(csvfile)
        users = [row for row in reader]

    def on_start(self):
        user = random.choice(self.users)
        self.email = user["email"]
        password = user["password"]

        response = self.client.post("/[요청]/[경로]v1/login", json={
            "email": self.email,
            "password": password
        })

        print(f"[LOGIN] {self.email} => {response.status_code}")
        if response.status_code != 200:
            print(f"[FAIL] 로그인 실패: {self.email} | {response.text}")

    @task
    def visit_project_page(self):
        endpoints = [
            "/[요청]/[경로]/case/1",
            "/[요청]/[경로]/case/2",
             "/[요청]/[경로]/case/3"
        ]

        for url in endpoints:
            with self.client.get(url, catch_response=True) as response:
                if response.status_code == 200:
                    print(f"[OK] {url} | {self.email}")
                    response.success()
                else:
                    print(f"[FAIL] {url} | {self.email} | {response.status_code}")
                    response.failure(f"실패: {response.status_code}")

        self.stop(True)

 

 

 

3. CSV 사용자 계정 파일 (user.csv 예시)

email,password
abc@abc.com abcabc!
abcd@abdc.com abcdabcd!
...

 

 

4. pyhton 코드 실행

locust -f locust.py --host=https://{domain 주소} --headless -u 100 -r 100 -t 30s

 

이렇게 요청의 실패율, 평균, 최소, 최고, 중간 요청 시간 등을 볼 수 있다!!!

 

 

 

클래스 정의

class ProjectPageUser(HttpUser): 
	wait_time = between(1, 2)

• ProjectPageUser: 테스트할 가상의 사용자 역할
• wait_time = between(1, 2): 각각의 작업 사이에 1~2초 랜덤한 대기 시간을 두어 현실적인 사용자 행동을 시뮬레이션을 위함.

 

사용자 CSV 파일 로딩

    users = []
    with open("[csv파일]/[경로]/mong/user.csv", newline='') as csvfile:
        reader = csv.DictReader(csvfile)
        users = [row for row in reader]

• user.csv: 사전에 생성한 사용자 목록임. 예를 들어 100명의 이메일/비밀번호 조합이 담긴 CSV!
• csv.DictReader: 각 줄을 딕셔너리 형태로 읽어옴. ({"email": ..., "password": ...} 형태)
• users는 클래스 변수로 정의되어, 테스트 사용자마다 파일을 반복해서 읽지 않고 한 번만 읽어서 공유함.

 

로그인 시나리오 (on_start)

    def on_start(self):
        user = random.choice(self.users)
        self.email = user["email"]
        password = user["password"]

        response = self.client.post("/[요청]/[경로]v1/login", json={
            "email": self.email,
            "password": password
        })

        print(f"[LOGIN] {self.email} => {response.status_code}")
        if response.status_code != 200:
            print(f"[FAIL] 로그인 실패: {self.email} | {response.text}")

• on_start(): Locust가 유저를 시작할 때 자동으로 호출되는 메서드
• 랜덤 유저 한 명을 뽑아 로그인 요청을 /v1/login 엔드포인트로 보냄.
• 여기서 세션이나 토큰이 잘 오고 있는지 확인할 수 있는 코드도 추가할 수 있겠져...?! 
• response.status_code == 200일 경우 성공으로 간주하지만, 이 토큰을 저장하지 않기 때문에 이후 인증이 필요한 경우에는 실패 가능성 있음

 

메인 Task - 프로젝트 페이지 방문

    @task
    def visit_project_page(self):
        endpoints = [
            "/[요청]/[경로]/case/1",
            "/[요청]/[경로]/case/2",
            "/[요청]/[경로]/case/3"
        ]

• 이 task는 각 유저가 case/1, case/2, case/3 페이지를 순차적으로 방문하는 것을 시뮬레이션함.

        for url in endpoints:
            with self.client.get(url, catch_response=True) as response:
                if response.status_code == 200:
                    print(f"[OK] {url} | {self.email}")
                    response.success()
                else:
                    print(f"[FAIL] {url} | {self.email} | {response.status_code}")
                    response.failure(f"실패: {response.status_code}")

 

• 각 요청에 대해 성공/실패 여부를 콘솔에 출력함.

 

테스트 종료

        self.stop(True)

• 각 유저는 위 3개 페이지를 다 방문한 뒤 테스트를 중단함.
• self.stop(True)는 해당 유저의 테스트를 즉시 종료하게 함.(즉, 한 번만 동작하고 종료됨).

 

 

회고 : 처음에는 로드 밸런싱을 통해 두 개의 백엔드 서버로 부하를 분산하였지만, 이미지 로드가 간헐적으로 불가능한 경우가 발생하였다.. (이건 또 따로 포스팅을 하도록 하겠다 ㅠㅠ) 결국 백엔드 서버 로직을 고치지 않는 이상, 다중 백엔드 구성이 불가능하다고 판단되어, 결국 하나의 백엔드 서버만을 사용하는 방식으로 결정하였다. 테스트 초기에는 로그인 요청의 평균 응답 시간이 3초 이상으로 측정되어, 성능 개선을 위해 서버의 코어 수를 확장하는 방향으로 대응하였다. 😢 출장 날짜가 얼마 남지 않아 그냥 하드웨어를 업그레이드 시켰지만, 돌아오면 백엔드 로직 개선이나 부하를 더 분산할 수 있는 방법을 조사해봐야겠다.