# 壓力測試 Locust教學
## 前言
由於耶誕節快要到各校都會舉辦耶誕舞會或者相關活動 也因此團隊也和不少活動進行合作 為了避免忽然衝高的使用人數超過伺服器承受上限進而導致系統癱瘓 因此我們需要在活動進行前做壓力測試檢視系統的耐受度
## 常見壓力測試軟體種類
### JMeter
Apache JMeter 是Apache 開發的壓力測試開源套件，以JAVA寫成。Apache JMeter可用於測試靜態和動態資源，Web動態應用程式的性能。 它可用於類比伺服器，伺服器組，網路或物件上的繁重負載，以測試其強度或分析不同負載類型下的整體性能。
https://jmeter.apache.org/
### Locust
Locust是一個開源負載測試工具。 使用 Python 代碼定義用戶行為，也可以模擬百萬個使用者。 Locust 是非常簡單易用，分散式，用戶負載測試工具。 Locust 主要為網站或者其他系統進行負載測試，能測試出一個系統可以併發處理多少使用者Locust 是完全基於時間的，因此單個機器支援幾千個併發使用者。 相比其他許多事件驅動的應用，Locust 不使用回調，而是使用輕量級的處理方式協程。
## Locust教學
以下是官方範例程式
```python=
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/")
```
上述的程式碼主要可以分成兩個部分，分別是
1. User class
2. Tasks
### User class
Loust 會依據User class 定義的各種task, event...產生一群user去模擬多個用戶
#### wait_time attribute
我們可以透過更改wait_time method 簡單的模擬用戶在操作多個task 之間的等待時間，主要以兩種方法
* const: 等待固定時間
* between: 等待在範圍內的隨機時間
也可以自定義wait_time 函式
```python=
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會產生相同數量的使用者
```python=
class WebUser(User):
weight = 3
...
class MobileUser(User):
weight = 1
...
```
也可以透過指定fixed_count來確保locust產生的user為固定數量(若有攝氏會忽視weight)
```python=
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
```python=
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的兩倍
```python=
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，這裡就只貼一些範例程式碼
```python=
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連線，而後續操作做就是針對該連線做接收及傳送訊息
```python=
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來自定義何時觸發回應成功的成功或失敗。
```python=
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)
```
也可以使用`@event` decorator
```python=
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 file我們可以透過以下指令執行他
```shell=
$locust
```
接著瀏覽器就會出現以下畫面
我們可以根據需要測試的人數以及生成速度打上去
畫面上就會出現目前的發送量、平均回復時間、回復成功數......等
也會有相應的圖表
#### Running without the web UI
```shell=
locust -f locust_files/my_locust_file.py --headless -u 1000 -r 100
```
-u: 人數
-r: 生成速度
-t: 執行時間
### 參考資料
http://docs.locust.io/en/stable/
https://gist.github.com/yamionp/9112dd6e54694d594306
###### tags: `Locust` `壓力測試`
Sign in with Wallet
Connect another wallet