EDA with NATS: Observability and SLOs for JetStream Projection Pipelines

Parts 1–4 of this series built the messaging and projection stack. Part 1 covered subjects and queue groups. Part 2 added durable streams and consumers. Part 4 walked through a generic orders pipeline: PostgreSQL as truth, a CDC mirror into NATS KV, and an indexer fan-in to OpenSearch.

This article answers the operational question those pieces leave open: what do you measure, what SLOs do you set, and what do you do when alerts fire?

Researched against NATS monitoring documentation, JetStream advisories, the Prometheus NATS Exporter, and OpenTelemetry messaging semantic conventions on July 6, 2026.

TL;DR

• Enable nats-server HTTP monitoring (http_port: 8222) and scrape /varz, /connz, and /jsz — or use the official Prometheus NATS Exporter with -varz -jsz=all.
• Watch server health (/healthz, connections, memory), JetStream capacity (stream messages/bytes, account storage limits), and consumer state (num_pending, num_ack_pending, num_redelivered).
• slow_consumers in /varz signals Core NATS subscribers that cannot keep up; JetStream uses MaxAckPending flow control instead of dropping connections.
• Application SLOs from Part 4 — projection_lag_seconds, kv_mirror_age_seconds, cdc_slot_lag_bytes, opensearch_bulk_failures — catch end-to-end projection drift that server metrics alone will miss.
• Subscribe to JetStream advisories on $JS.EVENT.ADVISORY.> for MAX_DELIVERIES, MSG_NAKED, and quorum events.
• OpenTelemetry does not ship inside nats-server; propagate W3C traceparent through NATS message headers manually or via a client wrapper.

What You Will Learn Here

• A layered observability model: server → JetStream → application projections
• Which NATS monitoring endpoints and exporter flags to enable
• Consumer lag semantics: num_pending vs num_ack_pending vs num_redelivered
• SLO definitions for the orders/CDC/indexer pipeline from Part 4
• Prometheus scrape config, Grafana dashboards, and alert rule examples
• Runbook hints for common failure modes
• JetStream advisory subjects worth storing and alerting on
• OpenTelemetry trace propagation across publish → stream → consumer

Observability Layers

Projection pipelines fail in three places: the broker, the consumer workers, and the downstream stores. Monitor all three.

  LAYER 1 — NATS SERVER              LAYER 2 — JETSTREAM              LAYER 3 — APPLICATION
  ─────────────────────              ───────────────────              ───────────────────────

  /varz  connections,              /jsz   stream bytes,            projection_lag_seconds
         slow_consumers,                    messages,                 kv_mirror_age_seconds
         in_msgs/out_msgs                   consumer lag              cdc_slot_lag_bytes
  /connz per-connection              advisories:                    opensearch_bulk_failures
         pending bytes                      MAX_DELIVERIES
  /healthz  accept connections             MSG_NAKED
                                             QUORUM_LOST

  ORDERS PROJECTION PIPELINE — WHERE METRICS ATTACH
  ==================================================

  PostgreSQL (truth)
       |
       |  cdc_slot_lag_bytes  <<<<<<<<<<<<<<<<<<<<<<  replication slot health
       v
  CDC Mirror Worker
       |
       |  kv_mirror_age_seconds  <<<<<<<<<<<<<<<<<<<<  mirror lag
       v
  NATS KV (app-objects / orders-v1.*)
       |
       |  JetStream stream bytes, /jsz consumer state
       v
  index.order  ──►  INDEX_SIGNALS stream
       |
       |  num_pending, num_ack_pending, num_redelivered  <<  indexer consumer
       v
  Indexer Worker
       |
       |  projection_lag_seconds, opensearch_bulk_failures
       v
  OpenSearch (search read model)

The diagram is generic: swap “orders” for catalog, users, or inventory — the metric attachment points stay the same.

Server Health: What NATS Exposes

nats-server ships a lightweight HTTP monitoring server. Per official documentation, the conventional port is 8222.

Enable monitoring in server config:

# nats-server.conf
http_port: 8222

jetstream {
  store_dir: /data/jetstream
  max_memory_store: 1GB
  max_file_store: 10GB
}

