LocustでDBの負荷テストを行う
はじめに
Locust は HTTP リクエストに対する負荷テストを行うことを前提としており、DB への負荷テストを行うためには、少し工夫が必要です。
本記事では、Locust を使って DB に対する負荷テストを行う方法を紹介します。
ここでは、MySQL を使用した DB に対して負荷テストを行う例を示しますが、DB へ接続するモジュールを変更することで、他の DB に対しても同様のテストを行うことができます。
Locust とは
Locust は、Python で書かれたオープンソースの負荷テストツールです。
Web アプリケーションや API の負荷テストを行うことができます。
Locust は、テストのためのシナリオを Python で記述することができます。
条件
環境
- os: macOS Sonoma 14.2.1
- Python 3.12.3
DB
- MySQL Ver 9.1.0 (Docker コンテナ)
- DB 名: sample_db
- テーブル名: employees
- レコード:
id name age department 1 Alice 30 Sales 2 Bob 25 Marketing 3 Charlie 35 HR
手順
1. 仮想環境(venv)の作成
ここでは、仮想環境を作成して、必要なパッケージをインストールします。
仮想環境を使用しない場合は、この手順は不要です。
プロジェクトディレクトリを作成
mkdir locust_db_test
cd locust_db_test
仮想環境の作成
python3 -m venv venv
仮想環境の有効化
source venv/bin/activate
windows の場合は以下のコマンドを実行してください。
venv\Scripts\activate
2. 必要なパッケージのインストール
Locust のインストール
pip install locust
MySQL ドライバのインストール
負荷テストを行う DB に合わせて、適切なドライバをインストールしてください。
pip install mysql-connector-python
3. Locust ファイルの作成
locustfile.py を作成
locust-db-test ディレクトリに locustfile.py を作成します。
from locust import User, TaskSet, task, between
import mysql.connector
# MySQL への接続設定
def create_mysql_connection():
# MySQL への接続オブジェクトを返す
return mysql.connector.connect(
host="localhost", # MySQL サーバのホスト（コンテナやリモートホストの場合、IP アドレスを設定）
user="root", # MySQL のユーザー名
password="password", # MySQL のパスワード
database="sample_db", # 使用するデータベース名
)
class MySQLQueryTaskSet(TaskSet):
@task
def execute_query(self):
# データベースに接続
connection = create_mysql_connection()
cursor = connection.cursor()
# クエリを実行
query = "SELECT * FROM employees;"
cursor.execute(query)
# 結果をフェッチ
rows = cursor.fetchall()
print(f"result: {rows}")
# コネクションを閉じる
cursor.close()
connection.close()
class MySQLUser(User):
tasks = [MySQLQueryTaskSet] # タスクを指定
wait_time = between(1, 5) # リクエスト間の待機時間を設定（秒単位）
-
MySQLQueryTaskSetクラス
クエリを実行するためのクラスです。 -
MySQLUserクラス
Locust の User クラスを継承し、タスクを実行するユーザーを定義します。 -
tasksプロパティ
ユーザーが実行するタスクを指定します。 -
wait_timeプロパティ
リクエスト間の待機時間を指定した範囲でランダムに設定します。
例:between(1, 5)は、1 秒から 5 秒の間でランダムに待機します。
4. Locust の起動
locustfile.py があるディレクトリでコマンドを実行
locust -f locustfile.py
ブラウザで http://localhost:8089 にアクセス
テストの設定を入力して START ボタンをクリック
- Number of users (peak concurrency): 同時接続ユーザー数 (ピーク時)を指定
- Ramp up (users started/second): ユーザーの増加速度 (1 秒あたりのユーザーの増加数)
- Host: テスト対象のホスト名
- Advanced options: その他のオプション
Host は未入力でもテストは実行できますが、テスト対象のホスト名を入力することで、テスト結果にホスト名が表示されます。
5. テストの結果の確認
ターミナルでテストの結果を確認
上記の通りのコードを記載している場合、ターミナルにはクエリの結果が表示されます。
result: [(1, 'Alice', 30, 'Sales'), (2, 'Bob', 25, 'Marketing'), (3, 'Charlie', 35, 'HR')]
result: [(1, 'Alice', 30, 'Sales'), (2, 'Bob', 25, 'Marketing'), (3, 'Charlie', 35, 'HR')]
result: [(1, 'Alice', 30, 'Sales'), (2, 'Bob', 25, 'Marketing'), (3, 'Charlie', 35, 'HR')]
result: [(1, 'Alice', 30, 'Sales'), (2, 'Bob', 25, 'Marketing'), (3, 'Charlie', 35, 'HR')]
…
ブラウザでテスト結果を確認
ブラウザの方に戻り、「CHARTS」タブをクリックすると、テストの結果をグラフで確認できます。
しかし、「Total Request per Second」や「Response Time」などのグラフに変化がありません。
これは、Locust がデフォルトで HTTP リクエストに対するテストを行うためです。
そのため、DB に対するクエリの実行結果は表示されません。
クエリの実行結果をグラフに表示させるために、少しプログラムを変更する必要があります。
6. クエリの実行結果をグラフに表示
locustfile.py を以下のように変更
- from locust import User, TaskSet, task, between
+ from locust import User, TaskSet, task, between, events
import mysql.connector
+ import time
# MySQLへの接続設定
def create_mysql_connection():
return mysql.connector.connect(
host="localhost", # MySQL サーバのホスト（コンテナやリモートホストの場合、IP アドレスを設定）
user="root", # MySQL のユーザー名
password="password", # MySQL のパスワード
database="sample_db", # 使用するデータベース名
)
class MySQLQueryTaskSet(TaskSet):
@task
def execute_query(self):
# データベースに接続
connection = create_mysql_connection()
cursor = connection.cursor()
+ # クエリを実行する際に時間を計測
+ start_time = time.time()
+ try:
query = "SELECT * FROM employees;"
cursor.execute(query)
# 結果をフェッチ
rows = cursor.fetchall()
print(f"result: {rows}")
+ # 成功した場合、経過時間を報告
+ total_time = (time.time() - start_time) * 1000 # ミリ秒に変換
+ events.request.fire(
+ request_type="mysql",
+ name="execute_query",
+ response_time=total_time,
+ response_length=len(rows),
+ )
+ except Exception as e:
+ # エラーが発生した場合、エラーを報告
+ total_time = (time.time() - start_time) * 1000 # ミリ秒に変換
+ events.request.fire(
+ request_type="mysql",
+ name="execute_query",
+ response_time=total_time,
+ exception=e,
+ )
+ finally:
# コネクションを閉じる
cursor.close()
connection.close()
class MySQLUser(User):
tasks = [MySQLQueryTaskSet] # タスクを指定
wait_time = between(1, 5) # リクエスト間の待機時間を設定（秒単位）
-
events.request.fire()
クエリの実行結果をグラフに表示するためのメソッドです。-
request_type
リクエストの種類を指定します。本来はGETやPOSTなどの HTTP メソッドが指定されますが、DB に対するクエリの場合はmysqlなど、わかりやすい名前を指定すると良いでしょう。 -
name
リクエストの名前を指定します。 -
response_time
リクエストの応答時間を指定します。 -
response_length
レスポンスの長さを指定します。 -
exception
エラーが発生した場合、エラーを指定します。
-
再度 Locust を起動してテストを実行
下記のようにグラフにクエリの実行結果が表示されます。
まとめ
本記事では、Locust を使って DB に対する負荷テストを行う方法を紹介しました。
Locust は HTTP リクエストに対する負荷テストを行うことを前提としているため、DB に対する負荷テストを行う場合は、少し工夫が必要です。
DB に対する負荷テストを行う際は、DB への接続設定やクエリの実行結果をグラフに表示するためのコードを記載することで、テスト結果を詳細に確認することができます。
Discussion