負荷試験ツール Locust を使ってみました

＊Locust とは

Python製のWeb負荷試験ツールです。シナリオをPythonのコードで実装し、何秒間に何回アクセスさせるかなどPythonで設定することができます。実行結果はLocustのGUIで確認でき、結果をCSVファイルとしてダウンロードできる機能もあります。

また、タスクを順番に実行させることもできます。セッションの保持といったこともできるので、複数ユーザーを想定したシナリオの作成も可能です。

負荷試験ツールとしては他にもJmeterなどが有名ですが、シナリオがXMLファイルなのでコードを自分で実装するのはかなり困難で手間が掛かります。（GUIから作成する必要があります）
その点、Locustはシナリオ作成もPythonで書くことができるので、シナリオの追加や機能の拡張も簡単ですし、コードを見ただけで何をするかを理解することができます。

使ったことがなかったので実際に試してみました。

＊参考

＊環境

MacOS
Python 3.6
locustio 0.9.0

＊インストール

Macの場合は最初に下記コマンドを実行する必要があります。

$ brew install libev

下記コマンドを実行してインストールします。

$ python3 -m pip install locustio

＊基本的な使い方

任意のファイル名を指定して実行することもできますが、ファイル名を指定しない場合、デフォルトではlocustfile.pyが実行されます。

<locustfile.py>

from locust import HttpLocust, TaskSet, task

class UserBehavior(TaskSet):
    def on_start(self):
        self.login()

    def on_stop(self):
        self.logout()

    def login(self):
        self.client.post("/login", {"username":"ellen_key", "password":"education"})

    def logout(self):
        self.client.post("/logout", {"username":"ellen_key", "password":"education"})

    @task(2)
    def index(self):
        self.client.get("/")

    @task(1)
    def profile(self):
        self.client.get("/profile")

class WebsiteUser(HttpLocust):
    task_set = UserBehavior
    min_wait = 5000
    max_wait = 9000

TaskSetにシナリオを記述しこれが繰り返し実行される
min_waitとmax_waitの間でランダムに実行される（単位はミリ秒）
@taskで重み付けをすることが可能（@task(3)は@task(1)の3倍の頻度で実行される）
on_start()はタスクの開始時に呼ばれる（APIの実行に必要なHTTPヘッダーなどはここで設定）
on_stop()はタスクの終了時に呼ばれる

下記コマンドを実行して Locust のGUIを起動します。

$ locust --host=http://localhost:8080

# ファイル名を指定する場合
$ locust -f locust_files/my_locust_file.py --host=http://localhost:8080

下記URLをひらいて Locust GUI にアクセスします。
http://localhost:8089

Number of users to simulateにクライアントの数、
Hatch rateに何秒間にクライアントを作成するかの設定をし、Startを実行するとテストが開始されます。

＊タスクを順番に実行

TaskSetクラスをTaskSequenceに変更し、各タスクに
@seq_taskデコレータを付与して実行順を指定します。

<locustfile.py>

from locust import HttpLocust, seq_task, task, TaskSequence


class UserBehavior(TaskSequence):

    @seq_task(1)
    @task(1)
    def login(self):
        self.client.post("/login", {"username":"ellen_key", "password":"education"})

    @seq_task(2)
    @task(1)
    def view(self):
        self.client.get('/view')
 

    @seq_task(3)
    @task(1)
    def logout(self):
        self.client.post("/logout", {"username":"ellen_key", "password":"education"})


class WebsiteUser(HttpLocust):
    task_set = UserBehavior
    min_wait = 5000
    max_wait = 9000

複数ユーザーでの実行

ユーザー情報を辞書型のリストで定義しておき、この情報をシナリオの最初に呼ばれるon_start()で取り出すようにします。

<data/user_info.py>

USER_INFO = [
    {
        "username": "Tom",
        "password": "hogehoge"
    },
    {
        "username": "Bob",
        "password": "hugahuga"
    }
]

<locustfile.py>

from locust import HttpLocust, seq_task, task, TaskSequence
from data.user_info import USER_INFO


class UserBehavior(TaskSequence):
    def __init__(self, parent):
        super().__init__(parent)
        self.token = ''
        self.user_info = ''

    def on_start(self):
        if len(USER_INFO) > 0:
            self.user_info = USER_INFO.pop()

    @seq_task(1)
    @task(1)
    def login_access(self):
        res = self.client.get('/accounts/login')
        self.token = res.cookies['csrftoken']

    @seq_task(2)
    @task(1)
    def login(self):
        payload = {
            'token': self.csrftoken,
            'username': self.user_info.get('username'),
            'password': self.user_info.get('password')
        }
        self.client.post('/accounts/login', data=payload, headers={'X-CSRFToken': self.token})

    @seq_task(3)
    @task(1)
    def view_upload(self):
        payload = {'token': self.token}
        self.client.post('/view/', data=payload, headers={'X-CSRFToken': self.token})

    @seq_task(4)
    @task(1)
    def process_upload(self):
       payload = {'token': self.token}
       self.client.post('/view/process_upload', data=payload, headers={'X-CSRFToken': self.token})

    @seq_task(5)
    @task(1)
    def register(self):
        payload = {'token': self.token}
        self.client.post('/register', data=payload, headers={'X-CSRFToken': self.token})


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

後続のシナリオを同じユーザーで実行させるためにトークンの保持してHTTPヘッダーに設定するようにしました。
HTTPのGETメソッドのレスポンスからres.cookies['csrftoken']でトークンを取り出し、その後のPOSTメソッドでheadersにトークンを設定しています。

＊所感

LocustはGUIでの操作も軽いですし、Pythonのコードで簡単に実装できるという点が非常に便利でした。シナリオを後から見返すときにもコードを見れば理解できるので、保守の観点でも魅力的でした。
一方で、APIのパラメータの確認がGUIからできないなど他ツールよりも機能が充実していなかったり、日本語のドキュメントが少ないので情報収集に時間がかかるといった懸念点はありますが、使い慣れてしまえば問題ないかと思います。
（現時点では実行したパラメータをGUIで確認することはできないので、サーバーの実行ログで確認するかテストコードを書いてレスポンスの値を確認するしかできないようです。）
今後は業務での負荷テストでも積極的に使っていきたいと思っています。

もぐもぐプログラミング

＊Locust とは

＊参考

＊環境

＊インストール

＊基本的な使い方

＊タスクを順番に実行

複数ユーザーでの実行

＊所感

人気の投稿