Security note (from NATS docs): the monitoring port has no built-in authentication and binds to all interfaces by default. Restrict it to localhost or a private network — do not expose port 8222 to the public internet.

Connections and slow consumers

/varz exposes connections (current count) and slow_consumers (cumulative count of subscribers cut off for not processing fast enough). The slow_consumer_stats object breaks this down by connection kind (clients, routes, gateways, leaf nodes).

Per NATS slow consumer documentation, a slow consumer is a subscriber whose buffer fills because it cannot process messages at the publish rate. For Core NATS, the server may eventually close the connection. For JetStream, flow control via MaxAckPending suspends delivery instead — but slow_consumers on Core paths (live watches, request/reply) still matters.

/connz?sort=pending surfaces connections with the largest outbound pending buffers — useful for finding a hot subscription before the server drops it.

JetStream Metrics: Streams and Consumers

The /jsz endpoint reports JetStream state. In clustered deployments, NATS documentation recommends querying the stream leader (leader-only=true) for accurate consumer positions.

Useful query parameters:

Stream-level signals

From a stream_detail[].state object:

• messages / bytes — current retention; compare against max_msgs, max_bytes, max_age limits configured on the stream (Part 2)
• first_seq / last_seq — sequence range; sudden first_seq jumps may indicate discard-old retention under pressure
• consumer_count — how many consumers are bound

Alert when stream bytes approach configured limits. With discard: old, the server drops the oldest messages — which can make catch-up impossible for a lagging CDC or indexer consumer.

Consumer lag: the three num_* fields

When you request consumers=true, each consumer_detail entry includes fields documented in the /jsz response schema:

For the Part 4 search-indexer durable consumer on INDEX_SIGNALS:

• Rising num_pending → indexer cannot keep up with index.> publish rate
• High num_ack_pending sustained near MaxAckPending → indexer is slow or blocked on OpenSearch; JetStream stops delivering more until acks arrive (consumer flow control)
• Climbing num_redelivered → handlers are NAKing, timing out, or crashing before ack — check opensearch_bulk_failures and handler logs

The delivered and ack_floor objects show the consumer sequence cursor vs the stream sequence — useful for debugging stuck consumers after a deploy.

Application SLOs for Projection Pipelines

Server metrics tell you the broker and JetStream consumers are healthy. They do not tell you whether a user searching OpenSearch sees data written to PostgreSQL ten seconds ago. That requires application-level SLOs tied to the Part 4 pipeline.

SLO targets (editorial starting points)

Part 4 listed expected lag budgets per read path. The table below turns those into measurable SLOs. Treat thresholds as starting points — tune from your traffic baseline.

These are editorial judgments informed by Part 4’s lag table, not NATS defaults. Instrument them in your CDC mirror, indexer, and a synthetic canary that writes a marker row and polls OpenSearch.

Implementing application metrics

Emit counters and histograms from your workers. Example using Prometheus client conventions:

import { Counter, Histogram, Gauge, Registry } from "prom-client";

const registry = new Registry();

const projectionLag = new Histogram({
  name: "projection_lag_seconds",
  help: "Seconds from DB commit to OpenSearch visibility",
  buckets: [0.5, 1, 2, 5, 10, 30, 60, 120],
  registers: [registry],
});

const kvMirrorAge = new Histogram({
  name: "kv_mirror_age_seconds",
  help: "Seconds from DB commit to KV mirror update",
  buckets: [0.1, 0.5, 1, 2, 5, 10],
  registers: [registry],
});

const bulkFailures = new Counter({
  name: "opensearch_bulk_failures_total",
  help: "Failed OpenSearch bulk operations",
  registers: [registry],
});

const dlqDepth = new Gauge({
  name: "indexer_dlq_depth",
  help: "Messages that hit MaxDeliver on the search-indexer consumer",
  registers: [registry],
});

// In the indexer, after a successful OpenSearch upsert:
async function recordProjection(commitTs: Date) {
  const lagSec = (Date.now() - commitTs.getTime()) / 1000;
  projectionLag.observe(lagSec);
}

