From 3d8572a69f99b304dce7d55be1d869e326c25319 Mon Sep 17 00:00:00 2001 From: windyboy Date: Sun, 16 Nov 2025 11:55:13 +0800 Subject: [PATCH] =?UTF-8?q?=E2=9C=A8=20Implement=20dead-letter=20queue=20(?= =?UTF-8?q?DLQ)=20enhancements=20by=20adding=20configuration=20options=20a?= =?UTF-8?q?nd=20validation=20logic.=20Update=20observability=20metrics=20t?= =?UTF-8?q?o=20track=20DLQ=20message=20counts=20and=20publish=20failures.?= =?UTF-8?q?=20Enhance=20documentation=20to=20reflect=20new=20metrics=20and?= =?UTF-8?q?=20DLQ=20behavior=20in=20JetStream=20mode.?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- configs/config.testdefaults.toml | 4 ++ docs/observability.md | 6 ++ docs/reliability.md | 7 +- internal/infra/nats/consumer.go | 79 ++++++++++++++++++++++- internal/observability/metrics/metrics.go | 37 +++++++++-- 5 files changed, 127 insertions(+), 6 deletions(-) diff --git a/configs/config.testdefaults.toml b/configs/config.testdefaults.toml index 3c198a4..69ec52b 100644 --- a/configs/config.testdefaults.toml +++ b/configs/config.testdefaults.toml @@ -22,6 +22,10 @@ format = "json" [telemetry] enabled = false +[dlq] +enabled = false +subject = "" + [monitoring] addr = ":0" enable_metrics = false diff --git a/docs/observability.md b/docs/observability.md index e03d18a..d73d349 100644 --- a/docs/observability.md +++ b/docs/observability.md @@ -22,6 +22,12 @@ The service exposes Prometheus metrics via the monitoring HTTP server (default ` - `caatsm_db_query_latency_seconds{operation}` DB operation latency. +- `caatsm_dlq_messages_total{stream,consumer}` + Count of messages successfully routed to the DLQ. + +- `caatsm_dlq_publish_failures_total{stream,consumer}` + Count of failures when attempting to publish messages to the DLQ. + Additional OTEL metrics are emitted via the configured OTEL endpoint, including: - `caatsm_messages_processed_total` diff --git a/docs/reliability.md b/docs/reliability.md index 2f4115c..05fab68 100644 --- a/docs/reliability.md +++ b/docs/reliability.md @@ -15,6 +15,10 @@ subject = "caatsm.dlq" - When `dlq.enabled` is `true` and `dlq.subject` is non-empty, **permanent** failures are routed to the DLQ subject. - A permanent failure is indicated by wrapping an error with `app.Permanent` and is treated as a **poison message**. +> Note: DLQ routing only applies when the NATS mode is `jetstream`. In `core` +> mode, the consumer does not attempt to publish to the DLQ even if it is +> configured. + Behaviour (JetStream mode): 1. The NATS consumer calls `processor.Handle`. @@ -33,7 +37,8 @@ The DLQ subject should be consumed by an offline repair/analysis tool or operati - Inspect poison messages. - Decide whether to fix and re-publish, or discard with justification. -- Track DLQ volume over time for alerting. +- Track DLQ volume over time for alerting (e.g. via `caatsm_dlq_messages_total` + and `caatsm_dlq_publish_failures_total` metrics). ### Transient Errors and Backoff diff --git a/internal/infra/nats/consumer.go b/internal/infra/nats/consumer.go index e68ed0b..c4d7d2b 100644 --- a/internal/infra/nats/consumer.go +++ b/internal/infra/nats/consumer.go @@ -72,7 +72,12 @@ func ProvideConsumer( streamName = "TELEGRAM" } - dlqSubject := strings.TrimSpace(cfg.DLQ.Subject) + // DLQ routing is only meaningful in JetStream mode. Respect dlq.enabled to allow + // environments to opt out cleanly even if a subject is configured. + dlqSubject := "" + if cfg.DLQ.Enabled { + dlqSubject = strings.TrimSpace(cfg.DLQ.Subject) + } ackWait := cfg.NATS.ConsumerRules.AckWait if ackWait == 0 { @@ -120,6 +125,9 @@ func ProvideConsumer( if err := consumer.ensureConsumer(); err != nil { return nil, fmt.Errorf("failed to ensure consumer: %w", err) } + // Validate DLQ configuration early so misconfiguration is visible at startup + // rather than only when the first poison message appears. + consumer.validateDLQ() } else { logger.Info("Running consumer in core NATS mode", zap.String("subject", subject), @@ -189,6 +197,64 @@ func (c *Consumer) ensureConsumer() error { return nil } +// validateDLQ verifies whether DLQ routing should be enabled and, if so, whether +// the configured DLQ subject is bound to a JetStream stream. If validation fails, +// DLQ routing is disabled (by clearing c.dlqSubject) and a warning is logged, +// but the consumer is still allowed to start. +func (c *Consumer) validateDLQ() { + if c == nil { + return + } + + // DLQ routing is only active in JetStream mode. + if c.mode != "jetstream" { + return + } + + // If DLQ is not enabled in config, make sure we don't accidentally route to it. + if !c.cfg.DLQ.Enabled { + if strings.TrimSpace(c.dlqSubject) != "" { + c.logger.Info("DLQ subject configured but dlq.enabled is false; DLQ routing disabled", + zap.String("dlq_subject", c.dlqSubject), + ) + } + c.dlqSubject = "" + return + } + + subject := strings.TrimSpace(c.dlqSubject) + if subject == "" { + c.logger.Warn("DLQ enabled but dlq.subject is empty; DLQ routing disabled") + return + } + + if c.js == nil { + c.logger.Warn("DLQ enabled but JetStream context is nil; DLQ routing disabled", + zap.String("dlq_subject", subject), + ) + c.dlqSubject = "" + return + } + + // Ensure the DLQ subject is actually bound to a JetStream stream. This avoids + // the opaque `nats: no response from stream` error later when publishing. + obsmetrics.RecordJSAPICall("dlq_validate_stream") + streamName, err := c.js.StreamNameBySubject(subject) + if err != nil || strings.TrimSpace(streamName) == "" { + c.logger.Warn("DLQ subject not bound to any JetStream stream; DLQ routing disabled", + zap.String("dlq_subject", subject), + zap.Error(err), + ) + c.dlqSubject = "" + return + } + + c.logger.Info("DLQ configuration validated", + zap.String("dlq_subject", subject), + zap.String("dlq_stream", streamName), + ) +} + // Start starts consuming messages func (c *Consumer) Start(ctx context.Context) error { if c.mode == "core" { @@ -507,9 +573,20 @@ func (c *Consumer) routeToDLQ(ctx context.Context, msg *nats.Msg, cause error) e } if _, err := c.js.Publish(c.dlqSubject, data); err != nil { + // nats.ErrNoResponders typically means that no JetStream stream is + // configured to receive this subject, or JetStream is temporarily + // unavailable. Surface this explicitly to make operational diagnosis + // easier. + if errors.Is(err, nats.ErrNoResponders) { + obsmetrics.RecordDLQPublishFailure(c.streamName, c.consumerName) + return fmt.Errorf("publish to dlq subject %s: no JetStream stream found for subject or JetStream unavailable: %w", c.dlqSubject, err) + } + obsmetrics.RecordDLQPublishFailure(c.streamName, c.consumerName) return fmt.Errorf("publish to dlq subject %s: %w", c.dlqSubject, err) } + obsmetrics.RecordDLQMessage(c.streamName, c.consumerName) + return nil } diff --git a/internal/observability/metrics/metrics.go b/internal/observability/metrics/metrics.go index 2f3b17b..fc2aad0 100644 --- a/internal/observability/metrics/metrics.go +++ b/internal/observability/metrics/metrics.go @@ -25,6 +25,8 @@ const ( MetricJSAPICallsTotal = "caatsm_js_api_calls_total" MetricDBQueriesTotal = "caatsm_db_queries_total" MetricDBQueryLatencySeconds = "caatsm_db_query_latency_seconds" + MetricDLQMessagesTotal = "caatsm_dlq_messages_total" + MetricDLQPublishFailures = "caatsm_dlq_publish_failures_total" // Common label keys. LabelStatus = "status" @@ -60,10 +62,12 @@ var ( parseLatency *prometheus.HistogramVec // Message handling metrics (per stream / consumer). - messagesTotal *prometheus.CounterVec - handleLatency *prometheus.HistogramVec - retriesTotal *prometheus.CounterVec - jsAPICallsTotal *prometheus.CounterVec + messagesTotal *prometheus.CounterVec + handleLatency *prometheus.HistogramVec + retriesTotal *prometheus.CounterVec + jsAPICallsTotal *prometheus.CounterVec + dlqMessagesTotal *prometheus.CounterVec + dlqPublishFailures *prometheus.CounterVec // Database metrics. dbQueriesTotal *prometheus.CounterVec @@ -107,6 +111,16 @@ func initCollectors() { Help: "Total number of message retries (negative acknowledgements), labelled by stream, consumer and reason.", }, []string{LabelStream, LabelConsumer, LabelReason}) + dlqMessagesTotal = prometheus.NewCounterVec(prometheus.CounterOpts{ + Name: MetricDLQMessagesTotal, + Help: "Total number of messages routed to the DLQ, labelled by stream and consumer.", + }, []string{LabelStream, LabelConsumer}) + + dlqPublishFailures = prometheus.NewCounterVec(prometheus.CounterOpts{ + Name: MetricDLQPublishFailures, + Help: "Total number of failures when publishing to the DLQ, labelled by stream and consumer.", + }, []string{LabelStream, LabelConsumer}) + jsAPICallsTotal = prometheus.NewCounterVec(prometheus.CounterOpts{ Name: MetricJSAPICallsTotal, Help: "Count of JetStream API calls made by the receiver.", @@ -132,6 +146,8 @@ func initCollectors() { handleLatency, retriesTotal, jsAPICallsTotal, + dlqMessagesTotal, + dlqPublishFailures, dbQueriesTotal, dbQueryLatency, ) @@ -179,6 +195,19 @@ func RecordRetry(stream, consumer, reason string) { retriesTotal.WithLabelValues(labelValue(stream), labelValue(consumer), labelValue(reason)).Inc() } +// RecordDLQMessage increments the DLQ message counter for a successfully routed message. +func RecordDLQMessage(stream, consumer string) { + ensureCollectors() + dlqMessagesTotal.WithLabelValues(labelValue(stream), labelValue(consumer)).Inc() +} + +// RecordDLQPublishFailure increments the DLQ publish failure counter when a DLQ +// publish attempt fails. +func RecordDLQPublishFailure(stream, consumer string) { + ensureCollectors() + dlqPublishFailures.WithLabelValues(labelValue(stream), labelValue(consumer)).Inc() +} + // RecordDBQuery records metrics for a single database operation. // Operation examples: "insert_one", "insert_batch", "insert_raw". // Result is usually "ok" or "error".