I am trying to setup Locust for generating load (in a distributed manner) for my service. I am using a custom shape to ramp up the traffic, sustain it for a time period and then ramp down.
Now my use case is that I need custom requests to be sent (these requests are generated by a separate PySpark process and there are multiple files) and want those files to be roughly equally distributed amongst each worker in a non-overlapping way.
My problem is the on_start
method is getting called per User created instead of per worker so I am unable to figure out a way to share one file per worker to be used by multiple users spawned on the same worker.
Below is my code currently.
from locust import HttpUser, task, between, LoadTestShape, constant_pacing
from locust.env import Environment
pacing_factor = 0.5 # 2 requests per sec. Each request having a gap of 500ms.
rps_per_user = 1 / pacing_factor
class CustomLoadShape(LoadTestShape):
ramp_up_duration = 30 # Time to ramp up (seconds)
hold_duration = 90 # Hold at peak load (seconds)
ramp_down_duration = 30 # Time to ramp down (seconds)
peak_rps = 2000 # Target requests per second
min_rps = 20 # Start and end RPS
def tick(self):
run_time = self.get_run_time()
total_duration = self.ramp_up_duration + self.hold_duration + self.ramp_down_duration
if run_time < self.ramp_up_duration:
# Ramp up phase: Interpolates between min_rps and peak_rps
target_rps = self.min_rps + ((run_time / self.ramp_up_duration) * (self.peak_rps - self.min_rps))
elif run_time < self.ramp_up_duration + self.hold_duration:
# Steady-state phase
target_rps = self.peak_rps
elif run_time < total_duration:
# Ramp down phase: Interpolates between peak_rps and min_rps
elapsed_ramp_down = run_time - (self.ramp_up_duration + self.hold_duration)
target_rps = self.peak_rps - ((elapsed_ramp_down / self.ramp_down_duration) * (self.peak_rps - self.min_rps))
else:
return None # Stop the test
# Convert target RPS to users per worker. This assumes a 1 RPS per user. If its more (based on pacing_factor then
# we should adjust the final number of users.
users_per_worker = int(target_rps)
users_per_worker = users_per_worker / rps_per_user
return users_per_worker, users_per_worker # (user count, spawn rate)
class MyUser(HttpUser):
wait_time = constant_pacing(pacing_factor)
host = "http://127.0.0.1:8000"
def on_start(self):
self.worker_id = self.environment.runner.worker_index
print(f"I'm worker {self.worker_id}")
@task
def send_request(self):
self.client.get("/students")
Is there any example (I couldn't find any) or has anybody done something that achieves the same ?