// In the CDC mirror, after kv.put:
function recordMirrorLag(commitTs: Date) {
  kvMirrorAge.observe((Date.now() - commitTs.getTime()) / 1000);
}

For cdc_slot_lag_bytes, query PostgreSQL directly:

SELECT slot_name,
       pg_wal_lsn_diff(pg_current_wal_lsn(), confirmed_flush_lsn) AS lag_bytes
FROM pg_replication_slots
WHERE slot_name = 'cdc_mirror_orders';

Expose the result as a Prometheus gauge from your CDC worker or a postgres_exporter custom query.

Concrete Implementation: Prometheus and Grafana

The Prometheus NATS Exporter aggregates NATS monitoring endpoints into a single /metrics scrape target. The official walkthrough covers the full Prometheus + Grafana setup.

Launch the exporter

# NATS server with monitoring enabled
nats-server -c nats-server.conf   # http_port: 8222

# Exporter: general + full JetStream detail
prometheus-nats-exporter \
  -varz -connz -healthz \
  -jsz=all \
  http://localhost:8222

The exporter listens on 0.0.0.0:7777/metrics by default. Flags map directly to monitoring endpoints: -varz for server stats, -jsz=all for accounts, streams, and consumers.

Import the bundled Grafana dashboards from the exporter repository:

• grafana-nats-dash.json — Core server metrics
• grafana-jetstream-dash.json — Stream and consumer panels

If you deploy via the NATS Helm chart, metric prefixes change from gnatsd/jetstream to nats — use the Helm-specific dashboard variants noted in the walkthrough.

Prometheus scrape config

# prometheus.yml
scrape_configs:
  - job_name: nats
    static_configs:
      - targets: ["nats-exporter:7777"]

  - job_name: projection-workers
    static_configs:
      - targets: ["cdc-mirror:9090", "search-indexer:9090"]

Grafana alert rules

Prometheus recording rules are optional; these alert expressions are starting points. Adjust thresholds after baselining your environment.

# alerts/nats-jetstream.yml
groups:
  - name: nats-server
    rules:
      - alert: NatsServerUnhealthy
        expr: gnatsd_healthz_status != 1
        for: 2m
        labels:
          severity: critical
        annotations:
          summary: "NATS /healthz failing on {{ $labels.server_id }}"
          runbook: "Check nats-server logs, disk space on JetStream store_dir, and cluster quorum."

      - alert: NatsSlowConsumersRising
        expr: increase(gnatsd_varz_slow_consumers[5m]) > 0
        for: 5m
        labels:
          severity: warning
        annotations:
          summary: "Slow consumers detected on {{ $labels.server_id }}"
          runbook: "Run /connz?sort=pending. Identify subscription. Scale consumer or reduce publish rate."

  - name: jetstream-consumers
    rules:
      - alert: IndexerConsumerLagHigh
        expr: jetstream_consumer_num_pending{consumer="search-indexer"} > 1000
        for: 5m
        labels:
          severity: warning
        annotations:
          summary: "search-indexer has {{ $value }} pending messages"
          runbook: "Scale indexer replicas (queue group) or check OpenSearch latency. Verify num_ack_pending near MaxAckPending."

      - alert: IndexerRedeliveryStorm
        expr: jetstream_consumer_num_redelivered{consumer="search-indexer"} > 50
        for: 3m
        labels:
          severity: warning
        annotations:
          summary: "{{ $value }} messages redelivering on search-indexer"
          runbook: "Check indexer logs for NAK loops. Inspect opensearch_bulk_failures. Review AckWait vs handler p99."

      - alert: JetStreamStoragePressure
        expr: |
          jetstream_stream_bytes / jetstream_stream_max_bytes > 0.85
        for: 10m
        labels:
          severity: warning
        annotations:
          summary: "Stream {{ $labels.stream }} at {{ $value | humanizePercentage }} of max_bytes"
          runbook: "Review retention (max_age, max_bytes). Check for stuck consumers preventing ack-based deletion in workqueue streams."

  - name: projection-slos
    rules:
      - alert: ProjectionLagSLOBreach
        expr: histogram_quantile(0.99, rate(projection_lag_seconds_bucket[5m])) > 10
        for: 5m
        labels:
          severity: warning
        annotations:
          summary: "Search projection p99 lag {{ $value }}s (SLO: 10s)"
          runbook: "Correlate with indexer num_pending, OpenSearch cluster health, and cdc_slot_lag_bytes."

      - alert: CDCSlotLagCritical
        expr: cdc_slot_lag_bytes > 104857600
        for: 5m
        labels:
          severity: critical
        annotations:
          summary: "CDC replication slot lag {{ $value | humanize }}B"
          runbook: "Scale CDC mirror workers. If lag exceeds WAL retention, plan full rebuild from PostgreSQL snapshot."

      - alert: IndexerPoisonMessages
        expr: indexer_dlq_depth > 0
        for: 1m
        labels:
          severity: critical
        annotations:
          summary: "{{ $value }} messages exhausted MaxDeliver"
          runbook: "Inspect message payload. Fix handler bug. Replay from stream sequence or re-publish index signal after fix."

