LocustでDBの負荷テストを行う

mkdir mysql_load_test
cd mysql_load_test

python3 -m venv venv

source venv/bin/activate

pip install locust mysql-connector-python

touch locustfile.py

from locust import User, TaskSet, task, between
import mysql.connector
import random

# 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()

        # クエリを実行
        query = "SELECT * FROM employees WHERE age > %s"
        age_threshold = random.choice([20, 25, 30, 35, 40])
        cursor.execute(query, (age_threshold,))

        # 結果をフェッチ
        rows = cursor.fetchall()

        # コネクションを閉じる
        cursor.close()
        connection.close()

class MySQLUser(User):
    tasks = [MySQLQueryTaskSet]
    wait_time = between(1, 5)  # リクエスト間の待機時間を設定（秒単位）

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)  # リクエスト間の待機時間を設定（秒単位）