目的

locust（負荷試験ツール）を利用し、ローカル環境で負荷試験を実施できる環境を構築する
基本的な loucst での負荷試験シナリオ作成ができるようになる

目次

環境構築
docker-compose.yaml の作成
ちょっとしたシナリオ tips
WebGUI での実行
実行結果
最後に

環境構築

今回の実行環境は、Windows + WSL2 上での実行環境となります。
locust は Docker での構築を行う為、DockerDesktop を使用し WSL2 上で docker コマンドが使えるようになっている状態を前提とします。
今回の記事では、Docker 環境構築については省略しますので、こちらは別途 Windows + WSL2 での Docker 環境構築をご参照ください。

構築する DIR

.
├── docker-compose.yaml
├── locust
│   ├── locustfile.py
│   └── users.csv

docker-compose.yaml の作成

最新の locust イメージファイルを使用し、locust の master と worker のイメージを立ち上げます。
docker-compose.yaml サンプル

version: "3.8"

services:
  locust_master:
    image: locustio/locust # Locustの公式Dockerイメージを使用
    ports:
      - "8089:8089" # Web UI用のポート
    volumes:
      - ./docker/locust/:/mnt/locust # 作業DIRをコンテナにマウント
    command: -f /mnt/locust/locustfile.py --master # locustの実行はWeb UIから行うので、docker compose up でテスト実行はしないようにします

  locust_worker1:
    image: locustio/locust
    volumes:
      - ./docker/locust/:/mnt/locust # 作業DIRをコンテナにマウント
    command: -f /mnt/locust/locustfile.py --worker; WORKER_ID=1 --master-host locust_master # locustfile.py上でWorker単位での処理分岐をする為に、WORKER_IDという識別子を環境変数に設定しておく

worker を複数増やしたい場合は以下のようにサービス名を変えて、docker-compose.yml の定義を増やしてください。

  locust_worker2:
    image: locustio/locust
    volumes:
      - ./docker/locust/:/mnt/locust # 作業DIRをコンテナにマウント
    command: -f /mnt/locust/locustfile.py --worker; WORKER_ID=2 --master-host locust_master # locustfile.py上でWorker単位での処理分岐をする為に、WORKER_IDという識別子を環境変数に設定しておく

locustfile.py の作成

概要
- ユーザ CSV から取り出した ID を元にログインを行うシナリオ
- Worker を複数にした場合、同じユーザ CSV を利用すると同一 ID でのログインが行われるので、環境変数を利用して CSV 読み込みを制御
- シナリオの処理開始時にユーザを 1 件ずつ取り出して処理を行う
サンプルコードは下記になります

from locust import HttpUser, task, between, SequentialTaskSet
from requests.auth import HTTPBasicAuth
import os
import re
import csv
import itertools
from urllib.parse import unquote

# 環境変数からワーカーIDを取得（デフォルトは1）
worker_id = os.getenv('WORKER_ID', '1')

# ワーカーIDに基づいてCSVファイル名を設定
csv_filename = f"users_{worker_id}.csv"

# CSVからユーザリストを読み込む処理
def read_users_from_csv(filename=csv_filename):
    # コンテナ内のマウント先ディレクトリを指定
    base_path = "/mnt/locust"
    file_path = os.path.join(base_path, filename)

    # ユーザリストの読み込み
    user_list = []
    try:
        with open(file_path, 'r', encoding='utf-8') as f:
            reader = csv.DictReader(f)
            for row in reader:
                user_list.append(row)
    except FileNotFoundError:
        print(f"エラー: {filename} が見つかりません。")
        # ファイルがない場合はLocustを終了させる
        exit(1)

    if not user_list:
        print(f"エラー: {filename} が空か、ヘッダーしかありません。")
        exit(1)

    return user_list

# スクリプト読み込み時に一度だけCSVを読み込む
all_users = read_users_from_csv()

# ユーザリストのイテレーター化
user_pool = iter(all_users)

# ユーザリストをループさせたい場合、イテレーターをcycleで作成
# user_pool = itertools.cycle(all_users)