Note: Exact metric names from the exporter depend on version and -jsz filter. Validate names with curl localhost:7777/metrics | grep jetstream before deploying rules. Helm deployments use the nats_ prefix instead of gnatsd_/jetstream_.

JetStream Advisories

Beyond polled metrics, JetStream publishes real-time events on $JS.EVENT.ADVISORY.> (monitoring JetStream). Store them in a dedicated advisory stream for audit and alerting.

View advisories live:

nats event --js-advisory

For programmatic consumption, the jsm.go package provides typed advisory structs. JSON schemas are available via nats schema show <kind>.

  ADVISORY FLOW (optional but valuable)
  =====================================

  JetStream server
       |
       |  publish advisory JSON
       v
  $JS.EVENT.ADVISORY.>
       |
       +──► nats event --js-advisory  (debugging)
       |
       +──► ADVISORY_LOG stream       (retained audit)
       |
       +──► alert router              (PagerDuty / Slack)

OpenTelemetry: Tracing Publish → Stream → Consumer

nats-server does not emit OpenTelemetry spans. Distributed tracing requires client-side instrumentation that injects and extracts W3C Trace Context headers on NATS messages.

NATS messages support headers (traceparent, tracestate). The propagation pattern — documented in OpenTelemetry’s messaging spans semantic conventions — is:

1. Publisher starts a PRODUCER span and injects context into message headers
2. Consumer extracts context and starts a CONSUMER child span
3. For JetStream, add attributes: messaging.system=nats, messaging.destination.name=<subject>, stream name, consumer name

Go example (manual propagation, no third-party wrapper):

import (
    "context"

    "github.com/nats-io/nats.go"
    "go.opentelemetry.io/otel"
    "go.opentelemetry.io/otel/attribute"
    "go.opentelemetry.io/otel/propagation"
    semconv "go.opentelemetry.io/otel/semconv/v1.27.0"
    "go.opentelemetry.io/otel/trace"
)

func publishIndexSignal(ctx context.Context, js nats.JetStreamContext, subject string, payload []byte) error {
    ctx, span := tracer.Start(ctx, "publish index signal",
        trace.WithSpanKind(trace.SpanKindProducer),
        trace.WithAttributes(
            semconv.MessagingSystemKey.String("nats"),
            semconv.MessagingDestinationName(subject),
            attribute.String("messaging.nats.stream", "INDEX_SIGNALS"),
        ),
    )
    defer span.End()

    hdr := nats.Header{}
    otel.GetTextMapPropagator().Inject(ctx, propagation.HeaderCarrier(hdr))

    _, err := js.PublishMsg(&nats.Msg{Subject: subject, Header: hdr, Data: payload})
    return err
}

func handleIndexMessage(ctx context.Context, msg *nats.Msg) {
    ctx = otel.GetTextMapPropagator().Extract(ctx, propagation.HeaderCarrier(msg.Header))
    ctx, span := tracer.Start(ctx, "indexer consume",
        trace.WithSpanKind(trace.SpanKindConsumer),
        trace.WithAttributes(
            semconv.MessagingSystemKey.String("nats"),
            semconv.MessagingDestinationName(msg.Subject),
            attribute.String("messaging.nats.consumer", "search-indexer"),
        ),
    )
    defer span.End()

    // ... build OpenSearch doc, ack/nak ...
}

