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

Question

We plan on doing load testing on a sequence of APIs using locust. So we have a set of APIs we would be calling sequentially on locust using SequentialTaskSet. By default it would run indefinitely if not stopped manually or I can set a timer in the command line to stop after a pre-defined interval. But what we plan to do is, we would be gradually increasing the number of concurrent users. Then we plan to call a separate API every 5 minutes or so which would give us memory consumed on the servers where our sequential APIs are hosted. If the value of memory consumed reaches a threshold, we want the locust script to stop automatically. I tried reading the documentation but couldn't find how we would go about doing this. Any tips?

Basic code we have so far:

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)```

Basic code we have so far:

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)```

user_1591921 · Answer

In a User you can call

self.environment.runner.quit()

In distributed runs, that will only shut down that particular worker, so it isnt perfect.

If you dont want to run the check in a User, you can create your own greenlet that checks your metric in the background (the example checks a locust stats value, but you could just call your API there)

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)

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

1 Answers