From 8914156a58b51fa999d7be4d62cd3fae1cb0d9e8 Mon Sep 17 00:00:00 2001 From: windyboy Date: Wed, 19 Nov 2025 16:49:53 +0800 Subject: [PATCH] =?UTF-8?q?=E2=9C=A8=20Update=20Go=20module=20dependencies?= =?UTF-8?q?=20and=20enhance=20NATS=20consumer=20metrics.=20Add=20`github.c?= =?UTF-8?q?om/kylelemons/godebug`=20as=20an=20indirect=20dependency.=20Upg?= =?UTF-8?q?rade=20`github.com/prometheus/common`=20to=20version=200.67.3.?= =?UTF-8?q?=20Change=20NATS=20consumer=20mode=20to=20"jetstream"=20in=20th?= =?UTF-8?q?e=20development=20configuration.=20Improve=20observability=20by?= =?UTF-8?q?=20ensuring=20metrics=20for=20pending=20messages=20are=20initia?= =?UTF-8?q?lized=20and=20recorded=20correctly=20in=20JetStream=20mode.=20U?= =?UTF-8?q?pdate=20documentation=20to=20reflect=20changes=20in=20metrics?= =?UTF-8?q?=20handling=20and=20consumer=20behavior.?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- configs/config.dev.toml | 2 +- docs/observability.md | 3 +- docs/reliability.md | 19 +++--- go.mod | 3 +- internal/infra/metrics/metrics.go | 16 ++++- internal/infra/metrics/metrics_test.go | 77 +++++++++++++++++++---- internal/infra/nats/consumer.go | 80 +++++++++++++++++++++--- internal/infra/nats/message_processor.go | 31 +++++++-- internal/infra/nats/stream_manager.go | 77 ++++++++++++++++++++--- test/integration/nats_recovery_test.go | 2 +- 10 files changed, 261 insertions(+), 49 deletions(-) diff --git a/configs/config.dev.toml b/configs/config.dev.toml index 64922d3..15505a6 100644 --- a/configs/config.dev.toml +++ b/configs/config.dev.toml @@ -15,7 +15,7 @@ url = "nats://localhost:4222" # * Dead-letter queue (DLQ) support # * Batch processing and consumer monitoring # * Recommended for production environments -mode = "core" +mode = "jetstream" client = "serial-client" cluster = "tele-cluster" stream = "TELEGRAM" diff --git a/docs/observability.md b/docs/observability.md index d657208..4aab9fb 100644 --- a/docs/observability.md +++ b/docs/observability.md @@ -32,7 +32,7 @@ The service exposes Prometheus metrics via the monitoring HTTP server (default ` Count of general publish failures (not DLQ-specific), labelled by message category. - `caatsm_nats_consumer_pending_messages{stream,consumer}` - Current pending message count for each JetStream consumer (useful for lag/backlog alerts). + Current pending message count for each JetStream consumer (useful for lag/backlog alerts). A zero sample is emitted only once at startup so the series exists; if JetStream stats queries fail later, the last known value is preserved rather than force-setting the gauge to `0`, which prevents false “queue cleared” alerts. Additional OTEL metrics are emitted via the configured OTEL endpoint, including: @@ -282,4 +282,3 @@ Logging is done with Zap. The `internal/infra/log` package standardises fields v Handler and consumer logs should always be emitted through `WithMessageContext` to ensure these fields are present where applicable. - diff --git a/docs/reliability.md b/docs/reliability.md index 6d76b96..0b93efc 100644 --- a/docs/reliability.md +++ b/docs/reliability.md @@ -107,14 +107,16 @@ Recommended pattern: restarted and returns `ErrNoResponders`), the consumer uses an exponential backoff when retrying `Fetch` calls (roughly `1s, 2s, 4s, ...` up to around `30s`) to avoid log spam while allowing the system to recover. -- In dev/test environments, if the stream or consumer is detected as missing at - runtime (for example after `docker compose down -v`), the consumer calls the - shared `EnsureStream` and `ensureConsumer` logic to recreate them and - re-establish subscriptions. -- In production environments, missing streams/consumers are treated as - configuration or operational errors: - - They are **not** auto-recreated. - - Errors are logged prominently so operators can diagnose and fix the issue. +- At startup the receiver always calls `StreamManager.EnsureStream` and + `ConsumerManager.EnsureConsumer`. If the JetStream account allows it, + missing streams are created with the configured retention limits + (`max_msgs`, `max_bytes`, `max_age`, discard/storage policy, replicas) before + the durable consumer is created. This keeps dev/test clusters self-healing + after `docker compose down -v` and removes the race where a consumer was + created without its stream. +- When the JetStream account lacks permissions to create protected resources + (a common production posture), the same code path fails fast with a clear + error message so operators know they must provision the stream out-of-band. - On the publishing side, JetStream `ErrNoResponders` and similar errors are treated as temporary by the processor: - Such errors cause the consumer to NAK messages and rely on the configured @@ -142,4 +144,3 @@ Dashboards should combine: - NATS consumer statistics (pending, redelivered, ack_pending). - DB health indicators (latency, error counts, connection usage). - diff --git a/go.mod b/go.mod index e2d051f..5eb28ab 100644 --- a/go.mod +++ b/go.mod @@ -63,6 +63,7 @@ require ( github.com/jackc/puddle/v2 v2.2.2 // indirect github.com/klauspost/compress v1.18.1 // indirect github.com/knadh/koanf/maps v0.1.2 // indirect + github.com/kylelemons/godebug v1.1.0 // indirect github.com/lufia/plan9stats v0.0.0-20211012122336-39d0f177ccd0 // indirect github.com/magiconair/properties v1.8.7 // indirect github.com/mitchellh/copystructure v1.2.0 // indirect @@ -82,7 +83,7 @@ require ( github.com/pmezard/go-difflib v1.0.0 // indirect github.com/power-devops/perfstat v0.0.0-20210106213030-5aafc221ea8c // indirect github.com/prometheus/client_model v0.6.2 // indirect - github.com/prometheus/common v0.67.2 // indirect + github.com/prometheus/common v0.67.3 // indirect github.com/prometheus/procfs v0.19.2 // indirect github.com/russross/blackfriday/v2 v2.1.0 // indirect github.com/shirou/gopsutil/v3 v3.23.12 // indirect diff --git a/internal/infra/metrics/metrics.go b/internal/infra/metrics/metrics.go index 91ba0de..52e2f48 100644 --- a/internal/infra/metrics/metrics.go +++ b/internal/infra/metrics/metrics.go @@ -255,7 +255,21 @@ func RecordJSAPICall(operation string) { // JetStream consumer as a gauge, enabling backlog / lag alerts. func RecordNATSConsumerPending(stream, consumer string, pending uint64) { ensureCollectors() - natsConsumerPending.WithLabelValues(labelValue(stream), labelValue(consumer)).Set(float64(pending)) + + // Validate inputs to ensure metric is recorded correctly + streamLabel := labelValue(stream) + consumerLabel := labelValue(consumer) + + // Ensure metric is always set, even with empty labels (will be "unknown") + if streamLabel == "" { + streamLabel = "unknown" + } + if consumerLabel == "" { + consumerLabel = "unknown" + } + + // Set the metric value + natsConsumerPending.WithLabelValues(streamLabel, consumerLabel).Set(float64(pending)) } func labelValue(value string) string { diff --git a/internal/infra/metrics/metrics_test.go b/internal/infra/metrics/metrics_test.go index dc1278e..04011f0 100644 --- a/internal/infra/metrics/metrics_test.go +++ b/internal/infra/metrics/metrics_test.go @@ -3,9 +3,12 @@ package metrics import ( "net/http" "net/http/httptest" + "sync" "testing" "time" + "github.com/prometheus/client_golang/prometheus/testutil" + . "github.com/onsi/ginkgo/v2" . "github.com/onsi/gomega" ) @@ -16,6 +19,10 @@ func TestMetrics(t *testing.T) { } var _ = Describe("Metrics", func() { + BeforeEach(func() { + resetMetricsForTest() + }) + Describe("Handler", func() { It("serves metrics with the correct content type", func() { req := httptest.NewRequest(http.MethodGet, "/metrics", nil) @@ -29,25 +36,69 @@ var _ = Describe("Metrics", func() { }) Describe("labelValue helper", func() { - It("returns unknown for empty values and lowercases input", func() { + It("normalizes empty and uppercase values", func() { Expect(labelValue(" ")).To(Equal("unknown")) Expect(labelValue("SOME_VALUE")).To(Equal("some_value")) }) }) Describe("record helpers", func() { - It("can be invoked without panicking", func() { - Expect(func() { - RecordProcessed("parsed", "ARR", 150*time.Millisecond) - RecordFailure("parser") - RecordMessageHandled("TEST", "consumer", ResultOK, 205*time.Millisecond) - RecordRetry("TEST", "consumer", RetryReasonProcessorError) - RecordDLQMessage("TEST", "consumer") - RecordDLQPublishFailure("TEST", "consumer") - RecordDBQuery("insert", DBResultOK, 10*time.Millisecond) - RecordJSAPICall("publish") - RecordNATSConsumerPending("TEST", "consumer", 7) - }).NotTo(Panic()) + It("record counters and histograms for message processing", func() { + RecordProcessed("PARSED", "ARR", 150*time.Millisecond) + RecordFailure("parser") + RecordMessageHandled("stream1", "consumer1", ResultOK, 80*time.Millisecond) + RecordRetry("stream1", "consumer1", RetryReasonProcessorError) + RecordDLQMessage("stream1", "consumer1") + RecordDLQPublishFailure("stream1", "consumer1") + RecordPublishFailure("flightPlan") + RecordDBQuery("insert_one", DBResultOK, 10*time.Millisecond) + RecordJSAPICall("publish") + + Expect(testutil.ToFloat64(processedCounter.WithLabelValues("parsed", "arr"))).To(BeNumerically("==", 1)) + Expect(testutil.CollectAndCount(parseLatency, MetricParseLatencySeconds)).To(Equal(1)) + + Expect(testutil.ToFloat64(failureCounter.WithLabelValues("parser"))).To(BeNumerically("==", 1)) + + Expect(testutil.ToFloat64(messagesTotal.WithLabelValues("stream1", "consumer1", ResultOK))).To(BeNumerically("==", 1)) + Expect(testutil.CollectAndCount(handleLatency, MetricHandleLatencySeconds)).To(Equal(1)) + + Expect(testutil.ToFloat64(retriesTotal.WithLabelValues("stream1", "consumer1", RetryReasonProcessorError))).To(BeNumerically("==", 1)) + + Expect(testutil.ToFloat64(dlqMessagesTotal.WithLabelValues("stream1", "consumer1"))).To(BeNumerically("==", 1)) + Expect(testutil.ToFloat64(dlqPublishFailures.WithLabelValues("stream1", "consumer1"))).To(BeNumerically("==", 1)) + + Expect(testutil.ToFloat64(publishFailuresTotal.WithLabelValues("flightplan"))).To(BeNumerically("==", 1)) + + Expect(testutil.ToFloat64(dbQueriesTotal.WithLabelValues("insert_one", DBResultOK))).To(BeNumerically("==", 1)) + Expect(testutil.CollectAndCount(dbQueryLatency, MetricDBQueryLatencySeconds)).To(Equal(1)) + + Expect(testutil.ToFloat64(jsAPICallsTotal.WithLabelValues("publish"))).To(BeNumerically("==", 1)) + }) + + It("records gauge values for pending messages", func() { + RecordNATSConsumerPending("STREAM_A", "consumerA", 42) + Expect(testutil.ToFloat64(natsConsumerPending.WithLabelValues("stream_a", "consumera"))).To(BeNumerically("==", 42)) + + RecordNATSConsumerPending("STREAM_A", "consumerA", 5) + Expect(testutil.ToFloat64(natsConsumerPending.WithLabelValues("stream_a", "consumera"))).To(BeNumerically("==", 5)) }) }) }) + +func resetMetricsForTest() { + once = sync.Once{} + registry = nil + processedCounter = nil + failureCounter = nil + parseLatency = nil + messagesTotal = nil + handleLatency = nil + retriesTotal = nil + jsAPICallsTotal = nil + dlqMessagesTotal = nil + dlqPublishFailures = nil + publishFailuresTotal = nil + dbQueriesTotal = nil + dbQueryLatency = nil + natsConsumerPending = nil +} diff --git a/internal/infra/nats/consumer.go b/internal/infra/nats/consumer.go index 11ee7e6..6eb0049 100644 --- a/internal/infra/nats/consumer.go +++ b/internal/infra/nats/consumer.go @@ -76,6 +76,27 @@ func ProvideConsumer( } consumer.initCollaborators() + // Initialize the pending messages metric early (set to 0) so it appears in Prometheus + // even before the consumer starts. This ensures the metric is always visible. + // We do this first, before any operations that might fail, to ensure the metric exists. + // Initialize the metric unconditionally when in JetStream mode, even if js is nil, + // as it will be updated later when js becomes available. + if normCfg.mode == "jetstream" { + logger.Info("Initializing NATS consumer pending messages metric", + zap.String("stream", normCfg.streamName), + zap.String("consumer", normCfg.consumerName), + zap.Uint64("pending", 0), + zap.Bool("js_available", js != nil), + ) + // Always initialize the metric in JetStream mode to ensure it appears in Prometheus + // The metric will be updated with actual values when the consumer starts + obsmetrics.RecordNATSConsumerPending(normCfg.streamName, normCfg.consumerName, 0) + } else { + logger.Debug("Skipping NATS consumer pending messages metric initialization (not JetStream mode)", + zap.String("mode", normCfg.mode), + ) + } + // Initialize managers if consumer.config.mode == "jetstream" { consumer.consumerManager = NewConsumerManager(js, normCfg.streamName, normCfg.consumerName, normCfg.subject, logger) @@ -93,6 +114,19 @@ func ProvideConsumer( fetcher.streamManager = consumer.streamManager } + // Ensure stream exists before creating consumer + streamCfg := &StreamConfig{ + MaxMsgs: cfg.NATS.StreamLimits.MaxMsgs, + MaxBytes: cfg.NATS.StreamLimits.MaxBytes, + MaxAge: cfg.NATS.StreamLimits.MaxAge, + Discard: cfg.NATS.StreamLimits.Discard, + Storage: cfg.NATS.StreamLimits.Storage, + Replicas: cfg.NATS.StreamLimits.Replicas, + } + if err := consumer.streamManager.EnsureStream(streamCfg); err != nil { + return nil, fmt.Errorf("failed to ensure stream: %w", err) + } + // Create consumer if it doesn't exist consumerConfig := consumer.buildConsumerConfig() if err := consumer.consumerManager.EnsureConsumer(consumerConfig); err != nil { @@ -269,14 +303,9 @@ func (c *Consumer) startCore(ctx context.Context) error { } handler := func(msg *nats.Msg) { - if err := c.batchProcessor.ProcessMessage(ctx, msg); err != nil { - isPermanent := app.IsPermanent(err) - c.logger.Error("Failed to process message (core mode)", - zap.String("subject", msg.Subject), - zap.Error(err), - zap.Bool("permanent", isPermanent), - ) - } + // Use ProcessBatch to ensure metrics are recorded via processSingleMessage + // ProcessBatch handles error recording and metrics for both success and failure cases + c.batchProcessor.ProcessBatch(ctx, []*nats.Msg{msg}) } sub, err := c.conn.QueueSubscribe(c.config.subject, queueGroup, handler) @@ -335,6 +364,23 @@ func (c *Consumer) startJetStream(ctx context.Context) error { zap.String("stream", c.config.streamName), ) + // Record initial pending messages metric immediately + // This ensures the metric appears in Prometheus right away + if info, err := c.js.ConsumerInfo(c.config.streamName, c.config.consumerName); err == nil { + c.logger.Info("Recording initial NATS consumer pending messages metric", + zap.String("stream", c.config.streamName), + zap.String("consumer", c.config.consumerName), + zap.Uint64("pending", info.NumPending), + ) + obsmetrics.RecordNATSConsumerPending(c.config.streamName, c.config.consumerName, info.NumPending) + } else { + c.logger.Warn("Failed to fetch initial consumer info for pending messages metric", + zap.String("stream", c.config.streamName), + zap.String("consumer", c.config.consumerName), + zap.Error(err), + ) + } + statsCtx, statsCancel := context.WithCancel(ctx) defer statsCancel() go c.emitConsumerStats(statsCtx) @@ -379,6 +425,14 @@ func (c *Consumer) emitConsumerStats(ctx context.Context) { ticker := time.NewTicker(c.config.monitorInterval) defer ticker.Stop() + // Record initial metric (0) to ensure it appears in Prometheus even before first tick + c.logger.Info("Starting NATS consumer stats emission goroutine, recording initial pending metric", + zap.String("stream", c.config.streamName), + zap.String("consumer", c.config.consumerName), + zap.Duration("interval", c.config.monitorInterval), + ) + obsmetrics.RecordNATSConsumerPending(c.config.streamName, c.config.consumerName, 0) + for { select { case <-ctx.Done(): @@ -386,9 +440,19 @@ func (c *Consumer) emitConsumerStats(ctx context.Context) { case <-ticker.C: info, err := c.js.ConsumerInfo(c.config.streamName, c.config.consumerName) if err != nil { + c.logger.Warn("Failed to fetch consumer info for pending messages metric", + zap.String("stream", c.config.streamName), + zap.String("consumer", c.config.consumerName), + zap.Error(err), + ) continue } // Record pending messages for monitoring + c.logger.Debug("Recording NATS consumer pending messages metric", + zap.String("stream", c.config.streamName), + zap.String("consumer", c.config.consumerName), + zap.Uint64("pending", info.NumPending), + ) obsmetrics.RecordNATSConsumerPending(c.config.streamName, c.config.consumerName, info.NumPending) } } diff --git a/internal/infra/nats/message_processor.go b/internal/infra/nats/message_processor.go index bcb30cb..655f9ef 100644 --- a/internal/infra/nats/message_processor.go +++ b/internal/infra/nats/message_processor.go @@ -74,11 +74,19 @@ func (p *defaultBatchProcessor) processSingleMessage(ctx context.Context, msg *n *p.consecutiveProcessErrors = 0 } - // ACK the message - if ackErr := msg.Ack(); ackErr != nil { - p.logger.Error("Failed to ACK message", zap.Error(ackErr)) + elapsed := time.Since(start) + + // ACK the message (only in JetStream mode; Core NATS doesn't support ACK) + if p.mode == "jetstream" { + if ackErr := msg.Ack(); ackErr != nil { + p.logger.Error("Failed to ACK message", zap.Error(ackErr)) + // Still record metrics even if ACK fails + p.telemetry.RecordMessageHandled(ctx, p.streamName, p.consumerName, "ok", elapsed) + } else { + p.telemetry.RecordMessageHandled(ctx, p.streamName, p.consumerName, "ok", elapsed) + } } else { - elapsed := time.Since(start) + // Core NATS mode: record metrics without ACK (ACK not supported) p.telemetry.RecordMessageHandled(ctx, p.streamName, p.consumerName, "ok", elapsed) } } @@ -206,6 +214,14 @@ func (p *defaultBatchProcessor) handlePermanentError(ctx context.Context, msg *n if p.consecutiveProcessErrors != nil { *p.consecutiveProcessErrors = 0 } + + if p.mode != "jetstream" { + p.logger.Debug("Permanent-error message in core mode; skipping DLQ/ACK (not supported)", + zap.String("subject", msg.Subject), + ) + return + } + // Poison/permanent message: route to DLQ if configured, then ACK if p.dlqHandler != nil { if dlqErr := p.dlqHandler.RouteToDLQ(ctx, msg, err); dlqErr != nil { @@ -243,6 +259,13 @@ func (p *defaultBatchProcessor) handleTransientError(ctx context.Context, msg *n } } + if p.mode != "jetstream" { + p.logger.Debug("Transient-error message in core mode; skipping retry (ACK/NAK unsupported)", + zap.String("subject", msg.Subject), + ) + return + } + // Transient error: request redelivery with optional delay p.telemetry.RecordRetry(ctx, p.streamName, p.consumerName, obsmetrics.RetryReasonProcessorError) if nakErr := p.nakWithStrategy(msg); nakErr != nil { diff --git a/internal/infra/nats/stream_manager.go b/internal/infra/nats/stream_manager.go index 79d0200..335dbd0 100644 --- a/internal/infra/nats/stream_manager.go +++ b/internal/infra/nats/stream_manager.go @@ -1,7 +1,9 @@ package nats import ( + "errors" "fmt" + "time" "github.com/nats-io/nats.go" "go.uber.org/zap" @@ -15,6 +17,16 @@ type StreamManager struct { logger *zap.Logger } +// StreamConfig holds configuration for creating a JetStream stream +type StreamConfig struct { + MaxMsgs int64 + MaxBytes int64 + MaxAge time.Duration + Discard string // "old" or "new" + Storage string // "file" or "memory" + Replicas int +} + // NewStreamManager creates a new stream manager func NewStreamManager(js nats.JetStreamContext, streamName string, subjects []string, logger *zap.Logger) *StreamManager { return &StreamManager{ @@ -25,16 +37,63 @@ func NewStreamManager(js nats.JetStreamContext, streamName string, subjects []st } } -// EnsureStream ensures that the configured JetStream stream exists -func (sm *StreamManager) EnsureStream() error { +// EnsureStream ensures that the configured JetStream stream exists, creating it if necessary +func (sm *StreamManager) EnsureStream(cfg *StreamConfig) error { + // Check if stream already exists _, err := sm.js.StreamInfo(sm.streamName) - if err != nil { - return fmt.Errorf("stream %s not found or inaccessible: %w", sm.streamName, err) + if err == nil { + sm.logger.Info("JetStream stream verified", + zap.String("stream", sm.streamName), + zap.Strings("subjects", sm.subjects), + ) + return nil } - sm.logger.Info("JetStream stream verified", - zap.String("stream", sm.streamName), - zap.Strings("subjects", sm.subjects), - ) - return nil + // If stream doesn't exist, create it + if errors.Is(err, nats.ErrStreamNotFound) { + streamCfg := &nats.StreamConfig{ + Name: sm.streamName, + Subjects: sm.subjects, + } + + // Apply limits if provided + if cfg != nil { + if cfg.MaxMsgs > 0 { + streamCfg.MaxMsgs = cfg.MaxMsgs + } + if cfg.MaxBytes > 0 { + streamCfg.MaxBytes = cfg.MaxBytes + } + if cfg.MaxAge > 0 { + streamCfg.MaxAge = cfg.MaxAge + } + if cfg.Discard == "new" { + streamCfg.Discard = nats.DiscardNew + } else { + streamCfg.Discard = nats.DiscardOld + } + if cfg.Storage == "memory" { + streamCfg.Storage = nats.MemoryStorage + } else { + streamCfg.Storage = nats.FileStorage + } + if cfg.Replicas > 0 { + streamCfg.Replicas = cfg.Replicas + } + } + + _, err := sm.js.AddStream(streamCfg) + if err != nil { + return fmt.Errorf("failed to create stream %s: %w", sm.streamName, err) + } + + sm.logger.Info("Created JetStream stream", + zap.String("stream", sm.streamName), + zap.Strings("subjects", sm.subjects), + ) + return nil + } + + // Other error (e.g., permission denied) + return fmt.Errorf("stream %s not found or inaccessible: %w", sm.streamName, err) } diff --git a/test/integration/nats_recovery_test.go b/test/integration/nats_recovery_test.go index d85567f..a67a872 100644 --- a/test/integration/nats_recovery_test.go +++ b/test/integration/nats_recovery_test.go @@ -39,7 +39,7 @@ func TestNATSConsumerRecovery(t *testing.T) { // Test stream recovery streamManager := natsinfra.NewStreamManager(js, "TEST_STREAM", []string{"test.subject"}, nil) - err = streamManager.EnsureStream() + err = streamManager.EnsureStream(nil) // nil uses default stream configuration assert.NoError(t, err) // Test consumer recovery