AInimal 人工智慧社群養成

由於耶誕節快要到各校都會舉辦耶誕舞會或者相關活動也因此團隊也和不少活動進行合作為了避免忽然衝高的使用人數超過伺服器承受上限進而導致系統癱瘓因此我們需要在活動進行前做壓力測試檢視系統的耐受度

常見壓力測試軟體種類

JMeter

Apache JMeter 是 Apache 開發的壓力測試開源套件，以 JAVA 寫成。Apache JMeter 可用於測試靜態和動態資源，Web 動態應用程式的性能。它可用於類比伺服器，伺服器組，網路或物件上的繁重負載，以測試其強度或分析不同負載類型下的整體性能。 https://jmeter.apache.org/

Locust

Locust 是一個開源負載測試工具。使用 Python 代碼定義用戶行為，也可以模擬百萬個使用者。 Locust 是非常簡單易用，分散式，用戶負載測試工具。 Locust 主要為網站或者其他系統進行負載測試，能測試出一個系統可以併發處理多少使用者 Locust 是完全基於時間的，因此單個機器支援幾千個併發使用者。相比其他許多事件驅動的應用，Locust 不使用回調，而是使用輕量級的處理方式協程。

Locust 教學


from locust import HttpUser, between, task
class WebsiteUser(HttpUser):
    wait_time = between(5, 15)
    def on_start(self):
        self.client.post("/login", {
            "username": "test_user",
            "password": ""
        })
    @task
    def index(self):
        self.client.get("/")
        self.client.get("/static/assets.js")
    @task
    def about(self):
        self.client.get("/about/")

User class

Loust 會依據 User class 定義的各種 task, event...產生一群 user 去模擬多個用戶

wait_time attribute

我們可以透過更改 wait_time method 簡單的模擬用戶在操作多個 task 之間的等待時間，主要以兩種方法


class MyUser(User):
    last_wait_time = 0
    def wait_time(self):
        self.last_wait_time += 1
        return self.last_wait_time
    ...

weight and fixed_count attributes

如果在同一份文件中定義了多個 user class，可以透過指定 count 數來讓 locust 依照比例產生不同 user，若是沒有指定，locust 會產生相同數量的使用者


class WebUser(User):
    weight = 3
    ...
class MobileUser(User):
    weight = 1
    ...

也可以透過指定 fixed_count 來確保 locust 產生的 user 為固定數量(若有攝氏會忽視 weight)


class AdminUser(User):
    wait_time = constant(600)
    fixed_count = 1
    @task
    def restart_app(self):
        ...
class WebUser(User):
    ...

on_start and on_stop methods

User class 中可以定義 on_start 和 on_stop 函式，on_start 會在開始時執行，on_stop 會在 user 停止時執行(被 interrupt 或 killed)

Tasks

User class 中可以透過@task decorator 來讓 user 做指定任務，當 user 被生成時，會產生一個 thread 去選擇要做的 task，做完時未再選擇下一個(或新的)task

@task decorator


from locust import User, task, constant
class MyUser(User):
    wait_time = constant(1)
    @task
    def my_task(self):
        print("User instance (%r) executing my_task" % self)

我們也可以設定各個 task 的比重來讓 user 可以依據不同的比重執行各個 tasks，如下面所示，task2 被執行到的機會會是 task1 的兩倍


from locust import User, task, between
class MyUser(User):
    wait_time = between(5, 15)
    @task(3)
    def task1(self):
        pass
    @task(6)
    def task2(self):
        pass

HttpUser class

HttpUser class 除了繼承了原有 user class，還擁有了 client 的屬性，來建立 HTTP requests (理論上在原有的 user class 使用 requests 也能達到相同的效果) 而 client 的使用方法也類似 requests，這裡就只貼一些範例程式碼


from locust import HttpUser, task, between
class MyUser(HttpUser):
    wait_time = between(5, 15)
    @task(4)
    def index(self):
        self.client.get("/")
    @task(1)
    def about(self):
        self.client.get("/about/")
    @task(2)
    def index2(self):
        with self.client.get("/", catch_response=True) as response:
            if response.text != "Success":
                response.failure("Got wrong response")
            elif response.elapsed.total_seconds() > 0.5:
                response.failure("Request took too long")
    @task(2)
    def index3(self):
        with self.client.post("/", json={"foo": 42, "bar": None}, catch_response=True) as response:
            try:
                if response.json()["greeting"] != "hello":
                    response.failure("Did not get expected value in greeting")
            except JSONDecodeError:
                response.failure("Response could not be decoded as JSON")
            except KeyError:
                response.failure("Response did not contain expected key 'greeting'")

更進一步

由於當初在壓力測試軟體時有被要求要連同 websocket 的部分要一起測，因此就來大致說一下如何在 locust 中測試 websocket。

Custom client(websocket)

要建立 websocket 的 load test 首先我們要先了解 websocket 的運作機制

從上圖可以發現 websocket 和 http 最大的差異在於 websocket 在正式傳送訊息前會有一個 handshaking 建立連線的動作。因此在建構 user class 時，我會在 on_start 建立 websocket 連線，而後續操作做就是針對該連線做接收及傳送訊息


from websocket import create_connection
from locust import HttpLocust, TaskSet, task
class TestUser(User):
    def on_start(self):
        ws = create_connection('ws://127.0.0.1:5000/echo')
        self.ws = ws

Event

雖然我們可以透過自己額外的套件來達成 websocket 的連線測試，但是卻無法紀錄連線資訊。幸好 Locust 提供 Event 來自定義事件，使用者可以在 init 時或 test_start 以及 test_stop 時要做的事，也可以透過 event 來自定義何時觸發回應成功的成功或失敗。


from locust.events import request_success
def _receive():
    while True:
        res = ws.recv()
        data = json.loads(res)
        end_at = time.time()
        #使用event裡面的request_sucess來紀錄成功事件
        response_time = int((end_at - data['start_at']) * 1000000)
        request_success.fire(
            request_type='WebSocket Recv',
            name='test/ws/echo',
            response_time=response_time,
            response_length=len(res),
        )
gevent.spawn(_receive)


from locust import events
@events.test_start.add_listener
def on_test_start(environment, **kwargs):
    print("A new test is starting")
@events.test_stop.add_listener
def on_test_stop(environment, **kwargs):
    print("A new test is ending")

更多 event 的用法可參考官方網站 docs.locust.io/en/stable/extending-locust.html

如何使用

Running in web UI


$locust

接著瀏覽器就會出現以下畫面

我們可以根據需要測試的人數以及生成速度打上去畫面上就會出現目前的發送量、平均回復時間、回復成功數......等

也會有相應的圖表

Running without the web UI


locust -f locust_files/my_locust_file.py --headless -u 1000 -r 100

壓力測試 Locust 教學

前言