Editorial note: Community wrappers exist (for example otel-instrumentation-nats for Python, natstrace for Go), but NATS does not ship an official OpenTelemetry SDK as of July 2026. Validate header preservation across your client library version — JetStream redelivery should carry the original traceparent, letting you see redelivery chains in your trace backend.

A useful trace for the Part 4 pipeline spans: Order API write → CDC mirror consume → KV put → index signal publish → indexer consume → OpenSearch bulk. Correlate trace IDs with projection_lag_seconds exemplars when your metrics backend supports it.

Runbook Hints

Engineer-focused first steps when alerts fire.

IndexerConsumerLagHigh (num_pending rising)

1. Check num_ack_pending — if near MaxAckPending, indexer is slow, not starved
2. Check opensearch_bulk_failures and OpenSearch cluster health (yellow/red shards)
3. Scale indexer replicas if using a queue group on the pull consumer
4. Verify publish rate spike vs normal baseline (rate(jetstream_stream_total_messages))

IndexerRedeliveryStorm (num_redelivered climbing)

1. Pull recent indexer logs — NAK reasons (OpenSearch timeout, bad document, missing KV key)
2. Compare handler p99 latency to AckWait — if p99 > AckWait, increase AckWait or fix the slow step
3. Check for deploy-induced schema mismatch (mapping change without reindex)

CDCSlotLagCritical

1. Verify CDC mirror pods are running and connected to NATS
2. Check PostgreSQL pg_replication_slots — active, restart_lsn, WAL disk usage
3. If lag exceeds retention, stop guessing — plan a KV rebuild from database snapshot and reset the slot with ops approval

IndexerPoisonMessages (MaxDeliver advisory / dlq_depth > 0)

1. Fetch the message from the stream by sequence (nats stream view INDEX_SIGNALS <seq>)
2. Determine if payload is corrupt or handler has a bug
3. Fix handler, deploy, then replay: reset consumer to earlier sequence or re-publish index.order signals from PostgreSQL

NatsSlowConsumersRising (Core NATS path)

1. /connz?sort=pending — find the subscription
2. Common in Part 4: a watch("orders-v1.>") cache warmer that cannot deserialize fast enough
3. Scale the watcher or reduce per-message work; do not only increase client buffer limits (NATS docs warn this postpones the problem)

How This Ties to Parts 1–4

  MONITORING CHECKLIST FOR A NEW PROJECTION
  ==========================================

  [ ] http_port enabled, monitoring not public
  [ ] Prometheus exporter: -varz -jsz=all -healthz
  [ ] Grafana dashboards imported, alerts baselined
  [ ] Per-consumer: num_pending, num_ack_pending, num_redelivered
  [ ] Per-stream: messages, bytes vs limits
  [ ] Application: projection_lag_seconds, cdc_slot_lag_bytes
  [ ] Advisory stream on $JS.EVENT.ADVISORY.>
  [ ] Trace propagation on publish + consume (optional but high value)
  [ ] Runbook linked from each alert annotation

Next in series: NATS vs Kafka vs AWS SQS (Part 6)

Sources

• NATS Monitoring — /varz, /connz, /jsz, /healthz endpoints
• Monitoring JetStream — advisories on $JS.EVENT.ADVISORY.>
• NATS Slow Consumers — detection, client buffers, server write deadlines
• Slow Consumers (Server Tuning) — scale, meter, tune guidance
• JetStream Consumers — MaxAckPending, flow control, pull consumer pacing
• Prometheus NATS Exporter — official exporter, Grafana dashboards, walkthrough
• NATS Surveyor — alternative System-account-based monitoring
• OpenTelemetry Messaging Spans — PRODUCER/CONSUMER span kinds, NATS attributes
• W3C Trace Context — traceparent / tracestate header format
• PostgreSQL Replication Slots — cdc_slot_lag_bytes foundation
• Core NATS (Part 1)
• JetStream (Part 2)
• CQRS Projections (Part 4) — pipeline architecture and application metrics