Use of different LoadTestShape classes in the same locust file

[irawildjunior]

Describe the solution you'd like

Customers need to periodically check if they can scale their web applications X times the current volume of requests per second to plan for scalability.
Thus, an observability tool is used to obtain throughput and latency SLAs, and the values ​​obtained are used as a baseline to validate the ability to scale.

For this, it would be necessary, in the same test package, to have different users that run different tasks sets, with different LoadShapes, as in the example below:

from locust import (
    HttpUser, 
    LoadTestShape, 
    constant_throughput, 
    task, 
    SequentialTaskSet
)

class ProductShape(LoadTestShape):
    user_count = 1000
    spawn_rate = 1000

    def tick(self):
        return (self.user_count, self.spawn_rate)


class OrderShape(LoadTestShape):
    user_count = 100
    spawn_rate = 100

    def tick(self):
        return (self.user_count, self.spawn_rate)


class ProductTaskSet(SequentialTaskSet):

    @task
    def get_home(self):
        self.client.get("/products/")
    

class OrderTaskSet(SequentialTaskSet):

    @task
    def get_home(self):
        self.client.get("/orders/")


class ProductSearchUser(HttpUser):
    shape = ProductShape
    wait_time = constant_throughput(0.1)
    tasks = [
        ProductTaskSet,
    ]


class OrderSearchUser(HttpUser):
    shape = OrderShape
    wait_time = constant_throughput(0.5)
    tasks = [
        OrderTaskSet,
    ]

[irawildjunior]

I was able to test scalability using just one LoadTestShape class, independently calculating the target troughput of each task, for a given target scale, using the function constant_throughput_at_scale(throughput).

from locust import (
    HttpUser, 
    LoadTestShape, 
    constant_throughput, 
    task, 
    SequentialTaskSet
)

time_limit = 120
scale = 10
user_classes_count = 2
users_per_class = 100


class ScaleTestShape(LoadTestShape):
    user_count = user_classes_count * users_per_class
    spawn_rate = user_classes_count * users_per_class

    def tick(self):
        run_time = self.get_run_time()

        if run_time >= time_limit:
            return None

        return (self.user_count, self.spawn_rate)


class ProductTaskSet(SequentialTaskSet):

    @task
    def get_home(self):
        self.client.get("/products")
    

class OrderTaskSet(SequentialTaskSet):

    @task
    def get_home(self):
        self.client.get("/orders")


def constant_throughput_at_scale(throughput):
    return constant_throughput(throughput * scale / users_per_class)


class ProductSearchUser(HttpUser):
    throughput = 0.67 #  SLA obtained in observability tool
    wait_time = constant_throughput_at_scale(throughput)
    tasks = [
        ProductTaskSet,
    ]


class OrderSearchUser(HttpUser):
    throughput = 2.5 #  SLA obtained in observability tool
    wait_time = constant_throughput_at_scale(throughput)
    tasks = [
        OrderTaskSet,
    ]

As we can see, the final throughput in the report (req / s) is 10 x (target scale) the SLAs.