# 負荷試験シナリオ
# ログイン＞ユーザ情報取得APIのように流れを固定する場合、シナリオクラスを用意
class Scenario(SequentialTaskSet):

    # シナリオ単位の変数宣言
    # AccessTokenやCsrfTokenなど
    access_token = None
    csrf_token = None
    user_id = None
    user_credentials = None  # ユーザー情報を格納する変数

    # 認証用のヘッダー生成
    # 認証が必要なAPIを呼び出す際はこのヘッダー関数を通してヘッダーをセット
    def auth_headers(self):
        print(f"auth_headers: User {self.user_id} login successful. csrfToken: {self.csrf_token[:10]}...")
        return {
            "Content-Type": "application/json",
            "Authorization": f"Bearer {self.access_token}",
            "X-CSRF-Token": unquote(self.csrf_token)  # CSRFトークンをヘッダーに含める
        }

    # ログイン用のヘッダー生成
    def login_headers(self):
        return {
            "Content-Type": "application/json"
        }

    # シナリオ開始時に一度だけ実行される処理
    # CSVからのユーザデータ取得など、1回だけ行う処理はここに記載
    def on_start(self):
        try:
            self.user_credentials = next(user_pool)
            print(f"Starting user: {self.user_credentials['user_id']}")
        except StopIteration:
            # user_poolが空の場合に発生するが、cycleを使っているため通常は到達しない
            print("すべてのユーザーを使い切りました。")
            self.environment.runner.quit()


    # ログインAPIを叩き、access_tokenを取得するタスク
    # サンプルとしてログインAPIからはアクセストークンとCSRFトークンが返ってくるものとして記載
    @task
    def login(self):
        self.user_id = self.user_credentials['user_id']

        # エンドポイント（例: /api/login）にPOSTリクエストを送信
        with self.client.get("/auth/debug_stress_test", headers=self.login_headers(), catch_response=True) as response:

            # ステータスコードが200番台でない場合は失敗としてマーク
            if not response.ok:
                response.failure(f"Login failed with status {response.status_code}, {response.json()} ")
                return

            try:
                # レスポンスから次のAPI呼び出し
                json_response = response.json()
                access_token = json_response.get("access_token")
                csrf_token = response.cookies.get("XSRF-TOKEN")

                if access_token:
                    # 成功としてマーク
                    response.success()
                    print(f"User {self.user_id} login successful. accessToken: {access_token[:10]}...")
                    # 取得したトークンを後続のタスクで利用するため、変数に保存
                    self.access_token = access_token
                else:
                    response.failure("access_token not found in response")

                if csrf_token:
                    # 成功としてマーク
                    response.success()

                    # 取得したトークンを後続のタスクで利用するため、変数に保存
                    self.csrf_token = csrf_token
                    print(f"User {self.user_id} login successful. csrfToken: {self.csrf_token[:10]}...")
                else:
                    response.failure("csrf_token not found in response")

            except ValueError:
                response.failure("Invalid JSON received")

    # シナリオ終了時に一度だけ実行される処理
    def on_stop(self):
        print(f"User {self.user_id} is stopping.")

# シナリオの実行ユーザクラス
class ScenarioUser(HttpUser):
    tasks = [Scenario]
    wait_time = between(1, 2)  # 各タスク間の待機時間を1～2秒に設定

ちょっとしたシナリオ tips

HTTP メソッド

■GET
self.client.get("sample/path") as response

■POST
self.client.post("sample/path") as response

■DELETE
self.client.delete("sample/path") as response

■PUT
self.client.put("sample/path") as response

■OPTIONS
self.client.options("sample/path") as response

catch_response=trueの利用

catch_responseをtrueで設定すると、HTTPコードによる自動結果判定が行われない
レスポンスを用いて、結果が正しいか失敗かを記録する必要があります
使用例としては、HTTPコードが404であることを期待するようなケース、HTTPコードが200だがレスポンスパラメータの中身で成功・失敗を判断したいケース等

■成功として記録
response.success()

■失敗として記録!
response.failuer("failure message")

レスポンスの取得

■jsonレスポンスの取得
json_response = response.json()

■jsonのkey指定でデータ取得
access_token = json_response.get("access_token")

■cookieのkey指定でデータ取得
csrf_token = response.cookies.get("XSRF-TOKEN")

■xml/text類（XMLはtextで平文を取得した後に、パースすることで利用可能）
text_response = response.text

POST データの設定

■payloadの作成
payload = {
            "data_list": [
                {"label_no":1, "label_value":1},
                {"label_no":2, "label_value":2},
                {"label_no":3, "label_value":1},
                {"label_no":4, "label_value":1},
                {"label_no":5, "label_value":2},
            ]
        }

■jsonにpayloadを引き渡し
self.client.post("/data/regist", headers=self.auth_headers(), json=payload, catch_response=True) as response

multipart/form-data の値を送信する

#タプル形式でパラメータを作る
image = ("sample.jpg", open("path/sample.jpg", "rb"), "image/jpeg")
name = (None, "sample_name")
payload = {
            "image_file" : image
            "user_name" : name
        }

# files引数にpayloadを引き渡す
# Headerに対してmultipart/form-dataの指定は不要（locust側で自動で付与します）
self.client.post("/upload/form_data", files=payload);

WebGUI での実行

本手順だと localhost の port:8089 で GUI が立ち上がります。
http://localhost:8089/ にアクセス
下記のように locust の画面が表示されます
設定項目

実行結果

負荷試験を実行すると、下記の画面が表示されます
主要な項目
- STATISTICS：各 URL の実行結果、リクエスト数や失敗数、レスポンスタイムの中央値や平均値が出力されます
- CHARTS：リクエストに対する成否のグラフ、レスポンスタイムググラフ、実行ユーザ数の経過グラフが出力されます
- FAILURES：locustfile.py 内で response.failure を出力すると、ここに結果が記録されます。
- DOWNLOAD DATA：実行結果の CSV の DL などが行えます
locustfile.py 内の print は Docker のコンテナ側のログで確認できます。Windows であれば Docker Desktop の Logs で確認できるので、適宜確認してください。

最後に

Docker で簡単に立ち上げて実行できるのがありがたいです。
必要なセットアップさえ覚えてしまえば、環境立上げ・環境破棄ともに気軽に行えます。
コードベースのシナリオ作成になるので、JMeter の GUI でシナリオを組む工程と比較して、やりやすいかは人それぞれだと思いますが、シンプルな処理(単純な get/post 程度)であれば覚える事も少なくやりやすいかなと思います。