Add new benchmarks

[ApusBerliozi]

Description

Two new scenarios per broker (Kafka, Confluent, NATS, RabbitMQ, Redis), each as a FastStream-vs-raw-client pair:

1. test_metrics — consume/echo loop with observability enabled: *PrometheusMiddleware + *TelemetryMiddleware on the FastStream side, and the equivalent Prometheus MetricsManager calls + an OpenTelemetry process span/histogram hand-rolled around the raw-client handler. Measures the throughput cost of metrics + tracing.

2. test_sql — same loop plus a Postgres lookup (SELECT ... FROM users WHERE name = $1) on every message via asyncpg, sharing one self.sql_pool per case. Measures throughput when the handler does real I/O.

The existing test_basic / test_pydantic / test_msgspec cases were also restructured to the same TestFaststream<Broker><Variant>Case / TestPure<Broker><Variant>Case layout (the standalone test_aiokafka / test_nats / test_aiopika / test_confluent files were folded in), so every scenario now compares FastStream against the underlying client directly.

Supporting pieces:

• metrics.py — shared Prometheus CollectorRegistry (scrape endpoint on :8001) and an OTLP tracer provider exporting to http://localhost:4318/v1/traces.
• sql.py — find_user_by_name(name, pool) helper and the BENCHMARK_PG_DSN config.
• seed_db.sql — creates and seeds the users table (John/Mike + 10k filler rows, indexed on name); auto-runs via the postgres service.
• docker-compose.yaml — added postgres (mounts the seed script) and otel-collector services.
• pyproject.toml — new benchmarks dependency group with the broker clients, asyncpg, msgspec, OTLP exporter, and test runner.

Each *_cases/__init__.py now re-exports its case classes and a CASES dict for iteration.

Fixes #3084

Type of change

• New feature (a non-breaking change that adds functionality)

Checklist

• My code adheres to the style guidelines of this project (just lint shows no errors)
• I have conducted a self-review of my own code
• I have made the necessary changes to the documentation
• My changes do not generate any new warnings
• I have added tests to validate the effectiveness of my fix or the functionality of my new feature
• Both new and existing unit tests pass successfully on my local environment by running just test-coverage
• I have ensured that static analysis tests are passing by running just static-analysis
• I have included code examples to illustrate the modifications

[borisalekseev]

Please, add custom compose profile or separate compose for benchmarks

[Lancetnik]

I would not merge new scenarios on top of the current harness: the ping-pong loop measures broker round-trip, not FastStream, and in its current state the branch does not run. test_sql as the "everything on" case (pydantic + otel + prometheus + postgres) is the right scenario and stays.

Blocking

• benchmarks/kafka_cases/test_basic.py:24, :69 and every other case: setup_method became async. Neither pytest nor bench.py:69 awaits it, so self.EVENTS_PROCESSED and self.broker never exist and every case fails with AttributeError. The "tests pass" checkbox cannot be true.
• benchmarks/bench.py:65 imports rabbit_cases.test_aiopika.TestRabbitCase, which this PR deletes. The new *_CASES dicts are not used anywhere.
• benchmarks/kafka_cases/test_basic.py:90: the raw consumer uses auto_offset_reset="earliest" with no group_id, FastStream defaults to latest. After one 10-minute run the topic holds a few hundred thousand echoes, so the raw case starts with all of them in flight and the FastStream case with one. The two numbers are not comparable.
• benchmarks/README.md:16 still installs faststream==0.6.0rc0 from PyPI. That benchmarks a release, not the branch.
• benchmarks/kafka_cases/test_sql.py:11-16 mixes import roots: from metrics import needs benchmarks/ on sys.path, from benchmarks.sql import needs the repo root.

nit: test_sql.py:41 has the comment copied from test_metrics, so the two rows are indistinguishable in the CSV. nit: metrics.py:10 starts an HTTP server at import time; otel-collector is pinned to latest.

Methodology

With one message in flight, EPS = 1 / RTT. FastStream, the middlewares and even the Postgres query add a few serial milliseconds on top of a round-trip the broker owns. CPU never becomes the bottleneck, and CPU is where library overhead lives. So the harness needs two families of runs, labelled separately in the results and never compared to each other.

