How to divide request data in a non-overlapping manner per worker in Locust - Stack Overflow

admin管理员组

文章数量:1318028

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 ?

本文标签： How to divide request data in a nonoverlapping manner per worker in LocustStack Overflow