Кастомные профили нагрузки через LoadTestShape.tick(), step load, spike, ramp-up/down паттерны.
LoadTestShapeпозволяет полностью контролировать, сколько пользователей активно в каждую секунду теста. Вместо простого линейного нарастания вы описываете любой профиль нагрузки — step load, spike, ramp-up/down — прямо в коде.
Стандартный запуск Locust (--users 100 --spawn-rate 10) создаёт линейный рост от 0 до 100 пользователей и держит их до остановки. Реальные нагрузочные сценарии сложнее:
Все эти профили реализуются через LoadTestShape.
Чтобы создать кастомный профиль, нужно наследоваться от 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, то есть сколько пользователей добавлять/убирать в секунду при изменении целевого числа.
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.
Типичный производственный профиль для проверки стабильности: нагрузка медленно растёт, держится на плато, затем снижается:
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().
Иногда нужно ограничить тест по времени без сложного профиля. Для этого используйте --run-time в CLI или проверяйте get_run_time() в tick():
locust -f locustfile.py --headless --users 100 --spawn-rate 10 --run-time 5mЕсли LoadTestShape уже возвращает None по таймеру — флаг --run-time избыточен, но не вредит.
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 Nonetick() может принимать решения на основе текущей статистики теста — например, остановиться, если 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, 10self.runner.stats.total.fail_ratio возвращает долю ошибочных запросов от всех. Это мощный инструмент для автоматического обнаружения точки деградации.
Вопросы ещё не добавлены
Вопросы для этой подтемы ещё не добавлены.