Go to Qiita Advent Calendar Top
LoginSignup
0
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?

@sotaheavymetal21(sota heavymetal21)

[Locust🦗でバッタを跳ばせ] Locust で気軽に負荷テストを実行🚀

Last updated at Posted at 2025-12-04

やりたかったこと

SMS送信システムが1時間に1000件さばけるか検証したくて、Locustで負荷テストを実施してみました。送信処理も履歴取得も両方ね。

「本当に耐えられるの?」っていうのを数字でちゃんと示したかったんですよね。実際の運用前にボトルネックを見つけておきたいし。

環境構成

  • テスト環境: ステージング環境
  • SMS送信: LocalStack(AWS SNS模擬)使用 ← 本番環境の課金を避けるため
  • バックエンド: Django REST Framework + PostgreSQL
  • インフラ: AWS ECS(Fargate)

テストユーザーはpython manage.py loaddataコマンドでフィクスチャから流し込んでます

[
    {
        "model": "accounts.User",
        "pk": 1,
        "fields": {
            "username": "admin",
            "email": "admin@example.com",
            "is_staff": true,
            "is_superuser": true
        }
    },
    {
        "model": "accounts.Profile",
        "fields": {
            "user": 1,
            "employee_id": "EMP001",
            "full_name": "管理者ユーザー"
        }
    }
]

テストシナリオの設計

実際のユーザー行動をシミュレートすることが大事なので、こんな流れにしました

  1. ログイン: 管理者としてログイン
  2. 顧客登録: 新規顧客を登録
  3. トークン生成: SMS送信用のワンタイムトークンを生成
  4. テンプレート登録: SMS送信用のテンプレートを登録
  5. SMS送信: 実際にSMSを送信
  6. 履歴取得: 送信履歴を確認

各ステップの間に2〜5秒のランダムな待機時間を入れることで、実際のユーザー操作に近い負荷をかけられるようにしてます。

実装のポイント

CSRF対応

Django REST FrameworkはAPI専用でもCSRF保護が有効になってるケースがあります。特にSessionAuthenticationを使ってる場合は要注意。

def on_start(self):
    """初回ログイン時にCSRFトークンを取得"""
    res = self.client.post(
        "/api/auth/login",
        json={"username": "admin", "password": "password123"}
    )
    # Cookieからトークン取得
    self._csrf = res.cookies.get("csrftoken")
    # 以降のリクエストで使用するヘッダー
    self._headers = {
        "X-CSRFToken": self._csrf,
        "Referer": "http://localhost:3000"  # RefererもDjangoが検証する
    }

Refererヘッダーも一緒に送るのがポイント。DjangoのCSRF保護はCSRF_TRUSTED_ORIGINSとの照合も行うので。

完全なコード

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

class SMSLoadTest(HttpUser):
    wait_time = between(1, 1)
    
    def on_start(self):
        """テスト開始時に一度だけ実行される"""
        self._login()
    
    def _login(self):
        """ログイン処理"""
        res = self.client.post(
            "/api/auth/login",
            json={"username": "admin", "password": "password123"}
        )
        self._csrf = res.cookies.get("csrftoken")
        self._headers = {
            "X-CSRFToken": self._csrf,
            "Referer": "http://localhost:3000"
        }
    
    def _make_request(self, method, url, **kwargs):
        """共通のリクエスト処理"""
        kwargs.setdefault("headers", self._headers)
        kwargs.setdefault("cookies", {"csrftoken": self._csrf})
        return getattr(self.client, method)(url, **kwargs)
    
    @task
    def sms_flow(self):
        """メインのSMS送信フロー"""
        # 1. 顧客登録
        res = self._make_request(
            "post",
            "/api/customers",
            json={"phone_number": "09012345678"}
        )
        customer_id = res.json()["id"]
        
        # 2. トークン生成
        res = self._make_request(
            "post",
            "/api/tokens",
            json={"customer_id": customer_id}
        )
        token = res.json()["token"]
        
        # 3. テンプレート登録
        res = self._make_request(
            "post",
            f"/api/customers/{customer_id}/templates",
            json={
                "name": "welcome_message",
                "content": "ご登録ありがとうございます。"
            }
        )
        template_id = res.json()["id"]
        
        # 4. SMS送信
        self._make_request(
            "post",
            "/api/sms/send",
            json={
                "phone_number": "09012345678",
                "token": token,
                "customer_id": customer_id,
                "template_id": template_id
            }
        )
        
        # 5. ユーザーの操作を模擬した待機
        time.sleep(random.randrange(2, 5))
        
        # 6. 履歴取得
        self._make_request("get", "/api/sms/history")

