目的
- locust（負荷試験ツール）を利用し、ローカル環境で負荷試験を実施できる環境を構築する
- 基本的な loucst での負荷試験シナリオ作成ができるようになる
目次
環境構築
- 今回の実行環境は、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 の画面が表示されます
-
設定項目
- Number of users：シナリオを実行するユーザ数の最大値
- ここで指定した分だけユーザが立ち上がる為、ユーザ ID が有限の場合のシナリオを実行する際は、ユーザ ID 数を超えないように調整してください
- Worker が複数の場合、各 Worker 上でこのユーザ数が立ち上がります。その為、ユーザ CSV を共通のものにした場合、同一ユーザ ID のシナリオが実行される為、留意してください。
- Ramp up：1 秒間あたりに増加するユーザ数
- Host：負荷試験対象の URL
- ローカルの Docker に対して負荷をかける場合、http://host.docker.internal を URL として指定してください
- Number of users：シナリオを実行するユーザ数の最大値
実行結果
-
負荷試験を実行すると、下記の画面が表示されます
-
主要な項目
- 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 程度)であれば覚える事も少なくやりやすいかなと思います。