locust-best-practices

Locust シナリオビルダー

Python で Locust のロードテストスクリプトに一貫した本番環境対応のパターンを適用します。現実的なユーザー動作、カスタム負荷形状、適切なレスポンス検証を含む HTTP/REST API を対象とします。

出力形式

locustfile を作成または修正する場合は、必ず次の 3 点を提示してください。

1. 完全に実行可能な locustfile — 部分的なスニペットは提示しないでください。
2. 必要なすべてのフラグ（--headless、-u、-r、-t、--host）を含む正確な実行コマンド。
3. 選択した wait_time 戦略と、それが負荷目標に適している理由を説明する1 行の説明。

---

Step 1 — コンテキストを確認する

不明な点だけを質問してください。

• 目標： 新しい locustfile · 既存の locustfile の修正 · 特定の API に関する質問
• プロトコル： HTTP/REST（デフォルト） · WebSocket · gRPC
• ワークロード： 固定の同時ユーザー数、または固定の到着レート（RPS）？
• 認証： なし · 起動時にログイン · 環境変数からのトークン？
• データ： 静的 · CSV ファイル · ユーザーごとに生成？

ユーザーがカスタム負荷プロファイル、複数段階のランプアップ、スパイクのシミュレーション、または LoadTestShape について尋ねた場合に限り、references/SHAPES.md を読み込んでください。

ユーザーが WebSocket、gRPC、または HTTP 以外のプロトコルのテストについて言及した場合に限り、references/PROTOCOLS.md を読み込んでください。 カスタムクライアントラッパー、events.request.fire() のパターン、プロトコルの互換性に関する注意事項が含まれています。

ユーザーがプロジェクト構成、複数のユーザークラス、TaskSet、共通の認証ヘルパー、大規模な CSV データ読み込み、モジュール化された locustfile、または複数ファイルの構成について尋ねた場合に限り、references/DESIGN-PATTERNS.md を読み込んでください。

ユーザーがヘッドレスモード、CI/CD 統合、終了コード、SLA に基づくビルド失敗、分散テスト、マスター／ワーカーのセットアップ、複数コア、ファイルへのログ出力、またはカスタムログレベルについて尋ねた場合に限り、references/RUNNING.md を読み込んでください。

---

Step 2 — 4 ブロックパターンを適用する

すべての locustfile は次の構成に従う必要があります。まず完全なスケルトンを生成し、その後に詳細を埋めてください。

Block 1 → Imports + config    locust imports, constants, env vars
Block 2 → User class          HttpUser with host, wait_time, on_start, @task methods
Block 3 → Shape (optional)    LoadTestShape for custom ramp profiles
Block 4 → Run command         locust CLI with all flags as a comment at the bottom

---

Step 3 — HttpUser のルール

3.1 検証には必ず catch_response=True を使用する

HTTP ステータスだけに頼らないでください。必ずレスポンス本文を検証してください。

# Wrong — only checks HTTP status, misses business logic errors
self.client.get("/api/users")

# Correct — validates response content
with self.client.get("/api/users", catch_response=True) as response:
    if response.status_code != 200:
        response.failure(f"Expected 200, got {response.status_code}")
    elif "users" not in response.json():
        response.failure("Missing 'users' key in response")

3.2 動的 URL は name= でグループ化する

グループ化しないと、固有の URL ごとに個別の統計エントリーが作成され、レポートが埋め尽くされます。

# Wrong — creates N separate stats rows
for user_id in range(100):
    self.client.get(f"/user/{user_id}")

# Correct — groups all under one stats entry
for user_id in range(100):
    self.client.get(f"/user/{user_id}", name="/user/[id]")

3.3 認証には on_start を使用する — ユーザー間でトークンを共有しない

各仮想ユーザーは個別に認証する必要があります。

# Wrong — shared token is a race condition and not realistic
shared_token = None

class MyUser(HttpUser):
    def on_start(self):
        global shared_token  # ❌ shared state across users

# Correct — each user gets its own token
class MyUser(HttpUser):
    def on_start(self):
        response = self.client.post("/auth/login", json={
            "username": os.environ["TEST_USER"],
            "password": os.environ["TEST_PASSWORD"],
        })
        self.token = response.json()["access_token"]

    @task
    def get_profile(self):
        self.client.get("/profile", headers={"Authorization": f"Bearer {self.token}"})

3.4 負荷目標に基づいて wait_time を選択する

目標	wait_time	理由
思考時間を設けて実際のユーザーをシミュレートする	between(1, 5)	現実的なペース
レイテンシにかかわらず固定 RPS を維持する	constant_throughput(2)	オープンモデル — ユーザー 1 人あたり毎秒 2 タスク
正確なユーザーペースを再現する	constant(1)	固定間隔

wait_time = constant(0) は絶対に使用しないでください — 思考時間をゼロにすると、実際のユーザーの 10～100 倍の負荷が発生し、非現実的な結果になります。

3.5 @task の重みで実際のトラフィック分布を反映する

class ShopUser(HttpUser):
    wait_time = between(1, 3)

    @task(10)   # 10× more common — most users just browse
    def browse_products(self):
        self.client.get("/products")

    @task(3)    # 3× — some users search
    def search(self):
        self.client.get("/search?q=shoes", name="/search")

    @task(1)    # 1× — few users actually purchase
    def checkout(self):
        self.client.post("/checkout", json={"cart_id": self.cart_id})

---

Step 4 — 標準的な Locustfile テンプレート

import os
import json
from locust import HttpUser, task, between, events

