前言
在進行性能測試時，我們需要對測試結(jié)果進行監(jiān)控和分析，以便于及時發(fā)現(xiàn)問題并進行優(yōu)化。
Locust在內(nèi)存中維護了一個時間序列數(shù)據(jù)結(jié)構(gòu)，用于存儲每個事件的統(tǒng)計信息。 這個數(shù)據(jù)結(jié)構(gòu)允許我們在Charts標簽頁中查看不同時間點的性能指標，但是正因為Locust WebUI上展示的數(shù)據(jù)實際上是存儲在內(nèi)存中的。所以在Locust測試結(jié)束后，這些數(shù)據(jù)將不再可用。 如果我們需要長期保存以便后續(xù)分析測試數(shù)據(jù)，可以考慮將Locust的測試數(shù)據(jù)上報到外部的數(shù)據(jù)存儲系統(tǒng)，如InfluxDB，并使用Grafana等可視化工具進行展示和分析。
本文將介紹如何使用Locust進行負載測試，并將測試數(shù)據(jù)上報到InfluxDB。同時，我們將使用Grafana對測試數(shù)據(jù)進行展示和分析。
最終效果：
influxDB
InfluxDB是一款開源的時間序列數(shù)據(jù)庫，專為處理大量的時間序列數(shù)據(jù)而設(shè)計。時間序列數(shù)據(jù)通常是按照時間順序存儲的數(shù)據(jù)點，每個數(shù)據(jù)點都包含一個時間戳和一個或多個與之相關(guān)的值。這種數(shù)據(jù)類型在許多場景下都非常常見，如監(jiān)控系統(tǒng)、物聯(lián)網(wǎng)設(shè)備、金融市場數(shù)據(jù)等。在這些場景下，數(shù)據(jù)上報是一種關(guān)鍵的需求，因為它可以幫助我們實時了解系統(tǒng)的狀態(tài)和性能。
注： InfluxDB 開源的時單機版本，集群版本并未開元，但是對于普通用戶的日常場景已經(jīng)完全夠用。
以下是關(guān)于InfluxDB的關(guān)鍵特性和優(yōu)勢的表格：
安裝運行InfluxDB
如果你已經(jīng)安裝了Docker，可以直接使用官方的InfluxDB鏡像來運行InfluxDB：
docker run -p 8086:8086 -v $PWD:/var/lib/influxdb influxdb:1.8
使用Python 上報數(shù)據(jù)到influxdb
首先，確保已經(jīng)安裝了influxdb庫：
pip install influxdb
然后，使用以下代碼上報數(shù)據(jù)到InfluxDB：
以下是一個使用Python操作InfluxDB上報數(shù)據(jù)的示例，對照MySQL進行注釋：
import time
from influxdb import InfluxDBClient
# 連接到InfluxDB（類似于連接到MySQL數(shù)據(jù)庫）
client = InfluxDBClient(host='localhost', port=8086)
# 創(chuàng)建數(shù)據(jù)庫（類似于在MySQL中創(chuàng)建一個新的數(shù)據(jù)庫）
client.create_database('mydb')
# 切換到創(chuàng)建的數(shù)據(jù)庫（類似于在MySQL中選擇一個數(shù)據(jù)庫）
client.switch_database('mydb')
# 上報數(shù)據(jù)（類似于在MySQL中插入一條記錄）
data = [
{
# 在InfluxDB中，measurement相當于MySQL中的表名
"measurement": "cpu_load",
# tags相當于MySQL中的索引列，用于快速查詢
"tags": {
"host": "server01",
"region": "us-west"
},
# time為時間戳，是InfluxDB中的關(guān)鍵字段
"time": int(time.time_ns()),
# fields相當于MySQL中的數(shù)據(jù)列，用于存儲實際的數(shù)據(jù)值
"fields": {
"value": 0.64
}
}
]
# 寫入數(shù)據(jù)（類似于在MySQL中執(zhí)行INSERT語句）
client.write_points(data)
在這個示例中，我們首先連接到InfluxDB（類似于連接到MySQL數(shù)據(jù)庫），然后創(chuàng)建一個名為mydb的數(shù)據(jù)庫（類似于在MySQL中創(chuàng)建一個新的數(shù)據(jù)庫），并切換到創(chuàng)建的數(shù)據(jù)庫（類似于在MySQL中選擇一個數(shù)據(jù)庫）。接著，我們準備了一條名為cpu_load的數(shù)據(jù)（在InfluxDB中，measurement相當于MySQL中的表名），并為數(shù)據(jù)添加了host和region標簽（類似于MySQL中的索引列）。最后，我們將數(shù)據(jù)寫入到InfluxDB中（類似于在MySQL中執(zhí)行INSERT語句）。
執(zhí)行上面的代碼后我們可以看到我們的操作成功了：
如果我們安裝了influx-cli就可以在命令行中直接查詢剛才寫入的數(shù)據(jù)：
bingohe@MacBook-Pro ~ $ /usr/local/Cellar/influxdb@1/1.11.1/bin/influx
Connected to http://localhost:8086 version 1.8.10
InfluxDB shell version: 1.11.1
> show databases;
name: databases
name
----
_internal
mydb
> use mydb
Using database mydb
> show measurements;
name: measurements
name
----
cpu_load
> select * from cpu_load;
name: cpu_load
time host region value
---- ---- ------ -----
1688874870046897000 server01 us-west 0.64
Locust 數(shù)據(jù)寫入到 influx
在 【Python】萬字長文，Locust 性能測試指北（上） 中我們提到過Locust的生命周期，我們也通過Locust生命周期實現(xiàn)了集合點的功能。現(xiàn)在我們一起來通過它實現(xiàn)測試數(shù)據(jù)的實時展示。
Locust的生命周期
test_start：測試開始時觸發(fā)。spawning_start：生成用戶時觸發(fā)。user_add：每個用戶被添加時觸發(fā)。spawning_complete：所有用戶生成完成時觸發(fā)。request：每個請求發(fā)生時觸發(fā)。test_stop：測試停止時觸發(fā)。
上報數(shù)據(jù)
我們先來看看常用的事件里面可以獲取到的數(shù)據(jù)：
import time
from locust import HttpUser, task, between, events
@events.request.add_listener
def request_handler(*args, **kwargs):
print(f"request args: {args}")
print(f"request kwargs: {kwargs}")
@events.worker_report.add_listener
def worker_report_handlers(*args, **kwargs):
print(f"worker_report args: {args}")
print(f"worker_report kwargs: {kwargs}")
@events.test_start.add_listener
def test_start_handlers(*args, **kwargs):
print(f"test_start args: {args}")
print(f"test_start kwargs: {kwargs}")
@events.test_stop.add_listener
def test_stop_handlers(*args, **kwargs):
print(f"test_stop args: {args}")
print(f"test_stop kwargs: {kwargs}")
class QuickstartUser(HttpUser):
wait_time = between(1, 2)
@task
def root(self):
with self.client.get("/", json={"time": time.time()}, catch_response=True) as rsp:
rsp_json = rsp.json()
if rsp_json["id"] != 5:
# 失敗時上報返回的數(shù)據(jù)
rsp.failure(f"{rsp_json}")
運行一次測試時能看到這些生命周期內(nèi)的Locust 對外暴露的數(shù)據(jù)：
test_start args: ()
test_start kwargs: {'environment': <locust.env.Environment object at 0x10c426c70>}
request args: ()
request kwargs: {'request_type': 'GET', 'response_time': 2.6886250000011103, 'name': '/', 'context': {}, 'response': <Response [200]>, 'exception': None, 'start_time': 1688888321.896039, 'url': 'http://0.0.0.0:10000/', 'response_length': 8}
request args: ()
request kwargs: {'request_type': 'GET', 'response_time': 2.735957999998817, 'name': '/', 'context': {}, 'response': <Response [200]>, 'exception': CatchResponseError("{'id': 6}"), 'start_time': 1688888323.421389, 'url': 'http://0.0.0.0:10000/', 'response_length': 8}
test_stopping args: ()
test_stopping kwargs: {'environment': <locust.env.Environment object at 0x10c426c70>}
test_stop args: ()
test_stop kwargs: {'environment': <locust.env.Environment object at 0x10c426c70>}
從上面的監(jiān)控我們可以看到，每次任務(wù)啟動和停止的時候會分別調(diào)用@events.test_start.add_listener和@events.test_stop.add_listener裝飾的函數(shù)，每次請求發(fā)生的的時候都會調(diào)用@events.request.add_listener 監(jiān)聽器裝飾的函數(shù)，我們就是要利用這一點來進行數(shù)據(jù)的上報。
通過查看 Locust 的 EventHook 源碼注釋我們可以看到標準的使用方法：
#.../site-packages/locust/event.py
...
class EventHook:
"""
Simple event class used to provide hooks for different types of events in Locust.
Here's how to use the EventHook class::
my_event = EventHook()
def on_my_event(a, b, **kw):
print("Event was fired with arguments: %s, %s" % (a, b))
my_event.add_listener(on_my_event)
my_event.fire(a="foo", b="bar")
If reverse is True, then the handlers will run in the reverse order
that they were inserted
"""
...
結(jié)合前面的寫數(shù)據(jù)到 influxDB的實現(xiàn)，上報數(shù)據(jù)這一項一下子就變簡單了：
簡單實現(xiàn)每次請求數(shù)據(jù)上報 到 influxDB
下面的代碼運行Locust測試后會自動創(chuàng)建一個locust_requests的 measurement，然后將每次請求的數(shù)據(jù)上報。
import time
from datetime import datetime
from influxdb import InfluxDBClient
from locust import HttpUser, task, between, events
client = InfluxDBClient(host='localhost', port=8086, database="mydb")
def request(request_type, name, response_time, response_length, response, context, exception, url, start_time):
_time = datetime.utcnow()
was_successful = True
if response:
was_successful = 199 < response.status_code < 400
tags = {
'request_type': request_type,
'name': name,
'success': was_successful,
'exception': str(exception),
}
fields = {
'response_time': response_time,
'response_length': response_length,
}
data = {"measurement": 'locust_requests', "tags": tags, "time": _time, "fields": fields}
client.write_points([data])
# 在每次請求的時候通過前面定義的request函數(shù)寫數(shù)據(jù)到 DB
events.request.add_listener(request)
class QuickstartUser(HttpUser):
wait_time = between(1, 2)
@task
def root(self):
with self.client.get("/", json={"time": time.time()}, catch_response=True) as rsp:
rsp_json = rsp.json()
if rsp_json["id"] != 5:
rsp.failure(f"{rsp_json}")
優(yōu)化升級
上面的這個上報很粗糙，每次請求會上報一次數(shù)據(jù)，會影響實際的壓測，如果我們將要上報的數(shù)據(jù)放在一個數(shù)據(jù)結(jié)構(gòu)中中，異步的上報這個數(shù)據(jù)將極大的提升性能
# 將 __flush_points 方法中的寫入操作放到一個單獨的線程中，避免阻塞主線程，提高性能。
self.write_thread = threading.Thread(target=self.__write_points_worker)
# 批量寫入
if len(self.write_batch) >= self.batch_size or time.time() - self.last_flush_time >= self.interval_ms / 1000:
# 使用 gzip 壓縮上報的數(shù)據(jù)
influxdb_writer = InfluxDBWriter('localhost', 8086, 'mydb', batch_size=1000, gzip_enabled=True)
...
配置Grafana
在測試數(shù)據(jù)被上報到InfluxDB之后，可以通過Grafana進行數(shù)據(jù)展示和分析。需要先在Grafana中配置InfluxDB數(shù)據(jù)源，然后創(chuàng)建相應的圖表和儀表盤。
在創(chuàng)建圖表和儀表盤時，可以選擇InfluxDB作為數(shù)據(jù)源，并使用InfluxQL查詢語言進行數(shù)據(jù)查詢和過濾?？梢愿鶕?jù)需要選擇不同的圖表類型和顯示方式，以展示測試結(jié)果數(shù)據(jù)的趨勢和變化。
總結(jié)
本文介紹了如何將Locust測試數(shù)據(jù)上報到InfluxDB，并通過Grafana進行展示和分析。通過將測試數(shù)據(jù)與監(jiān)控工具相結(jié)合，可以更好地了解系統(tǒng)的性能和穩(wěn)定性，及時發(fā)現(xiàn)問題并進行優(yōu)化，也可以方便后續(xù)進行測試數(shù)據(jù)分析。希望本文能對大家有所幫助。
