Locust Event System Overview and Usage

Locust's event system provides a mechanism to execute custom code at different stages of a load test, useful for extending functionality like logging, notifications, or special data processing.

init: Triggered when the Locust process starts, suitable for setting global configuration or initializing resources.

from locust import events
@events.init.add_listener
def on_locust_init(environment, **kwargs):
    print("Locust has started")

test_start and test_stop: Fired before a test begins and after it ends, allowing preparation of the test environment or cleanup of resources.

@events.test_start.add_listener
def on_test_start(environment, **kwargs):
    print("Test is starting")
@events.test_stop.add_listener
def on_test_stop(environment, **kwargs):
    print("Test has stopped")

request_success: Triggered when a request completes successfully, providing request type, name, response time, and length for performance metrics or logging.

@events.request_success.add_listener
def on_request_success(request_type, name, response_time, response_length, **kwargs):
    print(f"Success: {name}, Response time: {response_time}")

request_failure: Triggered when a request fails, supplying request type, name, response time, and exception details for error analysis.

@events.request_failure.add_listener
def on_request_failure(request_type, name, response_time, exception, **kwargs):
    print(f"Failure: {name}, Exception: {exception}")

quitting: Fired just before all Locust processes exit, useful for ensuring resources are released or final state is saved.

@events.quitting.add_listener
def on_quitting(environment, **kwargs):
    print("Locust is quitting")

worker_report: In distributed mode, triggered when a worker reports statistics to the master, suitable for custom reporting or cross‑node data synchronization.

@events.worker_report.add_listener
def on_worker_report(client_id, data, **kwargs):
    print(f"Received report from worker: {client_id}")

hatch_complete: Fired when all simulated users have been spawned and are ready, allowing confirmation that full load has been applied.

@events.hatch_complete.add_listener
def on_hatch_complete(user_count, **kwargs):
    print(f"All users have been hatched: {user_count}")

These events enable fine‑grained control and customization of Locust's behavior, supporting complex workflows such as integrating external monitoring tools, detailed performance logging, or dynamic test adjustments.