LoadTestShape: профили нагрузки и spike-тесты

LoadTestShape позволяет полностью контролировать, сколько пользователей активно в каждую секунду теста. Вместо простого линейного нарастания вы описываете любой профиль нагрузки — step load, spike, ramp-up/down — прямо в коде.

#1. Зачем нужен LoadTestShape

Стандартный запуск Locust (--users 100 --spawn-rate 10) создаёт линейный рост от 0 до 100 пользователей и держит их до остановки. Реальные нагрузочные сценарии сложнее:

Step load test — нагрузка ступенями (100 → 200 → 300 → ...) для поиска точки деградации
Spike test — резкий скачок нагрузки в 5–10 раз для проверки устойчивости к пиковому трафику
Soak test — умеренная нагрузка на длительное время для обнаружения утечек памяти
Ramp-up/down — плавный подъём, удержание и спуск для измерения стабильности

Все эти профили реализуются через LoadTestShape.

#2. Базовый интерфейс

Чтобы создать кастомный профиль, нужно наследоваться от LoadTestShape и реализовать метод tick(). Locust вызывает tick() каждую секунду и ожидает кортеж (user_count, spawn_rate). Если tick() возвращает None, тест останавливается:

from locust import LoadTestShape

class StepLoadShape(LoadTestShape):
    """
    Ступенчатый рост нагрузки: каждые 60 секунд добавляем 100 пользователей.
    """
    step_duration = 60   # секунды на ступень
    step_users = 100     # пользователей на ступень
    max_users = 500      # максимум пользователей

    def tick(self):
        run_time = self.get_run_time()
        current_step = run_time // self.step_duration
        user_count = (current_step + 1) * self.step_users

        if user_count > self.max_users:
            return None  # тест завершён

        return user_count, self.step_users

Метод get_run_time() возвращает количество секунд с начала теста. Второй элемент кортежа — spawn_rate, то есть сколько пользователей добавлять/убирать в секунду при изменении целевого числа.

#3. Spike test

Spike test проверяет, как система переживает внезапный скачок нагрузки. Классический профиль: низкая базовая нагрузка → мгновенный пик → возврат к норме:

from locust import LoadTestShape

class SpikeShape(LoadTestShape):
    """
    Базовая нагрузка 50 пользователей. Через 60 секунд — пик 500. Через 30 секунд пика — возврат к 50.
    """
    def tick(self):
        run_time = self.get_run_time()

        if run_time < 60:
            return 50, 10     # базовая нагрузка
        elif run_time < 90:
            return 500, 200   # пик: быстро нагоняем до 500
        elif run_time < 180:
            return 50, 200    # возврат: быстро снижаем до 50
        else:
            return None       # конец теста

Высокий spawn_rate (200) при пике означает, что 500 пользователей появятся почти мгновенно — именно это нужно для spike test.

#4. Ramp-up с plateau

Типичный производственный профиль для проверки стабильности: нагрузка медленно растёт, держится на плато, затем снижается:

from locust import LoadTestShape

class RampUpPlateau(LoadTestShape):
    """
    Фаза 1 (0–120 с): рост от 0 до 200 пользователей.
    Фаза 2 (120–420 с): удержание 200 пользователей (плато).
    Фаза 3 (420–540 с): снижение до 0.
    """
    stages = [
        {"duration": 120, "users": 200, "spawn_rate": 5},
        {"duration": 420, "users": 200, "spawn_rate": 5},
        {"duration": 540, "users":   0, "spawn_rate": 5},
    ]

    def tick(self):
        run_time = self.get_run_time()
        for stage in self.stages:
            if run_time < stage["duration"]:
                return stage["users"], stage["spawn_rate"]
        return None

Этот паттерн удобен тем, что список stages легко модифицировать без изменения логики tick().

#5. Использование run_time limit

Иногда нужно ограничить тест по времени без сложного профиля. Для этого используйте --run-time в CLI или проверяйте get_run_time() в tick():

locust -f locustfile.py --headless --users 100 --spawn-rate 10 --run-time 5m

Если LoadTestShape уже возвращает None по таймеру — флаг --run-time избыточен, но не вредит.

#6. Совместное использование с классами пользователей

LoadTestShape работает вместе с классами пользователей в одном locustfile. При использовании нескольких классов пользователей user_count в tick() задаёт суммарное число — Locust распределяет их по классам пропорционально атрибуту weight:

from locust import HttpUser, LoadTestShape, task, between

class ReadUser(HttpUser):
    weight = 3
    wait_time = between(1, 2)

    @task
    def read(self):
        self.client.get("/api/items")

class WriteUser(HttpUser):
    weight = 1
    wait_time = between(2, 5)

    @task
    def write(self):
        self.client.post("/api/items", json={"name": "item"})

class MyShape(LoadTestShape):
    def tick(self):
        run_time = self.get_run_time()
        if run_time < 120:
            return 100, 10  # 75 ReadUser + 25 WriteUser
        return None

#7. Доступ к статистике внутри tick

tick() может принимать решения на основе текущей статистики теста — например, остановиться, если error rate превысил порог:

from locust import LoadTestShape

class AdaptiveShape(LoadTestShape):
    def tick(self):
        run_time = self.get_run_time()
        if run_time > 300:
            return None

        stats = self.runner.stats.total
        if stats.num_requests > 100 and stats.fail_ratio > 0.1:
            # Более 10% ошибок — останавливаем тест досрочно
            return None

        user_count = min(int(run_time / 10) * 10, 200)
        return user_count, 10

self.runner.stats.total.fail_ratio возвращает долю ошибочных запросов от всех. Это мощный инструмент для автоматического обнаружения точки деградации.