From 7e7b8ca412e8efb9a20b7b6ff4f8f89e5c7dfd2b Mon Sep 17 00:00:00 2001 From: windyboy Date: Sun, 16 Nov 2025 10:34:53 +0800 Subject: [PATCH] =?UTF-8?q?=E2=9C=A8=20Update=20observability=20configurat?= =?UTF-8?q?ion=20by=20enabling=20monitoring=20in=20the=20development=20env?= =?UTF-8?q?ironment,=20adding=20a=20new=20Prometheus=20scrape=20job=20for?= =?UTF-8?q?=20the=20CAATSM=20receiver,=20and=20enhancing=20documentation?= =?UTF-8?q?=20with=20recommended=20scrape=20configurations=20and=20validat?= =?UTF-8?q?ion=20steps=20for=20metrics.=20Refactor=20NATS=20consumer=20ini?= =?UTF-8?q?tialization=20to=20ensure=20idempotent=20consumer=20creation.?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- configs/config.dev.toml | 1 + configs/prometheus.dev.yml | 7 +++ docs/observability.md | 104 ++++++++++++++++++++++++++++++++ internal/infra/nats/consumer.go | 34 +++++++---- 4 files changed, 135 insertions(+), 11 deletions(-) diff --git a/configs/config.dev.toml b/configs/config.dev.toml index c04094e..1432001 100644 --- a/configs/config.dev.toml +++ b/configs/config.dev.toml @@ -57,6 +57,7 @@ endpoint = "localhost:4318" insecure = true [monitoring] +disabled = false addr = ":2112" enable_metrics = true enable_health = true diff --git a/configs/prometheus.dev.yml b/configs/prometheus.dev.yml index 3def892..705dbe5 100644 --- a/configs/prometheus.dev.yml +++ b/configs/prometheus.dev.yml @@ -12,3 +12,10 @@ scrape_configs: - targets: - "nats-exporter:7777" + - job_name: "caatsm-receiver" + static_configs: + - targets: + # When go-caatsm runs on the host/WSL and Prometheus runs in Docker, + # use host.docker.internal to reach the host monitoring server. + - "host.docker.internal:2112" + diff --git a/docs/observability.md b/docs/observability.md index 3170ac0..e03d18a 100644 --- a/docs/observability.md +++ b/docs/observability.md @@ -39,6 +39,49 @@ These metrics are intended to be scraped by Prometheus (either directly or via t - NATS consumer backlog and redelivery counts. - DB query rates and latencies. +#### Prometheus scrape configuration + +In the local dev environment, metrics are typically scraped by the Prometheus +container defined in `docker-compose.dev.yml` using `configs/prometheus.dev.yml`. + +A recommended scrape configuration for the receiver is: + +```yaml +scrape_configs: + - job_name: "otel-collector" + static_configs: + - targets: + - "otel-collector:8888" + + - job_name: "nats-exporter" + static_configs: + - targets: + - "nats-exporter:7777" + + - job_name: "caatsm-receiver" + static_configs: + - targets: + # go-caatsm running on host/WSL, Prometheus in Docker + - "host.docker.internal:2112" +``` + +When you run the receiver directly on the host/WSL, ensure the monitoring +server listens on all interfaces so that Docker can reach it, for example via: + +```bash +export CAATSM_MONITORING_ADDR=0.0.0.0:2112 +export CAATSM_MONITORING_ENABLE_METRICS=true +export CAATSM_MONITORING_ENABLE_HEALTH=true +``` + +Alternative topologies: + +- **Receiver and Prometheus in the same Docker network** + Expose the monitoring server via a container port and use the container + name as the scrape target, e.g. `caatsm-receiver:2112`. +- **Receiver behind a reverse proxy / load balancer** + Point Prometheus at the proxy address and path that forwards to `/metrics`. + #### CAATSM – Receiver Overview Dashboard The `caatsm-overview` Grafana dashboard (provisioned from `configs/grafana-dashboards.dev/caatsm-overview.json`) focuses on the CAATSM receiver service and surfaces: @@ -50,6 +93,36 @@ The `caatsm-overview` Grafana dashboard (provisioned from `configs/grafana-dashb - **Retry and permanent failure rates** – from `caatsm_retries_total` and `caatsm_messages_total{result="permanent_fail"}`. - **Publish failures** – from `caatsm_publish_failures_total`. +To validate that the dashboard is receiving data: + +1. Check the monitoring endpoint directly: + + ```bash + curl -s http://localhost:2112/metrics | grep caatsm_messages_total || true + ``` + +2. In Prometheus (`http://localhost:9090`), run: + + ```text + caatsm_messages_total + ``` + + and + + ```text + rate(caatsm_messages_total[5m]) + ``` + +3. In Grafana, open the **CAATSM – Receiver Overview** dashboard and + verify that: + + - “Messages by result (5m rate)” shows time series for `ok`, `fail`, + and `permanent_fail`. + - “Messages per stream/consumer” shows series labelled by `stream` + and `consumer`. + - DB-related panels show non-zero values based on + `caatsm_db_queries_total` and `caatsm_db_query_latency_seconds`. + ### Health and Readiness The monitoring server exposes: @@ -72,6 +145,37 @@ Tracing is configured via the `telemetry` section: - `telemetry.endpoint` – OTLP HTTP endpoint (e.g. `localhost:4318`). - `telemetry.insecure` – disables TLS for local/dev. +#### OTEL vs Prometheus metrics + +The receiver reports two complementary sets of metrics: + +- **Prometheus metrics via `/metrics`** + Implemented in `internal/observability/metrics`, covering: + - End-to-end message handling (`caatsm_messages_total`, + `caatsm_handle_latency_seconds`, `caatsm_retries_total`) + - DB activity (`caatsm_db_queries_total`, + `caatsm_db_query_latency_seconds`) + - Legacy per-telegram metrics + +- **OpenTelemetry metrics via OTLP** + Implemented using `otel.Meter` in the NATS consumer and app processor, + including: + - `caatsm_messages_processed_total` + - `caatsm_parse_duration_ms` + - `caatsm_publish_failures_total` + - `caatsm_nats_consumer_ack_pending` + - `caatsm_nats_consumer_redelivered` + - `caatsm_nats_consumer_pending` + - `caatsm_nats_consumer_delivered` + +Prometheus only sees the metrics exposed on `/metrics`. OTEL metrics are +exported to the configured OTEL collector (`telemetry.endpoint`) via OTLP and +are, by default, forwarded to Jaeger (traces) and logs (metrics) according to +`configs/otel-collector.dev.yaml`. If you want OTEL metrics to appear in +Prometheus as well, you can extend the collector configuration with a +`prometheus` or `prometheusremotewrite` exporter and add a corresponding +scrape or remote-write configuration. + Key spans: - `caatsm/nats` diff --git a/internal/infra/nats/consumer.go b/internal/infra/nats/consumer.go index 8a02be8..21a7189 100644 --- a/internal/infra/nats/consumer.go +++ b/internal/infra/nats/consumer.go @@ -130,7 +130,7 @@ func ProvideConsumer( return consumer, nil } -// ensureConsumer creates the consumer if it doesn't exist +// ensureConsumer creates the consumer if it doesn't exist; if it already exists, it is reused. func (c *Consumer) ensureConsumer() error { consumerConfig := &nats.ConsumerConfig{ Durable: c.consumerName, @@ -158,21 +158,33 @@ func (c *Consumer) ensureConsumer() error { } } - _, err := c.js.AddConsumer(c.streamName, consumerConfig) - if err != nil && err != nats.ErrConsumerNameAlreadyInUse { - return fmt.Errorf("failed to create consumer: %w", err) - } - - if err == nil { - c.logger.Info("Created JetStream consumer", + // First check if the consumer already exists to make this initialization idempotent. + info, err := c.js.ConsumerInfo(c.streamName, c.consumerName) + if err == nil && info != nil { + c.logger.Info("Using existing JetStream consumer", zap.String("consumer", c.consumerName), zap.String("stream", c.streamName), zap.String("subject", c.subject), - zap.Duration("ack_wait", c.ackWait), - zap.String("deliver_policy", c.cfg.NATS.ConsumerRules.DeliverPolicy), - zap.String("replay_policy", c.cfg.NATS.ConsumerRules.ReplayPolicy), ) + return nil } + if err != nil && !errors.Is(err, nats.ErrConsumerNotFound) { + return fmt.Errorf("failed to fetch consumer info: %w", err) + } + + // Consumer does not exist; create it. + if _, err := c.js.AddConsumer(c.streamName, consumerConfig); err != nil { + return fmt.Errorf("failed to create consumer: %w", err) + } + + c.logger.Info("Created JetStream consumer", + zap.String("consumer", c.consumerName), + zap.String("stream", c.streamName), + zap.String("subject", c.subject), + zap.Duration("ack_wait", c.ackWait), + zap.String("deliver_policy", c.cfg.NATS.ConsumerRules.DeliverPolicy), + zap.String("replay_policy", c.cfg.NATS.ConsumerRules.ReplayPolicy), + ) return nil }