locustでLINEメッセージ受信サーバに負荷試験

はじめに

業務でLINE Messeging APIを使用しているのですが、先日突然メッセージ受信サーバ(LINEからのWebhookを受け付けるサーバ)に対して、負荷試験を実施してほしいと先輩に依頼されました。
負荷試験ツールではjMeterが有名ですがノウハウがなく、もっと簡単に実施できるものが無いか調べたところ、 locustというPythonで記述できるツールがあることを知りました。
その内容を、自分用メモレベルですが備忘としてまとめます。

locustのインストール

公式サイト

Locust - A modern load testing framework

LocustはPython3.6までにしか対応していないので、Python3.6系の環境をあらかじめpyenvなどで準備しておきます。

Locust supports Python 2.7, 3.4, 3.5, and 3.6.

Macにてインストールする場合、libevが必要なので先にインストールしておきます。

$ brew install libev
$ pip install locustio

基本的な使い方は上記公式サイトや以下サイトを参照しました。

シナリオ作成

LINEから送られるWebhookを模してJSONのhttpヘッダーにx-line-signatureパラメータを追加して送っています。
リクエスト先はLine developerで設定しているcallbackURLのパスを指定しています。(今回は/callback
メッセージIDは、受け付けるサーバ側(今回負荷をかけるサーバ)でユニークなものでないと処理できない仕様になっていたようなので、タイムスタンプを設定して送っています。
Messaging API - LINE Developers

from locust import HttpLocust, TaskSet, task
import base64
import hashlib
import hmac
import datetime
import json

channel_secret = '  channnel secret   '

class UserBehavior(TaskSet):
    @task(1)
    def index(self):
        payload = {
                    "events": 
                    [
                        {
                            "replyToken": "0f377",
                            "type": "message",
                            "timestamp": 1462629479859,
                            "source": {
                                "type": "user",
                                "userId": "Ue636d0*****************"
                            },
                            "message": {
                                "id": str(int(datetime.datetime.now().timestamp()*1000000)),
                                "type": "text",
                                "text": "性能試験実施中"
                            }
                        }
                    ],
                    "destination":"U6415f13f***********"
                }
        hash = hmac.new(channel_secret.encode('utf-8'),json.dumps(payload).encode('utf-8'), hashlib.sha256).digest()
        signature = base64.b64encode(hash)
        headers = {'content-type': 'application/json', 'x-line-signature': signature}
        r = self.client.post("/callback", data=json.dumps(payload), headers=headers, catch_response=True)
        print(r)

class WebsiteUser(HttpLocust):
    task_set = UserBehavior
    min_wait = 1000
    max_wait = 1000

以下コマンドで起動します。

$ locust -H http://"負荷をかけるサーバURL"

画面が起動して

  • Number of users to simulate:
    何クライアント作成するか
  • Hatch rate:
    クライアントの作成スピード(毎秒)

を入力するとリクエストが生成され、負荷試験できました!