Locust：将当前用户和当前用户发送到 DB(influxDB)

Question

经过几天的尝试，我决定在这方面请求您的帮助。我很难将 locust 的实时数据（如 current rps、current active users 计数发送到 influxDB。

用户类：starter.py

class Users(HttpUser):
wait_time = constant(1)
host = 'https://DOMAIN.jp'
tasks = [UserBehaviour.get_miniapp]

events.request_success.add_listener(FluxdbEvents.success_events_handle)
events.request_failure.add_listener(FluxdbEvents.failed_events_handle)
events.quitting.add_listener(FluxdbEvents.quitting_events_handle)

任务类：sdkApis.py

from locust import task, TaskSet

class UserBehaviour(TaskSet):
@task
def get_miniapp(self):
    self.client.headers = {"apiKey": "ras-xxxxxxxxxxxx"}
    proxies = {
        'http': 'http://XXX.XXX.XXX.XXX:10080',
        'https': 'http://XXX.XXX.XXX.XXX:10080',
    }
    with self.client.get(
            url='/novalue-stg/novalue-publishing',
            name='get mini app details',
            auth=None,
            catch_response=True,
            proxies=proxies
    ) as response:
        if response.status_code == 200:
            response.success()
        else:
            response.failure("Failed")

数据库连接类：influxdb.py

from influxdb import InfluxDBClient
import json
import socket
import datetime
import pytz
from locust.stats import stats_printer
import gevent

class FluxdbEvents:

client = InfluxDBClient(host="localhost", port="8086", username="telegraf", password="xxxxxx", database="miniapp_metrics")

def success_events_handle(request_type, name, response_time, response_length, **kwargs):
    SUCCESS_TEMPLATE = '[{"measurement":"%s","tags":{"hostname":"%s","requestName":"%s","requestType":"%s","status":"%s","excution_num":"1"},"time":"%s","fields":{"responseTime":"%s","responseLength":"%s", "other":"%s"}}]'
    json_string = SUCCESS_TEMPLATE%("runtime_stats", socket.gethostname(), name, request_type, "OK", datetime.datetime.now(tz=pytz.UTC), response_time, response_length, kwargs)
    gevent.spawn(stats_printer(env.stats))
    FluxdbEvents.client.write_points((json.loads(json_string)))

def failed_events_handle(request_type, name, response_time, response_length, exception, **kwargs):
    FAIL_TEMPLATE = '[{"measurement":"%s", "tags":{"hostname":"%s","requestName":"%s","requestType":"%s","exception":"%s","status":"%s","excution_num":"1"}, "time":"%s", "fields": {"responseTime":"%s","responseLength":"%s","other":"%s"}}]'
    json_string = FAIL_TEMPLATE%("runtime_stats", socket.gethostname(), name, request_type, exception, "FAILED", datetime.datetime.now(tz=pytz.UTC), response_time, response_length, kwargs)
    FluxdbEvents.client.write_points((json.loads(json_string)))

def quitting_events_handle(environment):
    SUCCESS_TEMPLATE = '[{"measurement":"%s","tags":{"hostname":"%s","requestName":"%s","requestType":"%s","status":"%s","startTime":"%s"},"time":"%s","fields":{"name": "%s", "method":"%s", "num_requests":"%s", "num_failures":"%s", "avg_response_time":"%s", "min_response_time":"%s", "max_response_time": "%s", "median_response_time":"%s", "current_rps":"%s", "total_rps":"%s", "fail_ratio":"%s","50percentile":"%s", "66percentile":"%s","75percentile":"%s", "80percentile":"%s","90percentile":"%s", "95percentile":"%s", "98percentile":"%s","99percentile":"%s", "999percentile":"%s", "9999percentile":"%s", "100_percentile":"%s" }}]'
    json_string = SUCCESS_TEMPLATE%("global_stats", socket.gethostname(), environment.stats.total.name, environment.stats.total.method, "OK", environment.stats.total.start_time, datetime.datetime.now(tz=pytz.UTC), environment.stats.total.name, environment.stats.total.method, environment.stats.total.num_requests, environment.stats.total.num_failures,environment.stats.total.avg_response_time, environment.stats.total.min_response_time, environment.stats.total.max_response_time, environment.stats.total.median_response_time, environment.stats.total.current_rps, environment.stats.total.total_rps, environment.stats.total.fail_ratio, environment.stats.total.get_response_time_percentile(0.50), environment.stats.total.get_response_time_percentile(0.66), environment.stats.total.get_response_time_percentile(0.75), environment.stats.total.get_response_time_percentile(0.80), environment.stats.total.get_response_time_percentile(0.90),  environment.stats.total.get_response_time_percentile(0.95), environment.stats.total.get_response_time_percentile(0.98), environment.stats.total.get_response_time_percentile(0.99), environment.stats.total.get_response_time_percentile(0.999), environment.stats.total.get_response_time_percentile(0.9999),  environment.stats.total.get_response_time_percentile(1))
    FluxdbEvents.client.write_points((json.loads(json_string)))

如果您注意到，我的 success_events_handle 和 failed_events_handle 仅返回几个指标，例如 request_type、name、response_time、response_length、有什么方法可以发送更多数据，例如 当前 rps 和 当前用户数，或者有没有更好的方法来处理这个问题？

请注意 quitting.add_listener 完美运行并将统计指标发送到数据库。

请帮帮我

Answer 1

我没有复制粘贴解决方案，但基本上有两个选项，都在 Locust 插件中实现

• 运行一个单独的 greenlet 来记录当前用户的数量，就像 TimescaleListener 一样。 https://github.com/SvenskaSpel/locust-plugins/blob/d7cbe469dd165085aec00315f84c3c7c5eb12d8a/locust_plugins/listeners.py#L112

• 使用附加数据记录您自己的事件，就像 JmeterListener 一样。看看这里：https://github.com/SvenskaSpel/locust-plugins/blob/master/locust_plugins/jmeter_listener.py（不是我写的，所以不要问我细节。不确定它是否适用于分布式运行）