リファクタリングのポイント

  • _make_requestメソッドで共通処理を集約
  • エラーハンドリングはLocustが自動でやってくれる
  • タスクの重み付けもできる(今回は単一タスクなので不要)

実行方法

基本的な起動

# Locust起動
locust -f locustfile.py

# ブラウザで http://0.0.0.0:8089 を開く

Web UIで設定するパラメータ

  • Number of users: 3 (同時実行ユーザー数)
  • Ramp Up: 3 (1秒あたりの増加ユーザー数)
  • Host: ホスト

より実践的な実行方法

GUI使わずにコマンドラインで実行することもできます

# ヘッドレスモードで実行
locust -f locustfile.py \
    --headless \
    --users 10 \
    --spawn-rate 2 \
    --run-time 1h \
    --host https://staging.example.com \
    --html report.html

# CSV出力も可能
locust -f locustfile.py \
    --headless \
    --users 10 \
    --spawn-rate 2 \
    --run-time 30m \
    --host https://staging.example.com \
    --csv results

CI/CDパイプラインに組み込む場合はヘッドレスモードが便利です。

ハマったポイントと解決策

1. CSRFトークンの扱い

問題: DRFのAPI専用エンドポイントなのにCSRFエラーが出る

原因: SessionAuthenticationを使っている場合、CSRFチェックが有効になる

解決

# settings.py
REST_FRAMEWORK = {
    'DEFAULT_AUTHENTICATION_CLASSES': [
        'rest_framework.authentication.SessionAuthentication',
    ],
}

# LocustではRefererとCSRFトークンの両方を送信
self._headers = {
    "X-CSRFToken": self._csrf,
    "Referer": "http://localhost:3000"
}

2. LocalStackのSNS設定

問題: SMS送信APIを叩いても実際に送信されない

原因: LocalStackのエンドポイント設定が不正

解決:

# settings/development.py
import boto3

# LocalStack用のSNSクライアント設定
sns_client = boto3.client(
    'sns',
    endpoint_url='http://localstack:4566',  # LocalStackのエンドポイント
    region_name='ap-northeast-1',
    aws_access_key_id='test',
    aws_secret_access_key='test'
)

docker-compose.ymlでのLocalStack設定

services:
  localstack:
    image: localstack/localstack:latest
    ports:
      - "4566:4566"
    environment:
      - SERVICES=sns
      - DEBUG=1
      - DATA_DIR=/tmp/localstack/data

監視すべきメトリクス

負荷テスト中はこれらを監視してました

Locust側

  • Requests/s: 秒あたりのリクエスト数
  • Response Time (ms): 50%, 95%, 99%パーセンタイル
  • Failures: エラー率

AWS ECS側

  • CPU使用率: 80%超えたら要注意
  • メモリ使用率: スワップが発生してないか

PostgreSQL側

  • 接続数: max_connectionsに近づいてないか
  • クエリ実行時間: スロークエリログを確認
  • デッドロック: pg_stat_database.deadlocksをチェック

複数シナリオの実行

異なる負荷パターンをテストしたい場合

class NormalUser(HttpUser):
    wait_time = between(2, 5)
    weight = 8  # 80%のユーザー
    
    @task
    def normal_flow(self):
        # 通常の処理
        pass

class HeavyUser(HttpUser):
    wait_time = between(1, 2)
    weight = 2  # 20%のユーザー
    
    @task
    def heavy_flow(self):
        # 頻繁にアクセスする処理
        pass

まとめ

Locustを使った負荷テストで、SMS送信システムのパフォーマンスを定量的に評価できました。
まだまだ基礎的なところしか触っていませんが、Locust では色々できるようです。

ポイントは

  • 実際のユーザー行動に近いシナリオを作る
  • CSRF対応など、認証周りの実装を正確に
  • 複数のレイヤー(アプリ、DB、インフラ)で監視する
  • LocalStackで課金を気にせずテストできる

みなさんもぜひ試してみてください〜!✨

参考資料

0
0
0

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
  3. You can use dark theme
What you can do with signing up
Sign upLogin
0
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?