Locust – As an Open-Source Load Testing Tool in Python
- A practical experience sharing
Recently, we found an open-source tool called locust for load testing. We have used it in testing the Python endpoints of a GPU-based database for one of our clients. So, we are going to share a brief idea of load testing with locust.
The main advantage of this framework is that we can write the entire test script using pure Python. Locust uses fewer resources to simulate thousands or millions of concurrent users while performing load testing and this is another advantage of this framework.
Here we will walk you through an example of load testing with the command-line interface of locust.
There are three sections in this tutorial:
- Setup
- Usage
- Conclusion
Let’s get started!
- Setup
Here we will be using a machine with Ubuntu 18.04 installed.
We will be setting up a virtual environment for Python 3.
Please cd to a directory of your choice.
Install pip
sudo apt-get install python3-pip
Install virtualenv using pip3
sudo pip3 install virtualenv
Create a virtual environment
virtualenv venv
Activate your virtual environment
source venv/bin/activate
Install locust
pip install locust==1.3.1
- Usage
Here we will be testing a non-restful service, for example, some Python functions.
Within the present directory create the following files.
my_module.py
import random
import time
def func1():
time.sleep(random.randrange(1, 4)) # Sleep for 1 to 3 seconds
return random.randrange(0, 2) # Return 0 or 1
def func2():
time.sleep(random.randrange(1, 4)) # Sleep for 1 to 3 seconds
return random.randrange(0, 2) # Return 0 or 1
A locust load test is specified in a plain Python file.
locustfile.py
import time
from locust import TaskSet, User, constant, task, events
from my_module import func1, func2
class MyTaskSet(TaskSet):
user_count = 0
def on_start(self):
MyTaskSet.user_count += 1
self.user_id = 'User{}'.format(MyTaskSet.user_count)
@task
def run_func1(self):
print("{} Executing run_func1 task ...".format(self.user_id))
start_time = time.time()
# TODO: your code here!
return_code = func1()
end_time = time.time()
total_time = int((end_time - start_time) * 1000)
if return_code == 0:
events.request_success.fire(request_type='func1',
name=self.user_id,
response_time=total_time,
response_length=0)
else:
events.request_failure.fire(request_type='func1',
name=self.user_id,
response_time=total_time,
response_length=0,
exception=Exception())
print("{} DONE run_func1 task ...".format(self.user_id))
@task(3)
def run_func2(self):
print("{} Executing run_func2 task ...".format(self.user_id))
start_time = time.time()
# TODO: your code here!
return_code = func2()
end_time = time.time()
total_time = int((end_time - start_time) * 1000)
if return_code == 0:
events.request_success.fire(request_type='func2',
name=self.user_id,
response_time=total_time,
response_length=0)
else:
events.request_failure.fire(request_type='func2',
name=self.user_id,
response_time=total_time,
response_length=0,
exception=Exception())
print("{} DONE run_func2 task ...".format(self.user_id))
class MyUser(User):
tasks = [MyTaskSet]
wait_time = constant(0)
Let’s break it down
import time
from locust import TaskSet, User, constant, task, events
from my_module import func1, func2
A locust file is just a normal Python module, it can import code from other files or packages.
class MyUser(User):
tasks = [MyTaskSet]
wait_time = constant(0)
A locust file is a normal Python file. The only requirement is that it declares at least one class that inherits from the class User. A user class represents a user. Locust will spawn one instance of the User class for each user that is being simulated.
wait_time is an attribute that a User class may define. A User’s wait_time method is used to determine for how long a simulated user should wait between executing tasks. There are three built-in wait time functions:
- constant for a fixed amount of time
- between for a random time between a min and max value
- constant_pacing for an adaptive time that ensures the task runs (at most) once every X seconds
As we are doing load testing we don’t wait between tasks. This is achieved by using a built-in function constant and passing a zero (0) as a parameter.
tasks is an attribute to set the tasks to be performed by the User. The tasks attribute is either a list of Tasks or a <Task: int> dict, where Task is either a Python callable or a TaskSet class.
class MyTaskSet(TaskSet):For defining tasks, we are inheriting the TaskSet set class in a user-defined class named MyTaskSet.
class MyTaskSet(TaskSet):
...
def on_start(self):
MyTaskSet.user_count += 1
self.user_id = 'User{}'.format(MyTaskSet.user_count)
The on_start method is called when a simulated user starts executing that TaskSet. Similarly, there is an on_stop method which is called when the simulated user stops executing that TaskSet.
class MyTaskSet(TaskSet):
...
@task
def run_func1(self):
...
@task(3)
def run_func2(self):
....
To specify a method of the MyTaskSet class as a task for the User we have to decorate the method with the @task decorator. @task takes an optional weight argument that can be used to specify the task’s execution ratio. Here run_func2 will have thrice the chance of being picked as run_func1.
class MyTaskSet(TaskSet):
...
@task
def run_func1(self):
print("{} Executing run_func1 task ...".format(self.user_id))
start_time = time.time()
# TODO: your code here!
return_code = func1()
end_time = time.time()
total_time = int((end_time - start_time) * 1000)
if return_code == 0:
events.request_success.fire(request_type='func1',
name=self.user_id,
response_time=total_time,
response_length=0)
else:
events.request_failure.fire(request_type='func1',
name=self.user_id,
response_time=total_time,
response_length=0,
exception=Exception())
print("{} DONE run_func1 task ...".format(self.user_id))
Both the tasks here are similar except that the first one calls func1 and the second calls func2 from my_module.py at the point where it’s written TODO. In short, each task calculates the execution time of the function and fires either the success or failure request depending on the return value of the function.
Test
Let’s test it by running the following command.
locust -f locustfile.py --csv=locustdemo --headless -t 10m -u 10 -r 5 --stop-timeout 99
You should be able to see the output similar to the following:
[2020-10-31 15:59:04,235] user-desktop/INFO/locust.main: Run time limit set to 600 seconds
[2020-10-31 15:59:04,235] user-desktop/INFO/locust.main: Starting Locust 1.3.1
[2020-10-31 15:59:04,236] user-desktop/INFO/locust.runners: Spawning 10 users at the rate 5 users/s (0 users already running)...
The following parameters have been passed to the locust command.
-f: Absolute/Relative path to the locustfile to run.
--csv: Prefix of the three csv files generated
--headless: Required to run without the web interface
-t: Time for which the test should run. Here 10m means 10 minutes.
-u: Number of concurrent locust users
-r: The rate per second in which the users are spawned
--stop-timeout: Number of seconds to wait for a simulated user to complete any executing task before exiting
After the test runs you will find three CSV files generated – locustdemo_stats.csv, locustdemo_failures.csv, and locustdemo_stats_history.csv. The first two files will contain the stats and failures for the whole test run. The locustdemo_stats_history.csv will get new rows with the current (10-second sliding window) stats appended during the whole test run.
- Conclusion
Locust is a very nice and handy tool for doing load testing when it comes to Python. Testing REST APIs is very simple. But to test non-RESTful APIs might be a little harder. We have shown you one way to test non-RESTful APIs. There are other tools like JMeter for load testing. JMeter has a high learning curve that you face each time when it’s required to test something more complicated than a “Hello World” application. So, what you should choose quite depends on you.
Thanks for reading!!!
Author: Souvik Golui
