我怎样才能让蝗虫根据另一个 API 的响应停止负载测试？

Question

我们计划使用 Locust 对一系列 API 进行负载测试。所以我们有一组 APIs，我们将使用 SequentialTaskSet 在蝗虫上顺序调用。默认情况下，如果不手动停止它会运行无限期，或者我可以在命令行中设置一个计时器在预定义的时间间隔后停止。但我们计划做的是，我们会逐渐增加并发用户的数量。然后，我们计划每 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()

在分布式运行中，这只会关闭特定的 worker，因此它并不完美。

如果您不想运行签入用户，您可以创建自己的 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)

我怎样才能让蝗虫根据另一个 API 的响应停止负载测试？

How can I get locust to stop load testing based on response from another API?

python

load-testing

locust