What is locust?

Locust is a nice load testing tool using python. It can swarm your app with TONS of request or fewer easily. And it has a nice web UI!

Installing locust

pip install locust 🤷‍♂️

https://docs.locust.io/en/stable/installation.html

Hello World

First we create a locustfile.py, which locust automatically picks up.

I’ve filled mine with:

from locust import HttpUser, task

class HelloWorldUser(HttpUser):
@task
def hello_world(self):
self.client.get("/")
self.client.get("/thisismy404")

We then boot up locust by running the command locust from the same folder as the locustfile. in the output i get a localhost address for the UI, mine showed: http://0.0.0.0:8089. I open this in my browser and BAM!

I input some webserver I use for testing and see what I get when I press “start swarming”.

WARNING! This will spam your endpoint aggressively. Maybe use a local api to test.

Press STOP to stop it.

Here we see how many successful requests we got. There are also tabs here which contains some nice graphs, an overview of different error messages and much more. Pretty neat!

We can also stop it, press “new test” and for example spawn 1 user per second with up to 20 concurrent users. When that runs for a few seconds, the graph looks like this:

Fancy!

Control rate

Next up, a very useful feature is to control how many requests are spawned. The easiest way to control this is by setting the locust `wait_time` variable in the class.

from locust import HttpUser, task, between
import time

class HelloWorldUser(HttpUser):
wait_time = between(1, 5)
@task
def hello_world(self):
self.client.get("/")
self.client.get("/thisismy404")

This makes each “user” wait between 1 and 5 seconds between each request. Much more controllable!

Some different requests

Locust functions start a microthread per user. The user then picks a function decorated with @task at random, executes that, waits the wait_time and then picks another random task. You can however weigh them.

Let’s set up another task that is twice as likely to get picked. We want locust to test each of these:

Example:

class HelloWorldUser(HttpUser):
wait_time = between(1, 5)
@task
def hello_world(self):
self.client.get("/thisismy404")

@task(2)
def hello_isitmeyourelookingfor(self):
self.client.get("/?name=lionel")

We restart locust to reload the config and run it. My output looked like this:

Pretty cool! We technically now have everything we need to do some decent testing of endpoints.

Start and stop functions

I wanted to show another useful little thing. You can define an on_start and on_stop method, which will be executed once per user. Very useful!

For example:

from locust import HttpUser, task, between
import time

class HelloWorldUser(HttpUser):
wait_time = between(1, 5)
@task
def hello_world(self):
self.client.get("/thisismy404")

@task(2)
def hello_isitmeyourelookingfor(self):
self.client.get("/?name=lionel")


def on_start(self):
self.client.post("/", json={"username":"foo", "password":"bar"})

ran with 10 users for a few seconds, gives this output on my webserver:

Wohoo!

Last trick, random queries

Just a last little trick if you need to get some random requests! Remember that this is just standard python, so you can for example create a task like:

@task
def random_request(self):
my_random_string = ''.join(random.choices(string.ascii_uppercase, k=10))
print(my_random_string)
self.client.get(f"/{my_random_string}")

(remember to import random and import string).

It's just an example of how we can leverage python in our locust setup, but it's quite useful.

The End

I hope this helps you get started on load testing! Sure, it’s not automated and integrated into every CI-pipeline, but it’s a lot better than just throwing stuff into prod and see what happens.

Locust has tons more of awesome stuff. including a good command line interface. Check out the docs at https://docs.locust.io/en/stable/writing-a-locustfile.html# and the homepage at https://locust.io/.