From f01dd3bf1fe55acebbe19cecffae8e1de36a5b1f Mon Sep 17 00:00:00 2001 From: windyboy Date: Wed, 19 Nov 2025 14:53:58 +0800 Subject: [PATCH] =?UTF-8?q?=E2=9C=A8=20Refactor=20telemetry=20and=20metric?= =?UTF-8?q?s=20handling=20to=20enhance=20observability.=20Remove=20TLS=20c?= =?UTF-8?q?onfiguration=20from=20telemetry=20initialization,=20replacing?= =?UTF-8?q?=20it=20with=20an=20insecure=20option=20for=20HTTP.=20Update=20?= =?UTF-8?q?OpenTelemetry=20endpoint=20handling=20to=20normalize=20URLs.=20?= =?UTF-8?q?Introduce=20new=20metrics=20for=20publish=20failures=20categori?= =?UTF-8?q?zed=20by=20message=20type.=20Enhance=20Grafana=20dashboards=20w?= =?UTF-8?q?ith=20detailed=20descriptions=20and=20additional=20metrics=20fo?= =?UTF-8?q?r=20better=20monitoring=20insights.=20Update=20documentation=20?= =?UTF-8?q?to=20reflect=20changes=20in=20metrics=20and=20observability=20f?= =?UTF-8?q?eatures.?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- cmd/main/main.go | 19 +- .../caatsm-overview.json | 128 +++++++++++-- .../otel-collector.json | 180 +----------------- configs/otel-collector.dev.yaml | 6 - configs/prometheus.dev.yml | 1 + .../prometheus/targets/caatsm-receiver.json | 2 +- docs/nats.md | 2 +- docs/observability.md | 47 ++++- internal/infra/metrics/metrics.go | 15 ++ internal/infra/nats/jetstream.go | 9 +- internal/infra/telemetry/otel.go | 44 +++-- internal/infra/telemetry/telemetry.go | 5 +- pkg/di/wire_gen.go | 4 +- 13 files changed, 224 insertions(+), 238 deletions(-) diff --git a/cmd/main/main.go b/cmd/main/main.go index c1035ca..bd51c7b 100644 --- a/cmd/main/main.go +++ b/cmd/main/main.go @@ -5,7 +5,6 @@ import ( "caatsm/internal/infra/config" "caatsm/pkg/di" "context" - "crypto/tls" "errors" "fmt" "os" @@ -345,29 +344,19 @@ func initTelemetry(ctx context.Context, cfg *config.Config) (func(context.Contex return nil, fmt.Errorf("telemetry endpoint is required when telemetry.enabled=true") } - // Configure TLS settings - var tlsConfig *tls.Config - if cfg.Telemetry.Insecure { - tlsConfig = &tls.Config{ - InsecureSkipVerify: true, - MinVersion: tls.VersionTLS12, // TLS 1.2 minimum, TLS 1.3 preferred - } - } - traceOpts := []otlptracehttp.Option{ otlptracehttp.WithEndpoint(cfg.Telemetry.Endpoint), otlptracehttp.WithURLPath("/v1/traces"), } - if tlsConfig != nil { - traceOpts = append(traceOpts, otlptracehttp.WithTLSClientConfig(tlsConfig)) - } metricOpts := []otlpmetrichttp.Option{ otlpmetrichttp.WithEndpoint(cfg.Telemetry.Endpoint), otlpmetrichttp.WithURLPath("/v1/metrics"), } - if tlsConfig != nil { - metricOpts = append(metricOpts, otlpmetrichttp.WithTLSClientConfig(tlsConfig)) + + if cfg.Telemetry.Insecure { + traceOpts = append(traceOpts, otlptracehttp.WithInsecure()) + metricOpts = append(metricOpts, otlpmetrichttp.WithInsecure()) } traceExporter, err := otlptracehttp.New(ctx, traceOpts...) diff --git a/configs/grafana-dashboards.dev/caatsm-overview.json b/configs/grafana-dashboards.dev/caatsm-overview.json index 1ee6773..66be1e8 100644 --- a/configs/grafana-dashboards.dev/caatsm-overview.json +++ b/configs/grafana-dashboards.dev/caatsm-overview.json @@ -103,6 +103,7 @@ "id": 3, "type": "timeseries", "title": "End-to-end handle latency (P50/P95/P99)", + "description": "Latency breakdown by stream and consumer for detailed analysis.", "datasource": { "type": "prometheus", "uid": "prometheus-dev" @@ -123,18 +124,18 @@ "targets": [ { "refId": "P50", - "expr": "histogram_quantile(0.50, sum by (le) (rate(caatsm_handle_latency_seconds_bucket[5m])))", - "legendFormat": "P50" + "expr": "histogram_quantile(0.50, sum by (le, stream, consumer) (rate(caatsm_handle_latency_seconds_bucket[5m])))", + "legendFormat": "P50 ({{stream}}/{{consumer}})" }, { "refId": "P95", - "expr": "histogram_quantile(0.95, sum by (le) (rate(caatsm_handle_latency_seconds_bucket[5m])))", - "legendFormat": "P95" + "expr": "histogram_quantile(0.95, sum by (le, stream, consumer) (rate(caatsm_handle_latency_seconds_bucket[5m])))", + "legendFormat": "P95 ({{stream}}/{{consumer}})" }, { "refId": "P99", - "expr": "histogram_quantile(0.99, sum by (le) (rate(caatsm_handle_latency_seconds_bucket[5m])))", - "legendFormat": "P99" + "expr": "histogram_quantile(0.99, sum by (le, stream, consumer) (rate(caatsm_handle_latency_seconds_bucket[5m])))", + "legendFormat": "P99 ({{stream}}/{{consumer}})" } ] }, @@ -200,6 +201,7 @@ "id": 6, "type": "timeseries", "title": "Retries and permanent failures", + "description": "Retries broken down by stream, consumer, and reason. Permanent failures by stream/consumer.", "datasource": { "type": "prometheus", "uid": "prometheus-dev" @@ -220,13 +222,13 @@ "targets": [ { "refId": "retries", - "expr": "sum(rate(caatsm_retries_total[5m]))", - "legendFormat": "retries" + "expr": "sum by (stream, consumer, reason) (rate(caatsm_retries_total[5m]))", + "legendFormat": "retries ({{stream}}/{{consumer}}/{{reason}})" }, { "refId": "permanent_fail", - "expr": "sum(rate(caatsm_messages_total{result=\"permanent_fail\"}[5m]))", - "legendFormat": "permanent_fail" + "expr": "sum by (stream, consumer) (rate(caatsm_messages_total{result=\"permanent_fail\"}[5m]))", + "legendFormat": "permanent_fail ({{stream}}/{{consumer}})" } ] }, @@ -234,6 +236,7 @@ "id": 7, "type": "timeseries", "title": "Publish failures by category", + "description": "Prometheus-sourced publish failures. Panel 11 shows OTEL-sourced failures.", "datasource": { "type": "prometheus", "uid": "prometheus-dev" @@ -254,8 +257,8 @@ "targets": [ { "refId": "A", - "expr": "sum by (message_category) (rate(caatsm_publish_failures_total[5m]))", - "legendFormat": "{{message_category}}" + "expr": "sum by (category) (rate(caatsm_publish_failures_total[5m]))", + "legendFormat": "{{category}}" } ] }, @@ -287,6 +290,107 @@ "legendFormat": "{{stream}} / {{consumer}}" } ] + }, + { + "id": 9, + "type": "timeseries", + "title": "Messages Processed (OTEL - by status/category)", + "description": "OTEL metrics exported via collector. Labels: message_status, message_category (OTEL attributes converted from message.status, message.category).", + "datasource": { + "type": "prometheus", + "uid": "prometheus-dev" + }, + "gridPos": { "h": 7, "w": 12, "x": 12, "y": 26 }, + "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_status, message_category) (rate(caatsm_messages_processed_total[5m]))", + "legendFormat": "{{message_status}} / {{message_category}}" + } + ] + }, + { + "id": 10, + "type": "timeseries", + "title": "Parse Duration (OTEL - P50/P95/P99)", + "description": "OTEL histogram metrics for parse latency, broken down by message status and category.", + "datasource": { + "type": "prometheus", + "uid": "prometheus-dev" + }, + "gridPos": { "h": 7, "w": 12, "x": 0, "y": 33 }, + "fieldConfig": { + "defaults": { + "color": { "mode": "palette-classic" }, + "mappings": [], + "thresholds": { "mode": "absolute", "steps": [] }, + "unit": "s" + }, + "overrides": [] + }, + "options": { + "legend": { "displayMode": "table", "placement": "bottom" }, + "tooltip": { "mode": "single" } + }, + "targets": [ + { + "refId": "P50", + "expr": "histogram_quantile(0.50, sum by (le, message_status, message_category) (rate(caatsm_parse_duration_seconds_bucket[5m])))", + "legendFormat": "P50 ({{message_status}}/{{message_category}})" + }, + { + "refId": "P95", + "expr": "histogram_quantile(0.95, sum by (le, message_status, message_category) (rate(caatsm_parse_duration_seconds_bucket[5m])))", + "legendFormat": "P95 ({{message_status}}/{{message_category}})" + }, + { + "refId": "P99", + "expr": "histogram_quantile(0.99, sum by (le, message_status, message_category) (rate(caatsm_parse_duration_seconds_bucket[5m])))", + "legendFormat": "P99 ({{message_status}}/{{message_category}})" + } + ] + }, + { + "id": 11, + "type": "timeseries", + "title": "Publish Failures (OTEL - by category)", + "description": "OTEL-sourced publish failures. Panel 7 shows Prometheus-sourced failures.", + "datasource": { + "type": "prometheus", + "uid": "prometheus-dev" + }, + "gridPos": { "h": 7, "w": 12, "x": 12, "y": 33 }, + "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/configs/grafana-dashboards.dev/otel-collector.json b/configs/grafana-dashboards.dev/otel-collector.json index 33edd9b..cebed89 100644 --- a/configs/grafana-dashboards.dev/otel-collector.json +++ b/configs/grafana-dashboards.dev/otel-collector.json @@ -140,12 +140,12 @@ "targets": [ { "refId": "A", - "expr": "sum(rate(otelcol_exporter_sent_spans{job=\"otel-collector\",exporter=\"otlp/jaeger\"}[1m]))", + "expr": "sum(rate(otelcol_exporter_sent_spans{job=\"otel-collector\",exporter=\"otlphttp/jaeger\"}[1m]))", "legendFormat": "sent" }, { "refId": "B", - "expr": "sum(rate(otelcol_exporter_failed_spans{job=\"otel-collector\",exporter=\"otlp/jaeger\"}[1m]))", + "expr": "sum(rate(otelcol_exporter_failed_spans{job=\"otel-collector\",exporter=\"otlphttp/jaeger\"}[1m]))", "legendFormat": "failed" } ] @@ -199,7 +199,7 @@ "targets": [ { "refId": "A", - "expr": "avg(otelcol_exporter_queue_size{job=\"otel-collector\",exporter=\"otlp/jaeger\"})", + "expr": "avg(otelcol_exporter_queue_size{job=\"otel-collector\",exporter=\"otlphttp/jaeger\"})", "legendFormat": "queue size" } ] @@ -318,116 +318,6 @@ } ] }, - { - "id": 6, - "type": "timeseries", - "title": "Messages Processed Rate", - "datasource": { - "type": "prometheus", - "uid": "prometheus-dev" - }, - "fieldConfig": { - "defaults": { - "color": { - "mode": "palette-classic" - }, - "mappings": [], - "thresholds": { - "mode": "absolute", - "steps": [ - { - "color": "green", - "value": null - }, - { - "color": "red", - "value": 0 - } - ] - } - }, - "overrides": [] - }, - "gridPos": { - "h": 8, - "w": 12, - "x": 0, - "y": 12 - }, - "options": { - "legend": { - "displayMode": "list", - "placement": "bottom" - }, - "tooltip": { - "mode": "single" - } - }, - "targets": [ - { - "refId": "A", - "expr": "sum by (message_status) (rate(caatsm_messages_processed_total[1m]))", - "legendFormat": "{{message_status}}" - } - ] - }, - { - "id": 7, - "type": "timeseries", - "title": "Publish Failures Rate", - "datasource": { - "type": "prometheus", - "uid": "prometheus-dev" - }, - "fieldConfig": { - "defaults": { - "color": { - "mode": "palette-classic" - }, - "mappings": [], - "thresholds": { - "mode": "absolute", - "steps": [ - { - "color": "green", - "value": null - }, - { - "color": "orange", - "value": 100 - }, - { - "color": "red", - "value": 500 - } - ] - } - }, - "overrides": [] - }, - "gridPos": { - "h": 8, - "w": 12, - "x": 12, - "y": 12 - }, - "options": { - "legend": { - "displayMode": "list", - "placement": "bottom" - }, - "tooltip": { - "mode": "single" - } - }, - "targets": [ - { - "refId": "A", - "expr": "sum(rate(caatsm_publish_failures_total[1m]))", - "legendFormat": "failures/s" - } - ] - }, { "id": 8, "type": "timeseries", @@ -476,75 +366,16 @@ "targets": [ { "refId": "A", - "expr": "avg(otelcol_exporter_queue_size{job=\"otel-collector\",exporter=\"otlp/jaeger\"})", + "expr": "avg(otelcol_exporter_queue_size{job=\"otel-collector\",exporter=\"otlphttp/jaeger\"})", "legendFormat": "queue size" }, { "refId": "B", - "expr": "avg(otelcol_exporter_queue_capacity{job=\"otel-collector\",exporter=\"otlp/jaeger\"})", + "expr": "avg(otelcol_exporter_queue_capacity{job=\"otel-collector\",exporter=\"otlphttp/jaeger\"})", "legendFormat": "capacity" } ] }, - { - "id": 9, - "type": "timeseries", - "title": "Parse Duration (P95 / P50)", - "datasource": { - "type": "prometheus", - "uid": "prometheus-dev" - }, - "fieldConfig": { - "defaults": { - "color": { - "mode": "palette-classic" - }, - "mappings": [], - "thresholds": { - "mode": "absolute", - "steps": [ - { - "color": "green", - "value": null - }, - { - "color": "red", - "value": 0 - } - ] - }, - "unit": "ms" - }, - "overrides": [] - }, - "gridPos": { - "h": 8, - "w": 12, - "x": 12, - "y": 20 - }, - "options": { - "legend": { - "displayMode": "list", - "placement": "bottom" - }, - "tooltip": { - "mode": "single" - } - }, - "targets": [ - { - "refId": "A", - "expr": "histogram_quantile(0.95, sum(rate(caatsm_parse_duration_seconds_bucket[5m])) by (le))", - "legendFormat": "p95" - }, - { - "refId": "B", - "expr": "histogram_quantile(0.50, sum(rate(caatsm_parse_duration_seconds_bucket[5m])) by (le))", - "legendFormat": "p50" - } - ] - }, { "id": 10, "type": "timeseries", @@ -711,4 +542,3 @@ } ] } - diff --git a/configs/otel-collector.dev.yaml b/configs/otel-collector.dev.yaml index 7571bdb..2dc6612 100644 --- a/configs/otel-collector.dev.yaml +++ b/configs/otel-collector.dev.yaml @@ -4,10 +4,8 @@ receivers: http: endpoint: 0.0.0.0:4318 max_request_body_size: 20971520 # 20MB - max_concurrent_streams: 16 grpc: endpoint: 0.0.0.0:4317 - max_recv_msg_size: 4194304 # 4MB max_concurrent_streams: 16 processors: @@ -50,10 +48,6 @@ exporters: endpoint: "0.0.0.0:8889" const_labels: source: "otel-collector" - sending_queue: - queue_size: 10000 - retry_on_failure: - enabled: true service: pipelines: diff --git a/configs/prometheus.dev.yml b/configs/prometheus.dev.yml index 0422107..cbd06a9 100644 --- a/configs/prometheus.dev.yml +++ b/configs/prometheus.dev.yml @@ -6,6 +6,7 @@ scrape_configs: - job_name: "otel-collector" static_configs: - targets: + - "otel-collector:8888" - "otel-collector:8889" - job_name: "nats-exporter" static_configs: diff --git a/configs/prometheus/targets/caatsm-receiver.json b/configs/prometheus/targets/caatsm-receiver.json index c60b698..2e086da 100644 --- a/configs/prometheus/targets/caatsm-receiver.json +++ b/configs/prometheus/targets/caatsm-receiver.json @@ -1 +1 @@ -[{"labels":{"job":"caatsm-receiver","instance":"10.16.66.236:2112"},"targets":["10.16.66.236:2112"]}] +[{"labels":{"job":"caatsm-receiver","instance":"host.docker.internal:2112"},"targets":["host.docker.internal:2112"]}] diff --git a/docs/nats.md b/docs/nats.md index dd3e2b1..f9ff8e0 100644 --- a/docs/nats.md +++ b/docs/nats.md @@ -692,7 +692,7 @@ func main() { defer nc.Close() // 3. Get JetStream context - js, err := nats.ProvideJetStream(nc, zap.NewNop()) + js, err := nats.ProvideJetStream(nc, cfg, zap.NewNop()) if err != nil { panic(err) } diff --git a/docs/observability.md b/docs/observability.md index b748206..d657208 100644 --- a/docs/observability.md +++ b/docs/observability.md @@ -28,15 +28,22 @@ The service exposes Prometheus metrics via the monitoring HTTP server (default ` - `caatsm_dlq_publish_failures_total{stream,consumer}` Count of failures when attempting to publish messages to the DLQ. +- `caatsm_publish_failures_total{category}` + 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). Additional OTEL metrics are emitted via the configured OTEL endpoint, including: -- `caatsm_messages_processed_total` -- `caatsm_parse_duration_seconds` -- `caatsm_publish_failures_total` -- `caatsm_nats_consumer_pending_messages` +- `caatsm_messages_processed_total{message_status,message_category}` + Total number of telegrams processed, labelled by status and category. + +- `caatsm_parse_duration_seconds{message_status,message_category}` + Latency of parsing a telegram, in seconds, labelled by status and category. + +- `caatsm_publish_failures_total{message_category}` + Total number of telegram publish failures, labelled by category (also available in Prometheus as `caatsm_publish_failures_total{category}`). 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: @@ -92,12 +99,21 @@ Alternative topologies: The `caatsm-overview` Grafana dashboard (provisioned from `configs/grafana-dashboards.dev/caatsm-overview.json`) focuses on the CAATSM receiver service and surfaces: +**Prometheus Metrics (operational focus):** - **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`. +- **Publish failures (Prometheus)** – from `caatsm_publish_failures_total{category}`. +- **NATS consumer pending messages** – from `caatsm_nats_consumer_pending_messages`. + +**OTEL Metrics (business focus, scraped from collector):** +- **Messages Processed** – `caatsm_messages_processed_total` by `message_status` and `message_category`. +- **Parse Duration** – P50/P95/P99 percentiles from `caatsm_parse_duration_seconds_bucket`. +- **Publish Failures (OTEL)** – `caatsm_publish_failures_total` by `message_category`. + +All CAATSM metrics are consolidated in this dashboard for comprehensive service monitoring. To validate that the dashboard is receiving data: @@ -221,12 +237,14 @@ The receiver reports complementary metrics through both systems: - DB activity (`caatsm_db_queries_total`, `caatsm_db_query_latency_seconds`) - NATS consumer metrics (`caatsm_nats_consumer_pending_messages`) - DLQ operations (`caatsm_dlq_messages_total`, `caatsm_dlq_publish_failures_total`) +- Publish failures (`caatsm_publish_failures_total{category}`) **OpenTelemetry metrics via OTLP** (business focus): -- Message processing results (`caatsm_messages_processed_total`) -- Parse performance (`caatsm_parse_duration_seconds`) -- Publish reliability (`caatsm_publish_failures_total`) -- NATS consumer health metrics (ack pending, redelivered, delivered counts) +- Message processing results (`caatsm_messages_processed_total{message_status,message_category}`) +- Parse performance (`caatsm_parse_duration_seconds{message_status,message_category}`) +- Publish reliability (`caatsm_publish_failures_total{message_category}`) + +**Note**: `caatsm_publish_failures_total` is available in both Prometheus (with `category` label) and OTEL (with `message_category` label). The OTEL version is exported via the collector and scraped by Prometheus, where attribute keys are converted to label names (dots become underscores: `message_status`, `message_category`). #### Collector Integration @@ -235,7 +253,16 @@ OTEL metrics and traces are exported to the configured collector: - **Development**: `configs/otel-collector.dev.yaml` (batching, resource processing, retry logic) - **Production**: `configs/otel-collector.prod.yaml` (TLS, authentication, high availability) -To integrate OTEL metrics with Prometheus, extend the collector configuration with a `prometheusremotewrite` exporter. +The collector exports OTEL metrics to Prometheus via the `prometheus` exporter (default endpoint: `:8889`), making them available for Grafana dashboards. + +#### OpenTelemetry Collector Dashboard + +The `otel-collector-dev` Grafana dashboard (provisioned from `configs/grafana-dashboards.dev/otel-collector.json`) focuses on the OTEL collector infrastructure: + +- **OTEL Collector metrics** – receiver/exporter throughput, queue sizes, process metrics +- **Collector health** – memory, CPU, uptime, and queue capacity + +**Note**: CAATSM application metrics exported via OTEL are displayed in the **CAATSM – Receiver Overview** dashboard for consolidated service monitoring. The OTEL collector dashboard focuses solely on collector infrastructure metrics. ### Structured Logging Contract diff --git a/internal/infra/metrics/metrics.go b/internal/infra/metrics/metrics.go index 1068773..91ba0de 100644 --- a/internal/infra/metrics/metrics.go +++ b/internal/infra/metrics/metrics.go @@ -27,6 +27,7 @@ const ( MetricDBQueryLatencySeconds = "caatsm_db_query_latency_seconds" MetricDLQMessagesTotal = "caatsm_dlq_messages_total" MetricDLQPublishFailures = "caatsm_dlq_publish_failures_total" + MetricPublishFailuresTotal = "caatsm_publish_failures_total" MetricNATSConsumerPending = "caatsm_nats_consumer_pending_messages" // Common label keys. @@ -69,6 +70,7 @@ var ( jsAPICallsTotal *prometheus.CounterVec dlqMessagesTotal *prometheus.CounterVec dlqPublishFailures *prometheus.CounterVec + publishFailuresTotal *prometheus.CounterVec // Database metrics. dbQueriesTotal *prometheus.CounterVec @@ -125,6 +127,11 @@ func initCollectors() { Help: "Total number of failures when publishing to the DLQ, labelled by stream and consumer.", }, []string{LabelStream, LabelConsumer}) + publishFailuresTotal = prometheus.NewCounterVec(prometheus.CounterOpts{ + Name: MetricPublishFailuresTotal, + Help: "Total number of telegram publish failures, labelled by category.", + }, []string{LabelCategory}) + jsAPICallsTotal = prometheus.NewCounterVec(prometheus.CounterOpts{ Name: MetricJSAPICallsTotal, Help: "Count of JetStream API calls made by the receiver.", @@ -157,6 +164,7 @@ func initCollectors() { jsAPICallsTotal, dlqMessagesTotal, dlqPublishFailures, + publishFailuresTotal, dbQueriesTotal, dbQueryLatency, natsConsumerPending, @@ -218,6 +226,13 @@ func RecordDLQPublishFailure(stream, consumer string) { dlqPublishFailures.WithLabelValues(labelValue(stream), labelValue(consumer)).Inc() } +// RecordPublishFailure increments the publish failure counter for the given category. +// This tracks general publish failures (not DLQ-specific). +func RecordPublishFailure(category string) { + ensureCollectors() + publishFailuresTotal.WithLabelValues(labelValue(category)).Inc() +} + // RecordDBQuery records metrics for a single database operation. // Operation examples: "insert_one", "insert_batch", "insert_raw". // Result is usually "ok" or "error". diff --git a/internal/infra/nats/jetstream.go b/internal/infra/nats/jetstream.go index 06c167f..13fa5d1 100644 --- a/internal/infra/nats/jetstream.go +++ b/internal/infra/nats/jetstream.go @@ -72,7 +72,14 @@ func ProvideNATSConn(cfg *config.Config, logger *zap.Logger) (*nats.Conn, error) } // ProvideJetStream creates a JetStream context from a NATS connection. -func ProvideJetStream(nc *nats.Conn, logger *zap.Logger) (nats.JetStreamContext, error) { +// Returns nil when mode is "core" to indicate JetStream should not be used. +func ProvideJetStream(nc *nats.Conn, cfg *config.Config, logger *zap.Logger) (nats.JetStreamContext, error) { + // In core mode, return nil so that publishers/consumers use core NATS + if cfg.NATS.Mode == "core" { + logger.Debug("Skipping JetStream initialization (core mode)") + return nil, nil + } + js, err := nc.JetStream() if err != nil { logger.Error("failed to create JetStream context", zap.Error(err)) diff --git a/internal/infra/telemetry/otel.go b/internal/infra/telemetry/otel.go index 92480fb..f98dbbb 100644 --- a/internal/infra/telemetry/otel.go +++ b/internal/infra/telemetry/otel.go @@ -4,9 +4,9 @@ import ( "caatsm/internal/infra/buildinfo" "caatsm/internal/infra/config" "context" - "crypto/tls" "fmt" "os" + "strings" "sync" "time" @@ -39,8 +39,10 @@ func InitOTEL(ctx context.Context, cfg *config.Config, logger *zap.Logger) error // Log sensitive telemetry configuration to internal debug logs only if logger != nil { - logger.Debug("Initializing OpenTelemetry", - zap.String("telemetry.endpoint", cfg.Telemetry.Endpoint), + normalizedEndpoint := normalizeEndpoint(cfg.Telemetry.Endpoint) + logger.Info("Initializing OpenTelemetry", + zap.String("telemetry.endpoint.original", cfg.Telemetry.Endpoint), + zap.String("telemetry.endpoint.normalized", normalizedEndpoint), zap.Bool("telemetry.insecure", cfg.Telemetry.Insecure), ) } @@ -135,17 +137,33 @@ func createResource(ctx context.Context, cfg *config.Config) (*resource.Resource return resource.New(ctx, resource.WithAttributes(attrs...)) } +// normalizeEndpoint removes the scheme from the endpoint, returning just host:port +// The OpenTelemetry SDK's WithEndpoint() expects host:port, and WithInsecure() controls the protocol +func normalizeEndpoint(endpoint string) string { + if endpoint == "" { + return endpoint + } + + endpoint = strings.TrimSpace(endpoint) + + // Remove http:// or https:// scheme if present + endpoint = strings.TrimPrefix(endpoint, "http://") + endpoint = strings.TrimPrefix(endpoint, "https://") + + return endpoint +} + // initTracing sets up the trace provider with appropriate sampling func initTracing(ctx context.Context, cfg *config.Config, res *resource.Resource) error { var traceExporterOptions []otlptracehttp.Option - traceExporterOptions = append(traceExporterOptions, otlptracehttp.WithEndpoint(cfg.Telemetry.Endpoint)) + // Normalize endpoint to remove scheme (WithEndpoint expects host:port) + endpoint := normalizeEndpoint(cfg.Telemetry.Endpoint) + traceExporterOptions = append(traceExporterOptions, otlptracehttp.WithEndpoint(endpoint)) if cfg.Telemetry.Insecure { - traceExporterOptions = append(traceExporterOptions, otlptracehttp.WithTLSClientConfig(&tls.Config{ - InsecureSkipVerify: true, - MinVersion: tls.VersionTLS13, - })) + // Use HTTP instead of HTTPS when insecure is true + traceExporterOptions = append(traceExporterOptions, otlptracehttp.WithInsecure()) } traceExporter, err := otlptracehttp.New(ctx, traceExporterOptions...) @@ -174,13 +192,13 @@ func initTracing(ctx context.Context, cfg *config.Config, res *resource.Resource func initMetrics(ctx context.Context, cfg *config.Config, res *resource.Resource) error { var metricExporterOptions []otlpmetrichttp.Option - metricExporterOptions = append(metricExporterOptions, otlpmetrichttp.WithEndpoint(cfg.Telemetry.Endpoint)) + // Normalize endpoint to remove scheme (WithEndpoint expects host:port) + endpoint := normalizeEndpoint(cfg.Telemetry.Endpoint) + metricExporterOptions = append(metricExporterOptions, otlpmetrichttp.WithEndpoint(endpoint)) if cfg.Telemetry.Insecure { - metricExporterOptions = append(metricExporterOptions, otlpmetrichttp.WithTLSClientConfig(&tls.Config{ - InsecureSkipVerify: true, - MinVersion: tls.VersionTLS12, - })) + // Use HTTP instead of HTTPS when insecure is true + metricExporterOptions = append(metricExporterOptions, otlpmetrichttp.WithInsecure()) } metricExporter, err := otlpmetrichttp.New(ctx, metricExporterOptions...) diff --git a/internal/infra/telemetry/telemetry.go b/internal/infra/telemetry/telemetry.go index 9333d1e..083b0e0 100644 --- a/internal/infra/telemetry/telemetry.go +++ b/internal/infra/telemetry/telemetry.go @@ -183,8 +183,9 @@ func (p *promRecorder) RecordProcessingResult(ctx context.Context, status, categ } func (p *promRecorder) RecordPublishFailure(ctx context.Context, category string) { - // Prometheus metrics currently only expose failures via caatsm_failures_total, - // so we record the publisher failure there. + // Record publish failure with category label for detailed tracking. + obsmetrics.RecordPublishFailure(category) + // Also record to legacy failure counter for backward compatibility. obsmetrics.RecordFailure("publisher") } diff --git a/pkg/di/wire_gen.go b/pkg/di/wire_gen.go index e363f3a..cb53fa4 100644 --- a/pkg/di/wire_gen.go +++ b/pkg/di/wire_gen.go @@ -42,7 +42,7 @@ func buildAppComponents() (*appComponents, error) { if err != nil { return nil, err } - jetStreamContext, err := nats.ProvideJetStream(conn, logger) + jetStreamContext, err := nats.ProvideJetStream(conn, configConfig, logger) if err != nil { return nil, err } @@ -86,7 +86,7 @@ func buildAppComponentsWithConfig(cfg *config.Config) (*appComponents, error) { if err != nil { return nil, err } - jetStreamContext, err := nats.ProvideJetStream(conn, logger) + jetStreamContext, err := nats.ProvideJetStream(conn, cfg, logger) if err != nil { return nil, err }