feat(observability): Prometheus metrics, Grafana dashboards and batch-aware accounting - #113
Merged
Conversation
XTLine
force-pushed
the
perf/non-persistent-fanout-worker
branch
from
August 21, 2026 08:51
67a95ff to
c55198e
Compare
XTLine
force-pushed
the
perf/observability-stack
branch
2 times, most recently
from
August 21, 2026 10:10
786ab3d to
bd847d9
Compare
XTLine
force-pushed
the
perf/non-persistent-fanout-worker
branch
from
August 21, 2026 10:10
c55198e to
9e77f9e
Compare
Default options keep every SST open (max_open_files=-1) with an unbounded block cache, so accumulated data (hundreds of GB across perf runs) let anon memory grow past the 4G cgroup wall and OOM-kill the broker during high-load fanout scenarios. - max_open_files 512: resident table readers bounded - block cache LRU 256MB, index/filter blocks included - bounded memtable (64MB x2) and background jobs (4)
- Pass `cleanup=True` on restart in broker lib to clear /tmp data. - Ensure logs and timeseries are saved before storage is wiped. - Reset broker storage in a `finally` block in the stress test.
Reworks metric extraction to use the median of steady-state interval windows rather than relying solely on aggregated stats. Also adds detailed logging and exit-code explanations for E2E process failures to improve debuggability.
Adds an external broker backend to target standalone instances without lifecycle management. Also introduces local resource constraints via systemd-run and taskset, and adds a new 10GB solo-consumer backlog drain scenario.
Sample cgroup memory stats (anon/file) when broker is managed by systemd. Reset sampler between scenarios so metrics cover only the scenario window.
…ge-trigger - Unacked budget gate (max_unacked_messages=50000, aligned with Pulsar maxUnackedMessagesPerConsumer): dispatch pauses when a consumer's delivered-but-unacked set exceeds the budget, cutting the fuel of the redelivery/Flow feedback loop. - send-failure branches switch continue->break: a failed send no longer re-pops the same entry within the same pass, eliminating the read-fail-restore spin (150k entrylog reads/s of allocations). - Merge-trigger dispatch (Pulsar shouldRescheduleRead pattern): triggers arriving while a loop runs only set a flag; the pass checks it after releasing the guard and runs one more pass. No dropped triggers, no fixed time constant, empty passes never read storage. - Add Consumer::pending_acks_len accessor + unit test (shared_dispatch_pauses_when_unacked_budget_exceeded).
glibc malloc keeps freed memory in per-thread arenas and only returns the top chunk to the OS, so high-allocation-churn workloads (300k msg/s x ~5 allocs/message) pin RSS at the peak footprint. Under a 4G cgroup limit the broker OOM'd with 4.16GB anon while every data structure was empty (verified via depth instrumentation). jemalloc uses size-class buckets plus dirty-page decay: idle pages are munmapped back to the OS within ~10s, so RSS tracks live data. Same workload drops anon from 4.16GB to ~88MB and the full 10-scenario suite passes under 4G. - #[global_allocator] Jemalloc in main.rs - unprefixed_malloc_on_supported_platforms: RocksDB and other C libs allocate through jemalloc too - tikv-jemalloc-sys profiling feature enabled for MALLOC_CONF-based heap profiling (diagnostic runs only)
…e; move seek helpers into storage core
The per-Send hot path took Mutex<Storage> to clone the write-queue appender and held the topic write lock for the rate-limit check. Under 8-subscription fanout the storage mutex is saturated by dispatch reads and acks, so enqueueing producers queued behind it (P stuck at ~124k). - cache ConcurrentAppender once per connection at setup; Send path enqueues without locking storage - TopicPublishRateLimiter becomes interior-mutable (std Mutex behind &self) so validate_publish_rate runs under the topic read lock - producer publish_message rate check downgraded write->read lock
Shared-subscription acks past the mark-delete frontier were stored as one BTreeSet entry per position; 8-subscription fanout acking ~1M msg/s grew the set to gigabytes and OOM'd the broker. Store out-of-order acks as merged inclusive ranges (O(ranges) memory), mirroring Pulsar's individualDeletedMessages. - new RangeSet<K>: insert/contains/take_covering with adjacency merging - mark-delete advancement skips whole ranges in one step - persistence now encodes real ranges; decode coalesces legacy points
Layered diagnostic counters writing to /data/*_diag.txt to locate where the 16-consumer ack pipeline stalls (2100万 acks expected, storage sees <10k): - dispatch_diag: consumer send_message entry (per 100k) - ack_cmd_diag: handle_ack entry with ids/known/ack_type (per 100) - msg_acked_diag: message_acked entry with sub_type/is_persistent (per 100) - ack_drop_diag: dropped-ack points in message_acked (per 100k) - ack_enter_diag: storage ack_message/_shared entry (per 10k) - pending_acks_diag: PendingAcksMap len growth (per 100k insert) - cursor_diag: persist_state mark_delete/ranges CSV (per 100k) All probes are TEMPORARY and must be removed after root-cause.
ServerCnx::new gained a rocksdb-storage persistent_appender param in the enqueue-lock-removal change; build_test_connection was not updated, so the test profile failed to compile with rocksdb-storage.
…asts for clippy 1.98
…lock The backpressure change moved Flow handling into a spawned task, which split permit accounting in time: consumer-local permits were applied at frame processing while the dispatcher aggregate landed only when the task ran. Two races surfaced under CI timing: - key_shared: remove_consumer_with_recovery subtracts the consumer balance from the aggregate. With a refill Flow half-applied, the close zeroed the aggregate, and the flow task then dropped its own increment (contains_key guard), so the post-close redelivery dispatch exited on max_batch <= 0 and the surviving consumer never received the message. - non-persistent shared: send_messages dropped entries on a stale zero aggregate while the consumer actually held permits (the ack refill Flow from another connection had not been applied yet). Apply both permit layers synchronously under the subscription read lock (mutually exclusive with consumer removal writes), keep the dispatch trigger in its own task to preserve the event-loop starvation fix, and let per-consumer reservations instead of the aggregate fast path decide non-persistent drops. Verified locally: both previously-flaky tests now pass 30/30 runs (key_shared unacked redelivery preserves key owner, non-persist shared flow full consumer stops receiving); workspace tests and clippy 1.98 clean.
Rust keeps the OS default (Nagle on) for accepted sockets. The Pulsar protocol is request-response with small replies (Send -> SendReceipt, Ping -> Pong); with Nagle, each reply waits for the previous segment's ACK, adding ~1.4ms steady RTT. Netty defaults tcpNoDelay=true, so the Java broker never had this problem. Measured (8-sub non-persistent, 4 cores, same client): produce-only 417k -> 1000k msg/s (client cap), feed 397k -> 988k msg/s, p50 2.31ms -> 0.76ms.
The non-persistent SEND path ran fully inline in the connection task: topic.write() exclusive lock -> prepare -> dispatch_sequential().await (8 subscriptions, one lock + one dispatcher call per message) -> receipt, pipeline depth 1. Mirror the persistent write-queue shape instead: - per-topic bounded mpsc (4096) drained by one ordered worker; the connection task snapshots subscriptions under a short read lock, enqueues, and returns to reading frames (accept-ack receipt, same contract as the drop gate's fake receipts) - prepare_non_persistent_publish takes &self (the write lock was phantom: both callees only need &self) - worker drains up to 128 jobs per wakeup and groups entries per subscription: one lock + one vectorized send_non_persistent_entries per subscription per batch, amortizing wakeup and lock costs; per-subscription entry order still follows enqueue order (FIFO) - publish_message routes through the same queue with a oneshot completion reply so in-process callers and tests stay deterministic; topics built without a runtime keep the inline fallback - new test: fire-and-forget enqueue preserves per-subscription order (no-subscription publishes drain; 300-job FIFO; barrier semantics) Measured together with TCP_NODELAY fix (8-sub non-persistent, 4 cores): fanout feed 701k -> 988k msg/s, aggregate delivery 841k -> 1.18M msg/s, p99 119.5ms -> 17.5ms.
Clients send thousands of Flow commands per second under load; an INFO line each flooded the journal/PTY (10.5MB per 80s stress run) and the synchronous logger write stalled runtime workers under backpressure, observed as second-scale producer stalls. Ack handling already logs at debug.
XTLine
force-pushed
the
perf/observability-stack
branch
from
August 21, 2026 10:40
bd847d9 to
bdb3983
Compare
complete_persistent_send flushed a SendReceipt per completion, so with TCP_NODELAY every persistent message produced one small TCP segment (~200k pkt/s at max produce rate). That regressed bulk persistent produce ~29% (10GiB fill 204k -> 145k msg/s) when TCP_NODELAY landed. Feed each receipt into the write buffer instead and flush once per drained batch: the connection loop now drains up to 128 completions from conn_append_rx per wakeup, completes them all, and issues a single flush (mirroring write_message_batch on the consumer side). The bounded-drain path in handle_send and the test helper flush accordingly. Recovered from the 08-19 perf session (checkpoint b4c0b517 on the perf server); measured there at 10GiB fill 145k -> 277.9k msg/s (+36% over the pre-NODELAY baseline) with TCP_NODELAY retained. Verified locally: tests/persist 29/29, cargo test 128, clippy 1.98 clean.
Extract every Prometheus family into a dedicated workspace crate: - global registry + process collector served via prometheus-hyper - storage families: e2e/ledger write latency, entry size, write-queue batch metrics (observed from the single-writer worker thread) - PublishCommitObserver hook so the RocksDB write queue folds whole committed batches into one atomic update per topic - ledger-aware backlog_entries and stored_bytes on the storage contract with memory and RocksDB implementations plus reopen coverage
- pre-resolved TopicMetrics/SubscriptionMetrics handles on entities; label resolution happens once at creation, never per message - persistent publishes counted at write-queue batch commit, non-persistent in the fan-out worker, in-process at Producer::publish_message - consumer dispatch/ack hooks cache the handle via OnceLock + try_read so the dispatch path never awaits the subscription lock - 5s scrape aggregation task sets gauges (entity counts, backlog, unacked, storage size, windowed rates) entirely under try_read/try_lock with no broker lock held across awaits - [metrics] config section, GET /metrics listener in main, connection and rejection counters; dispatch-metrics log demoted to debug
Prometheus + Grafana with provisioning: topics dashboard (rates, backlog, unacked gate, latency/entry-size percentiles) and broker dashboard (connections, rejections, write-queue batches, process RSS/CPU)
_config_text rewrote every top-level `addr = ...` line, so the [metrics]
addr collapsed onto the protocol port: the two listeners raced for one
bind (broker exit, or a silently dead metrics endpoint), and the 127.0.0.1
binding was unreachable from the compose Prometheus via
host.docker.internal anyway.
- BrokerConfig.metrics_port derived as port + 1430 (6650 -> 8080,
6651 -> 8081, ... 6672 -> 8102); zero call-site churn.
- _config_text splits at the [metrics] header: main addr keeps
127.0.0.1:{port}, metrics addr becomes 0.0.0.0:{metrics_port}.
- compose Prometheus scrapes the six perf ports under job
pulsar-lite-perf (DOWN while no run is active - expected).
- grafana/README: live-view-during-perf-tests section.
Docker-backed perf runs use --network host, so the same ports apply.
…it paths Count client-visible messages (MessageMetadata.num_messages_in_batch, default 1) instead of entries everywhere: consumer stats, subscription metrics, non-persistent dispatcher dispatched/dropped counters, write-queue committed counters, and flow-control permits (try_reserve_dispatch now takes a permits argument; non-persistent paths deduct batch_count like the persistent shared/key_shared dispatchers already did). messages_in_batch moves to pulsar-lite-proto::codec so the storage layer can reuse it (added as an extra dep alongside managed-ledger in managed-ledger-rocksdb, whose types the crate still uses).
A failed try_lock during scrape meant "unknown", not zero: per-topic storage size, per-subscription backlog (now Option), and broker totals now skip the round instead of reporting a spurious dip to zero under publish-path contention.
XTLine
force-pushed
the
perf/observability-stack
branch
from
August 24, 2026 01:46
bdb3983 to
c0003ad
Compare
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
Add a dedicated
pulsar-lite-metricscrate and wire Prometheus export across broker layers, ship Grafana dashboards with a compose stack, and make message accounting batch-aware (count client-visible messages, not entries) across dispatch and commit paths. Tip PR of the perf/topic-write-queue stack.What Changed
pulsar-lite-metricscratePublishCommitObserverhook so the RocksDB write queue folds whole committed batches into one atomic update per topicbacklog_entries/stored_byteson the storage contract with memory and RocksDB implementations plus reopen coverageProducer::publish_messageOnceLock+try_readso the dispatch path never awaits the subscription lock[metrics]config section and aGET /metricslistener; connection and rejection countersMessageMetadata.num_messages_in_batch(default 1) instead of entries in consumer stats, subscription metrics, non-persistent dispatched/dropped counters, write-queue committed counters, and flow-control permits (try_reserve_dispatchtakes a permits argument)messages_in_batchmoves topulsar-lite-proto::codecso the storage layer can reuse itWhy
The perf work on this stack needed visibility into write-queue batching, backlog, unacked pressure, and memory that log scraping could not provide; pre-resolved handles and try-lock-only scraping keep the metrics path off the hot path. Batch-aware counts keep permits and rates meaningful for batched publishes, where entry counts under-count client-visible messages.
Verification
Developed and used locally as the observability layer for the perf runs on this stack (dashboards and scrape endpoints exercised during those runs; storage families covered by reopen tests). No fresh local run was performed for this PR; CI re-runs the workspace suite here.
Notes
perf/non-persistent-fanout-worker.pulsar-lite-perfjob shows DOWN while no perf run is active — expected.