検証用バックエンド実装

負荷テストの対象となる簡易REST APIを実装します。以下は商品検索とショッピングカートへのアイテム追加を処理するサンプルコードです。元の構造を再編成し、エンドポイント名やデータ形式を変更して実際のシステムに近い設計としています。

from flask import Flask, jsonify, request

flask_app = Flask(__name__)

@flask_app.route('/api/v1/items', methods=['GET'])
def fetch_catalog():
    items_data = [
        {'sku': 'A001', 'price': 1500, 'stock': 50},
        {'sku': 'B002', 'price': 2200, 'stock': 12}
    ]
    return jsonify(items_data)

@flask_app.route('/api/v1/cart', methods=['POST'])
def add_to_cart():
    payload = request.get_json()
    item_id = payload.get('item_id')
    quantity = payload.get('qty')
    
    if item_id and isinstance(quantity, int) and quantity > 0:
        return jsonify({'status': 'accepted', 'cart_id': 'CART-998'}), 201
    return jsonify({'status': 'rejected', 'reason': 'invalid_params'}), 400

if __name__ == '__main__':
    flask_app.run(host='127.0.0.1', port=8000)

負荷テストシナリオ定義

Locustフレームワークを使用して仮想ユーザーの操作フローを構築します。タスクに実行頻度の重み付けを適用し、セッション初期化を明確に分離しました。また、パフォーマンス劣化を検知するためのカスタムイベントフックも追加しています。

from locust import HttpUser, task, between, events

class ShoppingCartSimulator(HttpUser):
    # リクエスト間の待機時間（秒範囲）
    wait_time = between(2, 4)
    
    def on_start(self):
        # 認証トークン取得をシミュレート
        self.client.post('/auth/mock/token', json={'user': 'load_bot'}, name='/auth/mock/token')
        
    @task(5)
    def query_catalog(self):
        self.client.get('/api/v1/items')
        
    @task(2)
    def submit_order(self):
        payload = {'item_id': 'A001', 'qty': 1}
        self.client.post('/api/v1/cart', json=payload, headers={'X-Custom-Header': 'load-test'})

# レスポンスタイム監視フック
@events.request.add_listener
def track_slow_requests(environment, response_time, exception=None, **kwargs):
    url = kwargs.get('url', 'unknown')
    if response_time > 1500:
        print(f'[警告] 遅延検出: {url} ({response_time}ms)')

テスト実行と結果監視

依存パッケージをインストールした後、バックエンドアプリケーションを起動します。その後、ローカルホストを対象とするLocustエージェントを実行します。

pip install flask locust
python app_server.py
locust -f load_test.py --host=http://127.0.0.1:8000

エージェント起動完了後、任意のブラウザから管理ダッシュボード（http://localhost:8089）に接続します。設定画面で「Number of users to simulate（仮想ユーザー数）」と「Spawn rate（毎秒のユーザー増加数）」を入力してテストを開始します。ダッシュボード上では総リクエスト数、成功・失敗の比率、平均応答時間、並列処理能力をリアルタイムで可視化できます。イベントリスナー経由で閾値超過が発生した場合は、標準出力またはログファイルに注記が蓄積されます。