Testing other systems using custom clients

Locust was built with HTTP as its main use case but it can be extended to load test almost any system. You do this by writing a custom client that triggers request


Any protocol libraries that you use must be gevent-friendly (use the Python socket module or some other standard library function like subprocess), or your calls are likely to block the whole Locust/Python process.

Some C libraries cannot be monkey patched by gevent, but allow for other workarounds. For example, if you want to use psycopg2 to performance test PostgreSQL, you can use psycogreen.

Example: writing an XML-RPC User/client

Lets assume we had an XML-RPC server that we wanted to load test

import random
import time
from xmlrpc.server import SimpleXMLRPCServer

def get_time():
    return time.time()

def get_random_number(low, high):
    return random.randint(low, high)

server = SimpleXMLRPCServer(("localhost", 8877))
print("Listening on port 8877...")
server.register_function(get_time, "get_time")
server.register_function(get_random_number, "get_random_number")

We can build a generic XML-RPC client, by wrapping xmlrpc.client.ServerProxy

import time
from xmlrpc.client import ServerProxy, Fault

from locust import User, task

class XmlRpcClient(ServerProxy):
    XmlRpcClient is a wrapper around the standard library's ServerProxy.
    It proxies any function calls and fires the *request* event when they finish,
    so that the calls get recorded in Locust.

    def __init__(self, host, request_event):
        self._request_event = request_event

    def __getattr__(self, name):
        func = ServerProxy.__getattr__(self, name)

        def wrapper(*args, **kwargs):
            start_time = time.monotonic()
            request_meta = {
                "request_type": "xmlrpc",
                "name": name,
                "response_length": 0,  # calculating this for an xmlrpc.client response would be too hard
                "response": None,
                "context": {},  # see HttpUser if you actually want to implement contexts
                "exception": None,
                request_meta["response"] = func(*args, **kwargs)
            except Fault as e:
                request_meta["exception"] = e
            request_meta["response_time"] = (time.monotonic() - start_time) * 1000
            self._request_event.fire(**request_meta)  # This is what makes the request actually get logged in Locust
            return request_meta["response"]

        return wrapper

class XmlRpcUser(User):
    A minimal Locust user class that provides an XmlRpcClient to its subclasses

    abstract = True  # dont instantiate this as an actual user when running Locust

    def __init__(self, environment):
        self.client = XmlRpcClient(self.host, request_event=environment.events.request)

# The real user class that will be instantiated and run by Locust
# This is the only thing that is actually specific to the service that we are testing.
class MyUser(XmlRpcUser):
    host = ""

    def get_time(self):

    def get_random_number(self):
        self.client.get_random_number(0, 100)

For more examples, see locust-plugins