# ---------------------------------------------------------------------------
# Block 1 — Config
# ---------------------------------------------------------------------------
BASE_URL = os.environ.get("TARGET_HOST", "http://localhost:8080")

# ---------------------------------------------------------------------------
# Block 2 — User class
# ---------------------------------------------------------------------------
class ApiUser(HttpUser):
    host = BASE_URL
    wait_time = between(1, 3)

    def on_start(self):
        """Runs once per virtual user before any task executes."""
        with self.client.post(
            "/auth/login",
            json={
                "username": os.environ["TEST_USER"],
                "password": os.environ["TEST_PASSWORD"],
            },
            catch_response=True,
        ) as response:
            if response.status_code == 200:
                self.token = response.json()["access_token"]
            else:
                response.failure(f"Login failed: {response.status_code}")
                self.token = None

    def _auth_headers(self):
        return {"Authorization": f"Bearer {self.token}"}

    @task(3)
    def get_items(self):
        with self.client.get(
            "/api/items",
            headers=self._auth_headers(),
            catch_response=True,
        ) as response:
            if response.status_code != 200:
                response.failure(f"GET /api/items failed: {response.status_code}")

    @task(1)
    def create_item(self):
        with self.client.post(
            "/api/items",
            json={"name": "test-item", "value": 42},
            headers=self._auth_headers(),
            catch_response=True,
        ) as response:
            if response.status_code not in (200, 201):
                response.failure(f"POST /api/items failed: {response.status_code}")

# ---------------------------------------------------------------------------
# Block 3 — Events (optional)
# ---------------------------------------------------------------------------
@events.test_start.add_listener
def on_test_start(environment, **kwargs):
    print("Test started — verifying environment variables...")
    for var in ["TEST_USER", "TEST_PASSWORD"]:
        if not os.environ.get(var):
            raise ValueError(f"Missing required env var: {var}")

# ---------------------------------------------------------------------------
# Run command (headless):
# TEST_USER=alice TEST_PASSWORD=secret locust -f locustfile.py \
#   --headless -u 50 -r 5 -t 5m --host http://your-api.com \
#   --html report.html --csv results
# ---------------------------------------------------------------------------

---

Step 5 — よくある間違い

1. catch_response=True がない — エラーが見逃される

これがないと、有効な HTTP 接続で返された 500 レスポンスも、Locust の統計では成功としてカウントされます。必ず catch_response=True を使用し、response.failure() を明示的に呼び出してください。

2. 動的 URL に name= が指定されていない

URL パスが固有であるたびに、統計テーブルに新しい行が作成されます。/user/123、/user/456 のようなパラメーター付きパスは、レポートを膨大にします。name="/user/[id]" を使用してください。

3. ユーザー間で変更可能な状態が共有されている

Locust のユーザーはスレッドではなく greenlet（gevent）で実行されますが、共有状態はロジック上の競合状態を引き起こす可能性があります。ユーザーごとの状態（self.token、self.cart_id）は、必ずインスタンス属性として保存してください。

CSV から認証情報を読み込む場合は、1 つのフィールドだけでなく、すべてのフィールドを on_start でインスタンス属性として保存してください。

# Wrong — only stores username, password is lost
self.username = row["username"]

# Correct — store every field the user will need
self.username = row["username"]
self.password = row["password"]

4. wait_time = constant(0) または wait_time を指定しない

思考時間を省略すると非現実的な負荷が発生し、対象システムより先にテストマシンが飽和します。必ず適切な wait_time を定義してください。

5. ホストと認証情報のハードコード

host、ユーザー名、パスワードをハードコードしないでください。必ず環境変数を使用してください。

host = os.environ.get("TARGET_HOST", "http://localhost:8080")

6. レスポンスボディを確認せず、ステータスコードだけを検証する

{"error": "token expired"} が返されても、ステータスコードだけを確認するテストでは 200 が合格となります。ビジネスロジックが正しいことを確認するため、必ずレスポンスボディを検証してください。

7. トレードオフを理解せずに FastHttpUser を使用する

FastHttpUser は requests の代わりに geventhttpclient を使用します。より高速ですが、requests のすべての機能（Cookie Jar、一部の認証ヘルパーなど）をサポートしているわけではありません。対象システムではなく、テストランナー自体がスループットのボトルネックになっていることを確認できた場合にのみ使用してください。

---

ステップ 6 — コマンド実行リファレンス

# Basic headless run — 50 users, 5/sec ramp, 5 minutes
locust -f locustfile.py --headless -u 50 -r 5 -t 5m --host https://api.example.com

# With HTML report and CSV output
locust -f locustfile.py --headless -u 100 -r 10 -t 10m \
  --host https://api.example.com \
  --html report.html --csv results

# Run specific user class only
locust -f locustfile.py --headless -u 50 -r 5 -t 5m \
  --host https://api.example.com ApiUser

# With environment variables
TARGET_HOST=https://api.example.com \
TEST_USER=alice \
TEST_PASSWORD=secret \
locust -f locustfile.py --headless -u 50 -r 5 -t 5m

# Open web UI (no --headless) — useful for interactive exploration
locust -f locustfile.py --host https://api.example.com

フラグのリファレンス：

フラグ	意味
-u	同時に実行するユーザー数の最大値
-r	起動レート — 1 秒あたりに追加されるユーザー数
-t	実行時間（例：30s、5m、2h）
--headless	Web UI なしで実行し、自動的に終了する
--host	対象のベース URL
--html	指定したパスに HTML レポートを生成する
--csv	指定したプレフィックスで CSV 統計ファイルを生成する
--only-summary	インターバルごとの統計出力を抑制する