diff --git a/README.md b/README.md index 9b58466..dfbcf56 100644 --- a/README.md +++ b/README.md @@ -228,25 +228,41 @@ Critical overrides stay available through CLI flags; advanced tuning such as str - Configure server-side retry delays with `[nats.consumer].backoff = ["5s", "30s", "2m"]`; each duration becomes the delay before the next delivery attempt. - Combine `backoff` with `--ack-wait` to increase acknowledgement windows (e.g., `--ack-wait 2m`). -### Telemetry +### Observability -- Enable tracing/metrics via `[telemetry] enabled = true` and set `endpoint` to your OTLP/HTTP collector (e.g., `http://otel-collector:4318`). -- 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 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. +The processor exposes three complementary observability surfaces: -### Observability & Health +1. **OpenTelemetry (traces + metrics)** + - Enable via `[telemetry] enabled = true` and set `endpoint` to your OTLP/HTTP collector (e.g., `http://otel-collector:4318`). + - CLI overrides: + - `--telemetry-enabled` toggles exporters on/off. + - `--telemetry-endpoint` and `--telemetry-insecure` adjust the OTLP HTTP endpoint and TLS behavior. + - When enabled, the app emits: + - Traces for parser/repository/publisher spans (`caatsm/app`, `caatsm/postgres`, `caatsm/nats`). + - A focused set of metrics, including: + - `caatsm_messages_processed_total` (counter, by `message.status` / `message.category`) + - `caatsm_publish_failures_total` (counter) + - `caatsm_parse_duration_seconds` (histogram) + - Application code records these via a thin `telemetry.Recorder` abstraction, which fans out to OTEL and Prometheus backends as configured. -A lightweight monitoring server exposes both readiness information and Prometheus-friendly metrics: +2. **Prometheus metrics (`/metrics`)** + - Implemented in `internal/observability/metrics` and considered the primary source for SRE PromQL/SLOs. + - Key metric families: + - `caatsm_messages_total{stream,consumer,result}` – per-stream/consumer throughput and results. + - `caatsm_handle_latency_seconds_bucket{stream,consumer}` – end-to-end handling latency from NATS receive to handler completion. + - `caatsm_retries_total{stream,consumer,reason}` – JetStream retry/NAK counts. + - `caatsm_db_queries_total{operation,result}` and `caatsm_db_query_latency_seconds_bucket{operation}` – DB activity and latency. + - `caatsm_dlq_messages_total{stream,consumer}` and `caatsm_dlq_publish_failures_total{stream,consumer}` – DLQ routing success/failures. + - `caatsm_nats_consumer_pending_messages{stream,consumer}` – JetStream consumer backlog/lag. + - Prometheus scrapes `GET /metrics` on the monitoring server; Grafana dashboards under `configs/grafana-dashboards.dev` are wired to these series. -- `GET /healthz` probes PostgreSQL (connection ping) and NATS (connection status). It returns HTTP 200 when both dependencies respond within `monitoring.health_timeout`, otherwise 503. -- `GET /metrics` streams `caatsm_processed_total`, `caatsm_failures_total`, and `caatsm_parse_latency_seconds` counters/histograms from the built-in Prometheus registry. -- Configure the server via the `[monitoring]` block (defaults shown): +3. **Health and readiness endpoints** + - A lightweight monitoring server exposes: + - `GET /livez` – liveness endpoint: reports process and build information, does not call external dependencies. + - `GET /readyz` – readiness endpoint: pings PostgreSQL and checks NATS connection status within `monitoring.health_timeout`, returning 503 on failure. + - `GET /healthz` – backward-compatible alias currently sharing logic with `/readyz`. + - Responses include build metadata and dependency status/latency (see `docs/observability.md` for examples). + - Configure the server via the `[monitoring]` block (defaults shown): ```toml [monitoring] diff --git a/cmd/main/main.go b/cmd/main/main.go index a806fef..8fd8223 100644 --- a/cmd/main/main.go +++ b/cmd/main/main.go @@ -25,14 +25,21 @@ import ( semconv "go.opentelemetry.io/otel/semconv/v1.26.0" ) +// main is the entrypoint for the caatsm CLI. +// It delegates all logic to run so that startup behaviour can be tested. func main() { - app := setupApp() - if err := app.Run(os.Args); err != nil { - fmt.Printf("Error running application: %v\n", err) + if err := run(os.Args); err != nil { + fmt.Fprintf(os.Stderr, "caatsm failed: %v\n", err) os.Exit(1) } } +// run constructs the CLI application and executes it with the provided args. +func run(args []string) error { + app := setupApp() + return app.Run(args) +} + func setupApp() *cli.App { return &cli.App{ Name: "telegram message process", @@ -108,6 +115,9 @@ func setupApp() *cli.App { } } +// executeListen is the CLI handler for the "listen" command. +// It is responsible for loading configuration, applying CLI overrides, +// and delegating the main processing lifecycle to runListen. func executeListen(c *cli.Context) error { cfg, err := config.LoadConfig() if err != nil { @@ -116,10 +126,29 @@ func executeListen(c *cli.Context) error { applyCLIOverrides(cfg, c) + // Re-validate configuration after applying CLI overrides to ensure + // the resulting configuration is still consistent. + if err := cfg.Validate(); err != nil { + return fmt.Errorf("invalid configuration after CLI overrides: %w", err) + } + + return runListen(context.Background(), cfg) +} + +// runListen coordinates telemetry initialisation, dependency wiring, +// signal handling and graceful shutdown for the listener workflow. +func runListen(parentCtx context.Context, cfg *config.Config) error { + if cfg == nil { + return fmt.Errorf("config must not be nil") + } + + ctx, stop := signal.NotifyContext(parentCtx, os.Interrupt, syscall.SIGTERM) + defer stop() + shutdownTelemetry := func(context.Context) error { return nil } if cfg.Telemetry.Enabled { var telErr error - shutdownTelemetry, telErr = initTelemetry(context.Background(), cfg) + shutdownTelemetry, telErr = initTelemetry(ctx, cfg) if telErr != nil { return fmt.Errorf("failed to initialize telemetry: %w", telErr) } @@ -131,9 +160,6 @@ func executeListen(c *cli.Context) error { if err != nil { return fmt.Errorf("failed to initialize app: %w", err) } - // Create context with cancellation - ctx, cancel := context.WithCancel(context.Background()) - defer cancel() if monitorServer != nil { if err := monitorServer.Start(ctx); err != nil { @@ -142,32 +168,29 @@ func executeListen(c *cli.Context) error { defer monitorServer.Shutdown(context.Background()) } - // Handle graceful shutdown - sigChan := make(chan os.Signal, 1) - signal.Notify(sigChan, os.Interrupt, syscall.SIGTERM) - // Start consumer in a goroutine errChan := make(chan error, 1) go func() { - if err := consumer.Start(ctx); err != nil { + if err := consumer.Start(ctx); err != nil && !errors.Is(err, context.Canceled) { errChan <- fmt.Errorf("consumer error: %w", err) } }() var runErr error - // Wait for signal or error + // Wait for shutdown signal or consumer error select { - case sig := <-sigChan: - fmt.Printf("Received signal: %v, shutting down...\n", sig) - cancel() + case <-ctx.Done(): + fmt.Printf("Received shutdown signal: %v, shutting down...\n", ctx.Err()) case err := <-errChan: - cancel() - if err != nil && !errors.Is(err, context.Canceled) { + if err != nil { runErr = err } + // Ensure all downstream users of ctx see cancellation. + stop() } + // After cancellation, give the consumer a chance to finish cleanup. waitTimeout := 5 * time.Second select { case err := <-errChan: diff --git a/configs/grafana-dashboards.dev/caatsm-overview.json b/configs/grafana-dashboards.dev/caatsm-overview.json index 2d1422e..0ff49a0 100644 --- a/configs/grafana-dashboards.dev/caatsm-overview.json +++ b/configs/grafana-dashboards.dev/caatsm-overview.json @@ -258,6 +258,35 @@ "legendFormat": "{{message_category}}" } ] + }, + { + "id": 8, + "type": "timeseries", + "title": "NATS consumer pending messages (lag)", + "datasource": { + "type": "prometheus", + "uid": "prometheus-dev" + }, + "gridPos": { "h": 7, "w": 24, "x": 0, "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": "caatsm_nats_consumer_pending_messages", + "legendFormat": "{{stream}} / {{consumer}}" + } + ] } ] } diff --git a/configs/grafana-dashboards.dev/otel-collector.json b/configs/grafana-dashboards.dev/otel-collector.json index aae45b4..33edd9b 100644 --- a/configs/grafana-dashboards.dev/otel-collector.json +++ b/configs/grafana-dashboards.dev/otel-collector.json @@ -535,12 +535,12 @@ "targets": [ { "refId": "A", - "expr": "histogram_quantile(0.95, sum(rate(caatsm_parse_duration_ms_bucket[5m])) by (le))", + "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_ms_bucket[5m])) by (le))", + "expr": "histogram_quantile(0.50, sum(rate(caatsm_parse_duration_seconds_bucket[5m])) by (le))", "legendFormat": "p50" } ] diff --git a/docs/deploy-k8s.md b/docs/deploy-k8s.md index 2eec729..047a680 100644 --- a/docs/deploy-k8s.md +++ b/docs/deploy-k8s.md @@ -70,13 +70,13 @@ spec: mountPath: /etc/caatsm livenessProbe: httpGet: - path: /healthz + path: /livez port: monitoring initialDelaySeconds: 10 periodSeconds: 15 readinessProbe: httpGet: - path: /healthz + path: /readyz port: monitoring initialDelaySeconds: 5 periodSeconds: 15 @@ -105,5 +105,5 @@ spec: protocol: TCP ``` -Point Prometheus at the service above (or annotate it if you use `prometheus-operator`). The `/healthz` probe doubles as a readiness check and quickly surfaces upstream connectivity issues. +Point Prometheus at the service above (or annotate it if you use `prometheus-operator`). The `/readyz` probe surfaces upstream connectivity issues, while `/livez` is used solely for liveness. diff --git a/docs/deploy-systemd.md b/docs/deploy-systemd.md index dd6bf4e..93747ac 100644 --- a/docs/deploy-systemd.md +++ b/docs/deploy-systemd.md @@ -58,7 +58,7 @@ sudo systemctl enable --now caatsm ## 4. Observability Hooks - Expose `monitoring.addr = ":2112"` (default) and add firewall rules so Prometheus can scrape `http://host:2112/metrics`. -- systemd watchdogs can use `curl -sf http://127.0.0.1:2112/healthz`. +- systemd watchdogs can use `curl -sf http://127.0.0.1:2112/livez` for liveness and `curl -sf http://127.0.0.1:2112/readyz` for readiness. With these three files (binary, config, env) the service becomes repeatable and easy to operate. diff --git a/docs/dev-guide.md b/docs/dev-guide.md index a1797b7..b40b28f 100644 --- a/docs/dev-guide.md +++ b/docs/dev-guide.md @@ -120,6 +120,7 @@ Services: - 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. + - Note: `caatsm_parse_duration_ms` has been renamed to `caatsm_parse_duration_seconds` to align with Prometheus `_seconds` conventions. ### Customizing Collections & Dashboards diff --git a/docs/observability.md b/docs/observability.md index d73d349..403c3ff 100644 --- a/docs/observability.md +++ b/docs/observability.md @@ -28,10 +28,13 @@ 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_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_ms` +- `caatsm_parse_duration_seconds` - `caatsm_publish_failures_total` - `caatsm_nats_consumer_ack_pending` - `caatsm_nats_consumer_redelivered` @@ -133,15 +136,33 @@ To validate that the dashboard is receiving data: The monitoring server exposes: -- `/healthz` – basic liveness and dependency check. -- `/readyz` – readiness endpoint with the same logic as `/healthz`, intended for load balancers / orchestrators. +- `/livez` – lightweight liveness endpoint that reports process/build information without checking dependencies. +- `/healthz` – backward-compatible health endpoint used by existing deploys; currently shares logic with `/readyz`. +- `/readyz` – readiness endpoint that checks critical dependencies and should be used by load balancers / orchestrators. Checks performed: -- PostgreSQL: `pgxpool.Pool.Ping` with configurable timeout (`monitoring.health_timeout`). -- NATS: connection status must be `CONNECTED`. +- PostgreSQL: `pgxpool.Pool.Ping` with configurable timeout (`monitoring.health_timeout`), reporting `status` and `latency_ms`. +- NATS: connection status must be `CONNECTED`; otherwise the dependency is marked as unavailable. -A non-200 response indicates the service is not healthy/ready and should be removed from traffic. +Responses include build metadata and a dependency map, for example: + +```json +{ + "status": "ok", + "build": { + "version": "v0.4.3", + "rev": "abc1234", + "built_at": "2025-11-16T08:35:00Z" + }, + "dependencies": { + "postgres": {"status": "ok", "latency_ms": 4}, + "nats": {"status": "CONNECTED"} + } +} +``` + +A non-2xx response indicates the service is not healthy/ready and should be removed from traffic. ### Tracing @@ -167,7 +188,7 @@ The receiver reports two complementary sets of metrics: Implemented using `otel.Meter` in the NATS consumer and app processor, including: - `caatsm_messages_processed_total` - - `caatsm_parse_duration_ms` + - `caatsm_parse_duration_seconds` - `caatsm_publish_failures_total` - `caatsm_nats_consumer_ack_pending` - `caatsm_nats_consumer_redelivered` diff --git a/internal/app/processor.go b/internal/app/processor.go index 755f2ae..6b25fe5 100644 --- a/internal/app/processor.go +++ b/internal/app/processor.go @@ -5,7 +5,7 @@ import ( "caatsm/internal/adapter/parser" "caatsm/internal/model" obslogging "caatsm/internal/observability/logging" - obsmetrics "caatsm/internal/observability/metrics" + "caatsm/internal/observability/telemetry" "context" "fmt" "strings" @@ -14,7 +14,6 @@ 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" ) @@ -25,6 +24,7 @@ type MessageProcessor struct { repository adapter.Repository publisher adapter.Publisher logger *zap.Logger + telemetry telemetry.Recorder } // ProcessingStatus represents the outcome of the processing pipeline @@ -38,37 +38,12 @@ const ( ProcessingStatusPublishFailed ProcessingStatus = "publish_failed" ) -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, repository adapter.Repository, publisher adapter.Publisher, + rec telemetry.Recorder, logger *zap.Logger, ) *MessageProcessor { return &MessageProcessor{ @@ -76,6 +51,7 @@ func NewMessageProcessor( repository: repository, publisher: publisher, logger: logger, + telemetry: rec, } } @@ -149,14 +125,8 @@ func (p *MessageProcessor) Handle(ctx context.Context, raw []byte, msgID string) zap.Error(parseErr), ) latency := parsed.ParsedAt.Sub(receivedAt) - parseLatencyHistogram.Record(ctx, float64(latency.Milliseconds()), - metric.WithAttributes( - messageStatusAttrKey.String(string(parsed.Status)), - messageCategoryAttrKey.String(parsed.Category), - ), - ) - obsmetrics.RecordFailure("parser") - recordProcessedMetric(ctx, parsed, latency) + p.telemetry.RecordFailure("parser") + p.telemetry.RecordProcessingResult(ctx, string(parsed.Status), parsed.Category, latency) return Permanent(fmt.Errorf("parser error: %w", parseErr)) } parsed.ErrorReason = "" @@ -178,14 +148,8 @@ func (p *MessageProcessor) Handle(ctx context.Context, raw []byte, msgID string) span.RecordError(err) span.SetStatus(codes.Error, err.Error()) latency := parsed.ParsedAt.Sub(receivedAt) - parseLatencyHistogram.Record(ctx, float64(latency.Milliseconds()), - metric.WithAttributes( - messageStatusAttrKey.String(string(parsed.Status)), - messageCategoryAttrKey.String(parsed.Category), - ), - ) - obsmetrics.RecordFailure("repository") - recordProcessedMetric(ctx, parsed, latency) + p.telemetry.RecordFailure("repository") + p.telemetry.RecordProcessingResult(ctx, string(parsed.Status), parsed.Category, latency) return fmt.Errorf("failed to insert message: %w", err) } @@ -200,20 +164,10 @@ func (p *MessageProcessor) Handle(ctx context.Context, raw []byte, msgID string) pubSpan.RecordError(err) pubSpan.SetStatus(codes.Error, err.Error()) parsed.ErrorReason = err.Error() - messagePublishFailCounter.Add(ctx, 1, - metric.WithAttributes( - messageCategoryAttrKey.String(parsed.Category), - ), - ) + p.telemetry.RecordPublishFailure(ctx, parsed.Category) latency := parsed.ParsedAt.Sub(receivedAt) - parseLatencyHistogram.Record(ctx, float64(latency.Milliseconds()), - metric.WithAttributes( - messageStatusAttrKey.String(string(parsed.Status)), - messageCategoryAttrKey.String(parsed.Category), - ), - ) - obsmetrics.RecordFailure("publisher") - recordProcessedMetric(ctx, parsed, latency) + p.telemetry.RecordFailure("publisher") + p.telemetry.RecordProcessingResult(ctx, string(parsed.Status), parsed.Category, latency) p.persistRaw(ctx, parsed) // Mark as permanent so the consumer will ack instead of retrying pubSpan.End() @@ -222,13 +176,7 @@ func (p *MessageProcessor) Handle(ctx context.Context, raw []byte, msgID string) pubSpan.End() latency := parsed.ParsedAt.Sub(receivedAt) - parseLatencyHistogram.Record(ctx, float64(latency.Milliseconds()), - metric.WithAttributes( - messageStatusAttrKey.String(string(parsed.Status)), - messageCategoryAttrKey.String(parsed.Category), - ), - ) - recordProcessedMetric(ctx, parsed, latency) + p.telemetry.RecordProcessingResult(ctx, string(parsed.Status), parsed.Category, latency) return nil } @@ -269,19 +217,3 @@ func truncateContent(content string, limit int) string { } return content[:limit-3] + "..." } - -func recordProcessedMetric(ctx context.Context, msg *model.ParsedTelegram, elapsed time.Duration) { - if msg == nil { - return - } - messageProcessedCounter.Add(ctx, 1, - metric.WithAttributes( - messageStatusAttrKey.String(string(msg.Status)), - messageCategoryAttrKey.String(msg.Category), - ), - ) - if elapsed < 0 { - elapsed = 0 - } - obsmetrics.RecordProcessed(string(msg.Status), msg.Category, elapsed) -} diff --git a/internal/app/processor_test.go b/internal/app/processor_test.go index e4b6a91..a49936e 100644 --- a/internal/app/processor_test.go +++ b/internal/app/processor_test.go @@ -9,6 +9,7 @@ import ( "caatsm/internal/adapter" "caatsm/internal/adapter/parser" "caatsm/internal/model" + "caatsm/internal/observability/telemetry" "github.com/google/uuid" . "github.com/onsi/ginkgo/v2" @@ -121,7 +122,7 @@ var _ = Describe("MessageProcessor", func() { }, err: errors.New("parse failure"), } - proc = NewMessageProcessor(parserStub, repo, pub, logger) + proc = NewMessageProcessor(parserStub, repo, pub, telemetry.NewNoop(), logger) err := proc.Handle(ctx, []byte("raw"), "msg-6") Expect(err).To(HaveOccurred()) @@ -145,7 +146,7 @@ var _ = Describe("MessageProcessor", func() { }) func newTestProcessor(p parser.Parser, repo adapter.Repository, pub adapter.Publisher) *MessageProcessor { - return NewMessageProcessor(p, repo, pub, zap.NewNop()) + return NewMessageProcessor(p, repo, pub, telemetry.NewNoop(), zap.NewNop()) } type stubParser struct { diff --git a/internal/infra/buildinfo/buildinfo.go b/internal/infra/buildinfo/buildinfo.go new file mode 100644 index 0000000..7e3af78 --- /dev/null +++ b/internal/infra/buildinfo/buildinfo.go @@ -0,0 +1,19 @@ +package buildinfo + +// Version, Commit, and BuiltAt are populated via -ldflags at build time. They +// default to development-friendly values when not provided. +// +// Example: +// go build -ldflags "\ +// -X 'caatsm/internal/infra/buildinfo.Version=v0.4.3' \ +// -X 'caatsm/internal/infra/buildinfo.Commit=abc1234' \ +// -X 'caatsm/internal/infra/buildinfo.BuiltAt=2025-11-16T08:35:00Z' \ +// " + +var ( + Version = "dev" + Commit = "unknown" + BuiltAt = "" +) + + diff --git a/internal/infra/monitoring/server.go b/internal/infra/monitoring/server.go index cd054c4..b9b747f 100644 --- a/internal/infra/monitoring/server.go +++ b/internal/infra/monitoring/server.go @@ -1,6 +1,7 @@ package monitoring import ( + "caatsm/internal/infra/buildinfo" "caatsm/internal/infra/config" obsmetrics "caatsm/internal/observability/metrics" "context" @@ -47,11 +48,13 @@ 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. + // Liveness: cheap process check that does not hit external dependencies. + mux.HandleFunc("/livez", server.handleLive) + // Backward-compatible health endpoint. For now this keeps the same + // semantics as readiness but will remain stable for existing users. mux.HandleFunc("/healthz", server.handleHealth) - // Readiness: alias to the same implementation so consumers can adopt /readyz - // without breaking existing /healthz users. + // Readiness: dependency-aware check intended for load balancers and + // orchestrators. mux.HandleFunc("/readyz", server.handleHealth) routes++ } @@ -111,33 +114,52 @@ func (s *Server) Shutdown(ctx context.Context) error { func (s *Server) handleHealth(w http.ResponseWriter, r *http.Request) { status := http.StatusOK - result := map[string]interface{}{ - "postgres": "ok", - "nats": "ok", + deps := map[string]map[string]interface{}{ + "postgres": { + "status": "ok", + }, + "nats": { + "status": "ok", + }, } ctx, cancel := context.WithTimeout(r.Context(), s.healthTimeout()) defer cancel() if s.pool == nil { - result["postgres"] = "unconfigured" - status = http.StatusServiceUnavailable - } else if err := s.pool.Ping(ctx); err != nil { - result["postgres"] = err.Error() + deps["postgres"]["status"] = "unconfigured" status = http.StatusServiceUnavailable + } else { + start := time.Now() + if err := s.pool.Ping(ctx); err != nil { + deps["postgres"]["status"] = err.Error() + status = http.StatusServiceUnavailable + } else { + deps["postgres"]["latency_ms"] = time.Since(start).Milliseconds() + } } if s.conn == nil { - result["nats"] = "unconfigured" + deps["nats"]["status"] = "unconfigured" status = http.StatusServiceUnavailable } else if s.conn.Status() != nats.CONNECTED { - result["nats"] = s.conn.Status().String() + deps["nats"]["status"] = s.conn.Status().String() status = http.StatusServiceUnavailable } + payload := map[string]interface{}{ + "status": httpStatusLabel(status), + "build": map[string]interface{}{ + "version": buildinfo.Version, + "rev": buildinfo.Commit, + "built_at": buildinfo.BuiltAt, + }, + "dependencies": deps, + } + w.Header().Set("Content-Type", "application/json") w.WriteHeader(status) - _ = json.NewEncoder(w).Encode(result) + _ = json.NewEncoder(w).Encode(payload) } func (s *Server) healthTimeout() time.Duration { @@ -147,3 +169,28 @@ func (s *Server) healthTimeout() time.Duration { } return timeout } + +// handleLive reports basic process liveness and build information without +// consulting external dependencies. It is suitable for liveness probes. +func (s *Server) handleLive(w http.ResponseWriter, r *http.Request) { + payload := map[string]interface{}{ + "status": "ok", + "build": map[string]interface{}{ + "version": buildinfo.Version, + "rev": buildinfo.Commit, + "built_at": buildinfo.BuiltAt, + }, + } + + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusOK) + _ = json.NewEncoder(w).Encode(payload) +} + +func httpStatusLabel(code int) string { + if code >= 200 && code < 300 { + return "ok" + } + return "error" +} + diff --git a/internal/infra/nats/consumer.go b/internal/infra/nats/consumer.go index c4d7d2b..b9d16ad 100644 --- a/internal/infra/nats/consumer.go +++ b/internal/infra/nats/consumer.go @@ -5,6 +5,7 @@ import ( "caatsm/internal/infra/config" obslogging "caatsm/internal/observability/logging" obsmetrics "caatsm/internal/observability/metrics" + "caatsm/internal/observability/telemetry" "context" "encoding/json" "errors" @@ -28,6 +29,7 @@ type Consumer struct { processor *app.MessageProcessor cfg *config.Config logger *zap.Logger + telemetry telemetry.Recorder subject string consumerName string mode string @@ -53,6 +55,7 @@ func ProvideConsumer( js nats.JetStreamContext, processor *app.MessageProcessor, cfg *config.Config, + rec telemetry.Recorder, logger *zap.Logger, ) (*Consumer, error) { subject := cfg.EffectiveSubscriptionTopic() @@ -108,6 +111,7 @@ func ProvideConsumer( processor: processor, cfg: cfg, logger: logger, + telemetry: rec, subject: subject, consumerName: consumerName, mode: mode, @@ -238,7 +242,7 @@ func (c *Consumer) validateDLQ() { // Ensure the DLQ subject is actually bound to a JetStream stream. This avoids // the opaque `nats: no response from stream` error later when publishing. - obsmetrics.RecordJSAPICall("dlq_validate_stream") + c.telemetry.RecordJSAPICall("dlq_validate_stream") streamName, err := c.js.StreamNameBySubject(subject) if err != nil || strings.TrimSpace(streamName) == "" { c.logger.Warn("DLQ subject not bound to any JetStream stream; DLQ routing disabled", @@ -342,7 +346,7 @@ func (c *Consumer) startJetStream(ctx context.Context) error { if isPermanent { result = obsmetrics.ResultPermanentFail } - obsmetrics.RecordMessageHandled(c.streamName, c.consumerName, result, elapsed) + c.telemetry.RecordMessageHandled(ctx, c.streamName, c.consumerName, result, elapsed) if isPermanent { c.consecutiveProcessErrors = 0 @@ -376,7 +380,7 @@ func (c *Consumer) startJetStream(ctx context.Context) error { } // Transient error: request redelivery with optional delay - obsmetrics.RecordRetry(c.streamName, c.consumerName, obsmetrics.RetryReasonProcessorError) + c.telemetry.RecordRetry(ctx, c.streamName, c.consumerName, obsmetrics.RetryReasonProcessorError) if nakErr := c.nakWithStrategy(msg); nakErr != nil { c.logger.Error("Failed to NAK message", zap.Error(nakErr)) } @@ -393,7 +397,7 @@ func (c *Consumer) startJetStream(ctx context.Context) error { c.logger.Error("Failed to ACK message", zap.Error(ackErr)) } else { elapsed := time.Since(start) - obsmetrics.RecordMessageHandled(c.streamName, c.consumerName, "ok", elapsed) + c.telemetry.RecordMessageHandled(ctx, c.streamName, c.consumerName, "ok", elapsed) } } } @@ -531,6 +535,10 @@ func (c *Consumer) recordConsumerMetrics(ctx context.Context, info *nats.Consume if c.delivered != nil { c.delivered.Record(ctx, int64(info.Delivered.Stream)) } + + // Export an explicit pending messages gauge for Prometheus-based lag / + // backlog alerts. + obsmetrics.RecordNATSConsumerPending(c.streamName, c.consumerName, info.NumPending) } // routeToDLQ publishes a copy of the failed message to the configured DLQ subject, @@ -578,14 +586,14 @@ func (c *Consumer) routeToDLQ(ctx context.Context, msg *nats.Msg, cause error) e // unavailable. Surface this explicitly to make operational diagnosis // easier. if errors.Is(err, nats.ErrNoResponders) { - obsmetrics.RecordDLQPublishFailure(c.streamName, c.consumerName) + c.telemetry.RecordDLQPublishFailure(ctx, c.streamName, c.consumerName) return fmt.Errorf("publish to dlq subject %s: no JetStream stream found for subject or JetStream unavailable: %w", c.dlqSubject, err) } - obsmetrics.RecordDLQPublishFailure(c.streamName, c.consumerName) + c.telemetry.RecordDLQPublishFailure(ctx, c.streamName, c.consumerName) return fmt.Errorf("publish to dlq subject %s: %w", c.dlqSubject, err) } - obsmetrics.RecordDLQMessage(c.streamName, c.consumerName) + c.telemetry.RecordDLQMessage(ctx, c.streamName, c.consumerName) return nil } diff --git a/internal/observability/metrics/metrics.go b/internal/observability/metrics/metrics.go index fc2aad0..1068773 100644 --- a/internal/observability/metrics/metrics.go +++ b/internal/observability/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" + MetricNATSConsumerPending = "caatsm_nats_consumer_pending_messages" // Common label keys. LabelStatus = "status" @@ -72,6 +73,9 @@ var ( // Database metrics. dbQueriesTotal *prometheus.CounterVec dbQueryLatency *prometheus.HistogramVec + + // NATS consumer lag metrics. + natsConsumerPending *prometheus.GaugeVec ) func initCollectors() { @@ -138,6 +142,11 @@ func initCollectors() { Buckets: prometheus.DefBuckets, }, []string{LabelOperation}) + natsConsumerPending = prometheus.NewGaugeVec(prometheus.GaugeOpts{ + Name: MetricNATSConsumerPending, + Help: "Approximate number of pending messages for a JetStream consumer, labelled by stream and consumer.", + }, []string{LabelStream, LabelConsumer}) + registry.MustRegister( processedCounter, failureCounter, @@ -150,6 +159,7 @@ func initCollectors() { dlqPublishFailures, dbQueriesTotal, dbQueryLatency, + natsConsumerPending, ) } @@ -226,6 +236,13 @@ func RecordJSAPICall(operation string) { jsAPICallsTotal.WithLabelValues(labelValue(operation)).Inc() } +// RecordNATSConsumerPending records the current pending message count for a +// JetStream consumer as a gauge, enabling backlog / lag alerts. +func RecordNATSConsumerPending(stream, consumer string, pending uint64) { + ensureCollectors() + natsConsumerPending.WithLabelValues(labelValue(stream), labelValue(consumer)).Set(float64(pending)) +} + func labelValue(value string) string { value = strings.TrimSpace(value) if value == "" { diff --git a/internal/observability/telemetry/telemetry.go b/internal/observability/telemetry/telemetry.go new file mode 100644 index 0000000..9a4e1ab --- /dev/null +++ b/internal/observability/telemetry/telemetry.go @@ -0,0 +1,307 @@ +package telemetry + +import ( + "caatsm/internal/infra/config" + obsmetrics "caatsm/internal/observability/metrics" + "context" + "time" + + "go.opentelemetry.io/otel" + "go.opentelemetry.io/otel/attribute" + "go.opentelemetry.io/otel/metric" +) + +// Recorder provides a thin abstraction over telemetry backends (OpenTelemetry, +// Prometheus, etc.) so that application code does not need to import concrete +// metric libraries directly. +type Recorder interface { + // RecordProcessingResult captures the final processing status of a telegram + // along with the parser latency. + RecordProcessingResult(ctx context.Context, status, category string, parseLatency time.Duration) + + // RecordPublishFailure increments the publish failure counter for the given + // category. + RecordPublishFailure(ctx context.Context, category string) + + // RecordFailure records a high-level failure bucket (parser, repository, + // publisher, etc.). + RecordFailure(stage string) + + // RecordMessageHandled tracks end-to-end message handling for a particular + // stream/consumer pair. + RecordMessageHandled(ctx context.Context, stream, consumer, result string, elapsed time.Duration) + + // RecordRetry records a retry (negative acknowledgement) reason. + RecordRetry(ctx context.Context, stream, consumer, reason string) + + // RecordDLQMessage records a successfully routed DLQ message. + RecordDLQMessage(ctx context.Context, stream, consumer string) + + // RecordDLQPublishFailure records a DLQ publish failure. + RecordDLQPublishFailure(ctx context.Context, stream, consumer string) + + // RecordJSAPICall records a JetStream API call. + RecordJSAPICall(operation string) +} + +// ProvideRecorder wires a composite Recorder based on configuration flags. +// - When telemetry is enabled, an OpenTelemetry-backed recorder is included. +// - When metrics are enabled, a Prometheus-backed recorder is included. +// - When neither is enabled, a noop recorder is returned. +func ProvideRecorder(cfg *config.Config) Recorder { + if cfg == nil { + return NewNoop() + } + + var recorders []Recorder + + if cfg.Telemetry.Enabled { + recorders = append(recorders, newOTelRecorder()) + } + + if !cfg.Monitoring.Disabled && cfg.Monitoring.EnableMetrics { + recorders = append(recorders, newPromRecorder()) + } + + if len(recorders) == 0 { + return NewNoop() + } + return NewComposite(recorders...) +} + +// noopRecorder implements Recorder but performs no operations. +type noopRecorder struct{} + +func NewNoop() Recorder { + return &noopRecorder{} +} + +func (n *noopRecorder) RecordProcessingResult(ctx context.Context, status, category string, parseLatency time.Duration) { +} + +func (n *noopRecorder) RecordPublishFailure(ctx context.Context, category string) { +} + +func (n *noopRecorder) RecordFailure(stage string) { +} + +func (n *noopRecorder) RecordMessageHandled(ctx context.Context, stream, consumer, result string, elapsed time.Duration) { +} + +func (n *noopRecorder) RecordRetry(ctx context.Context, stream, consumer, reason string) { +} + +func (n *noopRecorder) RecordDLQMessage(ctx context.Context, stream, consumer string) { +} + +func (n *noopRecorder) RecordDLQPublishFailure(ctx context.Context, stream, consumer string) { +} + +func (n *noopRecorder) RecordJSAPICall(operation string) { +} + +// compositeRecorder fans out all calls to a slice of underlying recorders. +type compositeRecorder struct { + recorders []Recorder +} + +func NewComposite(recorders ...Recorder) Recorder { + // Filter out nils defensively. + var filtered []Recorder + for _, r := range recorders { + if r != nil { + filtered = append(filtered, r) + } + } + if len(filtered) == 0 { + return NewNoop() + } + return &compositeRecorder{recorders: filtered} +} + +func (c *compositeRecorder) RecordProcessingResult(ctx context.Context, status, category string, parseLatency time.Duration) { + for _, r := range c.recorders { + r.RecordProcessingResult(ctx, status, category, parseLatency) + } +} + +func (c *compositeRecorder) RecordPublishFailure(ctx context.Context, category string) { + for _, r := range c.recorders { + r.RecordPublishFailure(ctx, category) + } +} + +func (c *compositeRecorder) RecordFailure(stage string) { + for _, r := range c.recorders { + r.RecordFailure(stage) + } +} + +func (c *compositeRecorder) RecordMessageHandled(ctx context.Context, stream, consumer, result string, elapsed time.Duration) { + for _, r := range c.recorders { + r.RecordMessageHandled(ctx, stream, consumer, result, elapsed) + } +} + +func (c *compositeRecorder) RecordRetry(ctx context.Context, stream, consumer, reason string) { + for _, r := range c.recorders { + r.RecordRetry(ctx, stream, consumer, reason) + } +} + +func (c *compositeRecorder) RecordDLQMessage(ctx context.Context, stream, consumer string) { + for _, r := range c.recorders { + r.RecordDLQMessage(ctx, stream, consumer) + } +} + +func (c *compositeRecorder) RecordDLQPublishFailure(ctx context.Context, stream, consumer string) { + for _, r := range c.recorders { + r.RecordDLQPublishFailure(ctx, stream, consumer) + } +} + +func (c *compositeRecorder) RecordJSAPICall(operation string) { + for _, r := range c.recorders { + r.RecordJSAPICall(operation) + } +} + +// promRecorder delegates to the Prometheus metrics helpers in the +// internal/observability/metrics package. +type promRecorder struct{} + +func newPromRecorder() Recorder { + return &promRecorder{} +} + +func (p *promRecorder) RecordProcessingResult(ctx context.Context, status, category string, parseLatency time.Duration) { + if parseLatency < 0 { + parseLatency = 0 + } + obsmetrics.RecordProcessed(status, category, parseLatency) +} + +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. + obsmetrics.RecordFailure("publisher") +} + +func (p *promRecorder) RecordFailure(stage string) { + obsmetrics.RecordFailure(stage) +} + +func (p *promRecorder) RecordMessageHandled(ctx context.Context, stream, consumer, result string, elapsed time.Duration) { + obsmetrics.RecordMessageHandled(stream, consumer, result, elapsed) +} + +func (p *promRecorder) RecordRetry(ctx context.Context, stream, consumer, reason string) { + obsmetrics.RecordRetry(stream, consumer, reason) +} + +func (p *promRecorder) RecordDLQMessage(ctx context.Context, stream, consumer string) { + obsmetrics.RecordDLQMessage(stream, consumer) +} + +func (p *promRecorder) RecordDLQPublishFailure(ctx context.Context, stream, consumer string) { + obsmetrics.RecordDLQPublishFailure(stream, consumer) +} + +func (p *promRecorder) RecordJSAPICall(operation string) { + obsmetrics.RecordJSAPICall(operation) +} + +// otelRecorder creates and records OpenTelemetry metrics for the CAATSM +// processor. It intentionally focuses on a small set of high-value metrics to +// avoid duplicating the full Prometheus surface. +type otelRecorder struct { + meter metric.Meter + + messageStatusAttrKey attribute.Key + messageCategoryAttrKey attribute.Key + + messageProcessedCounter metric.Int64Counter + messagePublishFailCounter metric.Int64Counter + parseLatencyHistogram metric.Float64Histogram +} + +func newOTelRecorder() Recorder { + meter := otel.Meter("caatsm/app") + + statusKey := attribute.Key("message.status") + categoryKey := attribute.Key("message.category") + + messageProcessedCounter, _ := meter.Int64Counter( + "caatsm_messages_processed_total", + metric.WithDescription("Total number of telegrams processed by the CAATSM processor."), + ) + messagePublishFailCounter, _ := meter.Int64Counter( + "caatsm_publish_failures_total", + metric.WithDescription("Total number of telegram publish failures."), + ) + parseLatencyHistogram, _ := meter.Float64Histogram( + "caatsm_parse_duration_seconds", + metric.WithDescription("Latency of parsing a telegram, in seconds."), + metric.WithUnit("s"), + ) + + return &otelRecorder{ + meter: meter, + messageStatusAttrKey: statusKey, + messageCategoryAttrKey: categoryKey, + messageProcessedCounter: messageProcessedCounter, + messagePublishFailCounter: messagePublishFailCounter, + parseLatencyHistogram: parseLatencyHistogram, + } +} + +func (o *otelRecorder) RecordProcessingResult(ctx context.Context, status, category string, parseLatency time.Duration) { + if parseLatency < 0 { + parseLatency = 0 + } + o.messageProcessedCounter.Add(ctx, 1, + metric.WithAttributes( + o.messageStatusAttrKey.String(status), + o.messageCategoryAttrKey.String(category), + ), + ) + o.parseLatencyHistogram.Record(ctx, parseLatency.Seconds(), + metric.WithAttributes( + o.messageStatusAttrKey.String(status), + o.messageCategoryAttrKey.String(category), + ), + ) +} + +func (o *otelRecorder) RecordPublishFailure(ctx context.Context, category string) { + o.messagePublishFailCounter.Add(ctx, 1, + metric.WithAttributes( + o.messageCategoryAttrKey.String(category), + ), + ) +} + +func (o *otelRecorder) RecordFailure(stage string) { + // OpenTelemetry does not currently publish a dedicated failure counter; the + // Prometheus surface captures this. This method is a no-op here. +} + +func (o *otelRecorder) RecordMessageHandled(ctx context.Context, stream, consumer, result string, elapsed time.Duration) { + // High-cardinality stream/consumer labels are exposed via Prometheus + // metrics; OTEL can rely on traces and existing consumer metrics. +} + +func (o *otelRecorder) RecordRetry(ctx context.Context, stream, consumer, reason string) { +} + +func (o *otelRecorder) RecordDLQMessage(ctx context.Context, stream, consumer string) { +} + +func (o *otelRecorder) RecordDLQPublishFailure(ctx context.Context, stream, consumer string) { +} + +func (o *otelRecorder) RecordJSAPICall(operation string) { +} + + diff --git a/pkg/di/wire.go b/pkg/di/wire.go index e3b2450..6296de6 100644 --- a/pkg/di/wire.go +++ b/pkg/di/wire.go @@ -11,6 +11,7 @@ import ( "caatsm/internal/infra/monitoring" "caatsm/internal/infra/nats" "caatsm/internal/infra/postgres" + "caatsm/internal/observability/telemetry" "github.com/google/wire" ) @@ -49,6 +50,9 @@ var runtimeSet = wire.NewSet( // Parser parser.ProvideParser, + // Telemetry + telemetry.ProvideRecorder, + // App app.NewMessageProcessor, diff --git a/pkg/di/wire_gen.go b/pkg/di/wire_gen.go index 01fc818..7828639 100644 --- a/pkg/di/wire_gen.go +++ b/pkg/di/wire_gen.go @@ -14,6 +14,7 @@ import ( "caatsm/internal/infra/monitoring" "caatsm/internal/infra/nats" "caatsm/internal/infra/postgres" + "caatsm/internal/observability/telemetry" "github.com/google/wire" ) @@ -49,8 +50,9 @@ func buildAppComponents() (*appComponents, error) { if err != nil { return nil, err } - messageProcessor := app.NewMessageProcessor(parserParser, repository, publisher, logger) - consumer, err := nats.ProvideConsumer(conn, jetStreamContext, messageProcessor, configConfig, logger) + recorder := telemetry.ProvideRecorder(configConfig) + messageProcessor := app.NewMessageProcessor(parserParser, repository, publisher, recorder, logger) + consumer, err := nats.ProvideConsumer(conn, jetStreamContext, messageProcessor, configConfig, recorder, logger) if err != nil { return nil, err } @@ -92,8 +94,9 @@ func buildAppComponentsWithConfig(cfg *config.Config) (*appComponents, error) { if err != nil { return nil, err } - messageProcessor := app.NewMessageProcessor(parserParser, repository, publisher, logger) - consumer, err := nats.ProvideConsumer(conn, jetStreamContext, messageProcessor, cfg, logger) + recorder := telemetry.ProvideRecorder(cfg) + messageProcessor := app.NewMessageProcessor(parserParser, repository, publisher, recorder, logger) + consumer, err := nats.ProvideConsumer(conn, jetStreamContext, messageProcessor, cfg, recorder, logger) if err != nil { return nil, err } @@ -129,7 +132,7 @@ func InitializeAppWithConfig(cfg *config.Config) (*app.MessageProcessor, *nats.C return comps.Processor, comps.Consumer, comps.Monitoring, nil } -var runtimeSet = wire.NewSet(log.ProvideLogger, postgres.ProvideDB, postgres.ProvideRepository, nats.ProvideNATSConn, nats.ProvideJetStream, nats.ProvidePublisher, parser.ProvideParser, app.NewMessageProcessor, nats.ProvideConsumer, monitoring.ProvideServer) +var runtimeSet = wire.NewSet(log.ProvideLogger, postgres.ProvideDB, postgres.ProvideRepository, nats.ProvideNATSConn, nats.ProvideJetStream, nats.ProvidePublisher, parser.ProvideParser, telemetry.ProvideRecorder, app.NewMessageProcessor, nats.ProvideConsumer, monitoring.ProvideServer) type appComponents struct { Processor *app.MessageProcessor