如何让 locust 根据来自另一个 API 的响应停止负载测试？

Question

我们计划使用 locust 对一系列 API 进行负载测试。所以我们有一组 API，我们将使用 SequentialTask​​Set 在 locust 上按顺序调用。默认情况下，如果不手动停止，它将无限期运行，或者我可以在命令行中设置计时器以在预定义的时间间隔后停止。但我们计划做的是，我们将逐渐增加并发用户的数量。然后我们计划每 5 分钟左右调用一个单独的 API，这将使我们在托管顺序 API 的服务器上消耗内存。如果消耗的内存值达到阈值，我们希望 locust 脚本自动停止。我尝试阅读文档，但找不到我们将如何去做。有什么建议吗？

到目前为止我们的基本代码：

import json

class UserBehavior(SequentialTaskSet):

    @task
    def api_call(self):

        with self.client.get('/test_api') as response:
            print(response)

class WebsiteUser(HttpUser):
    tasks = [UserBehavior]
    wait_time = between(2, 10)```

Answer 1

在用户中可以调用

self.environment.runner.quit()

在分布式运行中，这只会关闭特定的工作人员，因此并不完美。

如果您不想在用户中运行检查，您可以创建自己的 greenlet 在后台检查您的指标（该示例检查蝗虫统计值，但您可以在那里调用您的 API）

https://docs.locust.io/en/stable/extending-locust.html#run-a-background-greenlet

from locust import events
from locust.runners import STATE_STOPPING, STATE_STOPPED, STATE_CLEANUP, WorkerRunner

def checker(environment):
    while not environment.runner.state in [STATE_STOPPING, STATE_STOPPED, STATE_CLEANUP]:
        time.sleep(1)
        if environment.runner.stats.total.fail_ratio > 0.2:
            print(f"fail ratio was {environment.runner.stats.total.fail_ratio}, quitting")
            environment.runner.quit()
            return


@events.init.add_listener
def on_locust_init(environment, **_kwargs):
    # only run this on master & standalone
    if not isinstance(environment.runner, WorkerRunner):
        gevent.spawn(checker, environment)