Ядро

Нагрузочное тестирование Pub/Sub

perfscale нагружает pub/sub-месседжинг через встроенный движок шагов: публикует пачки сообщений в subject, ждёт подходящие сообщения и измеряет как пропускную способность публикации, так и сквозную задержку (публикация → получение) — наряду с вашими метриками HTTP/gRPC/WebSocket.

Один шаг = один обмен: (опционально) подписка → (опционально) публикация → (опционально) ожидание count подходящих сообщений, ограниченное таймаутом. Когда заданы обе стороны, подписка устанавливается первой, поэтому круговой обмен (roundtrip) в пределах одного subject всегда видит свои собственные сообщения.

Эта страница — руководство. Параметры и выходные данные шага описаны в справочнике действий.

Драйверы

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

Драйвер	Транспорт
`memory` (по умолчанию)	Внутрипроцессная шина broadcast — брокер не нужен. Один канал на subject, общий для всех VU в процессе: публикация одного VU доходит до каждого VU, подписанного на этот subject (fan-out между VU — это возможность, а не утечка)
`nats`	Реальный сервер NATS (core NATS, без JetStream) через `async-nats`

Вендорские драйверы — Kafka, Redis, MQTT — это pro-capability, регистрируемые на том же драйверном шве; их настройки передаются в объекте options шага. Неизвестное значение driver завершает шаг ошибкой со списком зарегистрированных драйверов, так что всегда понятно, какую сборку вы запускаете.

Шине memory не нужны ни url, ни брокер — это самый быстрый способ собрать логику теста, а под нагрузкой она показывает чистый потолок движка. Направьте тот же YAML на nats (или pro-драйвер), чтобы добавить реальный круговой путь через брокер; сравнение двух запусков изолирует цену брокера.

Круговой обмен в одном шаге

Публикуем пачку и ждём её на том же subject — базовая проверка здоровья и задержки брокера:

steps:
  - name: order events roundtrip
    use: std/pubsub@v1
    with:
      driver: nats
      url: nats://127.0.0.1:4222
      subject: orders.created
      publish:
        - '{"id":"ord-1","total":42.50}'
        - '{"id":"ord-2","total":17.00}'
      subscribe:
        count: 2                    # ждём оба сообщения
        until_contains: '"id"'      # каждое учтённое сообщение должно совпасть
        timeout_ms: 2000
    check:
      body_contains: ord-2

Шаг падает при ошибке подключения, ошибке публикации или таймауте подписки — ошибка сообщает, сколько из count сообщений пришло и сколько было отклонено матчером until_contains. Недополучение при подписке — это провал проверки (assertion), а не транспортная ошибка: прогон продолжается, а упавшие обмены попадают в pubsub_e2e_ms_failed.

Нагрузка только от продюсера

Долбим брокер без потребления — успех означает, что каждая публикация была принята:

steps:
  - name: produce order events
    use: std/pubsub@v1
    with:
      driver: nats
      url: nats://nats.internal:4222
      subject: orders.created
      publish:
        - '{"id":"ord-${seq}","total":${randf(10,100,2)}}'

Полезная нагрузка сообщений интерполируется, как в любом другом шаге — ${seq}, ${randf(10, 100, 2)}, ${{ vars.* }}, выходы предыдущих шагов — так что каждая итерация может публиковать свежую нагрузку.

Потребитель с проверками

Шаг только с подпиской — чистый потребитель: ждём конкретное событие и проверяем нагрузку. outputs отдаёт последующим шагам published, received, duration_ms и body (совпавшие нагрузки, склеенные переводами строк):

steps:
  - name: await shipment event
    use: std/pubsub@v1
    with:
      driver: nats
      url: nats://nats.internal:4222
      subject: orders.shipped
      subscribe:
        count: 1
        until_contains: '"id":"ord-1"'
        timeout_ms: 3000
    check:
      body_contains: ord-1
    outputs: shipment

Пара продюсер / потребитель

Реалистичный паттерн с брокером: один процесс генерирует нагрузку, другие потребляют. Два тестовых файла используют общий конфиг; запускайте их параллельно против одного брокера:

# producer.yaml
steps:
  - name: produce order events
    use: std/pubsub@v1
    with:
      driver: nats
      url: nats://127.0.0.1:4222
      subject: orders.created
      publish: ['{"id":"ord-${seq}"}']

# consumer.yaml — 10 событий за итерацию, иначе шаг падает
steps:
  - name: consume order events
    use: std/pubsub@v1
    with:
      driver: nats
      url: nats://127.0.0.1:4222
      subject: orders.created
      subscribe: { count: 10, until_contains: '"id"', timeout_ms: 30000 }

# config.yaml
vus: 4
duration: 30s

perfscale run -f consumer.yaml -c config.yaml &
perfscale run -f producer.yaml -c config.yaml

Потребитель, который не успевает, роняет свой шаг по таймауту подписки — это тот самый сигнал, по которому вы подбираете размер пула потребителей. (С драйвером memory та же пара работает внутри одного процесса: шина делает fan-out между VU.)

Метрики и пороги

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

pubsub_msgs_published — счётчик (суммируется за прогон); сообщения, принятые транспортом. Эмитируется каждым завершённым обменом, при успехе и при ошибке (0 для шагов только с подпиской). Шаг, не сумевший подключиться, не эмитирует метрик вовсе.
pubsub_msgs_received — счётчик; сообщения, учтённые в subscribe.count. Эмитируется только когда у шага есть блок subscribe.
pubsub_e2e_ms — HDR-гистограмма в мс (процентили в сводке); начало фазы публикации → сообщение обработано, один сэмпл на каждое совпавшее сообщение. Эмитируется только с subscribe и только когда совпало хотя бы одно сообщение. Для шагов только с подпиской это время ожидания (якоря в виде публикации нет).
pubsub_e2e_ms_failed — производная доля ошибок: один сэмпл 0/1 на вызов, эмитировавший e2e-сэмпл (1 = шаг упал, например ожидание подписки не набрало count). Ожидание без единого совпавшего сообщения сэмпла не записывает — о механике вычисления см. metrics.md.

Проверяйте прогон по ним с помощью std/thresholds@v1:

  - use: std/thresholds@v1
    with:
      pubsub_e2e_ms:
        - "p(95)<50"              # 95% сообщений доставляются за 50 мс
      pubsub_e2e_ms_failed:
        - "rate<0.01"             # менее 1% обменов с таймаутом/ошибкой
      pubsub_msgs_received:
        - "count>=10000"          # потребители реально успевали

Модель подключений

Каждый обмен — это полный цикл: подключение → подписка → публикация → сбор → разрыв. Подключения не переиспользуются между шагами и итерациями VU — под нагрузкой каждый VU открывает свежее подключение к брокеру на каждую итерацию (кроме memory, который работает внутри процесса). Это держит измерение e2e честным — стоимость подключения включена, — но брокеры видят реальный оборот подключений; подбирайте интенсивность теста соответственно и предпочитайте прогоны, ограниченные duration, когда направляете нагрузку на общий staging-брокер.

Ограничения

Драйвер nats — это core NATS: никакого JetStream, персистентности, стримов и очередей — подписчик, который ещё не подключился, пропускает сообщения (шаг подписывается первым, поэтому круговые обмены внутри шага безопасны).
until_contains — это совпадение по подстроке, а не JSONPath/regex — для проверок на уровне полей читайте body через шаг std/check@v1.
Один subject на шаг; fan-in по нескольким subject требует одного шага на каждый subject.