feat(observability): Prometheus metrics, Grafana dashboards and batch-aware accounting

README.md

@@ -39,6 +39,7 @@ local setup cost:
 | Non-persistent topics | Dispatch-or-drop runtime semantics with coverage for flow control, disconnect/reconnect, ordering, dynamic consumers, and KeyShared routing. |
 | Persistent topics | RocksDB-backed managed-ledger style storage is available behind the `rocksdb-storage` feature. |
 | Subscription modes | Shared, Failover, Exclusive, and KeyShared are covered by Rust and Python integration tests. |
+| Monitoring | Prometheus metrics on `GET /metrics` (port 8080): broker/topic/subscription counters, backlog and unacked gauges, storage latency and entry-size histograms, process metrics. Optional Grafana stack under `grafana/`. |
 | Partitioned topics | Default partition metadata and partition topic routing are supported for local testing. |
 | Python package | Provides a small helper SDK that can start and manage a local broker process. |
 
@@ -110,6 +111,39 @@ Stop the broker:
 ../rust/pulsar-lite.sh stop
 ```
 
+## Monitoring
+
+The broker exposes Prometheus metrics on `GET /metrics` (default
+`0.0.0.0:8080`, same port model as native Pulsar's web service). Metric
+families reuse native Pulsar names and labels (`pulsar_rate_in`,
+`pulsar_subscription_back_log`, `pulsar_storage_write_latency`, ...), so
+existing dashboards and PromQL translate directly; extensions unique to
+Pulsar Lite use the `pulsar_lite_*` prefix.
+
+Quick check:
+
+```bash
+curl -s localhost:8080/metrics | grep -E '^pulsar_(broker|subscription)'
+```
+
+Configuration (`rust/pulsar-lite.toml`):
+
+```toml
+[metrics]
+enabled = true             # false: no listener, no scrape aggregation
+addr = "0.0.0.0:8080"
+cluster = "pulsar-lite"    # `cluster` label value on every family
+rate_window_secs = 60      # window for pulsar_rate_in-style gauges
+```
+
+A ready-to-run Prometheus + Grafana stack (with provisioned dashboards for
+topics and broker overview) lives under [`grafana/`](grafana/README.md):
+
+```bash
+docker compose -f grafana/docker-compose.yml up -d
+# Grafana http://localhost:3000 (admin/admin), Prometheus http://localhost:9090
+```
+
 ## Embedded Python Usage
 
 The Python helper can start a local broker for short-lived tests or examples:

grafana/README.md

@@ -0,0 +1,68 @@
+# Pulsar Lite Observability Stack
+
+A one-command Prometheus + Grafana stack that scrapes the broker's
+`GET /metrics` endpoint (default `0.0.0.0:8080`).
+
+## Quick start
+
+```bash
+# 1. Start the broker (metrics are on by default)
+../rust/target/release/pulsar-lite --config ../rust/pulsar-lite.toml
+
+# 2. Start the observability stack
+docker compose up -d
+
+# 3. Open
+#    Grafana    http://localhost:3000  (admin/admin; anonymous read enabled)
+#    Prometheus http://localhost:9090/targets — the pulsar-lite job must be UP
+```
+
+Dashboards are provisioned automatically (folder `Pulsar Lite`):
+
+- **Pulsar Lite / Topics** — publish/deliver rates and throughput, entity
+  counts, subscription backlog, unacked gate state, redelivery, storage
+  size, end-to-end and ledger write latency (P50/P99), entry-size
+  distribution.
+- **Pulsar Lite / Broker** — broker-level rates (counter `rate()` and
+  window-gauge views), connections and rejection reasons, backlog and
+  storage totals, write-queue batch sizes, process RSS/CPU.
+
+## Live view during perf tests
+
+The perf harness (`tests/perf/`) starts real brokers, each with a private
+metrics port derived from its protocol port (6651/6652 → 8081/8082,
+6661/6662 → 8091/8092, 6671/6672 → 8101/8102, bound on `0.0.0.0`). This
+stack scrapes all of them (`job="pulsar-lite-perf"`), so with the stack up
+you can watch a run live:
+
+```bash
+docker compose up -d                    # once
+python tests/perf/run_persistent_stress.py ...   # then run any scenario
+# Grafana http://localhost:3000 — dashboards show the running broker's series
+```
+
+Perf targets show DOWN in Prometheus `/targets` while no run is active —
+that is expected. Docker-backed perf runs use `--network host`, so the same
+ports apply.
+
+## Metric naming conventions
+
+- `pulsar_*` families reproduce native Apache Pulsar names and label sets
+  verbatim, so official dashboards and PromQL translate directly.
+- `pulsar_lite_*` families are extensions with no native counterpart
+  (error reasons, redelivery counters, write-queue batch metrics).
+- Histograms use the standard Prometheus shape (`_bucket{le=...}` +
+  `_sum` + `_count`); query percentiles with `histogram_quantile()`.
+
+## Broker configuration
+
+```toml
+[metrics]
+enabled = true            # false: no listener, no scrape aggregation
+addr = "0.0.0.0:8080"     # /metrics path; must be reachable by the scraper
+cluster = "pulsar-lite"   # cluster label value on every family
+rate_window_secs = 60     # window for pulsar_rate_in-style gauges
+```
+
+Remote broker: change the target in `prometheus/prometheus.yml` from
+`host.docker.internal:8080` to your broker address.

grafana/dashboards/pulsar-lite-broker.json

@@ -0,0 +1,223 @@
+{
+  "uid": "pulsar-lite-broker",
+  "title": "Pulsar Lite / Broker",
+  "tags": ["pulsar-lite"],
+  "timezone": "browser",
+  "schemaVersion": 39,
+  "version": 1,
+  "editable": true,
+  "refresh": "10s",
+  "time": { "from": "now-30m", "to": "now" },
+  "templating": {
+    "list": [
+      {
+        "name": "cluster",
+        "label": "Cluster",
+        "type": "query",
+        "datasource": { "type": "prometheus", "uid": "prometheus" },
+        "query": "label_values(pulsar_broker_rate_in, cluster)",
+        "refresh": 2,
+        "includeAll": false,
+        "multi": false,
+        "current": { "text": "pulsar-lite", "value": "pulsar-lite" }
+      }
+    ]
+  },
+  "panels": [
+    {
+      "id": 1,
+      "type": "timeseries",
+      "title": "Broker message rate (in / out)",
+      "datasource": { "type": "prometheus", "uid": "prometheus" },
+      "gridPos": { "h": 8, "w": 12, "x": 0, "y": 0 },
+      "targets": [
+        {
+          "expr": "pulsar_broker_rate_in{cluster=\"$cluster\"}",
+          "legendFormat": "in"
+        },
+        {
+          "expr": "pulsar_broker_rate_out{cluster=\"$cluster\"}",
+          "legendFormat": "out"
+        }
+      ]
+    },
+    {
+      "id": 2,
+      "type": "timeseries",
+      "title": "Broker throughput (bytes/s)",
+      "datasource": { "type": "prometheus", "uid": "prometheus" },
+      "gridPos": { "h": 8, "w": 12, "x": 12, "y": 0 },
+      "targets": [
+        {
+          "expr": "pulsar_broker_throughput_in{cluster=\"$cluster\"}",
+          "legendFormat": "in"
+        },
+        {
+          "expr": "pulsar_broker_throughput_out{cluster=\"$cluster\"}",
+          "legendFormat": "out"
+        }
+      ]
+    },
+    {
+      "id": 3,
+      "type": "timeseries",
+      "title": "Accepted / delivered totals (rate over counters)",
+      "datasource": { "type": "prometheus", "uid": "prometheus" },
+      "gridPos": { "h": 8, "w": 12, "x": 0, "y": 8 },
+      "targets": [
+        {
+          "expr": "rate(pulsar_broker_in_messages_total{cluster=\"$cluster\"}[1m])",
+          "legendFormat": "published msg/s"
+        },
+        {
+          "expr": "rate(pulsar_broker_out_messages_total{cluster=\"$cluster\"}[1m])",
+          "legendFormat": "delivered msg/s"
+        },
+        {
+          "expr": "rate(pulsar_broker_in_bytes_total{cluster=\"$cluster\"}[1m])",
+          "legendFormat": "published B/s"
+        },
+        {
+          "expr": "rate(pulsar_broker_out_bytes_total{cluster=\"$cluster\"}[1m])",
+          "legendFormat": "delivered B/s"
+        }
+      ]
+    },
+    {
+      "id": 4,
+      "type": "timeseries",
+      "title": "Entity counts",
+      "datasource": { "type": "prometheus", "uid": "prometheus" },
+      "gridPos": { "h": 8, "w": 6, "x": 12, "y": 8 },
+      "targets": [
+        {
+          "expr": "pulsar_broker_topics_count{cluster=\"$cluster\"}",
+          "legendFormat": "topics"
+        },
+        {
+          "expr": "pulsar_broker_subscriptions_count{cluster=\"$cluster\"}",
+          "legendFormat": "subscriptions"
+        },
+        {
+          "expr": "pulsar_broker_producers_count{cluster=\"$cluster\"}",
+          "legendFormat": "producers"
+        },
+        {
+          "expr": "pulsar_broker_consumers_count{cluster=\"$cluster\"}",
+          "legendFormat": "consumers"
+        }
+      ]
+    },
+    {
+      "id": 5,
+      "type": "timeseries",
+      "title": "Connections",
+      "datasource": { "type": "prometheus", "uid": "prometheus" },
+      "gridPos": { "h": 8, "w": 6, "x": 18, "y": 8 },
+      "targets": [
+        {
+          "expr": "pulsar_active_connections{cluster=\"$cluster\"}",
+          "legendFormat": "active"
+        },
+        {
+          "expr": "rate(pulsar_connection_created_total_count{cluster=\"$cluster\"}[5m])",
+          "legendFormat": "created /s"
+        },
+        {
+          "expr": "rate(pulsar_connection_closed_total_count{cluster=\"$cluster\"}[5m])",
+          "legendFormat": "closed /s"
+        }
+      ]
+    },
+    {
+      "id": 6,
+      "type": "timeseries",
+      "title": "Rejections by reason (/s)",
+      "datasource": { "type": "prometheus", "uid": "prometheus" },
+      "gridPos": { "h": 8, "w": 12, "x": 0, "y": 16 },
+      "targets": [
+        {
+          "expr": "sum by (reason) (rate(pulsar_lite_broker_errors_total{cluster=\"$cluster\"}[5m]))",
+          "legendFormat": "{{reason}}"
+        }
+      ]
+    },
+    {
+      "id": 7,
+      "type": "timeseries",
+      "title": "Backlog + storage size",
+      "datasource": { "type": "prometheus", "uid": "prometheus" },
+      "gridPos": { "h": 8, "w": 6, "x": 12, "y": 16 },
+      "targets": [
+        {
+          "expr": "pulsar_broker_msg_backlog{cluster=\"$cluster\"}",
+          "legendFormat": "backlog entries"
+        },
+        {
+          "expr": "pulsar_broker_storage_size{cluster=\"$cluster\"}",
+          "legendFormat": "stored bytes"
+        }
+      ]
+    },
+    {
+      "id": 8,
+      "type": "timeseries",
+      "title": "Publish rate-limit rejections (/s)",
+      "datasource": { "type": "prometheus", "uid": "prometheus" },
+      "gridPos": { "h": 8, "w": 6, "x": 18, "y": 16 },
+      "targets": [
+        {
+          "expr": "sum by (topic) (rate(pulsar_publish_rate_limit_times{cluster=\"$cluster\"}[5m]))",
+          "legendFormat": "{{topic}}"
+        }
+      ]
+    },
+    {
+      "id": 9,
+      "type": "timeseries",
+      "title": "Write-queue batch size (avg / P99)",
+      "datasource": { "type": "prometheus", "uid": "prometheus" },
+      "gridPos": { "h": 8, "w": 8, "x": 0, "y": 24 },
+      "targets": [
+        {
+          "expr": "rate(pulsar_lite_write_queue_batch_messages_total[1m]) / rate(pulsar_lite_write_queue_batches_total[1m])",
+          "legendFormat": "avg batch"
+        },
+        {
+          "expr": "histogram_quantile(0.99, sum by (le) (rate(pulsar_lite_write_queue_batch_size_bucket[5m])))",
+          "legendFormat": "P99 batch"
+        }
+      ]
+    },
+    {
+      "id": 10,
+      "type": "timeseries",
+      "title": "Broker RSS",
+      "datasource": { "type": "prometheus", "uid": "prometheus" },
+      "gridPos": { "h": 8, "w": 8, "x": 8, "y": 24 },
+      "fieldConfig": {
+        "defaults": { "unit": "bytes" },
+        "overrides": []
+      },
+      "targets": [
+        {
+          "expr": "process_resident_memory_bytes",
+          "legendFormat": "RSS"
+        }
+      ]
+    },
+    {
+      "id": 11,
+      "type": "timeseries",
+      "title": "Broker CPU (seconds/s)",
+      "datasource": { "type": "prometheus", "uid": "prometheus" },
+      "gridPos": { "h": 8, "w": 8, "x": 16, "y": 24 },
+      "targets": [
+        {
+          "expr": "rate(process_cpu_seconds_total[1m])",
+          "legendFormat": "cpu"
+        }
+      ]
+    }
+  ]
+}