From a96cb8a40b413c6147861a094bc8bbd746455f7e Mon Sep 17 00:00:00 2001 From: windyboy Date: Sun, 16 Nov 2025 09:15:37 +0800 Subject: [PATCH] =?UTF-8?q?=E2=9C=A8=20Implement=20dead-letter=20queue=20(?= =?UTF-8?q?DLQ)=20functionality=20for=20handling=20permanent=20failures=20?= =?UTF-8?q?in=20message=20processing.=20Update=20configuration=20to=20enab?= =?UTF-8?q?le=20DLQ=20and=20specify=20the=20subject=20for=20routing=20fail?= =?UTF-8?q?ed=20messages.=20Enhance=20observability=20by=20adding=20metric?= =?UTF-8?q?s=20for=20message=20handling,=20retries,=20and=20database=20ope?= =?UTF-8?q?rations.=20Introduce=20structured=20logging=20for=20better=20tr?= =?UTF-8?q?aceability=20of=20message=20processing=20events.?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- configs/config.dev.toml | 6 +- .../caatsm-overview.json | 265 ++++++++++++++++++ docs/architecture-ha.md | 74 +++++ docs/observability.md | 111 ++++++++ docs/reliability.md | 119 ++++++++ internal/app/processor.go | 39 ++- internal/infra/config/config.go | 9 + internal/infra/monitoring/server.go | 5 + internal/infra/nats/consumer.go | 131 ++++++++- internal/infra/postgres/repository.go | 70 +++++ internal/observability/logging/logger.go | 102 +++++++ internal/observability/metrics/metrics.go | 153 +++++++++- 12 files changed, 1053 insertions(+), 31 deletions(-) create mode 100644 configs/grafana-dashboards.dev/caatsm-overview.json create mode 100644 docs/architecture-ha.md create mode 100644 docs/observability.md create mode 100644 docs/reliability.md create mode 100644 internal/observability/logging/logger.go diff --git a/configs/config.dev.toml b/configs/config.dev.toml index ac2d030..c04094e 100644 --- a/configs/config.dev.toml +++ b/configs/config.dev.toml @@ -62,4 +62,8 @@ enable_metrics = true enable_health = true read_timeout = "5s" write_timeout = "5s" -health_timeout = "2s" \ No newline at end of file +health_timeout = "2s" + +[dlq] +enabled = true +subject = "caatsm.dlq" \ No newline at end of file diff --git a/configs/grafana-dashboards.dev/caatsm-overview.json b/configs/grafana-dashboards.dev/caatsm-overview.json new file mode 100644 index 0000000..2d1422e --- /dev/null +++ b/configs/grafana-dashboards.dev/caatsm-overview.json @@ -0,0 +1,265 @@ +{ + "id": null, + "uid": "caatsm-overview", + "title": "CAATSM – Receiver Overview", + "tags": ["caatsm", "receiver", "dev"], + "schemaVersion": 38, + "version": 1, + "refresh": "10s", + "time": { + "from": "now-15m", + "to": "now" + }, + "timepicker": {}, + "annotations": { + "list": [ + { + "builtIn": 1, + "datasource": "-- Grafana --", + "enable": true, + "hide": true, + "iconColor": "rgba(0, 211, 255, 1)", + "name": "Annotations & Alerts", + "type": "dashboard" + } + ] + }, + "templating": { + "list": [] + }, + "panels": [ + { + "id": 1, + "type": "stat", + "title": "Messages by result (5m rate)", + "datasource": { + "type": "prometheus", + "uid": "prometheus-dev" + }, + "gridPos": { "h": 4, "w": 8, "x": 0, "y": 0 }, + "options": { + "colorMode": "value", + "graphMode": "area", + "justifyMode": "auto", + "orientation": "auto", + "reduceOptions": { + "calcs": ["lastNotNull"], + "fields": "", + "values": false + }, + "textMode": "auto" + }, + "fieldConfig": { + "defaults": { + "mappings": [], + "thresholds": { + "mode": "absolute", + "steps": [ + { "color": "green", "value": null }, + { "color": "red", "value": 0 } + ] + } + }, + "overrides": [] + }, + "targets": [ + { + "refId": "A", + "expr": "sum by (result) (rate(caatsm_messages_total[5m]))", + "legendFormat": "{{result}}" + } + ] + }, + { + "id": 2, + "type": "timeseries", + "title": "Messages per stream/consumer", + "datasource": { + "type": "prometheus", + "uid": "prometheus-dev" + }, + "gridPos": { "h": 8, "w": 16, "x": 8, "y": 0 }, + "fieldConfig": { + "defaults": { + "color": { "mode": "palette-classic" }, + "mappings": [], + "thresholds": { "mode": "absolute", "steps": [] } + }, + "overrides": [] + }, + "options": { + "legend": { "displayMode": "table", "placement": "bottom" }, + "tooltip": { "mode": "single" } + }, + "targets": [ + { + "refId": "A", + "expr": "sum by (stream, consumer) (rate(caatsm_messages_total[5m]))", + "legendFormat": "{{stream}} / {{consumer}}" + } + ] + }, + { + "id": 3, + "type": "timeseries", + "title": "End-to-end handle latency (P50/P95/P99)", + "datasource": { + "type": "prometheus", + "uid": "prometheus-dev" + }, + "gridPos": { "h": 8, "w": 24, "x": 0, "y": 4 }, + "fieldConfig": { + "defaults": { + "color": { "mode": "palette-classic" }, + "mappings": [], + "thresholds": { "mode": "absolute", "steps": [] } + }, + "overrides": [] + }, + "options": { + "legend": { "displayMode": "table", "placement": "bottom" }, + "tooltip": { "mode": "single" } + }, + "targets": [ + { + "refId": "P50", + "expr": "histogram_quantile(0.50, sum by (le) (rate(caatsm_handle_latency_seconds_bucket[5m])))", + "legendFormat": "P50" + }, + { + "refId": "P95", + "expr": "histogram_quantile(0.95, sum by (le) (rate(caatsm_handle_latency_seconds_bucket[5m])))", + "legendFormat": "P95" + }, + { + "refId": "P99", + "expr": "histogram_quantile(0.99, sum by (le) (rate(caatsm_handle_latency_seconds_bucket[5m])))", + "legendFormat": "P99" + } + ] + }, + { + "id": 4, + "type": "timeseries", + "title": "DB query rate by operation/result", + "datasource": { + "type": "prometheus", + "uid": "prometheus-dev" + }, + "gridPos": { "h": 7, "w": 12, "x": 0, "y": 12 }, + "fieldConfig": { + "defaults": { + "color": { "mode": "palette-classic" }, + "mappings": [], + "thresholds": { "mode": "absolute", "steps": [] } + }, + "overrides": [] + }, + "options": { + "legend": { "displayMode": "table", "placement": "bottom" }, + "tooltip": { "mode": "single" } + }, + "targets": [ + { + "refId": "A", + "expr": "sum by (operation, result) (rate(caatsm_db_queries_total[5m]))", + "legendFormat": "{{operation}} / {{result}}" + } + ] + }, + { + "id": 5, + "type": "timeseries", + "title": "DB query latency P95 by operation", + "datasource": { + "type": "prometheus", + "uid": "prometheus-dev" + }, + "gridPos": { "h": 7, "w": 12, "x": 12, "y": 12 }, + "fieldConfig": { + "defaults": { + "color": { "mode": "palette-classic" }, + "mappings": [], + "thresholds": { "mode": "absolute", "steps": [] } + }, + "overrides": [] + }, + "options": { + "legend": { "displayMode": "table", "placement": "bottom" }, + "tooltip": { "mode": "single" } + }, + "targets": [ + { + "refId": "A", + "expr": "histogram_quantile(0.95, sum by (le, operation) (rate(caatsm_db_query_latency_seconds_bucket[5m])))", + "legendFormat": "{{operation}}" + } + ] + }, + { + "id": 6, + "type": "timeseries", + "title": "Retries and permanent failures", + "datasource": { + "type": "prometheus", + "uid": "prometheus-dev" + }, + "gridPos": { "h": 7, "w": 12, "x": 0, "y": 19 }, + "fieldConfig": { + "defaults": { + "color": { "mode": "palette-classic" }, + "mappings": [], + "thresholds": { "mode": "absolute", "steps": [] } + }, + "overrides": [] + }, + "options": { + "legend": { "displayMode": "table", "placement": "bottom" }, + "tooltip": { "mode": "single" } + }, + "targets": [ + { + "refId": "retries", + "expr": "sum(rate(caatsm_retries_total[5m]))", + "legendFormat": "retries" + }, + { + "refId": "permanent_fail", + "expr": "sum(rate(caatsm_messages_total{result=\"permanent_fail\"}[5m]))", + "legendFormat": "permanent_fail" + } + ] + }, + { + "id": 7, + "type": "timeseries", + "title": "Publish failures by category", + "datasource": { + "type": "prometheus", + "uid": "prometheus-dev" + }, + "gridPos": { "h": 7, "w": 12, "x": 12, "y": 19 }, + "fieldConfig": { + "defaults": { + "color": { "mode": "palette-classic" }, + "mappings": [], + "thresholds": { "mode": "absolute", "steps": [] } + }, + "overrides": [] + }, + "options": { + "legend": { "displayMode": "table", "placement": "bottom" }, + "tooltip": { "mode": "single" } + }, + "targets": [ + { + "refId": "A", + "expr": "sum by (message_category) (rate(caatsm_publish_failures_total[5m]))", + "legendFormat": "{{message_category}}" + } + ] + } + ] +} + + diff --git a/docs/architecture-ha.md b/docs/architecture-ha.md new file mode 100644 index 0000000..56e4f59 --- /dev/null +++ b/docs/architecture-ha.md @@ -0,0 +1,74 @@ +## High-Level Architecture and HA / Scaling + +### Components + +- **NATS / JetStream** – message broker providing durable storage and redelivery: + - Stream: `TELEGRAM` (configurable via `nats.stream`). + - Consumer: durable pull consumer per deployment (`nats.consumer`). + - Subjects: + - Inbound telegrams: `subscription.topic` (e.g. `telegram.serial`). + - Parsed telegrams (publisher): `publisher.topic` (e.g. `telegram.json`). + - Dead-letter: `dlq.subject` (e.g. `caatsm.dlq`). + +- **Receiver service (`caatsm`)**: + - NATS JetStream **pull consumer** (`internal/infra/nats/consumer.go`). + - Telegram parser and domain model (`internal/app`, `internal/parsers`, `internal/domain`). + - PostgreSQL repository (`internal/infra/postgres`). + - Monitoring/observability server (`internal/infra/monitoring`). + +### HA and Failover + +- NATS/JetStream is expected to run as a **cluster** with `replicas` configured on the stream to ensure message durability. +- The receiver service is stateless aside from DB side effects and can be deployed with multiple replicas: + - Each replica connects to the same NATS cluster and JetStream stream. + - Durability and at-least-once semantics are handled by JetStream. + +Consumer behaviour: + +- Pull-based consumption with configurable batch size/timeout (`app.batch_size`, `app.batch_timeout`). +- When a receiver instance stops or crashes: + - Its NATS connection is drained and closed. + - Remaining messages remain in the stream. + - Another healthy instance continues pulling from the durable consumer. + +### Scaling and Rebalancing + +Scaling out: + +- Increase the number of receiver replicas. +- All replicas share the same durable consumer name; for pull-based consumption, each instance independently fetches messages. +- JetStream distributes messages across fetch calls; with more instances, aggregate throughput increases. + +Scaling in / failure: + +- When replicas are reduced or fail, the remaining instances continue to fetch messages. +- No explicit rebalancing logic is required in the application; JetStream manages which messages are available for pull. + +Tuning: + +- **Per-instance throughput** is primarily influenced by: + - `app.batch_size` + - `app.batch_timeout` + - the number of concurrent instances + +- **Backpressure** is provided through: + - JetStream `backoff` and `max_deliver` settings. + - Additional sleeps in the consumer when many consecutive errors occur. + - Readiness checks exposing DB/NATS health. + +### Failure Scenarios + +1. **DB outage**: + - Insert operations fail and are treated as transient. + - Messages are NAKed with delay and the error streak causes additional consumer sleep. + - `/readyz` returns 503, signalling this instance should be removed from traffic. + +2. **NATS outage**: + - Connection events are logged via `ProvideNATSConn` callbacks. + - The consumer will stop fetching; once NATS is back and reconnected, consumption resumes. + +3. **Single instance crash**: + - Other instances continue consuming from JetStream. + - No messages are lost; unacked messages remain pending and will be fetched by surviving instances. + + diff --git a/docs/observability.md b/docs/observability.md new file mode 100644 index 0000000..3170ac0 --- /dev/null +++ b/docs/observability.md @@ -0,0 +1,111 @@ +## Observability + +### Metrics + +The service exposes Prometheus metrics via the monitoring HTTP server (default `:2112`): + +- `caatsm_messages_total{stream,consumer,result}` + Total number of messages handled by the receiver, labelled by stream/consumer and result (`ok`, `fail`, `permanent_fail`, `retry`). + +- `caatsm_handle_latency_seconds{stream,consumer}` + End-to-end handling latency from NATS receive to handler completion. + +- `caatsm_retries_total{stream,consumer,reason}` + Number of retries (NAKs) issued by the consumer, labelled by reason (e.g. `processor_error`). + +- `caatsm_js_api_calls_total{operation}` + JetStream API calls performed by the service. + +- `caatsm_db_queries_total{operation,result}` + Database operations grouped by operation (`insert_one`, `insert_batch`, `insert_raw`) and result (`ok`, `error`). + +- `caatsm_db_query_latency_seconds{operation}` + DB operation latency. + +Additional OTEL metrics are emitted via the configured OTEL endpoint, 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` + +These metrics are intended to be scraped by Prometheus (either directly or via the OTEL collector) and visualised in Grafana dashboards. Recommended dashboard panels include: + +- Per-stream/consumer message rate and error rate. +- Handling latency P50/P95/P99. +- NATS consumer backlog and redelivery counts. +- DB query rates and latencies. + +#### 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: + +- **Message throughput by result** – derived from `caatsm_messages_total{result}`. +- **Per stream/consumer rates** – `caatsm_messages_total{stream,consumer}`. +- **End-to-end handle latency** – P50/P95/P99 from `caatsm_handle_latency_seconds_bucket`. +- **DB query rate and latency** – from `caatsm_db_queries_total` and `caatsm_db_query_latency_seconds_bucket`. +- **Retry and permanent failure rates** – from `caatsm_retries_total` and `caatsm_messages_total{result="permanent_fail"}`. +- **Publish failures** – from `caatsm_publish_failures_total`. + +### Health and Readiness + +The monitoring server exposes: + +- `/healthz` – basic liveness and dependency check. +- `/readyz` – readiness endpoint with the same logic as `/healthz`, intended for load balancers / orchestrators. + +Checks performed: + +- PostgreSQL: `pgxpool.Pool.Ping` with configurable timeout (`monitoring.health_timeout`). +- NATS: connection status must be `CONNECTED`. + +A non-200 response indicates the service is not healthy/ready and should be removed from traffic. + +### Tracing + +Tracing is configured via the `telemetry` section: + +- `telemetry.enabled` – enables OTEL exporters. +- `telemetry.endpoint` – OTLP HTTP endpoint (e.g. `localhost:4318`). +- `telemetry.insecure` – disables TLS for local/dev. + +Key spans: + +- `caatsm/nats` + - `Consumer.processMessage` +- `caatsm/app` + - `MessageProcessor.Handle` + - `Publisher.Publish` +- `caatsm/postgres` + - `Repository.InsertOne` + - `Repository.InsertBatch` + - `Repository.InsertRaw` + +Important attributes: + +- `nats.subject`, `nats.msg_id`, `nats.js.stream_seq`, `nats.js.consumer_seq` +- `telegram.message_id`, `telegram.category`, `telegram.status` +- `db.table`, `db.inserted` + +### Structured Logging Contract + +Logging is done with Zap. The `internal/observability/logging` package standardises fields via `MessageFields`: + +- `service` – logical component (`caatsm-consumer`, `caatsm-processor` etc.). +- `transport_msg_id` – NATS/envelope message ID (derived from `Nats-Msg-Id` or JetStream sequence). +- `telegram_message_id` – business telegram message ID from the payload. +- `category` – telegram category (ARR, DEP, FPL, etc.). +- `stream`, `consumer`, `subject` – JetStream context. +- `nats_sequence` – JetStream stream sequence, when available. +- `request_id`, `trace_id` – correlation identifiers. +- `error_type` – high-level classification: + - `business` – payload/validation/domain issues; not suitable for retry. + - `transient` – network/DB/NATS glitches that may succeed on retry. + - `fatal` – programming errors, schema mismatches, or configuration issues requiring operator attention. + +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 new file mode 100644 index 0000000..2f4115c --- /dev/null +++ b/docs/reliability.md @@ -0,0 +1,119 @@ +## Reliability and Fault Handling + +This document summarises how the service handles failures, provides resilience, and avoids data loss or duplication. + +### Dead-Letter Queue (DLQ) and Poison Messages + +Configuration is defined under `dlq`: + +```toml +[dlq] +enabled = true +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**. + +Behaviour (JetStream mode): + +1. The NATS consumer calls `processor.Handle`. +2. If an error is returned and `app.IsPermanent(err)` is true: + - The original NATS message is copied into a DLQ payload with metadata: + - `transport_msg_id` (NATS message ID) + - `subject`, `stream`, `consumer` + - `nats_sequence`, `deliveries` + - `error` (stringified cause) + - `received_at` (DLQ event time) + - `body` (raw message body) + - The payload is published to `dlq.subject` using JetStream. + - The original message is **ACKed**, so it will not be redelivered. + +The DLQ subject should be consumed by an offline repair/analysis tool or operational dashboard that can: + +- Inspect poison messages. +- Decide whether to fix and re-publish, or discard with justification. +- Track DLQ volume over time for alerting. + +### Transient Errors and Backoff + +Transient errors (not marked permanent) result in: + +- Negative acknowledgements with delay (`NakWithDelay`) according to `nats.consumer_rules.backoff`. +- A retry streak counter inside the consumer: + - Each transient error increases `consecutiveProcessErrors`. + - After 10 or more consecutive errors, the consumer applies an additional **sleep**: + - `backoff = min(consecutive_errors * 100ms, 5s)`. + - A warning log with the sleep duration and error count is emitted. + +This combination provides **backpressure** when downstream systems (especially the DB) are in trouble, slowing down consumption instead of aggressively retrying. + +### Persistence and Idempotency + +The primary persistence path is `Repository.InsertOne` into `aviation.telegrams`. To avoid applying the same business event multiple times, a **minimal idempotency check** is implemented: + +- If both `message_id` and `date_time` are non-empty: + - `InsertOne` first calls `messageExists(message_id, date_time)`. + - If a row already exists, the insert is **skipped** and an informational log is written. + - Otherwise, the insert proceeds. + +This makes repeated delivery of the same telegram (same `message_id`/`date_time`) safe from a business perspective, even if JetStream redelivers messages or upstream replays. + +For higher guarantees in production environments, you may: + +- Add a unique index on `(message_id, date_time)` at the DB level, and treat any conflict as a duplicate. +- Extend the idempotency key with additional fields (e.g. originator, category) if required by the business model. + +### DB Degradation and Backpressure + +Database write failures in `Repository.InsertOne` and related methods are treated as **transient** by default: + +- Errors propagate back to the NATS consumer. +- The consumer issues a NAK (with delay) and increases the transient error counter. +- When errors persist, the added sleep in the consumer reduces message throughput and gives the DB time to recover. + +DB health also feeds into readiness: + +- The monitoring server hits `pgxpool.Pool.Ping` on `/readyz` and `/healthz`. +- If the DB is not reachable, the endpoints return `503`, signalling to orchestrators that this instance should be drained from traffic. + +Together, this yields: + +- **Backpressure** via reduced consumption rate and NATS-level backoff. +- **Degradation signalling** via health probes for external systems to act upon. + +### Retry and Max Deliver + +JetStream consumer configuration (via `nats.consumer_rules`) controls: + +- `max_deliver` – maximum number of redeliveries before JetStream gives up. +- `ack_wait` – how long JetStream waits for an ACK before considering the message pending. +- `backoff` – per-attempt delays for `NakWithDelay`. + +Recommended pattern: + +- Keep `max_deliver` modest (e.g. 5). +- Use a backoff array such as `[5s, 30s, 2m]`. +- Treat messages that still fail after `max_deliver` as candidates for DLQ, via the permanent error/poison message path where applicable. + +### Alerts and Dashboards + +Prometheus alert suggestions: + +- High failure rate: + - `rate(caatsm_messages_total{result!="ok"}[5m])` above a small threshold. + - `rate(caatsm_retries_total[5m])` above a threshold. + +- DLQ growth: + - Alerts on DLQ stream message count, using NATS/JetStream exporter metrics. + +- Readiness / health: + - Alert when `/readyz` fails or when DB/NATS checks start failing consistently. + +Dashboards should combine: + +- Message rates, error rates, and DLQ rates. +- NATS consumer statistics (pending, redelivered, ack_pending). +- DB health indicators (latency, error counts, connection usage). + + diff --git a/internal/app/processor.go b/internal/app/processor.go index 1f77396..ac9b04c 100644 --- a/internal/app/processor.go +++ b/internal/app/processor.go @@ -4,6 +4,7 @@ import ( "caatsm/internal/adapter" "caatsm/internal/adapter/parser" "caatsm/internal/domain" + obslogging "caatsm/internal/observability/logging" obsmetrics "caatsm/internal/observability/metrics" "context" "fmt" @@ -110,6 +111,20 @@ func (p *MessageProcessor) Handle(ctx context.Context, raw []byte, msgID string) } } + // Enrich span with parsed telegram information as soon as we have it. + span.SetAttributes( + attribute.String("telegram.message_id", parsed.MessageID), + attribute.String("telegram.category", parsed.Category), + attribute.String("telegram.status", string(parsed.Status)), + ) + + msgLogger := obslogging.WithMessageContext(p.logger, obslogging.MessageFields{ + Service: "caatsm-processor", + TransportMsgID: msgID, + BusinessMsgID: parsed.MessageID, + Category: parsed.Category, + }) + if parseErr != nil || !parsed.Parsed { if parsed.ErrorReason == "" && parseErr != nil { parsed.ErrorReason = parseErr.Error() @@ -117,12 +132,11 @@ func (p *MessageProcessor) Handle(ctx context.Context, raw []byte, msgID string) span.RecordError(parseErr) span.SetStatus(codes.Error, parseErr.Error()) p.persistRaw(ctx, parsed) - p.logger.Warn("Message failed to parse", - zap.String("msg_id", msgID), - zap.String("status", string(parsed.Status)), - zap.String("content_preview", truncateContent(parsed.Content, 256)), - zap.Error(parseErr), - ) + msgLogger.With(zap.String("status", string(parsed.Status))). + Warn("Message failed to parse", + zap.String("content_preview", truncateContent(parsed.Content, 256)), + zap.Error(parseErr), + ) latency := parsed.ParsedAt.Sub(receivedAt) parseLatencyHistogram.Record(ctx, float64(latency.Milliseconds()), metric.WithAttributes( @@ -143,10 +157,7 @@ func (p *MessageProcessor) Handle(ctx context.Context, raw []byte, msgID string) attribute.String("telegram.category", parsed.Category), ) - p.logger.Info("Message parsed successfully", - zap.String("msg_id", msgID), - zap.String("message_id", parsed.MessageID), - zap.String("category", parsed.Category), + msgLogger.Info("Message parsed successfully", zap.Time("received_at", parsed.ReceivedAt), zap.Time("parsed_at", parsed.ParsedAt), ) @@ -172,10 +183,10 @@ func (p *MessageProcessor) Handle(ctx context.Context, raw []byte, msgID string) _, pubSpan := tracer.Start(ctx, "Publisher.Publish") if err := p.publisher.Publish(parsed); err != nil { // Log error but don't fail the entire operation - p.logger.Error("Failed to publish message", - zap.String("msg_id", msgID), - zap.Error(err), - ) + msgLogger.With(zap.String("status", string(parsed.Status))). + Error("Failed to publish message", + zap.Error(err), + ) pubSpan.RecordError(err) pubSpan.SetStatus(codes.Error, err.Error()) parsed.Status = domain.MessageStatusPublishFail diff --git a/internal/infra/config/config.go b/internal/infra/config/config.go index 951921c..e1d762c 100644 --- a/internal/infra/config/config.go +++ b/internal/infra/config/config.go @@ -21,6 +21,7 @@ type Config struct { Publisher PublisherConfig `koanf:"publisher"` Telemetry TelemetryConfig `koanf:"telemetry"` Monitoring MonitoringConfig `koanf:"monitoring"` + DLQ DLQConfig `koanf:"dlq"` // Legacy fields for backward compatibility during migration Subscription SubscriptionConfig `koanf:"subscription"` Timeouts TimeoutsConfig `koanf:"timeouts"` @@ -93,6 +94,14 @@ type TelemetryConfig struct { Insecure bool `koanf:"insecure"` } +// DLQConfig defines the dead-letter queue routing for poison/permanent messages. +// If Enabled is true and Subject is non-empty, permanent failures will be published +// to the configured subject for offline processing. +type DLQConfig struct { + Enabled bool `koanf:"enabled"` + Subject string `koanf:"subject"` +} + // MonitoringConfig controls the lightweight HTTP server that exposes health and metrics endpoints. type MonitoringConfig struct { Disabled bool `koanf:"disabled"` diff --git a/internal/infra/monitoring/server.go b/internal/infra/monitoring/server.go index 223425c..cd054c4 100644 --- a/internal/infra/monitoring/server.go +++ b/internal/infra/monitoring/server.go @@ -47,7 +47,12 @@ func ProvideServer( routes := 0 if cfg.Monitoring.EnableHealth { + // Liveness: basic process check. For now this reuses the same implementation + // as readiness but can diverge in the future if we need a cheaper liveness probe. mux.HandleFunc("/healthz", server.handleHealth) + // Readiness: alias to the same implementation so consumers can adopt /readyz + // without breaking existing /healthz users. + mux.HandleFunc("/readyz", server.handleHealth) routes++ } if cfg.Monitoring.EnableMetrics { diff --git a/internal/infra/nats/consumer.go b/internal/infra/nats/consumer.go index 9a2cfd0..8a02be8 100644 --- a/internal/infra/nats/consumer.go +++ b/internal/infra/nats/consumer.go @@ -3,7 +3,10 @@ package nats import ( "caatsm/internal/app" "caatsm/internal/infra/config" + obslogging "caatsm/internal/observability/logging" + obsmetrics "caatsm/internal/observability/metrics" "context" + "encoding/json" "errors" "fmt" "strings" @@ -29,6 +32,7 @@ type Consumer struct { consumerName string mode string streamName string + dlqSubject string ackWait time.Duration batchSize int batchTimeout time.Duration @@ -38,6 +42,9 @@ type Consumer struct { redelivered metric.Int64Histogram pending metric.Int64Histogram delivered metric.Int64Histogram + + // simple backpressure / degradation state + consecutiveProcessErrors int } // ProvideConsumer creates a NATS consumer @@ -65,6 +72,8 @@ func ProvideConsumer( streamName = "TELEGRAM" } + dlqSubject := strings.TrimSpace(cfg.DLQ.Subject) + ackWait := cfg.NATS.ConsumerRules.AckWait if ackWait == 0 { ackWait = cfg.Timeouts.AckWait @@ -98,6 +107,7 @@ func ProvideConsumer( consumerName: consumerName, mode: mode, streamName: streamName, + dlqSubject: dlqSubject, ackWait: ackWait, batchSize: batchSize, batchTimeout: batchTimeout, @@ -227,31 +237,74 @@ func (c *Consumer) startJetStream(ctx context.Context) error { // Process each message // TODO: consider buffering messages to take advantage of Repository.InsertBatch for higher throughput. for _, msg := range msgs { + start := time.Now() + if err := c.processMessage(ctx, msg); err != nil { isPermanent := app.IsPermanent(err) + elapsed := time.Since(start) + c.logger.Error("Failed to process message", zap.String("subject", msg.Subject), zap.Error(err), zap.Bool("permanent", isPermanent), ) + result := obsmetrics.ResultFail if isPermanent { + result = obsmetrics.ResultPermanentFail + } + obsmetrics.RecordMessageHandled(c.streamName, c.consumerName, result, elapsed) + + if isPermanent { + c.consecutiveProcessErrors = 0 + // Poison/permanent message: route to DLQ if configured, then ACK + if err := c.routeToDLQ(ctx, msg, err); err != nil { + c.logger.Error("Failed to route permanent-error message to DLQ", zap.Error(err)) + } if ackErr := msg.Ack(); ackErr != nil { c.logger.Error("Failed to ACK permanent-error message", zap.Error(ackErr)) } continue } + // Transient error: increment error streak and apply simple backpressure if needed. + if c.consecutiveProcessErrors < 0 { + c.consecutiveProcessErrors = 0 + } + c.consecutiveProcessErrors++ + if c.consecutiveProcessErrors >= 10 { + // Apply a brief sleep to slow down consumption when the system + // is failing many messages in a row (e.g. DB unavailable). + backoff := time.Duration(c.consecutiveProcessErrors) * 100 * time.Millisecond + if backoff > 5*time.Second { + backoff = 5 * time.Second + } + c.logger.Warn("Applying backpressure due to consecutive processing errors", + zap.Int("consecutive_errors", c.consecutiveProcessErrors), + zap.Duration("sleep", backoff), + ) + time.Sleep(backoff) + } + // Transient error: request redelivery with optional delay + obsmetrics.RecordRetry(c.streamName, c.consumerName, obsmetrics.RetryReasonProcessorError) if nakErr := c.nakWithStrategy(msg); nakErr != nil { c.logger.Error("Failed to NAK message", zap.Error(nakErr)) } continue } + // Successful processing resets the error streak. + if c.consecutiveProcessErrors > 0 { + c.consecutiveProcessErrors = 0 + } + // ACK the message if ackErr := msg.Ack(); ackErr != nil { c.logger.Error("Failed to ACK message", zap.Error(ackErr)) + } else { + elapsed := time.Since(start) + obsmetrics.RecordMessageHandled(c.streamName, c.consumerName, "ok", elapsed) } } } @@ -359,16 +412,16 @@ func (c *Consumer) initMetrics() { meter := otel.Meter("caatsm/nats") c.meter = meter - if hist, err := meter.Int64Histogram("nats.consumer.ack_pending"); err == nil { + if hist, err := meter.Int64Histogram("caatsm_nats_consumer_ack_pending"); err == nil { c.ackPending = hist } - if hist, err := meter.Int64Histogram("nats.consumer.redelivered"); err == nil { + if hist, err := meter.Int64Histogram("caatsm_nats_consumer_redelivered"); err == nil { c.redelivered = hist } - if hist, err := meter.Int64Histogram("nats.consumer.pending"); err == nil { + if hist, err := meter.Int64Histogram("caatsm_nats_consumer_pending"); err == nil { c.pending = hist } - if hist, err := meter.Int64Histogram("nats.consumer.delivered"); err == nil { + if hist, err := meter.Int64Histogram("caatsm_nats_consumer_delivered"); err == nil { c.delivered = hist } } @@ -391,6 +444,52 @@ func (c *Consumer) recordConsumerMetrics(ctx context.Context, info *nats.Consume } } +// routeToDLQ publishes a copy of the failed message to the configured DLQ subject, +// including useful metadata for offline analysis. If DLQ is not configured or the +// consumer is not running in JetStream mode, this is a no-op. +func (c *Consumer) routeToDLQ(ctx context.Context, msg *nats.Msg, cause error) error { + if c == nil || c.js == nil { + return nil + } + if c.mode != "jetstream" { + return nil + } + if strings.TrimSpace(c.dlqSubject) == "" { + return nil + } + + meta, _ := msg.Metadata() + jsSeq := uint64(0) + deliveries := uint64(0) + if meta != nil { + jsSeq = meta.Sequence.Stream + deliveries = meta.NumDelivered + } + + payload := map[string]interface{}{ + "transport_msg_id": msg.Header.Get("Nats-Msg-Id"), + "subject": msg.Subject, + "stream": c.streamName, + "consumer": c.consumerName, + "nats_sequence": jsSeq, + "deliveries": deliveries, + "error": fmt.Sprint(cause), + "received_at": time.Now().UTC(), + "body": string(msg.Data), + } + + data, err := json.Marshal(payload) + if err != nil { + return fmt.Errorf("marshal dlq payload: %w", err) + } + + if _, err := c.js.Publish(c.dlqSubject, data); err != nil { + return fmt.Errorf("publish to dlq subject %s: %w", c.dlqSubject, err) + } + + return nil +} + func (c *Consumer) nakWithStrategy(msg *nats.Msg) error { backoff := c.cfg.NATS.ConsumerRules.Backoff if len(backoff) == 0 { @@ -439,10 +538,28 @@ func (c *Consumer) processMessage(ctx context.Context, msg *nats.Msg) error { ) } - c.logger.Debug("Processing message", - zap.String("subject", msg.Subject), - zap.String("msg_id", msgID), + // Attach structured logging context including stream/consumer and NATS metadata. + jsSeq := uint64(0) + if meta, metaErr := msg.Metadata(); metaErr == nil { + jsSeq = meta.Sequence.Stream + span.SetAttributes( + attribute.Int64("nats.js.stream_seq", int64(meta.Sequence.Stream)), + attribute.Int64("nats.js.consumer_seq", int64(meta.Sequence.Consumer)), + ) + } + + msgLogger := obslogging.WithMessageContext(c.logger, obslogging.MessageFields{ + Service: "caatsm-consumer", + TransportMsgID: msgID, + Stream: c.streamName, + Consumer: c.consumerName, + Subject: msg.Subject, + JSSequence: jsSeq, + }) + + msgLogger.Debug("Processing message", zap.Int("data_size", len(msg.Data)), + zap.String("msg_id_source", source), ) // Call processor diff --git a/internal/infra/postgres/repository.go b/internal/infra/postgres/repository.go index 55563a3..ae77522 100644 --- a/internal/infra/postgres/repository.go +++ b/internal/infra/postgres/repository.go @@ -4,6 +4,7 @@ import ( "caatsm/internal/adapter" "caatsm/internal/adapter/mapper" "caatsm/internal/domain" + obsmetrics "caatsm/internal/observability/metrics" "context" "encoding/json" "fmt" @@ -40,6 +41,25 @@ func (r *Repository) InsertOne(ctx context.Context, msg *domain.ParsedMessage) e defer span.End() span.SetAttributes(attribute.String("db.table", "aviation.telegrams")) + // Optional idempotency check based on business message identity. If we have a + // non-empty message ID and date/time, we can cheaply skip duplicates here to + // avoid applying the same business event multiple times. + if msg != nil && msg.MessageID != "" && msg.DateTime != "" { + exists, err := r.messageExists(ctx, msg.MessageID, msg.DateTime) + if err != nil { + span.RecordError(err) + span.SetStatus(codes.Error, err.Error()) + return fmt.Errorf("failed to check existing message: %w", err) + } + if exists { + r.logger.Info("Duplicate message detected by message_id/date_time; skipping insert", + zap.String("message_id", msg.MessageID), + zap.String("date_time", msg.DateTime), + ) + return nil + } + } + row, err := r.mapper.ToDBRow(msg) if err != nil { span.RecordError(err) @@ -59,16 +79,24 @@ func (r *Repository) InsertOne(ctx context.Context, msg *domain.ParsedMessage) e ON CONFLICT (uuid, received_at) DO NOTHING ` + start := time.Now() tag, err := r.pool.Exec(ctx, query, row[0], row[1], row[2], row[3], row[4], row[5], row[6], row[7], row[8], row[9], row[10], row[11], row[12], row[13], row[14], row[15], row[16], ) + elapsed := time.Since(start) + + result := obsmetrics.DBResultOK if err != nil { + result = obsmetrics.DBResultError span.RecordError(err) span.SetStatus(codes.Error, err.Error()) + obsmetrics.RecordDBQuery("insert_one", result, elapsed) return fmt.Errorf("failed to insert message: %w", err) } + obsmetrics.RecordDBQuery("insert_one", result, elapsed) + if tag.RowsAffected() == 0 { r.logger.Info("Duplicate message skipped", zap.String("uuid", msg.Uuid), @@ -106,6 +134,7 @@ func (r *Repository) InsertBatch(ctx context.Context, msgs []*domain.ParsedMessa } // Use CopyFrom for efficient batch insert + start := time.Now() copyCount, err := r.pool.CopyFrom( ctx, pgx.Identifier{"aviation", "telegrams"}, @@ -117,12 +146,19 @@ func (r *Repository) InsertBatch(ctx context.Context, msgs []*domain.ParsedMessa }, pgx.CopyFromRows(rows), ) + elapsed := time.Since(start) + + result := obsmetrics.DBResultOK if err != nil { + result = obsmetrics.DBResultError span.RecordError(err) span.SetStatus(codes.Error, err.Error()) + obsmetrics.RecordDBQuery("insert_batch", result, elapsed) return fmt.Errorf("failed to batch insert messages: %w", err) } + obsmetrics.RecordDBQuery("insert_batch", result, elapsed) + span.SetAttributes(attribute.Int64("db.inserted", copyCount)) r.logger.Info("Batch inserted messages", zap.Int("count", int(copyCount)), @@ -174,6 +210,7 @@ func (r *Repository) InsertRaw(ctx context.Context, msg *domain.ParsedMessage) e metadata = EXCLUDED.metadata ` + start := time.Now() _, err = r.pool.Exec(ctx, query, msg.Uuid, string(msg.Status), @@ -182,12 +219,19 @@ func (r *Repository) InsertRaw(ctx context.Context, msg *domain.ParsedMessage) e msg.ReceivedAt, metadataJSON, ) + elapsed := time.Since(start) + + result := obsmetrics.DBResultOK if err != nil { + result = obsmetrics.DBResultError span.RecordError(err) span.SetStatus(codes.Error, err.Error()) + obsmetrics.RecordDBQuery("insert_raw", result, elapsed) return fmt.Errorf("failed to insert raw telegram: %w", err) } + obsmetrics.RecordDBQuery("insert_raw", result, elapsed) + span.SetAttributes(attribute.String("telegram.uuid", msg.Uuid), attribute.String("telegram.status", string(msg.Status))) r.logger.Debug("Persisted raw telegram", zap.String("uuid", msg.Uuid), @@ -196,3 +240,29 @@ func (r *Repository) InsertRaw(ctx context.Context, msg *domain.ParsedMessage) e return nil } + +// messageExists performs a lightweight existence check for a business message, +// using (message_id, date_time) as the idempotency key. This avoids requiring a +// unique constraint at the database level while still preventing duplicate effects. +func (r *Repository) messageExists(ctx context.Context, messageID, dateTime string) (bool, error) { + if messageID == "" || dateTime == "" { + return false, nil + } + + const query = ` + SELECT 1 + FROM aviation.telegrams + WHERE message_id = $1 AND date_time = $2 + LIMIT 1 + ` + + var one int + if err := r.pool.QueryRow(ctx, query, messageID, dateTime).Scan(&one); err != nil { + if err == pgx.ErrNoRows { + return false, nil + } + return false, err + } + + return true, nil +} diff --git a/internal/observability/logging/logger.go b/internal/observability/logging/logger.go new file mode 100644 index 0000000..8ed0eb3 --- /dev/null +++ b/internal/observability/logging/logger.go @@ -0,0 +1,102 @@ +package logging + +import "go.uber.org/zap" + +// ErrorType represents a coarse-grained categorisation of errors for logging and alerting. +// Typical values: +// - business: validation failures, domain rule violations, payload issues. +// - transient: network / DB / NATS glitches that may succeed on retry. +// - fatal: programming bugs, schema mismatches, or conditions that require operator action. +type ErrorType string + +const ( + ErrorTypeBusiness ErrorType = "business" + ErrorTypeTransient ErrorType = "transient" + ErrorTypeFatal ErrorType = "fatal" +) + +// Canonical logging field names for structured logs produced by the CAATSM +// receiver. Using constants avoids scattering magic strings and keeps log +// analysis queries stable over time. +const ( + FieldService = "service" + FieldTransportMsgID = "transport_msg_id" + FieldTelegramMsgID = "telegram_message_id" + FieldCategory = "category" + FieldStream = "stream" + FieldConsumer = "consumer" + FieldSubject = "subject" + FieldNATSSequence = "nats_sequence" + FieldRequestID = "request_id" + FieldTraceID = "trace_id" + FieldErrorType = "error_type" +) + +// MessageFields captures the common structured logging contract for message-processing logs. +// All fields are optional; empty values will simply be skipped. +type MessageFields struct { + // Identifiers + Service string + TransportMsgID string + BusinessMsgID string + Category string + + // NATS / JetStream context + Stream string + Consumer string + Subject string + JSSequence uint64 + + // Correlation / tracing + RequestID string + TraceID string + + // Error classification + ErrorType ErrorType +} + +// WithMessageContext returns a logger pre-populated with the structured fields defined in MessageFields. +// This is the primary entry point for enforcing the logging contract in the codebase. +func WithMessageContext(logger *zap.Logger, mf MessageFields) *zap.Logger { + if logger == nil { + return zap.NewNop() + } + + fields := make([]zap.Field, 0, 12) + + if mf.Service != "" { + fields = append(fields, zap.String(FieldService, mf.Service)) + } + if mf.TransportMsgID != "" { + fields = append(fields, zap.String(FieldTransportMsgID, mf.TransportMsgID)) + } + if mf.BusinessMsgID != "" { + fields = append(fields, zap.String(FieldTelegramMsgID, mf.BusinessMsgID)) + } + if mf.Category != "" { + fields = append(fields, zap.String(FieldCategory, mf.Category)) + } + if mf.Stream != "" { + fields = append(fields, zap.String(FieldStream, mf.Stream)) + } + if mf.Consumer != "" { + fields = append(fields, zap.String(FieldConsumer, mf.Consumer)) + } + if mf.Subject != "" { + fields = append(fields, zap.String(FieldSubject, mf.Subject)) + } + if mf.JSSequence > 0 { + fields = append(fields, zap.Uint64(FieldNATSSequence, mf.JSSequence)) + } + if mf.RequestID != "" { + fields = append(fields, zap.String(FieldRequestID, mf.RequestID)) + } + if mf.TraceID != "" { + fields = append(fields, zap.String(FieldTraceID, mf.TraceID)) + } + if mf.ErrorType != "" { + fields = append(fields, zap.String(FieldErrorType, string(mf.ErrorType))) + } + + return logger.With(fields...) +} diff --git a/internal/observability/metrics/metrics.go b/internal/observability/metrics/metrics.go index f8d1341..2f3b17b 100644 --- a/internal/observability/metrics/metrics.go +++ b/internal/observability/metrics/metrics.go @@ -11,31 +11,130 @@ import ( "github.com/prometheus/client_golang/prometheus/promhttp" ) +// Metric and label key/value contracts for the CAATSM receiver. Centralising these +// names avoids scattering magic strings across the codebase and keeps PromQL and +// documentation aligned with the implementation. +const ( + // Metric names. + MetricProcessedTotal = "caatsm_processed_total" + MetricFailuresTotal = "caatsm_failures_total" + MetricParseLatencySeconds = "caatsm_parse_latency_seconds" + MetricMessagesTotal = "caatsm_messages_total" + MetricHandleLatencySeconds = "caatsm_handle_latency_seconds" + MetricRetriesTotal = "caatsm_retries_total" + MetricJSAPICallsTotal = "caatsm_js_api_calls_total" + MetricDBQueriesTotal = "caatsm_db_queries_total" + MetricDBQueryLatencySeconds = "caatsm_db_query_latency_seconds" + + // Common label keys. + LabelStatus = "status" + LabelCategory = "category" + LabelStage = "stage" + LabelStream = "stream" + LabelConsumer = "consumer" + LabelResult = "result" + LabelReason = "reason" + LabelOperation = "operation" + + // Standard result label values for caatsm_messages_total. + ResultOK = "ok" + ResultFail = "fail" + ResultPermanentFail = "permanent_fail" + + // Standard result values for DB operations. + DBResultOK = "ok" + DBResultError = "error" + + // Standard retry reasons. + RetryReasonProcessorError = "processor_error" +) + var ( - once sync.Once - registry *prometheus.Registry + once sync.Once + + registry *prometheus.Registry + + // Legacy metrics (kept for backward compatibility). processedCounter *prometheus.CounterVec failureCounter *prometheus.CounterVec parseLatency *prometheus.HistogramVec + + // Message handling metrics (per stream / consumer). + messagesTotal *prometheus.CounterVec + handleLatency *prometheus.HistogramVec + retriesTotal *prometheus.CounterVec + jsAPICallsTotal *prometheus.CounterVec + + // Database metrics. + dbQueriesTotal *prometheus.CounterVec + dbQueryLatency *prometheus.HistogramVec ) func initCollectors() { registry = prometheus.NewRegistry() + + // Legacy metrics. processedCounter = prometheus.NewCounterVec(prometheus.CounterOpts{ - Name: "caatsm_processed_total", + Name: MetricProcessedTotal, Help: "Count of telegrams processed by status and category.", - }, []string{"status", "category"}) + }, []string{LabelStatus, LabelCategory}) + failureCounter = prometheus.NewCounterVec(prometheus.CounterOpts{ - Name: "caatsm_failures_total", + Name: MetricFailuresTotal, Help: "Count of processor failures by stage (parser, repository, publisher).", - }, []string{"stage"}) + }, []string{LabelStage}) + parseLatency = prometheus.NewHistogramVec(prometheus.HistogramOpts{ - Name: "caatsm_parse_latency_seconds", + Name: MetricParseLatencySeconds, Help: "Latency between reception and parse completion.", Buckets: prometheus.DefBuckets, - }, []string{"status", "category"}) + }, []string{LabelStatus, LabelCategory}) - registry.MustRegister(processedCounter, failureCounter, parseLatency) + // New message handling metrics. + messagesTotal = prometheus.NewCounterVec(prometheus.CounterOpts{ + Name: MetricMessagesTotal, + Help: "Total number of messages handled by the receiver, labelled by stream, consumer and result.", + }, []string{LabelStream, LabelConsumer, LabelResult}) + + handleLatency = prometheus.NewHistogramVec(prometheus.HistogramOpts{ + Name: MetricHandleLatencySeconds, + Help: "Latency of end-to-end message handling in seconds, from NATS receive to handler completion.", + Buckets: prometheus.DefBuckets, + }, []string{LabelStream, LabelConsumer}) + + retriesTotal = prometheus.NewCounterVec(prometheus.CounterOpts{ + Name: MetricRetriesTotal, + Help: "Total number of message retries (negative acknowledgements), labelled by stream, consumer and reason.", + }, []string{LabelStream, LabelConsumer, LabelReason}) + + jsAPICallsTotal = prometheus.NewCounterVec(prometheus.CounterOpts{ + Name: MetricJSAPICallsTotal, + Help: "Count of JetStream API calls made by the receiver.", + }, []string{LabelOperation}) + + // Database metrics. + dbQueriesTotal = prometheus.NewCounterVec(prometheus.CounterOpts{ + Name: MetricDBQueriesTotal, + Help: "Total number of database operations, labelled by operation and result.", + }, []string{LabelOperation, LabelResult}) + + dbQueryLatency = prometheus.NewHistogramVec(prometheus.HistogramOpts{ + Name: MetricDBQueryLatencySeconds, + Help: "Latency of database operations in seconds, labelled by operation.", + Buckets: prometheus.DefBuckets, + }, []string{LabelOperation}) + + registry.MustRegister( + processedCounter, + failureCounter, + parseLatency, + messagesTotal, + handleLatency, + retriesTotal, + jsAPICallsTotal, + dbQueriesTotal, + dbQueryLatency, + ) } func ensureCollectors() { @@ -62,6 +161,42 @@ func RecordFailure(stage string) { failureCounter.WithLabelValues(labelValue(stage)).Inc() } +// RecordMessageHandled records end-to-end message handling metrics (per stream / consumer). +// Result is expected to be values such as "ok", "fail", or "retry". +func RecordMessageHandled(stream, consumer, result string, elapsed time.Duration) { + ensureCollectors() + if elapsed < 0 { + elapsed = 0 + } + messagesTotal.WithLabelValues(labelValue(stream), labelValue(consumer), labelValue(result)).Inc() + handleLatency.WithLabelValues(labelValue(stream), labelValue(consumer)).Observe(elapsed.Seconds()) +} + +// RecordRetry increments the retry counter for a message that is being negatively acknowledged. +// Reason can capture the high-level cause, e.g. "processor_error" or "nats_timeout". +func RecordRetry(stream, consumer, reason string) { + ensureCollectors() + retriesTotal.WithLabelValues(labelValue(stream), labelValue(consumer), labelValue(reason)).Inc() +} + +// RecordDBQuery records metrics for a single database operation. +// Operation examples: "insert_one", "insert_batch", "insert_raw". +// Result is usually "ok" or "error". +func RecordDBQuery(operation, result string, elapsed time.Duration) { + ensureCollectors() + if elapsed < 0 { + elapsed = 0 + } + dbQueriesTotal.WithLabelValues(labelValue(operation), labelValue(result)).Inc() + dbQueryLatency.WithLabelValues(labelValue(operation)).Observe(elapsed.Seconds()) +} + +// RecordJSAPICall increments the JetStream API call counter for the given operation. +func RecordJSAPICall(operation string) { + ensureCollectors() + jsAPICallsTotal.WithLabelValues(labelValue(operation)).Inc() +} + func labelValue(value string) string { value = strings.TrimSpace(value) if value == "" {