Drain. A pre-fill step publishes N messages, the consumer drains them, we measure how fast. Storage transports only: Kafka / Confluent topic, NATS JetStream stream, Redis Streams, durable RabbitMQ queue. broker_type says which ("NATS JetStream", "Redis Streams").

• Pre-fill is a separate script, not part of the case. It verifies the count afterwards (end offsets, nats stream info, XLEN, queue message_count) and that number is the run's N.
• Start with 200k to 1M and tune per broker. 10M pushes RabbitMQ into paging and flow control, and then we measure the broker under memory pressure.
• Fresh topic / stream / queue per run, unique name or explicit delete. Otherwise the leftover problem from test_basic.py:90 comes back.
• Every payload carries a sequence number. The run ends when message N is processed, not on a timer. The sequence numbers also give us drops and duplicates.
• Same work in the handler on both sides: parse, count, the SQL query in the sql case. No echo publish, or we are measuring the producer again.
• Redis: no MAXLEN ~ on pre-fill, approximate trimming drops part of the backlog silently. 1M × ~250 B is a few hundred MB of Redis memory, budget it in compose.

Latency. The current echo loop stays as a latency benchmark for core NATS and Redis pub/sub, which have no storage, and reports per-message RTT percentiles instead of EPS.

One subscriber, one queue, one process per run. Several subscribers or queues are a separate scenario later (10 subscribers on 10 subjects to see dispatch cost), not part of the baseline.

What to record per run

Column	How
messages_per_s	N / (t_last - t_first), connect and subscribe excluded
cpu_ms_per_1k	psutil.Process().cpu_times() user+system delta between first and last message, divided by N / 1000. The most stable overhead signal we can get, and it shows even when broker-bound
peak_rss_mb	resource.getrusage(RUSAGE_SELF).ru_maxrss at the end
rss_mb_median	memory_info().rss sampled once a second, median
rtt_p50_ms, rtt_p95_ms, rtt_p99_ms	latency family only: publish timestamp in the payload, perf_counter() on receive, all samples kept, percentiles at the end
dropped, duplicated	from sequence numbers
scenario, transport, implementation, N, run_idx, prefetch, batch, ack_mode	so one row is reproducible on its own

Five repetitions per (scenario, transport, implementation), median and stddev in the summary. Warm-up either as a separate 10k run or by discarding the first ~5% of messages. Consumer pinned to its own core (psutil.Process().cpu_affinity), broker, Postgres and otel-collector limited to other cores through compose cpus. Right now the BatchSpanProcessor export thread and Postgres share the core with the code under test.

Equal settings on both sides

Defaults differ between FastStream and the raw clients, so every one of these is set explicitly in both implementations:

Transport	Must match
Kafka / Confluent	auto_offset_reset="earliest", fresh group_id per run, max_poll_records, fetch_max_bytes / max_partition_fetch_bytes, one partition (or partitions == max_workers), commit mode
NATS JetStream	consumer type: FastStream is push unless pull_sub= is given, for drain use pull on both sides; DeliverPolicy.All, max_ack_pending, ack policy, pull batch size, fresh durable per run
Redis Streams	XREAD vs XREADGROUP, same on both; start id 0-0, or group created at 0 (StreamSub starts at $, or > with a group); same COUNT (FastStream: batch=True + max_records) and BLOCK; ack per message or none
RabbitMQ	prefetch_count, queue type (classic / quorum), durable, ack mode (msg.process() vs FastStream ack)

Handler concurrency has to be equal too. aio-pika and FastStream both run handlers concurrently up to prefetch, the Kafka loops are sequential on both sides. In the sql case the asyncpg pool must be at least prefetch on both sides, or the pool becomes the bottleneck for whichever side has more concurrency.

Instrumentation as well: the FastStream middlewares also emit publish spans and metrics with context propagation, the hand-rolled raw side only instruments process. Without the echo publish most of that difference disappears, but check that the raw side records the same set of spans and metrics per consumed message as the middleware does.

How to land it

Two PRs. First the harness: pre-fill script, the columns above, one broker (Kafka is the simplest to verify) with both implementations, bench.py iterating the cases, README running the branch through the benchmarks dependency group. Then port the other four brokers and the sql / metrics scenarios onto it. Doing all of it in this PR is possible, but the diff is already 3.8k lines and the harness change is the part that needs review.

Thanks for taking the benchmarks on. The direction, FastStream against the raw client for every scenario, is the right one.