Profiling with locust

This is part of the Semicolon&Sons Code Diary - consisting of lessons learned on the job. You're in the web-development category.

Last Updated: 2024-11-21

Locust is a python tool for profiling that has both a CLI and a web interface (for graphs). It works just as well if your site is not built in python

Here is a typical command locust --headless --host http://localhost:8000 --users 200 --spawn-rate 3 --run-time 300s --locustfile locustfile.py

Complications that this system needs to deal with: - rate-limiting - CSRF tokens - referer checking over HTTPS

A great thing about profiling like this is that the results can also be used as a tool for integration testing the website after far-reaching changes, such as Django upgrades.

from locust import HttpUser, between, tag, task

LOCUST_USER_AGENT = "Locust" # Within the source code you may disable some features if this is set

class LoggedInUser(HttpUser):
    """
    When a test starts, locust will create an instance of this class for every user that
    it simulates.

    The file contains multiple methods decorated with `@task` or `@task(weight)` - e.g. `@task(4)`.
    Locust randomly picks one of these methods to execute for each simulated user,
    preferring tasks with higher weights assigned. Any method not decorated with `@task` is
    a helper method.

    Some tasks are decorate with `@tag`. This can be used with the CLI to target or exclude
    these sets of tasks.

    If you need to debug this code, you can inspect the response of any `.get()` or `.post()`
    method and then call `.status_code` or `.text` on it.
    """

    # The user we log in as.
    USERNAME = "x"
    PASSWORD = "y"

    # This is how long the simulated user waits between executing each task.
    # These times were chosen to be within the bounds set by `rate_limit.py`
    wait_time = between(2, 10)

    def on_start(self):
        """
        This method is called when a simulated user starts up.
        """
        self.log_in()

    @task(1)
    def browse_static_pages(self):
        self.get("/")
        self.get("/about/")

    @task(5)
    def browse_blog(self):
        self.get("/blog/")

    @tag('admin')
    @task(1)
    def use_admin_area(self):
        for url in ADMIN_URLS:
            self.client.get(url)

    def get(self, url):
        headers = {
            "User-Agent": LOCUST_USER_AGENT,
            "Referer": self.client.base_url,
        }
        return self.client.get(url, headers=headers)

    def post(self, url, data):
        headers = {
            "User-Agent": LOCUST_USER_AGENT,
            "Referer": self.client.base_url,
        }
        # We return the response object in order to enable inspection in the caller code.
        return self.client.post(url, data, headers=headers)

    def log_in(self):
        response = self.get("/login/")
        csrftoken = response.cookies["csrftoken"]
        self.post(
            "login/",
            {
                "login": self.USERNAME,
                "password": self.PASSWORD,
                "csrfmiddlewaretoken": csrftoken,
            },
        )