Locustで合否判定結果をSlack通知する

概要

• システムの更新後、都度負荷試験を行っているが、めんどうになったので自動化したい😇
• Locustのheadlessモードを定期実行すれば行けそうだなと思ったので定期実行するlocust_fileを作ってみた

※ごりごりに実装しているだけですが、自分用のメモとして

通知内容

• 単なる情報は水色で、テスト合格であれば緑で、テスト不合格であれば赤色で通知
• 合否判定は「失敗件数」と「レスポンスの99%ile」
• 合否判定で、どちらかが失敗となったときに@hereでメンション
  • 性能に問題があるときだけ気付ける！

実装

locust_file.py

from locust import HttpUser, TaskSet, task, constant_pacing, events
import slackweb
import os

class UserTaskSet(TaskSet):
    @task(1)
    def sample(self):
        self.client.get(url="/sample", verify=False)

class WebsiteUser(HttpUser):
    tasks = [UserTaskSet]
    wait_time = constant_pacing(1)

    host = 'http://localhost:8080'

@events.test_stop.add_listener
def test_stop(**kwargs):
    slack = slackweb.Slack(url="<incoming webhookのURL>")

    # テストOKのカラー
    color_success = "#00ff00"
    # テストNGのカラー
    color_failed = "#cc3366"
    # infoのカラー
    color_info = "#99ffff"

    failure_threshold = int(os.environ.get("FAILURE_NG_THRESHOLD", 1))
    percentile_threshold = int(os.environ.get("PERCENTILE_NG_THRESHOLD", 1000))

    for v in kwargs['environment'].stats.entries:
        item = kwargs['environment'].stats.get(v[0], v[1])

        is_failure_ok = item.num_failures < failure_threshold
        is_percentile_ok = item.get_response_time_percentile(99) < percentile_threshold
        mention = "<!here>\n" if not is_failure_ok or not is_percentile_ok else ""

        rps = {
            "title": "rps (info)",
            "color": color_info,
            "text": "{:.1f} r/s".format(item.total_rps),
            "mrkdwn_in": ["text"]}
        failure = {
            "title": "Failure",
            "color": color_success if is_failure_ok else color_failed,
            "text": "{} ( {}% )".format(item.num_failures, item.fail_ratio),
            "mrkdwn_in": ["text"]}
        percentile = {
            "title": "99 Percentile",
            "color": color_success if is_percentile_ok else color_failed,
            "text": "{} ms".format(item.get_response_time_percentile(99)),
            "mrkdwn_in": ["text"]}
        # slack通知を行う
        slack.notify(
            text="{}Load test result of *{} {}*".format(mention, v[1], v[0]),
            attachments=[rps, failure, percentile])

実装ポイントやドキュメント

• locustのイベントハンドリング
  • https://docs.locust.io/en/stable/api.html#event-hooks
• slack通知（slackweb）
  • https://github.com/satoshi03/slack-python-webhook

あとは@events.test_stop.add_listenerのメソッドで受け取るkwargsで何が取得できるのかdir()を使ってひたすら調べました😇