EDA con NATS: observabilidad y SLOs para pipelines de proyección con JetStream

Las Partes 1–4 de esta serie construyeron el stack de mensajería y proyecciones. La Parte 1 cubrió subjects y queue groups. La Parte 2 agregó streams y consumidores durables. La Parte 4 presentó un pipeline genérico de orders: PostgreSQL como fuente de verdad, un mirror CDC hacia NATS KV y un indexer que consolida señales para OpenSearch.

Este artículo responde la pregunta operativa que esas piezas dejan abierta: ¿qué mides, qué SLOs defines y qué haces cuando se activan las alertas?

Investigado el July 6, 2026 con base en la documentación de monitoreo de NATS, los advisories de JetStream, el Prometheus NATS Exporter y las convenciones semánticas de mensajería de OpenTelemetry.

TL;DR

• Habilita el monitoreo HTTP de nats-server (http_port: 8222) y recopila métricas de /varz, /connz y /jsz, o usa el Prometheus NATS Exporter oficial con -varz -jsz=all.
• Vigila la salud del servidor (/healthz, conexiones, memoria), la capacidad de JetStream (messages/bytes del stream, límites de almacenamiento de la cuenta) y el estado de los consumidores (num_pending, num_ack_pending, num_redelivered).
• slow_consumers en /varz señala a los suscriptores de Core NATS que no pueden seguir el ritmo; JetStream usa control de flujo con MaxAckPending en lugar de cerrar conexiones.
• Los SLOs de aplicación de la Parte 4 —projection_lag_seconds, kv_mirror_age_seconds, cdc_slot_lag_bytes, opensearch_bulk_failures— detectan retrasos de proyección de extremo a extremo que las métricas del servidor por sí solas no captan.
• Suscríbete a los advisories de JetStream en $JS.EVENT.ADVISORY.> para recibir eventos de MAX_DELIVERIES, MSG_NAKED y quorum.
• OpenTelemetry no viene integrado en nats-server; propaga el encabezado W3C traceparent en los mensajes de NATS manualmente o mediante un wrapper del cliente.

Qué aprenderás aquí

• Un modelo de observabilidad por capas: servidor → JetStream → proyecciones de aplicación
• Qué endpoints de monitoreo de NATS y flags del exporter habilitar
• La semántica del retraso de los consumidores: num_pending vs num_ack_pending vs num_redelivered
• Definiciones de SLO para el pipeline orders/CDC/indexer de la Parte 4
• Configuración de recopilación de Prometheus, dashboards de Grafana y ejemplos de reglas de alerta
• Sugerencias de runbook para fallas comunes
• Subjects de advisories de JetStream que conviene almacenar y usar para generar alertas
• Propagación de trazas de OpenTelemetry a través de publicación → stream → consumidor

Capas de observabilidad

Los pipelines de proyección fallan en tres lugares: el broker, los workers consumidores y los almacenes de destino. Monitorea los tres.

  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)

El diagrama es genérico: cambia “orders” por catálogo, usuarios o inventario; los puntos donde se recopilan métricas siguen siendo los mismos.

Salud del servidor: qué expone NATS

nats-server incluye un servidor HTTP ligero de monitoreo. Según la documentación oficial, el puerto convencional es 8222.

Habilita el monitoreo en la configuración del servidor:

# nats-server.conf
http_port: 8222

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

Nota de seguridad (según la documentación de NATS): el puerto de monitoreo no tiene autenticación integrada y escucha en todas las interfaces por defecto. Limítalo a localhost o a una red privada; no expongas el puerto 8222 a internet.

Conexiones y consumidores lentos

/varz expone connections (conteo actual) y slow_consumers (conteo acumulado de suscriptores desconectados por no procesar mensajes con suficiente rapidez). El objeto slow_consumer_stats desglosa este conteo por tipo de conexión (clientes, rutas, gateways, nodos leaf).

Según la documentación de NATS sobre consumidores lentos, un consumidor lento es un suscriptor cuyo buffer se llena porque no puede procesar mensajes al ritmo de publicación. En Core NATS, el servidor puede terminar cerrando la conexión. En JetStream, el control de flujo mediante MaxAckPending suspende la entrega; aun así, slow_consumers en las rutas de Core (observación en vivo, request/reply) sigue siendo relevante.

/connz?sort=pending muestra las conexiones con los buffers de salida pendientes más grandes; es útil para encontrar una suscripción con mucho tráfico antes de que el servidor la desconecte.

Métricas de JetStream: streams y consumidores

El endpoint /jsz informa el estado de JetStream. En despliegues con clústeres, la documentación de NATS recomienda consultar al líder del stream (leader-only=true) para obtener posiciones precisas de los consumidores.

Parámetros de consulta útiles:

Señales a nivel de stream

De un objeto stream_detail[].state:

• messages / bytes — retención actual; compárala con los límites max_msgs, max_bytes, max_age configurados en el stream (Parte 2)
• first_seq / last_seq — rango de secuencias; saltos repentinos de first_seq pueden indicar que la retención está descartando mensajes antiguos bajo presión
• consumer_count — cantidad de consumidores asociados

Genera una alerta cuando los bytes del stream se acerquen a los límites configurados. Con discard: old, el servidor descarta los mensajes más antiguos, lo que puede impedir que un consumidor CDC o un indexer atrasado se ponga al día.

Retraso del consumidor: los tres campos num_*

Cuando solicitas consumers=true, cada entrada de consumer_detail incluye los campos documentados en el esquema de respuesta de /jsz:

Para el consumidor durable search-indexer de la Parte 4 en INDEX_SIGNALS:

• num_pending en aumento → el indexer no puede seguir el ritmo de publicación en index.>
• num_ack_pending alto y sostenido cerca de MaxAckPending → el indexer está lento o bloqueado por OpenSearch; JetStream deja de entregar mensajes hasta que lleguen las confirmaciones (control de flujo del consumidor)
• num_redelivered en aumento → los handlers están enviando NAK, agotando el tiempo de espera o fallando antes de confirmar; revisa opensearch_bulk_failures y los logs de los handlers

Los objetos delivered y ack_floor muestran el cursor de secuencia del consumidor frente a la secuencia del stream; son útiles para depurar consumidores bloqueados después de un deploy.

SLOs de aplicación para pipelines de proyección

Las métricas del servidor te indican que el broker y los consumidores de JetStream están sanos. No te indican si un usuario que busca en OpenSearch ve los datos escritos en PostgreSQL hace diez segundos. Eso requiere SLOs a nivel de aplicación vinculados al pipeline de la Parte 4.

Objetivos de SLO (puntos de partida editoriales)

La Parte 4 enumeró los presupuestos de retraso esperados para cada ruta de lectura. La siguiente tabla los convierte en SLOs medibles. Trata los umbrales como puntos de partida y ajústalos según el comportamiento habitual de tu tráfico.

Estos son criterios editoriales basados en la tabla de retrasos de la Parte 4, no valores predeterminados de NATS. Instrumenta estas métricas en tu mirror CDC, el indexer y un canario sintético que escriba una fila de marcador y consulte OpenSearch periódicamente.

Implementación de métricas de aplicación

Emite contadores e histogramas desde tus workers. Ejemplo con las convenciones del cliente de Prometheus:

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);
}

Para cdc_slot_lag_bytes, consulta PostgreSQL directamente:

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';

Expón el resultado como un gauge de Prometheus desde tu worker CDC o mediante una consulta personalizada de postgres_exporter.

Implementación concreta: Prometheus y Grafana

El Prometheus NATS Exporter agrega los endpoints de monitoreo de NATS en un único endpoint /metrics para la recopilación de métricas. La guía oficial cubre la configuración completa de Prometheus + Grafana.

Iniciar el 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

El exporter escucha en 0.0.0.0:7777/metrics por defecto. Los flags corresponden directamente a los endpoints de monitoreo: -varz para estadísticas del servidor, -jsz=all para cuentas, streams y consumidores.

Importa los dashboards de Grafana incluidos en el repositorio del exporter:

• grafana-nats-dash.json — métricas del servidor Core
• grafana-jetstream-dash.json — paneles de streams y consumidores

Si haces el deploy mediante el chart de Helm de NATS, los prefijos de métricas cambian de gnatsd/jetstream a nats; usa las variantes de dashboards específicas para Helm indicadas en la guía.

Configuración de recopilación de Prometheus

# 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"]

Reglas de alerta de Grafana

Las reglas de registro de Prometheus son opcionales; estas expresiones de alerta son puntos de partida. Ajusta los umbrales después de establecer una línea base para tu entorno.

# 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."

Nota: Los nombres exactos de las métricas del exporter dependen de la versión y del filtro -jsz. Valida los nombres con curl localhost:7777/metrics | grep jetstream antes de desplegar las reglas. Los despliegues con Helm usan el prefijo nats_ en lugar de gnatsd_/jetstream_.

Advisories de JetStream

Además de las métricas recopiladas periódicamente, JetStream publica eventos en tiempo real en $JS.EVENT.ADVISORY.> (monitoreo de JetStream). Almacénalos en un stream dedicado a advisories para auditoría y alertas.

Consulta los advisories en vivo:

nats event --js-advisory

Para consumirlos de forma programática, el paquete jsm.go proporciona structs tipados para advisories. Los esquemas JSON están disponibles mediante 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: trazas de publicación → stream → consumidor

nats-server no emite spans de OpenTelemetry. El trazado distribuido requiere instrumentación del lado del cliente que inyecte y extraiga encabezados de W3C Trace Context en los mensajes de NATS.

