diff --git a/README.md b/README.md index 58ae65c..5a4d9bb 100644 --- a/README.md +++ b/README.md @@ -221,7 +221,11 @@ Critical overrides stay available through CLI flags; advanced tuning such as str - CLI overrides: - `--telemetry-enabled` flips the feature on/off. - `--telemetry-endpoint` and `--telemetry-insecure` adjust the OTLP HTTP endpoint and TLS behavior. -- When enabled the app emits OpenTelemetry traces (parser/repository/publisher spans) and JetStream metrics (ack pending, deliveries) for dashboards and alerts. +- When enabled the app emits OpenTelemetry traces (parser/repository/publisher spans) and metrics. Custom OTLP metrics include: + - `caatsm_messages_processed_total` (counter, broken down by `message_status` / `message_category`) + - `caatsm_publish_failures_total` (counter) + - `caatsm_parse_duration_ms` (histogram) + These flow through the collector → Prometheus → Grafana dashboards in the dev stack. ## Development diff --git a/Taskfile.yml b/Taskfile.yml index abe5f4a..268ac84 100644 --- a/Taskfile.yml +++ b/Taskfile.yml @@ -108,7 +108,7 @@ tasks: desc: Start TimescaleDB + NATS dev stack (docker compose) cmds: - echo "Starting dev infrastructure..." - - docker compose -f docker-compose.dev.yml up -d postgres nats nats-box + - docker compose -f docker-compose.dev.yml up -d postgres nats nats-box nats-exporter - docker compose -f docker-compose.dev.yml up -d otel-collector jaeger prometheus grafana down: diff --git a/configs/grafana-dashboards.dev/nats-overview.json b/configs/grafana-dashboards.dev/nats-overview.json index 9eb03a3..c1fd78a 100644 --- a/configs/grafana-dashboards.dev/nats-overview.json +++ b/configs/grafana-dashboards.dev/nats-overview.json @@ -82,7 +82,7 @@ "targets": [ { "refId": "A", - "expr": "sum(nats_varz_total_connections)" + "expr": "sum(gnatsd_varz_connections{job=\"nats-exporter\"})" } ] }, @@ -136,7 +136,7 @@ "targets": [ { "refId": "A", - "expr": "sum(nats_varz_slow_consumers)" + "expr": "sum(gnatsd_varz_slow_consumers{job=\"nats-exporter\"})" } ] }, @@ -202,7 +202,7 @@ "targets": [ { "refId": "A", - "expr": "max(nats_varz_jetstream)" + "expr": "max(gnatsd_varz_jetstream_stats_accounts{job=\"nats-exporter\"})" } ] }, @@ -255,12 +255,12 @@ "targets": [ { "refId": "A", - "expr": "sum(rate(nats_varz_in_msgs[1m]))", + "expr": "sum(rate(gnatsd_varz_in_msgs{job=\"nats-exporter\"}[1m]))", "legendFormat": "in" }, { "refId": "B", - "expr": "sum(rate(nats_varz_out_msgs[1m]))", + "expr": "sum(rate(gnatsd_varz_out_msgs{job=\"nats-exporter\"}[1m]))", "legendFormat": "out" } ] @@ -315,12 +315,12 @@ "targets": [ { "refId": "A", - "expr": "sum(rate(nats_varz_in_bytes[1m]))", + "expr": "sum(rate(gnatsd_varz_in_bytes{job=\"nats-exporter\"}[1m]))", "legendFormat": "in" }, { "refId": "B", - "expr": "sum(rate(nats_varz_out_bytes[1m]))", + "expr": "sum(rate(gnatsd_varz_out_bytes{job=\"nats-exporter\"}[1m]))", "legendFormat": "out" } ] @@ -379,7 +379,193 @@ "targets": [ { "refId": "A", - "expr": "sum(nats_varz_jetstream_bytes)" + "expr": "sum(gnatsd_varz_jetstream_stats_storage{job=\"nats-exporter\"})" + } + ] + }, + { + "id": 7, + "type": "stat", + "title": "Total Subscriptions", + "datasource": { + "type": "prometheus", + "uid": "prometheus-dev" + }, + "fieldConfig": { + "defaults": { + "mappings": [], + "thresholds": { + "mode": "absolute", + "steps": [ + { + "color": "green", + "value": null + }, + { + "color": "red", + "value": 1000 + } + ] + } + }, + "overrides": [] + }, + "gridPos": { + "h": 4, + "w": 6, + "x": 0, + "y": 20 + }, + "options": { + "colorMode": "value", + "graphMode": "area", + "justifyMode": "auto", + "orientation": "auto", + "reduceOptions": { + "calcs": [ + "lastNotNull" + ], + "fields": "", + "values": false + }, + "textMode": "auto" + }, + "targets": [ + { + "refId": "A", + "expr": "sum(gnatsd_varz_subscriptions{job=\"nats-exporter\"})" + } + ] + }, + { + "id": 8, + "type": "timeseries", + "title": "JetStream API Calls / Errors", + "datasource": { + "type": "prometheus", + "uid": "prometheus-dev" + }, + "fieldConfig": { + "defaults": { + "color": { + "mode": "palette-classic" + }, + "custom": {}, + "mappings": [], + "thresholds": { + "mode": "absolute", + "steps": [ + { + "color": "green", + "value": null + }, + { + "color": "red", + "value": 0 + } + ] + } + }, + "overrides": [] + }, + "gridPos": { + "h": 8, + "w": 12, + "x": 6, + "y": 20 + }, + "options": { + "legend": { + "displayMode": "list", + "placement": "bottom" + }, + "tooltip": { + "mode": "single" + } + }, + "targets": [ + { + "refId": "A", + "expr": "sum(rate(gnatsd_varz_jetstream_stats_api_total{job=\"nats-exporter\"}[1m]))", + "legendFormat": "api total" + }, + { + "refId": "B", + "expr": "sum(rate(gnatsd_varz_jetstream_stats_api_errors{job=\"nats-exporter\"}[1m]))", + "legendFormat": "api errors" + } + ] + }, + { + "id": 9, + "type": "timeseries", + "title": "Server CPU / Memory", + "datasource": { + "type": "prometheus", + "uid": "prometheus-dev" + }, + "fieldConfig": { + "defaults": { + "color": { + "mode": "palette-classic" + }, + "custom": {}, + "mappings": [], + "thresholds": { + "mode": "absolute", + "steps": [ + { + "color": "green", + "value": null + }, + { + "color": "red", + "value": 0 + } + ] + }, + "unit": "none" + }, + "overrides": [ + { + "matcher": { + "id": "byName", + "options": "Memory (bytes)" + }, + "properties": [ + { + "id": "unit", + "value": "bytes" + } + ] + } + ] + }, + "gridPos": { + "h": 8, + "w": 12, + "x": 18, + "y": 20 + }, + "options": { + "legend": { + "displayMode": "list", + "placement": "bottom" + }, + "tooltip": { + "mode": "single" + } + }, + "targets": [ + { + "refId": "A", + "expr": "avg(gnatsd_varz_cpu{job=\"nats-exporter\"})", + "legendFormat": "CPU (%)" + }, + { + "refId": "B", + "expr": "sum(gnatsd_varz_mem{job=\"nats-exporter\"})", + "legendFormat": "Memory (bytes)" } ] } diff --git a/configs/grafana-dashboards.dev/otel-collector.json b/configs/grafana-dashboards.dev/otel-collector.json index a7b5c66..aae45b4 100644 --- a/configs/grafana-dashboards.dev/otel-collector.json +++ b/configs/grafana-dashboards.dev/otel-collector.json @@ -317,6 +317,397 @@ "expr": "rate(process_cpu_seconds_total{job=\"otel-collector\"}[5m])" } ] + }, + { + "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", + "title": "Exporter Queue Size vs Capacity", + "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": 20 + }, + "options": { + "legend": { + "displayMode": "list", + "placement": "bottom" + }, + "tooltip": { + "mode": "single" + } + }, + "targets": [ + { + "refId": "A", + "expr": "avg(otelcol_exporter_queue_size{job=\"otel-collector\",exporter=\"otlp/jaeger\"})", + "legendFormat": "queue size" + }, + { + "refId": "B", + "expr": "avg(otelcol_exporter_queue_capacity{job=\"otel-collector\",exporter=\"otlp/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_ms_bucket[5m])) by (le))", + "legendFormat": "p95" + }, + { + "refId": "B", + "expr": "histogram_quantile(0.50, sum(rate(caatsm_parse_duration_ms_bucket[5m])) by (le))", + "legendFormat": "p50" + } + ] + }, + { + "id": 10, + "type": "timeseries", + "title": "Metrics Export Rate (logging)", + "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": 28 + }, + "options": { + "legend": { + "displayMode": "list", + "placement": "bottom" + }, + "tooltip": { + "mode": "single" + } + }, + "targets": [ + { + "refId": "A", + "expr": "sum(rate(otelcol_exporter_sent_metrics{job=\"otel-collector\",exporter=\"logging\"}[1m]))", + "legendFormat": "sent metrics/s" + }, + { + "refId": "B", + "expr": "sum(rate(otelcol_exporter_failed_metrics{job=\"otel-collector\",exporter=\"logging\"}[1m]))", + "legendFormat": "failed metrics/s" + } + ] + }, + { + "id": 11, + "type": "stat", + "title": "Collector Uptime (s)", + "datasource": { + "type": "prometheus", + "uid": "prometheus-dev" + }, + "fieldConfig": { + "defaults": { + "mappings": [], + "thresholds": { + "mode": "absolute", + "steps": [ + { + "color": "green", + "value": null + } + ] + }, + "unit": "s" + }, + "overrides": [] + }, + "gridPos": { + "h": 4, + "w": 6, + "x": 12, + "y": 28 + }, + "options": { + "colorMode": "value", + "graphMode": "none", + "justifyMode": "auto", + "orientation": "auto", + "reduceOptions": { + "calcs": [ + "lastNotNull" + ], + "fields": "", + "values": false + }, + "textMode": "auto" + }, + "targets": [ + { + "refId": "A", + "expr": "otelcol_process_uptime{job=\"otel-collector\"}" + } + ] + }, + { + "id": 12, + "type": "timeseries", + "title": "Process Heap / Sys Memory", + "datasource": { + "type": "prometheus", + "uid": "prometheus-dev" + }, + "fieldConfig": { + "defaults": { + "color": { + "mode": "palette-classic" + }, + "mappings": [], + "thresholds": { + "mode": "absolute", + "steps": [ + { + "color": "green", + "value": null + } + ] + }, + "unit": "bytes" + }, + "overrides": [] + }, + "gridPos": { + "h": 8, + "w": 12, + "x": 18, + "y": 28 + }, + "options": { + "legend": { + "displayMode": "list", + "placement": "bottom" + }, + "tooltip": { + "mode": "single" + } + }, + "targets": [ + { + "refId": "A", + "expr": "sum(otelcol_process_runtime_heap_alloc_bytes{job=\"otel-collector\"})", + "legendFormat": "heap alloc" + }, + { + "refId": "B", + "expr": "sum(otelcol_process_runtime_total_sys_memory_bytes{job=\"otel-collector\"})", + "legendFormat": "sys memory" + } + ] } ] } diff --git a/configs/prometheus.dev.yml b/configs/prometheus.dev.yml index 01a7cef..3def892 100644 --- a/configs/prometheus.dev.yml +++ b/configs/prometheus.dev.yml @@ -7,10 +7,8 @@ scrape_configs: static_configs: - targets: - "otel-collector:8888" - - job_name: "nats" - metrics_path: /varz - scheme: http + - job_name: "nats-exporter" static_configs: - targets: - - "nats:8222" + - "nats-exporter:7777" diff --git a/docker-compose.dev.yml b/docker-compose.dev.yml index 00bf6d0..835b9d9 100644 --- a/docker-compose.dev.yml +++ b/docker-compose.dev.yml @@ -45,6 +45,21 @@ services: networks: - devnet + nats-exporter: + image: natsio/prometheus-nats-exporter:0.14.0 + depends_on: + - nats + command: + - "-varz" + - "-connz" + - "-routez" + - "-subz" + - "http://nats:8222" + ports: + - "7777:7777" + networks: + - devnet + nats-box: image: synadia/nats-box:latest entrypoint: ["sleep", "infinity"] diff --git a/docs/dev-guide.md b/docs/dev-guide.md index e761414..a1797b7 100644 --- a/docs/dev-guide.md +++ b/docs/dev-guide.md @@ -15,6 +15,7 @@ docker compose -f docker-compose.dev.yml up -d postgres nats nats-box - `postgres` seeds the `aviation` schema using `internal/repository/telegrams.ddl` and exposes port `5432`. - `nats` enables JetStream with client port `4222` and monitoring/UI on `8222`. - `nats-box` provides a toolbox container (`docker compose exec nats-box sh`) for publishing test messages or inspecting JetStream. +- `nats-exporter` scrapes the monitoring endpoints (`/varz`, `/connz`, `/routez`, `/subz`) and exposes them as Prometheus metrics on port `7777` for the Grafana dashboards. Prefer to run the Go application on your host for quick iteration while keeping infra in Docker: @@ -34,7 +35,7 @@ docker compose -f docker-compose.dev.yml down -v The `Taskfile.yml` includes helper targets that wrap the commands above: -- `task up` – starts PostgreSQL, NATS, and the observability stack (OpenTelemetry Collector, Jaeger, Prometheus, Grafana) using Docker Compose. +- `task up` – starts PostgreSQL, NATS (JetStream, toolbox, and Prometheus exporter), and the observability stack (OpenTelemetry Collector, Jaeger, Prometheus, Grafana) using Docker Compose. - `task dev-run` – ensures `task up` has run, exports the necessary `CAATSM_*` environment variables (including `CAATSM_NATS_MODE=jetstream`), and executes `go run ./cmd/main listen` with telemetry enabled. - `task down` – stops the entire stack and removes containers/volumes. @@ -114,10 +115,11 @@ Services: - `jaeger` - Receives OTLP traffic forwarded from the collector on `14250` gRPC and serves the UI at - `prometheus` - - Uses `configs/prometheus.dev.yml` to scrape the collector and NATS monitoring endpoint; UI available at + - Uses `configs/prometheus.dev.yml` to scrape the collector, `nats-exporter` (`http://nats-exporter:7777/metrics`), and application OTLP metrics forwarded via the collector; UI available at - `grafana` - Persists data in `grafana-data`, provisions datasources via `configs/grafana-datasources.dev.yml`, and listens on (login `admin` / `admin`) - Automatically loads dashboards from `configs/grafana-dashboards.dev/`, including OpenTelemetry Collector and NATS/JetStream overviews (find them under the **Dev Observability** folder) + - The OpenTelemetry dashboard also charts the CAATSM-specific metrics `caatsm_messages_processed_total`, `caatsm_publish_failures_total`, and `caatsm_parse_duration_ms` (percentiles) so you can track throughput and parsing latency. ### Customizing Collections & Dashboards diff --git a/internal/app/processor.go b/internal/app/processor.go index 2b1f41a..bd702f5 100644 --- a/internal/app/processor.go +++ b/internal/app/processor.go @@ -12,6 +12,7 @@ import ( "go.opentelemetry.io/otel" "go.opentelemetry.io/otel/attribute" "go.opentelemetry.io/otel/codes" + "go.opentelemetry.io/otel/metric" "go.opentelemetry.io/otel/trace" "go.uber.org/zap" ) @@ -24,6 +25,32 @@ type MessageProcessor struct { logger *zap.Logger } +var ( + appMeter = otel.Meter("caatsm/app") + messageStatusAttrKey = attribute.Key("message.status") + messageCategoryAttrKey = attribute.Key("message.category") + + messageProcessedCounter = mustInt64Counter("caatsm_messages_processed_total", "Total number of telegrams processed by the CAATSM processor.") + messagePublishFailCounter = mustInt64Counter("caatsm_publish_failures_total", "Total number of telegram publish failures.") + parseLatencyHistogram = mustFloat64Histogram("caatsm_parse_duration_ms", "Latency of parsing a telegram, in milliseconds.", metric.WithUnit("ms")) +) + +func mustInt64Counter(name, description string, opts ...metric.Int64CounterOption) metric.Int64Counter { + counter, err := appMeter.Int64Counter(name, append([]metric.Int64CounterOption{metric.WithDescription(description)}, opts...)...) + if err != nil { + panic(fmt.Sprintf("failed to create counter %s: %v", name, err)) + } + return counter +} + +func mustFloat64Histogram(name, description string, opts ...metric.Float64HistogramOption) metric.Float64Histogram { + hist, err := appMeter.Float64Histogram(name, append([]metric.Float64HistogramOption{metric.WithDescription(description)}, opts...)...) + if err != nil { + panic(fmt.Sprintf("failed to create histogram %s: %v", name, err)) + } + return hist +} + // NewMessageProcessor creates a new message processor func NewMessageProcessor( parser parser.Parser, @@ -95,6 +122,13 @@ func (p *MessageProcessor) Handle(ctx context.Context, raw []byte, msgID string) zap.String("content_preview", truncateContent(parsed.Content, 256)), zap.Error(parseErr), ) + parseLatencyHistogram.Record(ctx, float64(parsed.ParsedAt.Sub(receivedAt).Milliseconds()), + metric.WithAttributes( + messageStatusAttrKey.String(string(parsed.Status)), + messageCategoryAttrKey.String(parsed.Category), + ), + ) + recordProcessedMetric(ctx, parsed) return Permanent(fmt.Errorf("parser error: %w", parseErr)) } parsed.ErrorReason = "" @@ -118,6 +152,14 @@ func (p *MessageProcessor) Handle(ctx context.Context, raw []byte, msgID string) if err := p.repository.InsertOne(ctx, parsed); err != nil { span.RecordError(err) span.SetStatus(codes.Error, err.Error()) + parsed.Status = domain.MessageStatusRepositoryFail + parseLatencyHistogram.Record(ctx, float64(parsed.ParsedAt.Sub(receivedAt).Milliseconds()), + metric.WithAttributes( + messageStatusAttrKey.String(string(parsed.Status)), + messageCategoryAttrKey.String(parsed.Category), + ), + ) + recordProcessedMetric(ctx, parsed) return fmt.Errorf("failed to insert message: %w", err) } @@ -133,6 +175,18 @@ func (p *MessageProcessor) Handle(ctx context.Context, raw []byte, msgID string) pubSpan.SetStatus(codes.Error, err.Error()) parsed.Status = domain.MessageStatusPublishFail parsed.ErrorReason = err.Error() + messagePublishFailCounter.Add(ctx, 1, + metric.WithAttributes( + messageCategoryAttrKey.String(parsed.Category), + ), + ) + parseLatencyHistogram.Record(ctx, float64(parsed.ParsedAt.Sub(receivedAt).Milliseconds()), + metric.WithAttributes( + messageStatusAttrKey.String(string(parsed.Status)), + messageCategoryAttrKey.String(parsed.Category), + ), + ) + recordProcessedMetric(ctx, parsed) p.persistRaw(ctx, parsed) // Mark as permanent so the consumer will ack instead of retrying pubSpan.End() @@ -140,6 +194,14 @@ func (p *MessageProcessor) Handle(ctx context.Context, raw []byte, msgID string) } pubSpan.End() + parseLatencyHistogram.Record(ctx, float64(parsed.ParsedAt.Sub(receivedAt).Milliseconds()), + metric.WithAttributes( + messageStatusAttrKey.String(string(parsed.Status)), + messageCategoryAttrKey.String(parsed.Category), + ), + ) + recordProcessedMetric(ctx, parsed) + return nil } @@ -179,3 +241,15 @@ func truncateContent(content string, limit int) string { } return content[:limit-3] + "..." } + +func recordProcessedMetric(ctx context.Context, msg *domain.ParsedMessage) { + if msg == nil { + return + } + messageProcessedCounter.Add(ctx, 1, + metric.WithAttributes( + messageStatusAttrKey.String(string(msg.Status)), + messageCategoryAttrKey.String(msg.Category), + ), + ) +}