Testing other systems using custom clients¶
Locust was built with HTTP as its main target. However, it can easily be extended to load test
any request/response based system, by writing a custom client that triggers
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 will block the whole Locust 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, can use psycogreen.
Sample XML-RPC User client¶
Here is an example of a User class, XmlRpcUser, which provides an XML-RPC client, XmlRpcUser, and tracks all requests made:
import time from xmlrpc.client import ServerProxy, Fault from locust import User, task, between class XmlRpcClient(ServerProxy): """ Simple, sample XML RPC client implementation that wraps xmlrpclib.ServerProxy and fires locust events on request_success and request_failure, so that all requests gets tracked in locust's statistics. """ _locust_environment = None def __getattr__(self, name): func = ServerProxy.__getattr__(self, name) def wrapper(*args, **kwargs): start_time = time.time() try: result = func(*args, **kwargs) except Fault as e: total_time = int((time.time() - start_time) * 1000) self._locust_environment.events.request_failure.fire( request_type="xmlrpc", name=name, response_time=total_time, exception=e ) else: total_time = int((time.time() - start_time) * 1000) self._locust_environment.events.request_success.fire( request_type="xmlrpc", name=name, response_time=total_time, response_length=0 ) # In this example, I've hardcoded response_length=0. If we would want the response length to be # reported correctly in the statistics, we would probably need to hook in at a lower level return wrapper class XmlRpcUser(User): """ This is the abstract User class which should be subclassed. It provides an XML-RPC client that can be used to make XML-RPC requests that will be tracked in Locust's statistics. """ abstract = True def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) self.client = XmlRpcClient(self.host) self.client._locust_environment = self.environment class ApiUser(XmlRpcUser): host = "http://127.0.0.1:8877/" wait_time = between(0.1, 1) @task(10) def get_time(self): self.client.get_time() @task(5) def get_random_number(self): self.client.get_random_number(0, 100)
If you’ve written Locust tests before, you’ll recognize the class called
ApiUser which is a normal
User class that has a couple of tasks declared. However, the
ApiUser inherits from
XmlRpcUser that you can see right above
XmlRpcUser is marked as abstract
abstract = True which means that Locust will not try to create simulated users from that class
(only of classes that extend it).
XmlRpcUser provides an instance of XmlRpcClient under the
XmlRpcClient is a wrapper around the standard
xmlrpc.client.ServerProxy. It basically just proxies the function calls, but with the
important addition of firing
events, which will record all calls in Locust’s statistics.
Here’s an implementation of an XML-RPC server that would work as a server for the code above:
import random import time from xmlrpc.server import SimpleXMLRPCServer def get_time(): time.sleep(random.random()) return time.time() def get_random_number(low, high): time.sleep(random.random()) 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") server.serve_forever()
For more examples, see locust-plugins