Locust

Locust is an open source performance and load testing tool designed for HTTP and other protocols. It uses a developer-friendly approach where test scenarios are defined using regular Python code rather than XML, binary formats, or GUI-based configurations.

Key characteristics include:

• Python-based testing: Write scenarios using standard Python programming constructs (loops, conditionals, calculations). Because Locust runs users inside lightweight greenlets, you can write tests as normal blocking Python code without needing callbacks.
• Scalability: Built on an event-based architecture (using gevent), allowing a single process to handle thousands of concurrent users. It supports distributed testing across multiple machines to simulate hundreds of thousands of concurrent users.
• Execution Modes: Tests can be run via the command line (ideal for CI/CD) or through a web-based UI that provides real-time monitoring of throughput, response times, and errors. The UI also allows for dynamic load adjustment during a running test.
• Extensibility: The pluggable architecture allows you to test almost any system or protocol by writing a custom client, or by using community-created plugins.

You can customize the UI in two ways:

1. Extended Tabs: By default, these render a table based on extendedTables and extendedStats. However, you can provide a custom React component to render any UI within that tab.
2. The tabs prop: This allows you to completely control the order and visibility of tabs. You can use tabConfig to include base tabs (like stats or charts) and mix them with your own custom tab objects.

import { IRootState } from "locust-webui";
import { useSelector } from "react-redux";

// 1. Custom component using Redux state
function MyCustomTab() {
    const extendedStats = useSelector(
        ({ ui: { extendedStats } }: IRootState) => extendedStats
    );
    return <div>{JSON.stringify(extendedStats)}</div>;
}

// 2. Using extendedTabs with a custom component
const extendedTabs = [{
    title: "Content Length",
    key: "content-length",
    component: MyCustomTab
}];

// 3. Using the tabs prop for complete control
import LocustUi, { tabConfig } from "locust-ui";

const tabs = [
    tabConfig.stats,
    tabConfig.charts,
    {
        title: "Custom Tab",
        key: "custom-tab",
        component: MyCustomTab,
    },
];

function App() {
    return <LocustUi tabs={tabs} />;
}

Locust's default HttpUser uses python-requests, which is feature-rich but can be CPU-intensive for very high throughput. For scenarios requiring maximum requests per second (RPS) on limited hardware, use FastHttpUser.

FastHttpUser uses geventhttpclient instead of python-requests. It can increase the maximum RPS on a given hardware by as much as 5x-6x.

Key considerations:

• Response Times: FastHttpUser does not make individual requests faster; it only makes the load generator more efficient. Response times should be nearly identical to HttpUser as long as the load generator's CPU is not overloaded.
• CPU Monitoring: Check Locust's console output; it will log a warning if the load generator is limited by CPU.
• Scaling: For optimal performance, run one Locust process per CPU core.

from locust import task, FastHttpUser

class MyUser(FastHttpUser):
    @task
    def index(self):
        response = self.client.get("/")

MarkovTaskSet allows you to define a probabilistic sequence of tasks using a Markov chain. Instead of tasks being chosen randomly based on global weights, the next action is determined by the current state (the current task) and defined transition probabilities.

Requirements & Limitations:

• Reachability: All tasks must eventually be reachable from the first task.
• Transitions: At least one task must have transitions defined.
• Tags: Tags are not supported in MarkovTaskSet because they can invalidate the Markov chain logic.

Transition Decorators:

1. @transition(task_name, weight=1): Defines a single transition to a specific task.
2. @transitions(weights): Defines multiple possible next steps.
   • weights can be a dictionary: {"task_a": 3, "task_b": 1}
   • weights can be a list of names or tuples: ["task_a", ("task_b", 2)]

from locust import User, constant
from locust.user.markov_taskset import MarkovTaskSet, transition, transitions

class ShoppingBehavior(MarkovTaskSet):
    wait_time = constant(1)
    
    @transition("view_product")
    def browse_catalog(self):
        self.client.get("/catalog")
    
    @transitions({
        "add_to_cart": 3,  # 60% chance
        "browse_catalog": 1, # 20% chance
        "checkout": 1      # 20% chance
    })
    def view_product(self):
        self.client.get("/product/1")
    
    @transitions(["view_product", "checkout"])
    def add_to_cart(self):
        self.client.post("/cart/add", json={"product_id": 1})
    
    @transition("browse_catalog")
    def checkout(self):
        self.client.post("/checkout")

class ShopperUser(HttpUser):
    host = "http://localhost"
    tasks = {ShoppingBehavior: 1}

When standard user counts and spawn rates are insufficient, you can use a LoadTestShape class to gain full control over the load profile (e.g., creating spikes or custom ramps).

To implement a custom shape:

1. Inherit from LoadTestShape in your locust file.
2. Implement a tick() method. This method is called approximately once per second and must return a tuple: (user_count, spawn_rate). To stop the test, return None.
3. Use self.get_run_time() within tick() to determine how long the test has been running.
4. (Optional) Use self.get_current_user_count() to monitor active users, which is useful for ensuring a stage is fully loaded before moving to the next.

If you want to create a base class for reuse without Locust attempting to use it as the active shape, set abstract = True.

class MyCustomShape(LoadTestShape):
    time_limit = 600
    spawn_rate = 20
    
    def tick(self):
        run_time = self.get_run_time()

        if run_time < self.time_limit:
            # User count rounded to nearest hundred.
            user_count = round(run_time, -2)
            return (user_count, self.spawn_rate)

        return None

Locust provides an event system that allows you to hook into various stages of the test lifecycle. You can use @events.<event_name>.add_listener to register functions that execute when specific events occur.

Common use cases include:

• events.request: Triggered after a request is completed. Useful for custom logging or monitoring.
• events.test_start / events.test_stop: Triggered at the beginning and end of a test run. Useful for setup/cleanup.
• events.init: Triggered when Locust is initialized. Useful for adding web routes or spawning background tasks.
• events.init_command_line_parser: Used to add custom CLI arguments.

Note on Distributed Mode: When running in distributed mode, use isinstance(environment.runner, MasterRunner) to ensure certain logic (like setup or monitoring) only runs on the master node and not on every worker.

from locust import events
from locust.runners import MasterRunner

@events.request.add_listener
def my_request_handler(request_type, name, response_time, response_length, response, 
                       context, exception, start_time, url, **kwargs):
    if exception:
        print(f"Request to {name} failed with exception {exception}")
    else:
        print(f"Successfully made a request to: {name}")

@events.test_start.add_listener
def on_test_start(environment, **kwargs):
    if not isinstance(environment.runner, MasterRunner):
        print("Beginning test setup on worker")