使用每秒预定义请求的 Locust

Question

查看负载测试工具，我发现了 Locust，并立即发现它作为一个 Python 人很有吸引力，但我不完全确定我是否可以用它实现以下场景...

我的任务是使用真实流量对 REST API 进行负载测试。我已经从生产 Apache 日志中提取了 5 分钟的 GET 流量，我的想法是使用负载测试工具运行那些具有相同时间分布（每秒 1 到 36 个请求）的相同请求。为此，我构建了一个 Python 字典，其中一个相对时间戳（xx:xx，即 mins:secs）是关键，以及一个 URL 路径列表来请求那一秒是值。

我们将某个时间点的生产数据库转储恢复到我们的测试环境，请求来自创建转储后的接下来的五分钟。在测试运行之间，我正在更改 REST API 如何连接到数据库的参数，因此测试运行需要尽可能相同，以便指标可比。

我查看了 Locust 文档中的自定义负载形状，它们似乎可能有效，但我并不肯定。自定义 tick 方法实现能否实现这一点：

在 0 秒时，发出一组 4 个请求。
在 1 秒时，发出一组 2 个请求。
在 2 秒时，发出一组 12 个请求。
...时间流逝，每秒发生一组预定义的请求...
4分59秒，发出一组27个请求
在 5 分钟时，发出一组 14 个请求。

这将如何映射到 Locust 的功能上？产生多少用户并不重要，重要的是发出多少请求，以及在什么时间点发出请求。

我非常喜欢 Locust，因为它易于使用且语法熟悉，但它是否适合使用像这样的静态、可重复请求负载进行测试？

编辑： 因为看起来这种方法可能无法在没有很大困难（或根本没有）的情况下实现，我想出了另一种方法，不同的足以保证单独的问题：

Answer 1

你要的是custom Load Shape。您可以在代码中的任何给定时刻定义您希望 Locust 执行的操作。来自文档的示例：

class MyCustomShape(LoadTestShape):
    time_limit = 600
    spawn_rate = 20

    def tick(self):
        run_time = self.get_run_time()

        if run_time < self.time_limit:
            # User count rounded to nearest hundred.
            user_count = round(run_time, -2)
            return (user_count, spawn_rate)

        return None

但在您的情况下，您可以从字典中提取值而不是 user_count = round(run_time, -2)。如果你让 user_count 和 spawn_rate 相同，下一个效果将大致是你每次的用户数量（假设你有足够的工人能够每次产生差异）。如果需要，调整 spawn_rate 可以调整和消除差异，从而减少用户的突然变化。

Answer 2

简而言之，没有。 Locust 以运行一定数量的用户为中心，并且这些用户彼此相当独立。

您可以使用 wait_time = constant_throughput(x) (https://docs.locust.io/en/stable/writing-a-locustfile.html#wait-time-attribute). You can also check out locust-plugins's constant_total_ips that targets a global throughput (https://github.com/SvenskaSpel/locust-plugins/blob/master/examples/constant_total_ips_ex.py) 控制每个用户的吞吐量。

但是如果你想在一个准确的时间做准确个请求，我认为没有什么现成的。

您可以使用 python 在触发请求之前“同步”所有用户。我没有很好的解决方案，但这有点管用（不过你需要运行刚好有 10 个用户，否则你会运行陷入并发问题 - 很确定那里有更好的解决方案)

users_waiting = 0
lock = Semaphore(10)
...
    @task
    def t(self):
        global users_waiting
        with lock:
            users_waiting = users_waiting + 1 if users_waiting < 10 else 1
            while self.environment.runner and users_waiting < 10:
                time.sleep(0.1)
            # actually do your requests here

使用每秒预定义请求的 Locust

Using Locust with pre-defined requests per second

load-testing

locust