Los mensajes de NATS admiten encabezados (traceparent, tracestate). El patrón de propagación, documentado en las convenciones semánticas de spans de mensajería de OpenTelemetry, es:

1. El publicador inicia un span PRODUCER e inyecta el contexto en los encabezados del mensaje
2. El consumidor extrae el contexto e inicia un span hijo CONSUMER
3. Para JetStream, agrega atributos: messaging.system=nats, messaging.destination.name=<subject>, nombre del stream, nombre del consumidor

Ejemplo en Go (propagación manual, sin wrapper de terceros):

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 ...
}

Nota editorial: Existen wrappers de la comunidad (por ejemplo, otel-instrumentation-nats para Python, natstrace para Go), pero NATS no incluye un SDK oficial de OpenTelemetry hasta July 2026. Valida que tu versión de la biblioteca cliente conserve los encabezados: la reentrega de JetStream debería llevar el traceparent original, lo que te permite ver las cadenas de reentrega en tu backend de trazas.

Una traza útil para el pipeline de la Parte 4 abarca: escritura en la API de pedidos → consumo del mirror CDC → escritura en KV → publicación de la señal de indexación → consumo del indexer → operación bulk de OpenSearch. Correlaciona los IDs de traza con los ejemplares de projection_lag_seconds cuando tu backend de métricas lo permita.

Sugerencias de runbook

Primeros pasos para ingenieros cuando se activan las alertas.

IndexerConsumerLagHigh (num_pending en aumento)

1. Revisa num_ack_pending: si está cerca de MaxAckPending, el indexer está lento, no a la espera de trabajo
2. Revisa opensearch_bulk_failures y la salud del clúster de OpenSearch (shards amarillos/rojos)
3. Escala las réplicas del indexer si usas un queue group en el consumidor pull
4. Compara los picos de la tasa de publicación con la línea base normal (rate(jetstream_stream_total_messages))

IndexerRedeliveryStorm (num_redelivered en aumento)

1. Consulta los logs recientes del indexer: motivos de NAK (timeout de OpenSearch, documento incorrecto, clave de KV ausente)
2. Compara la latencia p99 del handler con AckWait: si p99 > AckWait, aumenta AckWait o corrige el paso lento
3. Busca incompatibilidades de esquema causadas por un deploy (cambio de mapping sin reindexación)

CDCSlotLagCritical

1. Verifica que los pods del mirror CDC estén funcionando y conectados a NATS
2. Revisa pg_replication_slots en PostgreSQL: active, restart_lsn, uso de disco del WAL
3. Si el retraso supera la retención, deja de especular: planifica una reconstrucción de KV a partir de un snapshot de la base de datos y restablece el slot con la aprobación de operaciones

IndexerPoisonMessages (advisory de MaxDeliver / dlq_depth > 0)

1. Obtén el mensaje del stream por secuencia (nats stream view INDEX_SIGNALS <seq>)
2. Determina si el payload está corrupto o si el handler tiene un bug
3. Corrige el handler, haz el deploy y luego reprocesa: restablece el consumidor a una secuencia anterior o vuelve a publicar las señales index.order desde PostgreSQL

NatsSlowConsumersRising (ruta de Core NATS)

1. /connz?sort=pending: encuentra la suscripción
2. Un caso común en la Parte 4: un proceso de precarga de caché con watch("orders-v1.>") que no puede deserializar con suficiente rapidez
3. Escala el watcher o reduce el trabajo por mensaje; no te limites a aumentar los límites del buffer del cliente (la documentación de NATS advierte que esto solo pospone el problema)

Cómo se relaciona con las Partes 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

Siguiente en la serie: NATS vs Kafka vs AWS SQS (Parte 6)

Fuentes

• Monitoreo de NATS — endpoints /varz, /connz, /jsz, /healthz
• Monitoreo de JetStream — advisories en $JS.EVENT.ADVISORY.>
• Consumidores lentos de NATS — detección, buffers del cliente, plazos de escritura del servidor
• Consumidores lentos (ajuste del servidor) — recomendaciones para escalar, medir y ajustar
• Consumidores de JetStream — MaxAckPending, control de flujo, ritmo de los consumidores pull
• Prometheus NATS Exporter — exporter oficial, dashboards de Grafana, guía
• NATS Surveyor — alternativa de monitoreo basada en la cuenta System
• Spans de mensajería de OpenTelemetry — tipos de span PRODUCER/CONSUMER, atributos de NATS
• W3C Trace Context — formato de los encabezados traceparent / tracestate
• Slots de replicación de PostgreSQL — fundamento de cdc_slot_lag_bytes
• Core NATS (Parte 1)
• JetStream (Parte 2)
• Proyecciones CQRS (Parte 4) — arquitectura del pipeline y métricas de aplicación