diff --git a/.cursor/rules/do.mdc b/.cursor/rules/do.mdc new file mode 100644 index 0000000..562b88b --- /dev/null +++ b/.cursor/rules/do.mdc @@ -0,0 +1,95 @@ +You are an expert in Go, microservices architecture, and clean backend development practices. Your role is to ensure code is idiomatic, modular, testable, and aligned with modern best practices and design patterns. + +### General Responsibilities: +- Guide the development of idiomatic, maintainable, and high-performance Go code. +- Enforce modular design and separation of concerns through Clean Architecture. +- Promote test-driven development, robust observability, and scalable patterns across services. + +### Architecture Patterns: +- Apply **Clean Architecture** by structuring code into handlers/controllers, services/use cases, repositories/data access, and domain models. +- Use **domain-driven design** principles where applicable. +- Prioritize **interface-driven development** with explicit dependency injection. +- Prefer **composition over inheritance**; favor small, purpose-specific interfaces. +- Ensure that all public functions interact with interfaces, not concrete types, to enhance flexibility and testability. + +### Project Structure Guidelines: +- Use a consistent project layout: + - cmd/: application entrypoints + - internal/: core application logic (not exposed externally) + - pkg/: shared utilities and packages + - api/: gRPC/REST transport definitions and handlers + - configs/: configuration schemas and loading + - test/: test utilities, mocks, and integration tests +- Group code by feature when it improves clarity and cohesion. +- Keep logic decoupled from framework-specific code. + +### Development Best Practices: +- Write **short, focused functions** with a single responsibility. +- Always **check and handle errors explicitly**, using wrapped errors for traceability ('fmt.Errorf("context: %w", err)'). +- Avoid **global state**; use constructor functions to inject dependencies. +- Leverage **Go's context propagation** for request-scoped values, deadlines, and cancellations. +- Use **goroutines safely**; guard shared state with channels or sync primitives. +- **Defer closing resources** and handle them carefully to avoid leaks. + +### Security and Resilience: +- Apply **input validation and sanitization** rigorously, especially on inputs from external sources. +- Use secure defaults for **JWT, cookies**, and configuration settings. +- Isolate sensitive operations with clear **permission boundaries**. +- Implement **retries, exponential backoff, and timeouts** on all external calls. +- Use **circuit breakers and rate limiting** for service protection. +- Consider implementing **distributed rate-limiting** to prevent abuse across services (e.g., using Redis). + +### Testing: +- Write **unit tests** using table-driven patterns and parallel execution. +- **Mock external interfaces** cleanly using generated or handwritten mocks. +- Separate **fast unit tests** from slower integration and E2E tests. +- Ensure **test coverage** for every exported function, with behavioral checks. +- Use tools like 'go test -cover' to ensure adequate test coverage. + +### Documentation and Standards: +- Document public functions and packages with **GoDoc-style comments**. +- Provide concise **READMEs** for services and libraries. +- Maintain a 'CONTRIBUTING.md' and 'ARCHITECTURE.md' to guide team practices. +- Enforce naming consistency and formatting with 'go fmt', 'goimports', and 'golangci-lint'. + +### Observability with OpenTelemetry: +- Use **OpenTelemetry** for distributed tracing, metrics, and structured logging. +- Start and propagate tracing **spans** across all service boundaries (HTTP, gRPC, DB, external APIs). +- Always attach 'context.Context' to spans, logs, and metric exports. +- Use **otel.Tracer** for creating spans and **otel.Meter** for collecting metrics. +- Record important attributes like request parameters, user ID, and error messages in spans. +- Use **log correlation** by injecting trace IDs into structured logs. +- Export data to **OpenTelemetry Collector**, **Jaeger**, or **Prometheus**. + +### Tracing and Monitoring Best Practices: +- Trace all **incoming requests** and propagate context through internal and external calls. +- Use **middleware** to instrument HTTP and gRPC endpoints automatically. +- Annotate slow, critical, or error-prone paths with **custom spans**. +- Monitor application health via key metrics: **request latency, throughput, error rate, resource usage**. +- Define **SLIs** (e.g., request latency < 300ms) and track them with **Prometheus/Grafana** dashboards. +- Alert on key conditions (e.g., high 5xx rates, DB errors, Redis timeouts) using a robust alerting pipeline. +- Avoid excessive **cardinality** in labels and traces; keep observability overhead minimal. +- Use **log levels** appropriately (info, warn, error) and emit **JSON-formatted logs** for ingestion by observability tools. +- Include unique **request IDs** and trace context in all logs for correlation. + +### Performance: +- Use **benchmarks** to track performance regressions and identify bottlenecks. +- Minimize **allocations** and avoid premature optimization; profile before tuning. +- Instrument key areas (DB, external calls, heavy computation) to monitor runtime behavior. + +### Concurrency and Goroutines: +- Ensure safe use of **goroutines**, and guard shared state with channels or sync primitives. +- Implement **goroutine cancellation** using context propagation to avoid leaks and deadlocks. + +### Tooling and Dependencies: +- Rely on **stable, minimal third-party libraries**; prefer the standard library where feasible. +- Use **Go modules** for dependency management and reproducibility. +- Version-lock dependencies for deterministic builds. +- Integrate **linting, testing, and security checks** in CI pipelines. + +### Key Conventions: +1. Prioritize **readability, simplicity, and maintainability**. +2. Design for **change**: isolate business logic and minimize framework lock-in. +3. Emphasize clear **boundaries** and **dependency inversion**. +4. Ensure all behavior is **observable, testable, and documented**. +5. **Automate workflows** for testing, building, and deployment. diff --git a/AGENTS.md b/AGENTS.md index fccec43..7bca49f 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -18,12 +18,21 @@ - **Generated code**: Never edit `/pkg/di/wire_gen.go` or other generated files - **Linting**: `golangci-lint run ./...` required; fix all issues before PR -## Testing Guidelines +## Testing & Architecture - **Unit tests**: Ginkgo BDD style next to implementation (`*_test.go`); declarative descriptions - **Integration**: Testcontainers in `test/integration`; spin up NATS/TimescaleDB - **Coverage**: Run `make coverage` before merging; address regressions - -## Architecture - **Structure**: Clean Architecture - `domain` (business logic), `app` (use cases), `adapter` (I/O), `infra` (framework deps) - **Entry point**: `cmd/main` - **Config**: `configs/config..toml`; secrets via `CAATSM_*` env vars + +## Cursor Rules (.cursor/rules/do.mdc) +- **Expertise**: Go, microservices, Clean Architecture, test-driven development +- **Architecture**: Clean Architecture with domain-driven design, interface-driven development +- **Project Structure**: cmd/, internal/, pkg/, api/, configs/, test/ layout +- **Best Practices**: Short focused functions, explicit error handling, context propagation, goroutine safety +- **Security**: Input validation, secure defaults, retries/backoff, circuit breakers +- **Testing**: Table-driven unit tests, mock interfaces, separate fast/slow tests +- **Observability**: Production-ready OpenTelemetry with environment-based sampling, comprehensive resource attributes, semantic span conventions, and dual telemetry (OTEL + Prometheus) +- **Performance**: Benchmarks, minimize allocations, profile before optimization +- **Tooling**: Go modules, linting, CI automation diff --git a/README.md b/README.md index 85533b2..0d4ad58 100644 --- a/README.md +++ b/README.md @@ -143,6 +143,11 @@ enabled = false endpoint = "http://otel-collector:4318" insecure = true +# Production configuration example: +# enabled = true +# endpoint = "otel-collector.company.com:4318" +# insecure = false # Use TLS in production + ### Timeouts and Ack Wait `[timeouts]` is optional, but if you plan to tune JetStream redelivery you should set `timeouts.ack_wait` and/or `[nats.consumer].ack_wait`. When neither is specified the application defaults both values to `30s`, ensuring predictable redelivery timing. @@ -470,20 +475,28 @@ Critical overrides stay available through CLI flags; advanced tuning such as str ### Observability -The processor exposes three complementary observability surfaces: +The processor exposes three complementary observability surfaces with production-ready OpenTelemetry implementation: 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. + - **Production-ready setup** with environment-based sampling, comprehensive resource attributes, and optimized batching. + - Enable via `[telemetry] enabled = true` and set `endpoint` to your OTLP/HTTP collector (e.g., `http://otel-collector:4318`). + - **Sampling strategy**: + - Production: 1% sampling (cost-effective) + - Staging: 10% sampling (balanced observability) + - Development/Test: 100% sampling (full debugging) + - CLI overrides: + - `--telemetry-enabled` toggles exporters on/off. + - `--telemetry-endpoint` and `--telemetry-insecure` adjust the OTLP HTTP endpoint and TLS behavior. + - **Comprehensive traces** with semantic attributes: + - `caatsm/app`: Message processing spans with `messaging.system`, `messaging.operation`, `caatsm.message.category` + - `caatsm/postgres`: Database operations with `db.system`, `db.operation`, `db.table` + - `caatsm/nats`: NATS operations with `messaging.destination`, `messaging.consumer.id` + - **Business metrics** (focused set for OTEL): + - `caatsm_messages_processed_total` (counter, by `message.status` / `message.category`) + - `caatsm_publish_failures_total` (counter) + - `caatsm_parse_duration_seconds` (histogram) + - **Resource attributes** include service metadata, environment, build info, and infrastructure details. + - Application code records telemetry via a thin `telemetry.Recorder` abstraction, which fans out to OTEL and Prometheus backends as configured. 2. **Prometheus metrics (`/metrics`)** - Implemented in `internal/infra/metrics` and considered the primary source for SRE PromQL/SLOs. diff --git a/cmd/main/main.go b/cmd/main/main.go index 50d87e5..bfa2491 100644 --- a/cmd/main/main.go +++ b/cmd/main/main.go @@ -4,6 +4,7 @@ import ( "caatsm/internal/infra/config" "caatsm/pkg/di" "context" + "crypto/tls" "errors" "fmt" "os" @@ -204,7 +205,10 @@ func runListen(parentCtx context.Context, cfg *config.Config) error { } // After cancellation, give the consumer a chance to finish cleanup. - waitTimeout := 5 * time.Second + // Consumer checks context before fetch and between messages, so it should exit quickly. + // Worst case: finishing current message (up to 1s for slow DB) + cleanup (~100ms) + // 1.5s provides safe buffer while keeping shutdown responsive. + waitTimeout := 1500 * time.Millisecond select { case err := <-errChan: if err != nil && !errors.Is(err, context.Canceled) { @@ -215,7 +219,10 @@ func runListen(parentCtx context.Context, cfg *config.Config) error { zap.Duration("timeout", waitTimeout)) } - if err := consumer.Shutdown(context.Background()); err != nil { + // Use a short timeout for shutdown since consumer should already be stopped + shutdownCtx, shutdownCancel := context.WithTimeout(context.Background(), 1*time.Second) + defer shutdownCancel() + if err := consumer.Shutdown(shutdownCtx); err != nil { if runErr == nil { runErr = fmt.Errorf("failed to drain NATS connection: %w", err) } @@ -329,17 +336,29 @@ 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 cfg.Telemetry.Insecure { - traceOpts = append(traceOpts, otlptracehttp.WithInsecure()) - metricOpts = append(metricOpts, otlpmetrichttp.WithInsecure()) + if tlsConfig != nil { + metricOpts = append(metricOpts, otlpmetrichttp.WithTLSClientConfig(tlsConfig)) } traceExporter, err := otlptracehttp.New(ctx, traceOpts...) @@ -351,31 +370,53 @@ func initTelemetry(ctx context.Context, cfg *config.Config) (func(context.Contex return nil, fmt.Errorf("init metric exporter: %w", err) } + // Get environment for sampling configuration env := os.Getenv("GO_ENV") if env == "" { env = "dev" } + + // Create comprehensive resource with service information res, err := resource.New(ctx, resource.WithFromEnv(), resource.WithProcess(), resource.WithOS(), resource.WithHost(), + resource.WithContainer(), resource.WithAttributes( semconv.ServiceName("caatsm"), + semconv.ServiceVersion("dev"), // TODO: Use build info + semconv.ServiceNamespace("airport"), + attribute.String("service.component", "receiver"), attribute.String("deployment.environment", env), + attribute.String("telemetry.endpoint", cfg.Telemetry.Endpoint), + attribute.Bool("telemetry.insecure", cfg.Telemetry.Insecure), ), ) if err != nil { return nil, fmt.Errorf("build telemetry resource: %w", err) } + // Configure sampling based on environment + sampler := getSamplerForEnvironment(env) + + // Configure tracer provider with batching and sampling tp := sdktrace.NewTracerProvider( - sdktrace.WithBatcher(traceExporter), + sdktrace.WithBatcher(traceExporter, + sdktrace.WithBatchTimeout(1*time.Second), + sdktrace.WithMaxExportBatchSize(512), + sdktrace.WithMaxQueueSize(2048), + ), sdktrace.WithResource(res), + sdktrace.WithSampler(sdktrace.ParentBased(sampler)), ) + + // Configure meter provider with periodic reader mp := sdkmetric.NewMeterProvider( sdkmetric.WithResource(res), - sdkmetric.WithReader(sdkmetric.NewPeriodicReader(metricExporter)), + sdkmetric.WithReader(sdkmetric.NewPeriodicReader(metricExporter, + sdkmetric.WithInterval(30*time.Second), + )), ) otel.SetTracerProvider(tp) @@ -395,3 +436,21 @@ func initTelemetry(ctx context.Context, cfg *config.Config) (func(context.Contex return shutdown, nil } + +// getSamplerForEnvironment returns appropriate sampling strategy for each environment +func getSamplerForEnvironment(env string) sdktrace.Sampler { + switch env { + case "prod", "production": + // 1% sampling in production to control costs and performance + return sdktrace.TraceIDRatioBased(0.01) + case "staging": + // 10% sampling in staging for better observability + return sdktrace.TraceIDRatioBased(0.1) + case "test", "testing": + // Always sample in testing for complete coverage + return sdktrace.AlwaysSample() + default: + // 100% sampling in development for debugging + return sdktrace.AlwaysSample() + } +} diff --git a/configs/otel-collector.dev.yaml b/configs/otel-collector.dev.yaml index 360e050..7571bdb 100644 --- a/configs/otel-collector.dev.yaml +++ b/configs/otel-collector.dev.yaml @@ -3,27 +3,66 @@ receivers: protocols: 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: + batch: + send_batch_size: 1024 + timeout: 1s + send_batch_max_size: 2048 + + resource: + attributes: + - key: service.instance.id + value: "${POD_NAME}" + action: upsert + - key: k8s.pod.name + value: "${POD_NAME}" + action: upsert + - key: k8s.namespace.name + value: "${NAMESPACE}" + action: upsert exporters: logging: loglevel: info + sampling_initial: 10 + sampling_thereafter: 100 + otlphttp/jaeger: endpoint: http://jaeger:4318 tls: insecure: true + sending_queue: + queue_size: 10000 + retry_on_failure: + enabled: true + initial_interval: 5s + max_interval: 30s + max_elapsed_time: 300s + prometheus: endpoint: "0.0.0.0:8889" const_labels: source: "otel-collector" + sending_queue: + queue_size: 10000 + retry_on_failure: + enabled: true service: pipelines: traces: receivers: [otlp] + processors: [resource, batch] exporters: [logging, otlphttp/jaeger] metrics: receivers: [otlp] + processors: [resource, batch] exporters: [logging, prometheus] diff --git a/configs/otel-collector.prod.yaml b/configs/otel-collector.prod.yaml new file mode 100644 index 0000000..0055f68 --- /dev/null +++ b/configs/otel-collector.prod.yaml @@ -0,0 +1,85 @@ +receivers: + otlp: + protocols: + grpc: + endpoint: 0.0.0.0:4317 + tls: + cert_file: /etc/ssl/certs/otel-collector.crt + key_file: /etc/ssl/private/otel-collector.key + auth: + authenticator: bearer_token + http: + endpoint: 0.0.0.0:4318 + tls: + cert_file: /etc/ssl/certs/otel-collector.crt + key_file: /etc/ssl/private/otel-collector.key + auth: + authenticator: bearer_token + +processors: + batch: + send_batch_size: 1024 + timeout: 1s + send_batch_max_size: 2048 + + resource: + attributes: + - key: service.instance.id + value: "${POD_NAME}" + action: upsert + - key: k8s.pod.name + value: "${POD_NAME}" + action: upsert + - key: k8s.namespace.name + value: "${NAMESPACE}" + action: upsert + +extensions: + health_check: + endpoint: 0.0.0.0:13133 + pprof: + endpoint: :1888 + zpages: + endpoint: :55679 + bearer_token: + token: ${OTEL_COLLECTOR_BEARER_TOKEN} + +exporters: + otlphttp/jaeger: + endpoint: https://jaeger.production.company.com:4318 + headers: + authorization: "Bearer ${JAEGER_API_TOKEN}" + tls: + insecure: false + ca_file: /etc/ssl/certs/ca.pem + sending_queue: + queue_size: 10000 + retry_on_failure: + enabled: true + initial_interval: 5s + max_interval: 30s + max_elapsed_time: 300s + + prometheusremotewrite: + endpoint: https://prometheus-prod.company.com/api/v1/write + headers: + authorization: "Bearer ${PROMETHEUS_API_TOKEN}" + tls: + insecure: false + ca_file: /etc/ssl/certs/ca.pem + sending_queue: + queue_size: 10000 + retry_on_failure: + enabled: true + +service: + extensions: [health_check, pprof, zpages, bearer_token] + pipelines: + traces: + receivers: [otlp] + processors: [resource, batch] + exporters: [otlphttp/jaeger] + metrics: + receivers: [otlp] + processors: [resource, batch] + exporters: [prometheusremotewrite] \ No newline at end of file diff --git a/docs/nats.md b/docs/nats.md new file mode 100644 index 0000000..047d8d6 --- /dev/null +++ b/docs/nats.md @@ -0,0 +1,323 @@ +# NATS Integration Architecture + +## Overview + +The NATS integration provides a robust, production-ready message processing system built on Clean Architecture principles. It supports both JetStream (persistent) and Core NATS (fire-and-forget) modes with comprehensive error handling, observability, and resilience features. + +## Architecture + +### Clean Architecture Layers + +``` +┌─────────────────────────────────────┐ +│ Port Interfaces │ +│ (Publisher, Consumer contracts) │ +├─────────────────────────────────────┤ +│ Application Layer │ +│ (Message processing logic) │ +├─────────────────────────────────────┤ +│ Infrastructure Layer │ +│ (NATS implementation details) │ +│ │ +│ ┌─────────────────────────────┐ │ +│ │ Consumer │ │ +│ │ ┌─────────────────────┐ │ │ +│ │ │ MessageFetcher │ │ │ +│ │ │ MessageProcessor │ │ │ +│ │ │ ErrorHandler │ │ │ +│ │ │ DLQHandler │ │ │ +│ │ └─────────────────────┘ │ │ +│ └─────────────────────────────┘ │ +│ │ +│ ┌─────────────────────────────┐ │ +│ │ Publisher │ │ +│ │ ┌─────────────────────┐ │ │ +│ │ │ MessageSerializer │ │ │ +│ │ │ HeaderEnricher │ │ │ +│ │ └─────────────────────┘ │ │ +│ └─────────────────────────────┘ │ +└─────────────────────────────────────┘ +``` + +## Core Components + +### Consumer + +The consumer handles message consumption with the following features: + +#### Modes +- **JetStream Mode**: Persistent, durable message processing with acknowledgments +- **Core Mode**: Fire-and-forget message processing for simple use cases + +#### Key Features +- **Batch Processing**: Configurable batch sizes and timeouts for efficient processing +- **Backpressure**: Automatic backpressure when processing errors accumulate +- **Dead Letter Queue (DLQ)**: Automatic routing of failed messages to DLQ +- **Advisory DLQ**: Handles messages that exceed MaxDeliver limits +- **Self-Healing**: Automatic recreation of missing streams/consumers in dev environments +- **Graceful Shutdown**: Proper cleanup and draining of connections + +#### Configuration +```toml +[NATS] +Mode = "jetstream" # or "core" +Stream = "TELEGRAM" +Consumer = "telegram-consumer" + +[NATS.ConsumerRules] +AckWait = "30s" +MaxDeliver = 3 +MaxAckPending = 1000 +DeliverPolicy = "all" +ReplayPolicy = "instant" +Backoff = ["1s", "2s", "5s", "10s"] + +[DLQ] +Enabled = true +Subject = "caatsm.dlq" + +[App] +BatchSize = 50 +BatchTimeout = "2s" +MonitorInterval = "30s" +``` + +### Publisher + +The publisher handles message publishing with deduplication and observability. + +#### Features +- **Message Deduplication**: Automatic UUID-based deduplication headers +- **JetStream/Core Mode Support**: Adapts based on available JetStream context +- **Structured Logging**: Comprehensive logging of publish operations +- **Error Classification**: Distinguishes between transient and permanent errors + +### Error Handling + +#### Error Types +- **Transient Errors**: Network issues, temporary unavailability (retried with backoff) +- **Permanent Errors**: Message format issues, business logic failures (routed to DLQ) +- **Resource Errors**: Missing streams/consumers (auto-recovered in dev, fail in prod) + +#### Recovery Strategies +- **Exponential Backoff**: Configurable backoff for transient failures +- **Circuit Breaker Pattern**: Prevents cascade failures +- **Resource Recreation**: Automatic recreation of missing JetStream resources +- **Graceful Degradation**: Continues processing other messages when one fails + +### Dead Letter Queue (DLQ) + +#### Features +- **Rich Metadata**: Includes original message, error details, delivery attempts +- **Stream Validation**: Validates DLQ stream exists at startup +- **Advisory Processing**: Handles MaxDeliver exhaustion automatically +- **Operational Visibility**: Comprehensive logging and metrics + +#### DLQ Message Format +```json +{ + "transport_msg_id": "uuid", + "subject": "original.subject", + "stream": "TELEGRAM", + "consumer": "telegram-consumer", + "nats_sequence": 12345, + "deliveries": 3, + "error": "processing failed: invalid format", + "received_at": "2024-01-01T12:00:00Z", + "body": "original message data" +} +``` + +## Observability + +### Metrics +- **Consumer Metrics**: ack_pending, redelivered, pending, delivered counts +- **Processing Metrics**: batch size, processing duration, error rates +- **DLQ Metrics**: messages routed to DLQ, publish failures +- **Connection Metrics**: connection health, reconnection events + +### Tracing +- **End-to-End Tracing**: Request correlation through trace IDs +- **Span Attributes**: Consumer name, stream name, batch size, error details +- **Context Propagation**: Trace context passed through processing pipeline + +### Logging +- **Structured Logs**: JSON format with correlation IDs +- **Log Levels**: Debug, Info, Warn, Error with appropriate detail levels +- **Operational Context**: Includes consumer, stream, and message metadata + +## Resilience Patterns + +### Backpressure +- **Error Accumulation**: Tracks consecutive processing errors +- **Adaptive Delay**: Increases delay based on error frequency +- **Circuit Breaking**: Stops processing when errors exceed threshold + +### Connection Management +- **Auto-Reconnection**: Built-in NATS reconnection logic +- **Graceful Shutdown**: Proper draining with timeouts +- **Resource Cleanup**: Ensures subscriptions and connections are closed + +### Self-Healing +- **Development Mode**: Auto-creates missing streams/consumers +- **Production Mode**: Fails fast on configuration issues +- **Recovery Logic**: Attempts to recreate resources on errors + +## Configuration + +### Environment Variables +```bash +CAATSM_NATS_URL=nats://localhost:4222 +CAATSM_NATS_MODE=jetstream +CAATSM_DLQ_ENABLED=true +CAATSM_DLQ_SUBJECT=caatsm.dlq +``` + +### Runtime Configuration +- **Hot Reload**: Configuration changes applied without restart +- **Validation**: Comprehensive validation at startup +- **Defaults**: Sensible defaults for all configuration options + +## Testing Strategy + +### Unit Tests +- **Pure Functions**: Configuration normalization, policy mapping +- **Mock Dependencies**: NATS connections, JetStream contexts +- **Table-Driven Tests**: Comprehensive coverage of edge cases + +### Integration Tests +- **Real NATS**: Testcontainers with actual NATS server +- **End-to-End**: Complete message processing pipelines +- **Failure Scenarios**: Network failures, resource unavailability + +### Test Categories +- **Happy Path**: Normal operation scenarios +- **Error Recovery**: Various failure and recovery scenarios +- **Performance**: Load testing and resource usage +- **Configuration**: Different configuration combinations + +## Usage Examples + +### Basic Consumer Setup +```go +consumer, err := natsinfra.ProvideConsumer( + natsConn, + jetStream, + messageProcessor, + config, + telemetryRecorder, + logger, +) +if err != nil { + return err +} + +ctx, cancel := context.WithCancel(context.Background()) +defer cancel() + +return consumer.Start(ctx) +``` + +### Publishing Messages +```go +publisher, err := natsinfra.ProvidePublisher( + jetStream, + natsConn, + config, + logger, +) +if err != nil { + return err +} + +err = publisher.Publish(&dto.ParsedTelegram{ + Uuid: uuid.NewString(), + Data: telegramData, +}) +``` + +### Custom Error Handling +```go +type CustomProcessor struct { + // implementation +} + +func (p *CustomProcessor) ProcessMessage(ctx context.Context, msg *nats.Msg) error { + // Business logic here + if shouldRetry := someCondition(); shouldRetry { + return app.NewTransientError("temporary failure") + } + if isInvalid := validateMessage(msg); isInvalid { + return app.NewPermanentError("invalid message format") + } + return nil +} +``` + +## Performance Considerations + +### Optimization Strategies +- **Batch Processing**: Reduces per-message overhead +- **Connection Pooling**: Reuses connections efficiently +- **Memory Management**: Proper buffer sizing and cleanup +- **Concurrent Processing**: Parallel message processing within batches + +### Monitoring Points +- **Throughput**: Messages processed per second +- **Latency**: End-to-end processing time +- **Resource Usage**: Memory, CPU, and network utilization +- **Error Rates**: Percentage of failed messages + +## Operational Guide + +### Deployment +1. **Configuration**: Set appropriate timeouts and limits +2. **Resource Provisioning**: Ensure sufficient NATS cluster capacity +3. **Monitoring Setup**: Configure alerts and dashboards +4. **DLQ Monitoring**: Set up DLQ message processing + +### Troubleshooting +- **High Latency**: Check batch sizes and processing logic +- **Message Loss**: Verify consumer acks and DLQ configuration +- **Connection Issues**: Check NATS cluster health and network connectivity +- **Resource Exhaustion**: Monitor memory usage and connection counts + +### Maintenance +- **Stream Cleanup**: Periodically clean up old streams +- **Consumer Recreation**: Recreate consumers for configuration changes +- **Performance Tuning**: Adjust batch sizes based on load patterns +- **Version Upgrades**: Test compatibility with NATS server versions + +## Security Considerations + +### Authentication +- **NATS Auth**: Use NATS built-in authentication mechanisms +- **TLS**: Enable TLS for encrypted communication +- **Token Auth**: Use NATS tokens for service authentication + +### Authorization +- **Subject Permissions**: Restrict publish/subscribe permissions +- **Stream Access**: Control access to specific streams +- **DLQ Security**: Secure DLQ access to prevent data leakage + +### Data Protection +- **Message Encryption**: Encrypt sensitive message data +- **Audit Logging**: Log all message operations for compliance +- **PII Handling**: Avoid logging sensitive information + +## Future Enhancements + +### Planned Features +- **Consumer Groups**: Horizontal scaling with multiple consumers +- **Message Filtering**: Subject-based and header-based filtering +- **Priority Queues**: High-priority message processing +- **Rate Limiting**: Per-consumer and per-subject rate limits +- **Message Transformation**: In-flight message modification +- **Multi-Region**: Cross-region message replication + +### Extensibility Points +- **Custom Serializers**: Pluggable message serialization +- **Middleware**: Request/response middleware support +- **Hooks**: Pre/post processing hooks +- **Metrics Backends**: Support for additional metrics systems +- **Storage Backends**: Alternative storage for DLQ messages \ No newline at end of file diff --git a/docs/observability.md b/docs/observability.md index f70bb55..df84481 100644 --- a/docs/observability.md +++ b/docs/observability.md @@ -166,60 +166,79 @@ A non-2xx response indicates the service is not healthy/ready and should be remo ### Tracing +The application implements production-ready OpenTelemetry tracing with comprehensive span coverage and semantic attributes. + +#### Configuration + Tracing is configured via the `telemetry` section: -- `telemetry.enabled` – enables OTEL exporters. -- `telemetry.endpoint` – OTLP HTTP endpoint (e.g. `localhost:4318`). -- `telemetry.insecure` – disables TLS for local/dev. +- `telemetry.enabled` – enables OTEL exporters (default: `false` in dev, `true` in prod) +- `telemetry.endpoint` – OTLP HTTP endpoint (e.g. `localhost:4318` for dev, `otel-collector.company.com:4318` for prod) +- `telemetry.insecure` – disables TLS for local/dev (default: `true` in dev, `false` in prod) -#### OTEL vs Prometheus metrics +#### Sampling Strategy -The receiver reports two complementary sets of metrics: +Environment-based sampling ensures cost-effective production monitoring: -- **Prometheus metrics via `/metrics`** - Implemented in `internal/infra/metrics`, covering: - - End-to-end message handling (`caatsm_messages_total`, - `caatsm_handle_latency_seconds`, `caatsm_retries_total`) - - DB activity (`caatsm_db_queries_total`, - `caatsm_db_query_latency_seconds`) - - Legacy per-telegram metrics +- **Production**: 1% sampling (cost-effective, maintains observability) +- **Staging**: 10% sampling (balanced observability for testing) +- **Development/Test**: 100% sampling (full debugging coverage) -- **OpenTelemetry metrics via OTLP** - Implemented using `otel.Meter` in the NATS consumer and app processor, - including: - - `caatsm_messages_processed_total` - - `caatsm_parse_duration_seconds` - - `caatsm_publish_failures_total` - - `caatsm_nats_consumer_ack_pending` - - `caatsm_nats_consumer_redelivered` - - `caatsm_nats_consumer_pending` - - `caatsm_nats_consumer_delivered` +#### Resource Attributes -Prometheus only sees the metrics exposed on `/metrics`. OTEL metrics are -exported to the configured OTEL collector (`telemetry.endpoint`) via OTLP and -are, by default, forwarded to Jaeger (traces) and logs (metrics) according to -`configs/otel-collector.dev.yaml`. If you want OTEL metrics to appear in -Prometheus as well, you can extend the collector configuration with a -`prometheus` or `prometheusremotewrite` exporter and add a corresponding -scrape or remote-write configuration. +All spans include comprehensive resource metadata: -Key spans: -- `caatsm/nats` - - `Consumer.processMessage` -- `caatsm/app` - - `MessageProcessor.Handle` - - `Publisher.Publish` -- `caatsm/postgres` - - `Repository.InsertOne` - - `Repository.InsertBatch` - - `Repository.InsertRaw` +#### Key Spans with Semantic Attributes -Important attributes: +**NATS Consumer (`caatsm/nats`)**: +- `Consumer.processMessage` + - `messaging.system: nats` + - `messaging.operation: receive` + - `messaging.destination: ` + - `messaging.consumer.id: ` + - `caatsm.stream: ` -- `nats.subject`, `nats.msg_id`, `nats.js.stream_seq`, `nats.js.consumer_seq` -- `telegram.message_id`, `telegram.category`, `telegram.status` -- `db.table`, `db.inserted` +**Application Processor (`caatsm/app`)**: +- `MessageProcessor.Handle` + - `messaging.system: nats` + - `messaging.operation: receive` + - `messaging.message_id: ` + - `caatsm.component: processor` + - `caatsm.message.category: ` + +**Database Operations (`caatsm/postgres`)**: +- `Repository.InsertOne`, `Repository.InsertBatch`, `Repository.InsertRaw` + - `db.system: postgresql` + - `db.operation: insert` + - `db.name: aviation` + - `db.table: telegrams` + - `caatsm.message.id: ` + +#### OTEL vs Prometheus Metrics + +The receiver reports complementary metrics through both systems: + +**Prometheus metrics via `/metrics`** (operational focus): +- End-to-end message handling (`caatsm_messages_total`, `caatsm_handle_latency_seconds`, `caatsm_retries_total`) +- 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`) + +**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) + +#### Collector Integration + +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. ### Structured Logging Contract diff --git a/docs/otel-best-practices.md b/docs/otel-best-practices.md new file mode 100644 index 0000000..7e09d00 --- /dev/null +++ b/docs/otel-best-practices.md @@ -0,0 +1,353 @@ +# OpenTelemetry Best Practices + +This document outlines the OpenTelemetry implementation and best practices for the CAATSM receiver service. + +## Architecture Overview + +The CAATSM receiver implements a dual-telemetry approach: + +1. **OpenTelemetry (OTEL)**: Business metrics and distributed tracing +2. **Prometheus**: Operational metrics and alerting + +## OTEL Implementation + +### SDK Configuration + +The application uses a production-ready OTEL SDK setup with: + +- **Environment-based sampling**: Cost-effective production monitoring +- **Comprehensive resource attributes**: Service identification and metadata +- **Optimized batching**: Efficient export with retry logic +- **TLS security**: Configurable secure connections + +### Sampling Strategy + +```go +Production: 1% // Cost-effective, maintains observability +Staging: 10% // Balanced observability for testing +Dev/Test: 100% // Full debugging coverage +``` + +### Resource Attributes + +All telemetry includes standardized resource metadata: + +```yaml +# Service identification +service.name: caatsm +service.version: dev +service.namespace: airport +service.component: receiver + +# Environment context +deployment.environment: prod|staging|dev + +# Build information +build.commit: +build.built_at: + +# Telemetry configuration +telemetry.endpoint: +telemetry.insecure: true|false +``` + +## Span Semantics + +### Messaging Spans + +**NATS Consumer Operations**: +```yaml +Span: Consumer.processMessage +Attributes: + messaging.system: nats + messaging.operation: receive + messaging.destination: telegram.serial + messaging.consumer.id: telegram-consumer + caatsm.stream: TELEGRAM +``` + +**Application Processing**: +```yaml +Span: MessageProcessor.Handle +Attributes: + messaging.system: nats + messaging.operation: receive + messaging.message_id: + caatsm.component: processor + caatsm.message.category: ARR|DEP|FPL|etc +``` + +### Database Spans + +**Repository Operations**: +```yaml +Span: Repository.InsertOne|InsertBatch|InsertRaw +Attributes: + db.system: postgresql + db.operation: insert + db.name: aviation + db.table: telegrams + caatsm.message.id: +``` + +## Metrics Strategy + +### OTEL Metrics (Business Focus) + +- `caatsm_messages_processed_total{message.status, message.category}` +- `caatsm_publish_failures_total{message.category}` +- `caatsm_parse_duration_seconds{message.status, message.category}` + +### Prometheus Metrics (Operational Focus) + +- `caatsm_messages_total{stream, consumer, result}` +- `caatsm_handle_latency_seconds{stream, consumer}` +- `caatsm_db_queries_total{operation, result}` +- `caatsm_nats_consumer_pending_messages{stream, consumer}` + +## Collector Configuration + +### Development Setup + +```yaml +# configs/otel-collector.dev.yaml +receivers: + otlp: + protocols: + http: + endpoint: 0.0.0.0:4318 + max_request_body_size: 20971520 + grpc: + endpoint: 0.0.0.0:4317 + +processors: + batch: + send_batch_size: 1024 + timeout: 1s + resource: + attributes: + - key: service.instance.id + value: "${POD_NAME}" + action: upsert + +exporters: + logging: + sampling_initial: 10 + sampling_thereafter: 100 + otlphttp/jaeger: + endpoint: http://jaeger:4318 + sending_queue: + queue_size: 10000 + retry_on_failure: + enabled: true + prometheus: + endpoint: "0.0.0.0:8889" + +service: + pipelines: + traces: + receivers: [otlp] + processors: [resource, batch] + exporters: [logging, otlphttp/jaeger] + metrics: + receivers: [otlp] + processors: [resource, batch] + exporters: [logging, prometheus] +``` + +### Production Setup + +```yaml +# configs/otel-collector.prod.yaml +receivers: + otlp: + protocols: + grpc: + endpoint: 0.0.0.0:4317 + tls: + cert_file: /etc/ssl/certs/otel-collector.crt + key_file: /etc/ssl/private/otel-collector.key + auth: + authenticator: bearer_token + http: + endpoint: 0.0.0.0:4318 + tls: + cert_file: /etc/ssl/certs/otel-collector.crt + key_file: /etc/ssl/private/otel-collector.key + auth: + authenticator: bearer_token + +extensions: + health_check: + endpoint: 0.0.0.0:13133 + pprof: + endpoint: :1888 + zpages: + endpoint: :55679 + +exporters: + otlphttp/jaeger: + endpoint: https://jaeger.company.com:4318 + headers: + authorization: "Bearer ${JAEGER_API_TOKEN}" + tls: + insecure: false + sending_queue: + queue_size: 10000 + retry_on_failure: + enabled: true + prometheusremotewrite: + endpoint: https://prometheus.company.com/api/v1/write + headers: + authorization: "Bearer ${PROMETHEUS_API_TOKEN}" + tls: + insecure: false + sending_queue: + queue_size: 10000 + retry_on_failure: + enabled: true + +service: + extensions: [health_check, pprof, zpages] + pipelines: + traces: + receivers: [otlp] + processors: [resource, batch] + exporters: [otlphttp/jaeger] + metrics: + receivers: [otlp] + processors: [resource, batch] + exporters: [prometheusremotewrite] +``` + +## Configuration Examples + +### Development Configuration + +```toml +[telemetry] +enabled = true +endpoint = "localhost:4318" +insecure = true +``` + +### Production Configuration + +```toml +[telemetry] +enabled = true +endpoint = "otel-collector.company.com:4318" +insecure = false +``` + +### CLI Overrides + +```bash +# Enable telemetry +./bin/receiver listen --telemetry-enabled + +# Custom endpoint +./bin/receiver listen --telemetry-endpoint https://otel-collector.prod:4318 + +# Insecure for development +./bin/receiver listen --telemetry-insecure +``` + +## Monitoring and Debugging + +### Health Checks + +```bash +# Collector health +curl http://otel-collector:13133 + +# Application metrics +curl http://localhost:2112/metrics + +# OTEL collector metrics +curl http://otel-collector:8888/metrics +``` + +### Tracing Verification + +```bash +# Jaeger UI +open http://localhost:16686 + +# Search for caatsm traces +Service: caatsm +Operation: Consumer.processMessage OR MessageProcessor.Handle +``` + +### Metrics Verification + +```bash +# Prometheus queries +caatsm_messages_processed_total +caatsm_parse_duration_seconds +rate(caatsm_messages_total[5m]) +``` + +## Best Practices + +### 1. Sampling Strategy +- Use environment-appropriate sampling rates +- Monitor sampling effectiveness +- Adjust based on cost and observability needs + +### 2. Resource Attributes +- Include comprehensive service metadata +- Use semantic conventions +- Add custom attributes for business context + +### 3. Span Attributes +- Follow OpenTelemetry semantic conventions +- Include relevant business context +- Avoid high-cardinality attributes + +### 4. Error Handling +- Always record errors on spans +- Set appropriate span status +- Include error context in attributes + +### 5. Performance +- Use batching to reduce export overhead +- Configure appropriate queue sizes +- Monitor exporter performance + +## Troubleshooting + +### Common Issues + +1. **No traces in Jaeger** + - Check collector logs: `docker logs otel-collector` + - Verify endpoint configuration + - Check network connectivity + +2. **High sampling rate** + - Adjust sampling configuration + - Monitor cost impact + - Consider head-based sampling + +3. **Missing metrics** + - Verify collector pipeline configuration + - Check Prometheus remote write configuration + - Validate metric names and labels + +4. **Performance impact** + - Review sampling rates + - Check batch configuration + - Monitor exporter queue sizes + +### Debug Commands + +```bash +# View collector configuration +docker exec otel-collector cat /etc/otel/config.yaml + +# Check collector metrics +curl -s http://otel-collector:8888/metrics | grep otel + +# View application telemetry logs +./bin/receiver listen --log-level=debug 2>&1 | grep -i telemetry +``` \ No newline at end of file diff --git a/internal/app/processor.go b/internal/app/processor.go index 9cc53c1..655952c 100644 --- a/internal/app/processor.go +++ b/internal/app/processor.go @@ -1,8 +1,8 @@ package app import ( - "caatsm/internal/adapter/parser" "caatsm/internal/adapter/dto" + "caatsm/internal/adapter/parser" "caatsm/internal/infra/log" "caatsm/internal/infra/telemetry" "caatsm/internal/port" @@ -64,7 +64,14 @@ func (p *MessageProcessor) Handle(ctx context.Context, raw []byte, msgID string) tracer := otel.Tracer("caatsm/app") ctx, span := tracer.Start(ctx, "MessageProcessor.Handle") defer span.End() - span.SetAttributes(attribute.String("nats.msg_id", msgID)) + + // Set semantic attributes following OpenTelemetry conventions + span.SetAttributes( + attribute.String("messaging.system", "nats"), + attribute.String("messaging.operation", "receive"), + attribute.String("messaging.message_id", msgID), + attribute.String("caatsm.component", "processor"), + ) receivedAt := time.Now() diff --git a/internal/infra/nats/advisory_dlq.go b/internal/infra/nats/advisory_dlq.go new file mode 100644 index 0000000..69d34e1 --- /dev/null +++ b/internal/infra/nats/advisory_dlq.go @@ -0,0 +1,167 @@ +package nats + +import ( + "context" + "encoding/json" + "fmt" + "time" + + "github.com/nats-io/nats.go" + "go.uber.org/zap" +) + +// MaxDeliveriesAdvisoryEvent represents the advisory message published when +// a message reaches MaxDeliver attempts. +type MaxDeliveriesAdvisoryEvent struct { + Type string `json:"type"` + Stream string `json:"stream"` + Consumer string `json:"consumer"` + StreamSeq uint64 `json:"stream_seq"` + Deliveries uint64 `json:"deliveries"` + Time string `json:"time"` +} + +// AdvisoryDLQHandler handles messages that exhaust MaxDeliver attempts +// by subscribing to JetStream advisory events. +type AdvisoryDLQHandler struct { + js nats.JetStreamContext + nc *nats.Conn + streamName string + consumerName string + dlqSubject string + logger *zap.Logger + telemetry TelemetryRecorder +} + +// TelemetryRecorder is an interface for recording telemetry events. +// This matches the telemetry.Recorder interface used by Consumer. +type TelemetryRecorder interface { + RecordDLQMessage(ctx context.Context, stream, consumer string) + RecordDLQPublishFailure(ctx context.Context, stream, consumer string) +} + +// NewAdvisoryDLQHandler creates a new advisory-based DLQ handler. +func NewAdvisoryDLQHandler( + js nats.JetStreamContext, + nc *nats.Conn, + streamName string, + consumerName string, + dlqSubject string, + logger *zap.Logger, + telemetry TelemetryRecorder, +) (*AdvisoryDLQHandler, error) { + return &AdvisoryDLQHandler{ + js: js, + nc: nc, + streamName: streamName, + consumerName: consumerName, + dlqSubject: dlqSubject, + logger: logger, + telemetry: telemetry, + }, nil +} + +// Start begins listening for advisory messages and routing failed messages to DLQ. +func (h *AdvisoryDLQHandler) Start(ctx context.Context) error { + // Subscribe to advisory subject pattern + // Format: $JS.EVENT.ADVISORY.CONSUMER.MAX_DELIVERIES.. + advisorySubject := fmt.Sprintf("$JS.EVENT.ADVISORY.CONSUMER.MAX_DELIVERIES.%s.%s", + h.streamName, h.consumerName) + + h.logger.Info("Starting advisory DLQ handler", + zap.String("advisory_subject", advisorySubject), + zap.String("stream", h.streamName), + zap.String("consumer", h.consumerName), + zap.String("dlq_subject", h.dlqSubject), + ) + + sub, err := h.nc.Subscribe(advisorySubject, func(msg *nats.Msg) { + h.handleAdvisory(ctx, msg) + }) + if err != nil { + return fmt.Errorf("failed to subscribe to advisory subject: %w", err) + } + + // Wait for context cancellation + go func() { + <-ctx.Done() + sub.Unsubscribe() + h.logger.Info("Stopped advisory DLQ handler") + }() + + return nil +} + +// handleAdvisory processes an advisory message about max deliveries. +func (h *AdvisoryDLQHandler) handleAdvisory(ctx context.Context, advisoryMsg *nats.Msg) { + var event MaxDeliveriesAdvisoryEvent + if err := json.Unmarshal(advisoryMsg.Data, &event); err != nil { + h.logger.Error("Failed to unmarshal advisory event", + zap.Error(err), + zap.String("data", string(advisoryMsg.Data)), + ) + return + } + + h.logger.Warn("Message reached MaxDeliver attempts", + zap.String("stream", event.Stream), + zap.String("consumer", event.Consumer), + zap.Uint64("stream_seq", event.StreamSeq), + zap.Uint64("deliveries", event.Deliveries), + ) + + // Retrieve the original message from the stream using GetMsg API + originalMsg, err := h.js.GetMsg(h.streamName, event.StreamSeq) + if err != nil { + h.logger.Error("Failed to retrieve original message from stream", + zap.Uint64("stream_seq", event.StreamSeq), + zap.Error(err), + ) + h.telemetry.RecordDLQPublishFailure(ctx, h.streamName, h.consumerName) + return + } + + // Extract message metadata + msgID := "" + subject := originalMsg.Subject + if originalMsg.Header != nil { + msgID = originalMsg.Header.Get("Nats-Msg-Id") + } + + // Create enriched DLQ payload (similar to existing routeToDLQ) + payload := map[string]interface{}{ + "transport_msg_id": msgID, + "subject": subject, + "stream": h.streamName, + "consumer": h.consumerName, + "nats_sequence": event.StreamSeq, + "deliveries": event.Deliveries, + "error": fmt.Sprintf("message exhausted max_deliver (%d) attempts", event.Deliveries), + "received_at": time.Now().UTC(), + "body": string(originalMsg.Data), + "advisory_source": true, // Flag to distinguish from immediate DLQ + } + + data, err := json.Marshal(payload) + if err != nil { + h.logger.Error("Failed to marshal advisory DLQ payload", zap.Error(err)) + h.telemetry.RecordDLQPublishFailure(ctx, h.streamName, h.consumerName) + return + } + + // Publish to DLQ + if _, err := h.js.Publish(h.dlqSubject, data); err != nil { + h.logger.Error("Failed to publish advisory message to DLQ", + zap.Uint64("stream_seq", event.StreamSeq), + zap.Error(err), + ) + h.telemetry.RecordDLQPublishFailure(ctx, h.streamName, h.consumerName) + return + } + + h.logger.Info("Routed max-deliveries message to DLQ", + zap.Uint64("stream_seq", event.StreamSeq), + zap.Uint64("deliveries", event.Deliveries), + ) + h.telemetry.RecordDLQMessage(ctx, h.streamName, h.consumerName) +} diff --git a/internal/infra/nats/consumer.go b/internal/infra/nats/consumer.go index 45f351b..49ece60 100644 --- a/internal/infra/nats/consumer.go +++ b/internal/infra/nats/consumer.go @@ -5,6 +5,8 @@ import ( "caatsm/internal/infra/config" "caatsm/internal/infra/telemetry" "context" + "encoding/json" + "errors" "fmt" "strings" "time" @@ -14,35 +16,57 @@ import ( "go.uber.org/zap" ) -// Consumer handles NATS JetStream message consumption -type Consumer struct { - conn *nats.Conn - js nats.JetStreamContext - processor *app.MessageProcessor - cfg *config.Config - logger *zap.Logger - telemetry telemetry.Recorder - subject string - consumerName string - mode string - streamName string - dlqSubject string - ackWait time.Duration - batchSize int - batchTimeout time.Duration - monitorInterval time.Duration - meter metric.Meter - ackPending metric.Int64Histogram - redelivered metric.Int64Histogram - pending metric.Int64Histogram - delivered metric.Int64Histogram +// MessageFetcher defines the interface for fetching messages from NATS +type MessageFetcher interface { + FetchBatch(ctx context.Context, sub *nats.Subscription) ([]*nats.Msg, error) + HandleFetchError(ctx context.Context, err error, sub **nats.Subscription, fetchErrorStreak *int) (bool, error) +} - // managers for resource lifecycle +// MessageProcessor defines the interface for processing message batches +type MessageProcessor interface { + ProcessBatch(ctx context.Context, msgs []*nats.Msg) +} + +// DLQHandler defines the interface for dead letter queue operations +type DLQHandler interface { + RouteToDLQ(ctx context.Context, msg *nats.Msg, cause error) error + ValidateDLQ() error +} + +// Consumer handles NATS JetStream message consumption with clean separation of concerns +type Consumer struct { + // Core dependencies + conn *nats.Conn + js nats.JetStreamContext + processor *app.MessageProcessor + cfg *config.Config + logger *zap.Logger + telemetry telemetry.Recorder + + // Configuration + config consumerConfig + + // Collaborators (injected for testability) + fetcher MessageFetcher + batchProcessor MessageProcessor + dlqHandler DLQHandler + errorHandler *ErrorHandler + + // Resource managers consumerManager *ConsumerManager streamManager *StreamManager - errorHandler *ErrorHandler - // simple backpressure / degradation state + // Advisory DLQ handler for messages exhausting MaxDeliver + advisoryDLQHandler *AdvisoryDLQHandler + + // Metrics + meter metric.Meter + ackPending metric.Int64Histogram + redelivered metric.Int64Histogram + pending metric.Int64Histogram + delivered metric.Int64Histogram + + // State consecutiveProcessErrors int } @@ -59,6 +83,397 @@ type consumerConfig struct { monitorInterval time.Duration } +// defaultMessageFetcher implements MessageFetcher interface +type defaultMessageFetcher struct { + batchSize int + batchTimeout time.Duration + logger *zap.Logger + conn *nats.Conn + js nats.JetStreamContext + consumerManager *ConsumerManager + streamManager *StreamManager + config *consumerConfig + cfg *config.Config +} + +func (f *defaultMessageFetcher) FetchBatch(ctx context.Context, sub *nats.Subscription) ([]*nats.Msg, error) { + return f.fetchBatch(ctx, sub) +} + +// fetchBatch fetches a batch of messages from the subscription with context awareness +func (f *defaultMessageFetcher) fetchBatch(ctx context.Context, sub *nats.Subscription) ([]*nats.Msg, error) { + // Check context before fetching + select { + case <-ctx.Done(): + return nil, ctx.Err() + default: + } + + // Use a shorter timeout for better responsiveness to cancellation + timeout := f.batchTimeout + if timeout > 500*time.Millisecond { + timeout = 500 * time.Millisecond + } + + return sub.Fetch(f.batchSize, nats.MaxWait(timeout)) +} + +func (f *defaultMessageFetcher) HandleFetchError(ctx context.Context, err error, sub **nats.Subscription, fetchErrorStreak *int) (bool, error) { + // Check context cancellation first + if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) { + f.logger.Info("Fetch error due to context cancellation", zap.Error(err)) + return false, err + } + + // Timeout errors are expected when no messages are available - not an error condition + if errors.Is(err, nats.ErrTimeout) { + return true, nil + } + + // Check connection health before proceeding + if f.conn != nil { + status := f.conn.Status() + if status != nats.CONNECTED { + f.logger.Warn("NATS connection not in CONNECTED state", + zap.String("status", status.String()), + zap.Error(err), + ) + // Connection is down - this is a transient error, apply backoff + *fetchErrorStreak++ + backoff := f.calculateExponentialBackoff(*fetchErrorStreak) + f.logger.Warn("Connection unhealthy, applying backoff before retry", + zap.String("status", status.String()), + zap.Int("error_streak", *fetchErrorStreak), + zap.Duration("backoff", backoff), + ) + if !sleepWithContext(ctx, backoff) { + return false, ctx.Err() + } + // Check if connection recovered after backoff + if f.conn.Status() == nats.CONNECTED { + *fetchErrorStreak = 0 + return true, nil + } + // Still not connected - continue with error handling + } + } + + // Check for connection closed errors + if errors.Is(err, nats.ErrConnectionClosed) { + f.logger.Error("NATS connection closed", + zap.Error(err), + zap.String("stream", f.config.streamName), + zap.String("consumer", f.config.consumerName), + ) + // Connection closed is fatal - cannot recover subscription + if *sub != nil { + (*sub).Unsubscribe() + *sub = nil + } + return false, fmt.Errorf("connection closed: %w", err) + } + + // JetStream API unavailable (e.g., NATS restarted or JetStream not ready) + if errors.Is(err, nats.ErrNoResponders) { + *fetchErrorStreak++ + backoff := f.calculateExponentialBackoff(*fetchErrorStreak) + backoff = min(backoff, 30*time.Second) + f.logger.Warn("JetStream not available, will retry with backoff", + zap.Error(err), + zap.String("stream", f.config.streamName), + zap.String("consumer", f.config.consumerName), + zap.Int("error_streak", *fetchErrorStreak), + zap.Duration("backoff", backoff), + ) + if !sleepWithContext(ctx, backoff) { + return false, ctx.Err() + } + return true, nil + } + + // Check for JetStream resource not found errors + if isJetStreamResourceNotFound(err) { + if isDevLikeEnv() && shouldBootstrapStream() { + f.logger.Warn("JetStream consumer or stream missing; attempting to recreate", + zap.Error(err), + zap.String("stream", f.config.streamName), + zap.String("consumer", f.config.consumerName), + ) + // Attempt to recover resources and recreate subscription + if f.consumerManager == nil || f.streamManager == nil { + return false, fmt.Errorf("cannot recover: consumer/stream manager not available: %w", err) + } + consumerConfig := f.buildConsumerConfig() + if recErr := f.consumerManager.RecoverResources(f.streamManager, consumerConfig); recErr != nil { + return false, fmt.Errorf("failed to recover JetStream resources: %w", recErr) + } + // Unsubscribe old subscription before creating new one + if *sub != nil { + (*sub).Unsubscribe() + } + // Create new subscription + newSub, subErr := f.consumerManager.CreatePullSubscription() + if subErr != nil { + return false, fmt.Errorf("failed to create pull subscription after recovery: %w", subErr) + } + *sub = newSub + *fetchErrorStreak = 0 + f.logger.Info("Successfully recovered subscription after resource recreation") + return true, nil + } + + // Production: treat as configuration/operational error - fatal + f.logger.Error("JetStream consumer or stream missing; not auto-recreating in this environment", + zap.Error(err), + zap.String("stream", f.config.streamName), + zap.String("consumer", f.config.consumerName), + ) + if *sub != nil { + (*sub).Unsubscribe() + *sub = nil + } + return false, fmt.Errorf("JetStream resource not found: %w", err) + } + + // Check for network/temporary errors + if f.isTemporaryError(err) { + *fetchErrorStreak++ + backoff := f.calculateExponentialBackoff(*fetchErrorStreak) + f.logger.Warn("Temporary network error, applying backoff", + zap.Error(err), + zap.Int("error_streak", *fetchErrorStreak), + zap.Duration("backoff", backoff), + ) + if !sleepWithContext(ctx, backoff) { + return false, ctx.Err() + } + // Verify subscription is still valid before returning success + if *sub != nil && f.conn != nil && f.conn.Status() == nats.CONNECTED { + return true, nil + } + // Subscription or connection invalid - attempt recovery + return f.attemptSubscriptionRecovery(ctx, sub, fetchErrorStreak) + } + + // Generic error path with exponential backoff + *fetchErrorStreak++ + backoff := f.calculateExponentialBackoff(*fetchErrorStreak) + f.logger.Error("Failed to fetch messages; backing off", + zap.Error(err), + zap.Int("error_streak", *fetchErrorStreak), + zap.Duration("backoff", backoff), + ) + if !sleepWithContext(ctx, backoff) { + return false, ctx.Err() + } + + // Verify subscription and connection health before returning success + if *sub == nil || (f.conn != nil && f.conn.Status() != nats.CONNECTED) { + return f.attemptSubscriptionRecovery(ctx, sub, fetchErrorStreak) + } + + return true, nil +} + +// calculateExponentialBackoff calculates exponential backoff duration with a cap +func (f *defaultMessageFetcher) calculateExponentialBackoff(streak int) time.Duration { + if streak <= 0 { + return 0 + } + // Exponential backoff: 2^(streak-1) seconds, capped at 30 seconds + backoff := time.Duration(1< 0 { - consumerConfig.OptStartSeq = c.cfg.NATS.ConsumerRules.StartSequence - } - if consumerConfig.DeliverPolicy == nats.DeliverByStartTimePolicy && strings.TrimSpace(c.cfg.NATS.ConsumerRules.StartTime) != "" { - startTime, err := time.Parse(time.RFC3339, c.cfg.NATS.ConsumerRules.StartTime) - if err != nil { - c.logger.Warn("Invalid start time, falling back to deliver policy defaults", - zap.String("start_time", c.cfg.NATS.ConsumerRules.StartTime), - zap.Error(err), - ) - } else { - consumerConfig.OptStartTime = &startTime - } - } - - return c.consumerManager.EnsureConsumer(consumerConfig) -} - // recoverJetStreamResources attempts to recreate the stream and consumer in // dev/test environments if they are missing. It is safe to call multiple times. func (c *Consumer) recoverJetStreamResources() error { @@ -98,14 +75,30 @@ func sleepWithContext(ctx context.Context, duration time.Duration) bool { } // fetchBatch fetches a batch of messages from the subscription. -func (c *Consumer) fetchBatch(sub *nats.Subscription) ([]*nats.Msg, error) { - return sub.Fetch(c.batchSize, nats.MaxWait(c.batchTimeout)) +// It respects context cancellation for faster shutdown. +func (c *Consumer) fetchBatch(ctx context.Context, sub *nats.Subscription) ([]*nats.Msg, error) { + // Check context before fetching + select { + case <-ctx.Done(): + return nil, ctx.Err() + default: + } + + // Use a shorter timeout for better responsiveness to cancellation + // The batchTimeout is still used, but we'll check context more frequently + timeout := c.config.batchTimeout + if timeout > 500*time.Millisecond { + // Cap at 500ms to improve responsiveness while still allowing batching + timeout = 500 * time.Millisecond + } + + return sub.Fetch(c.config.batchSize, nats.MaxWait(timeout)) } // handleFetchError handles errors during message fetching, including recovery logic. // Returns true if the error was handled and consumption should continue, false otherwise. func (c *Consumer) handleFetchError(ctx context.Context, err error, sub **nats.Subscription, fetchErrorStreak *int) (bool, error) { - result := c.errorHandler.HandleFetchError(ctx, err, sub, fetchErrorStreak, c.streamName, c.consumerName, func() (*nats.Subscription, error) { + result := c.errorHandler.HandleFetchError(ctx, err, sub, fetchErrorStreak, c.config.streamName, c.config.consumerName, func() (*nats.Subscription, error) { if recErr := c.recoverJetStreamResources(); recErr != nil { return nil, recErr } @@ -121,99 +114,6 @@ func (c *Consumer) handleFetchError(ctx context.Context, err error, sub **nats.S return result.ShouldContinue, result.Error } -// processBatch processes a batch of messages, handling errors and applying backpressure. -func (c *Consumer) processBatch(ctx context.Context, msgs []*nats.Msg) { - for _, msg := range msgs { - c.processSingleMessage(ctx, msg) - } -} - -// processSingleMessage processes a single message with error handling and backpressure. -func (c *Consumer) processSingleMessage(ctx context.Context, msg *nats.Msg) { - start := time.Now() - - if err := c.processMessage(ctx, msg); err != nil { - c.handleMessageError(ctx, msg, err, time.Since(start)) - return - } - - // Successful processing resets the error streak. - if c.consecutiveProcessErrors > 0 { - c.consecutiveProcessErrors = 0 - } - - // ACK the message - if ackErr := msg.Ack(); ackErr != nil { - c.logger.Error("Failed to ACK message", zap.Error(ackErr)) - } else { - elapsed := time.Since(start) - c.telemetry.RecordMessageHandled(ctx, c.streamName, c.consumerName, "ok", elapsed) - } -} - -// handleMessageError handles errors that occur during message processing. -func (c *Consumer) handleMessageError(ctx context.Context, msg *nats.Msg, err error, elapsed time.Duration) { - c.logger.Error("Failed to process message", - zap.String("subject", msg.Subject), - zap.Error(err), - zap.Bool("permanent", app.IsPermanent(err)), - ) - - result := obsmetrics.ResultFail - if app.IsPermanent(err) { - result = obsmetrics.ResultPermanentFail - } - c.telemetry.RecordMessageHandled(ctx, c.streamName, c.consumerName, result, elapsed) - - processingResult := c.errorHandler.HandleProcessingError(c.consecutiveProcessErrors, err, c.logger, msg.Subject) - - if processingResult.IsPermanent { - c.handlePermanentError(ctx, msg, err) - return - } - - c.handleTransientError(ctx, msg, processingResult) -} - -// handlePermanentError handles permanent/poison messages. -func (c *Consumer) handlePermanentError(ctx context.Context, msg *nats.Msg, err error) { - c.consecutiveProcessErrors = 0 - // Poison/permanent message: route to DLQ if configured, then ACK - if dlqErr := c.routeToDLQ(ctx, msg, err); dlqErr != nil { - c.logger.Error("Failed to route permanent-error message to DLQ", zap.Error(dlqErr)) - } - if ackErr := msg.Ack(); ackErr != nil { - c.logger.Error("Failed to ACK permanent-error message", zap.Error(ackErr)) - } -} - -// handleTransientError handles transient errors with backpressure and redelivery. -func (c *Consumer) handleTransientError(ctx context.Context, msg *nats.Msg, processingResult ProcessingErrorResult) { - // Increment error streak - if c.consecutiveProcessErrors < 0 { - c.consecutiveProcessErrors = 0 - } - c.consecutiveProcessErrors++ - - if processingResult.ShouldApplyBackpressure { - c.logger.Warn("Applying backpressure due to consecutive processing errors", - zap.Int("consecutive_errors", c.consecutiveProcessErrors), - zap.Duration("sleep", processingResult.BackpressureDelay), - ) - // Use context-aware sleep instead of blocking time.Sleep - if !sleepWithContext(ctx, processingResult.BackpressureDelay) { - // Context canceled, stop processing - return - } - } - - // Transient error: request redelivery with optional delay - 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)) - } -} - // startJetStream starts the JetStream consumer loop. func (c *Consumer) startJetStream(ctx context.Context) error { // Create pull subscription (with simple self-healing in dev/test). @@ -235,16 +135,16 @@ func (c *Consumer) startJetStream(ctx context.Context) error { defer cleanupSubscriber() c.logger.Info("Started consuming messages", - zap.String("subject", c.subject), - zap.String("consumer", c.consumerName), - zap.String("stream", c.streamName), + zap.String("subject", c.config.subject), + zap.String("consumer", c.config.consumerName), + zap.String("stream", c.config.streamName), ) c.logger.Info("Consumer pull configuration", - zap.Int("batch_size", c.batchSize), - zap.Duration("batch_timeout", c.batchTimeout), + zap.Int("batch_size", c.config.batchSize), + zap.Duration("batch_timeout", c.config.batchTimeout), zap.Int("max_deliver", c.cfg.NATS.ConsumerRules.MaxDeliver), - zap.Duration("ack_wait", c.ackWait), + zap.Duration("ack_wait", c.config.ackWait), zap.String("deliver_policy", c.cfg.NATS.ConsumerRules.DeliverPolicy), zap.String("replay_policy", c.cfg.NATS.ConsumerRules.ReplayPolicy), zap.Int("backoff_steps", len(c.cfg.NATS.ConsumerRules.Backoff)), @@ -254,6 +154,15 @@ func (c *Consumer) startJetStream(ctx context.Context) error { defer statsCancel() go c.emitConsumerStats(statsCtx) + // Start advisory DLQ handler in background if configured + if c.advisoryDLQHandler != nil { + go func() { + if err := c.advisoryDLQHandler.Start(ctx); err != nil { + c.logger.Error("Advisory DLQ handler failed", zap.Error(err)) + } + }() + } + var fetchErrorStreak int for { @@ -265,8 +174,13 @@ func (c *Consumer) startJetStream(ctx context.Context) error { } // Fetch messages in batch - msgs, err := c.fetchBatch(currentSub) + msgs, err := c.fetchBatch(ctx, currentSub) if err != nil { + // If context was cancelled, return immediately + if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) { + c.logger.Info("Stopping consumer due to context cancellation", zap.Error(err)) + return err + } shouldContinue, handleErr := c.handleFetchError(ctx, err, ¤tSub, &fetchErrorStreak) if !shouldContinue { return handleErr diff --git a/internal/infra/nats/consumer_jetstream_test.go b/internal/infra/nats/consumer_jetstream_test.go index 2c552bb..0060fcf 100644 --- a/internal/infra/nats/consumer_jetstream_test.go +++ b/internal/infra/nats/consumer_jetstream_test.go @@ -21,12 +21,14 @@ var _ = Describe("Consumer JetStream", func() { BeforeEach(func() { logger := zaptest.NewLogger(GinkgoT()) c = &Consumer{ - mode: "jetstream", - streamName: "TEST_STREAM", - consumerName: "test-consumer", - subject: "test.subject", - batchSize: 10, - batchTimeout: 2 * time.Second, + config: consumerConfig{ + mode: "jetstream", + streamName: "TEST_STREAM", + consumerName: "test-consumer", + subject: "test.subject", + batchSize: 10, + batchTimeout: 2 * time.Second, + }, logger: logger, errorHandler: NewErrorHandler(logger), cfg: &configpkg.Config{ diff --git a/internal/infra/nats/message_handler.go b/internal/infra/nats/consumer_message.go similarity index 56% rename from internal/infra/nats/message_handler.go rename to internal/infra/nats/consumer_message.go index 117dc6b..81d5790 100644 --- a/internal/infra/nats/message_handler.go +++ b/internal/infra/nats/consumer_message.go @@ -4,6 +4,7 @@ import ( "caatsm/internal/infra/log" "context" "fmt" + "time" "github.com/google/uuid" "github.com/nats-io/nats.go" @@ -13,11 +14,59 @@ import ( "go.uber.org/zap" ) +// processBatch processes a batch of messages, handling errors and applying backpressure. +// It checks context cancellation between messages for faster shutdown. +func (c *Consumer) processBatch(ctx context.Context, msgs []*nats.Msg) { + for _, msg := range msgs { + // Check context before processing each message + select { + case <-ctx.Done(): + c.logger.Info("Stopping batch processing due to cancellation", + zap.Int("remaining_messages", len(msgs)), + ) + return + default: + } + c.processSingleMessage(ctx, msg) + } +} + +// processSingleMessage processes a single message with error handling and backpressure. +func (c *Consumer) processSingleMessage(ctx context.Context, msg *nats.Msg) { + start := time.Now() + + if err := c.processMessage(ctx, msg); err != nil { + c.handleMessageError(ctx, msg, err, time.Since(start)) + return + } + + // Successful processing resets the error streak. + if c.consecutiveProcessErrors > 0 { + c.consecutiveProcessErrors = 0 + } + + // ACK the message + if ackErr := msg.Ack(); ackErr != nil { + c.logger.Error("Failed to ACK message", zap.Error(ackErr)) + } else { + elapsed := time.Since(start) + c.telemetry.RecordMessageHandled(ctx, c.config.streamName, c.config.consumerName, "ok", elapsed) + } +} + // processMessage processes a single message. func (c *Consumer) processMessage(ctx context.Context, msg *nats.Msg) error { ctx, span := otel.Tracer("caatsm/nats").Start(ctx, "Consumer.processMessage") defer span.End() - span.SetAttributes(attribute.String("nats.subject", msg.Subject)) + + // Set semantic messaging attributes + span.SetAttributes( + attribute.String("messaging.system", "nats"), + attribute.String("messaging.operation.name", "receive"), + attribute.String("messaging.destination.name", msg.Subject), + attribute.String("messaging.consumer.group.name", c.config.consumerName), + attribute.String("caatsm.stream", c.config.streamName), + ) msgID, source, err := c.resolveMsgID(msg) if err != nil { @@ -46,8 +95,8 @@ func (c *Consumer) processMessage(ctx context.Context, msg *nats.Msg) error { msgLogger := log.WithMessageContext(c.logger, log.MessageFields{ Service: "caatsm-consumer", TransportMsgID: msgID, - Stream: c.streamName, - Consumer: c.consumerName, + Stream: c.config.streamName, + Consumer: c.config.consumerName, Subject: msg.Subject, JSSequence: jsSeq, }) @@ -74,7 +123,7 @@ func (c *Consumer) resolveMsgID(msg *nats.Msg) (string, string, error) { return id, "header", nil } - if c.mode == "core" { + if c.config.mode == "core" { return uuid.NewString(), "generated", nil } @@ -85,4 +134,3 @@ func (c *Consumer) resolveMsgID(msg *nats.Msg) (string, string, error) { return fmt.Sprintf("js-%d", meta.Sequence.Stream), "metadata", nil } - diff --git a/internal/infra/nats/metrics.go b/internal/infra/nats/consumer_metrics.go similarity index 86% rename from internal/infra/nats/metrics.go rename to internal/infra/nats/consumer_metrics.go index 2f113d1..c889846 100644 --- a/internal/infra/nats/metrics.go +++ b/internal/infra/nats/consumer_metrics.go @@ -49,12 +49,12 @@ func (c *Consumer) recordConsumerMetrics(ctx context.Context, info *nats.Consume // Export an explicit pending messages gauge for Prometheus-based lag / // backlog alerts. - obsmetrics.RecordNATSConsumerPending(c.streamName, c.consumerName, info.NumPending) + obsmetrics.RecordNATSConsumerPending(c.config.streamName, c.config.consumerName, info.NumPending) } // emitConsumerStats periodically emits consumer statistics. func (c *Consumer) emitConsumerStats(ctx context.Context) { - ticker := time.NewTicker(c.monitorInterval) + ticker := time.NewTicker(c.config.monitorInterval) defer ticker.Stop() for { @@ -62,15 +62,15 @@ func (c *Consumer) emitConsumerStats(ctx context.Context) { case <-ctx.Done(): return case <-ticker.C: - info, err := c.js.ConsumerInfo(c.streamName, c.consumerName) + info, err := c.js.ConsumerInfo(c.config.streamName, c.config.consumerName) if err != nil { c.logger.Warn("Failed to fetch consumer info", zap.Error(err)) continue } c.logger.Debug("JetStream consumer metrics", - zap.String("stream", c.streamName), - zap.String("consumer", c.consumerName), + zap.String("stream", c.config.streamName), + zap.String("consumer", c.config.consumerName), zap.Uint64("num_ack_pending", uint64(info.NumAckPending)), zap.Uint64("num_redelivered", uint64(info.NumRedelivered)), zap.Uint64("num_pending", uint64(info.NumPending)), diff --git a/internal/infra/nats/dlq.go b/internal/infra/nats/dlq.go index 3f753ae..e8db18e 100644 --- a/internal/infra/nats/dlq.go +++ b/internal/infra/nats/dlq.go @@ -21,22 +21,22 @@ func (c *Consumer) validateDLQ() error { } // DLQ routing is only active in JetStream mode. - if c.mode != "jetstream" { + if c.config.mode != "jetstream" { return nil } // If DLQ is not enabled in config, make sure we don't accidentally route to it. if !c.cfg.DLQ.Enabled { - if strings.TrimSpace(c.dlqSubject) != "" { + if strings.TrimSpace(c.config.dlqSubject) != "" { c.logger.Info("DLQ subject configured but dlq.enabled is false; DLQ routing disabled", - zap.String("dlq_subject", c.dlqSubject), + zap.String("dlq_subject", c.config.dlqSubject), ) } - c.dlqSubject = "" + c.config.dlqSubject = "" return nil } - subject := strings.TrimSpace(c.dlqSubject) + subject := strings.TrimSpace(c.config.dlqSubject) if subject == "" { return fmt.Errorf("DLQ enabled but dlq.subject is empty") } @@ -68,10 +68,10 @@ func (c *Consumer) routeToDLQ(ctx context.Context, msg *nats.Msg, cause error) e if c == nil || c.js == nil { return nil } - if c.mode != "jetstream" { + if c.config.mode != "jetstream" { return nil } - if strings.TrimSpace(c.dlqSubject) == "" { + if strings.TrimSpace(c.config.dlqSubject) == "" { return nil } @@ -83,11 +83,11 @@ func (c *Consumer) routeToDLQ(ctx context.Context, msg *nats.Msg, cause error) e deliveries = meta.NumDelivered } - payload := map[string]interface{}{ + payload := map[string]any{ "transport_msg_id": msg.Header.Get("Nats-Msg-Id"), "subject": msg.Subject, - "stream": c.streamName, - "consumer": c.consumerName, + "stream": c.config.streamName, + "consumer": c.config.consumerName, "nats_sequence": jsSeq, "deliveries": deliveries, "error": fmt.Sprint(cause), @@ -98,42 +98,42 @@ func (c *Consumer) routeToDLQ(ctx context.Context, msg *nats.Msg, cause error) e data, err := json.Marshal(payload) if err != nil { c.logger.Error("failed to marshal DLQ payload", - zap.String("stream", c.streamName), - zap.String("consumer", c.consumerName), - zap.String("dlq_subject", c.dlqSubject), + zap.String("stream", c.config.streamName), + zap.String("consumer", c.config.consumerName), + zap.String("dlq_subject", c.config.dlqSubject), zap.Error(err), ) return fmt.Errorf("marshal dlq payload: %w", err) } - if _, err := c.js.Publish(c.dlqSubject, data); err != nil { + if _, err := c.js.Publish(c.config.dlqSubject, data); err != nil { // nats.ErrNoResponders typically means that no JetStream stream is // configured to receive this subject, or JetStream is temporarily // unavailable. Surface this explicitly to make operational diagnosis // easier. if errors.Is(err, nats.ErrNoResponders) { c.logger.Error("transient DLQ publish error (no responders)", - zap.String("stream", c.streamName), - zap.String("consumer", c.consumerName), - zap.String("dlq_subject", c.dlqSubject), + zap.String("stream", c.config.streamName), + zap.String("consumer", c.config.consumerName), + zap.String("dlq_subject", c.config.dlqSubject), zap.Int("payload_size", len(data)), zap.Error(err), ) - 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) + c.telemetry.RecordDLQPublishFailure(ctx, c.config.streamName, c.config.consumerName) + return fmt.Errorf("publish to dlq subject %s: no JetStream stream found for subject or JetStream unavailable: %w", c.config.dlqSubject, err) } c.logger.Error("failed to publish to DLQ", - zap.String("stream", c.streamName), - zap.String("consumer", c.consumerName), - zap.String("dlq_subject", c.dlqSubject), + zap.String("stream", c.config.streamName), + zap.String("consumer", c.config.consumerName), + zap.String("dlq_subject", c.config.dlqSubject), zap.Int("payload_size", len(data)), zap.Error(err), ) - c.telemetry.RecordDLQPublishFailure(ctx, c.streamName, c.consumerName) - return fmt.Errorf("publish to dlq subject %s: %w", c.dlqSubject, err) + c.telemetry.RecordDLQPublishFailure(ctx, c.config.streamName, c.config.consumerName) + return fmt.Errorf("publish to dlq subject %s: %w", c.config.dlqSubject, err) } - c.telemetry.RecordDLQMessage(ctx, c.streamName, c.consumerName) + c.telemetry.RecordDLQMessage(ctx, c.config.streamName, c.config.consumerName) return nil } diff --git a/internal/infra/nats/dlq_test.go b/internal/infra/nats/dlq_test.go index b5ef616..56c09ee 100644 --- a/internal/infra/nats/dlq_test.go +++ b/internal/infra/nats/dlq_test.go @@ -7,9 +7,9 @@ import ( configpkg "caatsm/internal/infra/config" "caatsm/internal/infra/telemetry" + "github.com/nats-io/nats.go" . "github.com/onsi/ginkgo/v2" . "github.com/onsi/gomega" - "github.com/nats-io/nats.go" "go.uber.org/zap" "go.uber.org/zap/zaptest" ) @@ -31,7 +31,9 @@ var _ = Describe("DLQ", func() { It("returns nil when mode is not jetstream", func() { c := &Consumer{ - mode: "core", + config: consumerConfig{ + mode: "core", + }, cfg: &configpkg.Config{ DLQ: configpkg.DLQConfig{ Enabled: true, @@ -45,8 +47,10 @@ var _ = Describe("DLQ", func() { It("clears dlqSubject when DLQ is disabled", func() { c := &Consumer{ - mode: "jetstream", - dlqSubject: "caatsm.dlq", + config: consumerConfig{ + mode: "jetstream", + dlqSubject: "caatsm.dlq", + }, cfg: &configpkg.Config{ DLQ: configpkg.DLQConfig{ Enabled: false, @@ -56,13 +60,15 @@ var _ = Describe("DLQ", func() { logger: logger, } Expect(c.validateDLQ()).To(Succeed()) - Expect(c.dlqSubject).To(Equal("")) + Expect(c.config.dlqSubject).To(Equal("")) }) It("returns error when DLQ is enabled but subject is empty", func() { c := &Consumer{ - mode: "jetstream", - dlqSubject: "", + config: consumerConfig{ + mode: "jetstream", + dlqSubject: "", + }, cfg: &configpkg.Config{ DLQ: configpkg.DLQConfig{ Enabled: true, @@ -78,9 +84,11 @@ var _ = Describe("DLQ", func() { It("returns error when DLQ is enabled but JetStream context is nil", func() { c := &Consumer{ - mode: "jetstream", - dlqSubject: "caatsm.dlq", - js: nil, + config: consumerConfig{ + mode: "jetstream", + dlqSubject: "caatsm.dlq", + }, + js: nil, cfg: &configpkg.Config{ DLQ: configpkg.DLQConfig{ Enabled: true, @@ -107,7 +115,9 @@ var _ = Describe("DLQ", func() { It("returns nil when mode is not jetstream", func() { c := &Consumer{ - mode: "core", + config: consumerConfig{ + mode: "core", + }, } ctx := context.Background() msg := &nats.Msg{} @@ -117,9 +127,11 @@ var _ = Describe("DLQ", func() { It("returns nil when dlqSubject is empty", func() { c := &Consumer{ - mode: "jetstream", - dlqSubject: "", - js: nil, // Can be nil for this test + config: consumerConfig{ + mode: "jetstream", + dlqSubject: "", + }, + js: nil, // Can be nil for this test } ctx := context.Background() msg := &nats.Msg{} @@ -128,4 +140,3 @@ var _ = Describe("DLQ", func() { }) }) }) - diff --git a/internal/infra/nats/error_handler.go b/internal/infra/nats/error_handler.go index d9b4bb5..dd32ead 100644 --- a/internal/infra/nats/error_handler.go +++ b/internal/infra/nats/error_handler.go @@ -47,9 +47,7 @@ func (h *ErrorHandler) HandleFetchError( if errors.Is(err, nats.ErrNoResponders) { *fetchErrorStreak++ backoff := time.Duration(*fetchErrorStreak) * time.Second - if backoff > 30*time.Second { - backoff = 30 * time.Second - } + backoff = min(backoff, 30*time.Second) h.logger.Warn("JetStream not available, will retry with backoff", zap.Error(err), zap.String("stream", streamName), @@ -91,9 +89,7 @@ func (h *ErrorHandler) HandleFetchError( // Generic error path with modest backoff. *fetchErrorStreak++ backoff := time.Duration(*fetchErrorStreak) * time.Second - if backoff > 10*time.Second { - backoff = 10 * time.Second - } + backoff = min(backoff, 10*time.Second) h.logger.Error("Failed to fetch messages; backing off", zap.Error(err), zap.Duration("backoff", backoff), @@ -133,9 +129,7 @@ func (h *ErrorHandler) HandleProcessingError( if consecutiveErrors >= 10 { result.ShouldApplyBackpressure = true result.BackpressureDelay = time.Duration(consecutiveErrors) * 100 * time.Millisecond - if result.BackpressureDelay > 5*time.Second { - result.BackpressureDelay = 5 * time.Second - } + result.BackpressureDelay = min(result.BackpressureDelay, 5*time.Second) } return result diff --git a/internal/infra/nats/jetstream.go b/internal/infra/nats/jetstream.go index 62719e7..a6411b9 100644 --- a/internal/infra/nats/jetstream.go +++ b/internal/infra/nats/jetstream.go @@ -2,9 +2,7 @@ package nats import ( "caatsm/internal/infra/config" - "errors" "fmt" - "os" "strings" "github.com/nats-io/nats.go" @@ -65,22 +63,7 @@ func ProvideJetStream(nc *nats.Conn, cfg *config.Config, logger *zap.Logger) (na return nil, fmt.Errorf("failed to get JetStream context: %w", err) } - // Ensure the stream exists and is minimally aligned with configuration. - if err := EnsureStream(js, cfg, logger); err != nil { - nc.Close() - return nil, err - } - - return js, nil -} - -// EnsureStream ensures that the configured JetStream stream exists and has -// at least the expected subjects bound. It is safe to call multiple times. -// -// In dev/test environments (see shouldBootstrapStream), the stream will be -// auto-created if it does not exist. In production, a missing stream results -// in an error so that operators can intervene. -func EnsureStream(js nats.JetStreamContext, cfg *config.Config, logger *zap.Logger) error { + // Ensure the stream exists using StreamManager streamName := cfg.NATS.Stream consumerSubject := cfg.EffectiveSubscriptionTopic() publisherSubject := strings.TrimSpace(cfg.Publisher.Topic) @@ -92,130 +75,15 @@ func EnsureStream(js nats.JetStreamContext, cfg *config.Config, logger *zap.Logg zap.String("consumer_subject", consumerSubject), zap.String("publisher_subject", publisherSubject), ) - return fmt.Errorf("no subjects configured for JetStream stream %s", streamName) + nc.Close() + return nil, fmt.Errorf("no subjects configured for JetStream stream %s", streamName) } - streamLimits := cfg.NATS.StreamLimits - storage := nats.FileStorage - switch strings.ToLower(streamLimits.Storage) { - case "memory": - storage = nats.MemoryStorage - case "file": - storage = nats.FileStorage + streamManager := NewStreamManagerWithConfig(js, streamName, streamSubjects, &cfg.NATS.StreamLimits, logger) + if err := streamManager.EnsureStream(); err != nil { + nc.Close() + return nil, err } - discard := nats.DiscardOld - if strings.EqualFold(streamLimits.Discard, "new") { - discard = nats.DiscardNew - } - - streamConfig := &nats.StreamConfig{ - Name: streamName, - Subjects: streamSubjects, - Retention: nats.LimitsPolicy, - MaxMsgs: streamLimits.MaxMsgs, - MaxBytes: streamLimits.MaxBytes, - MaxAge: streamLimits.MaxAge, - Discard: discard, - Storage: storage, - Replicas: streamLimits.Replicas, - } - - info, err := js.StreamInfo(streamName) - if err != nil { - if errors.Is(err, nats.ErrStreamNotFound) { - if shouldBootstrapStream() { - if _, err = js.AddStream(streamConfig); err != nil { - logger.Error("failed to create stream", - zap.String("stream", streamName), - zap.Strings("subjects", streamSubjects), - zap.Error(err), - ) - return fmt.Errorf("failed to create stream %s: %w", streamName, err) - } - logger.Info("Created JetStream stream", - zap.String("stream", streamName), - zap.Strings("subjects", streamSubjects), - ) - return nil - } - logger.Error("stream not found and auto-creation disabled", - zap.String("stream", streamName), - zap.Strings("expected_subjects", streamSubjects), - ) - return fmt.Errorf("stream %s not found and auto-creation disabled", streamName) - } - logger.Error("failed to fetch stream info", - zap.String("stream", streamName), - zap.Error(err), - ) - return fmt.Errorf("failed to fetch stream info for %s: %w", streamName, err) - } - - // Stream exists: validate subjects but do not fail hard if they differ. - validateStreamConfig(info, streamSubjects, logger) - return nil -} - -func shouldBootstrapStream() bool { - switch strings.ToLower(os.Getenv("GO_ENV")) { - case "", "dev", "development", "test", "testing": - return true - default: - return false - } -} - -func validateStreamConfig(info *nats.StreamInfo, expectedSubjects []string, logger *zap.Logger) { - if info == nil { - return - } - defer func() { - if len(expectedSubjects) == 0 { - expectedSubjects = []string{""} - } - }() - - missing := make([]string, 0) - for _, subj := range expectedSubjects { - if subj == "" { - continue - } - if !containsSubject(info.Config.Subjects, subj) { - missing = append(missing, subj) - } - } - if len(missing) > 0 { - logger.Warn("JetStream stream subjects missing expected entries", - zap.String("stream", info.Config.Name), - zap.Strings("stream_subjects", info.Config.Subjects), - zap.Strings("missing_subjects", missing), - ) - } -} - -func containsSubject(subjects []string, target string) bool { - for _, s := range subjects { - if s == target { - return true - } - } - return false -} - -func dedupeSubjects(subjects []string) []string { - seen := make(map[string]struct{}) - result := make([]string, 0, len(subjects)) - for _, subj := range subjects { - subj = strings.TrimSpace(subj) - if subj == "" { - continue - } - if _, ok := seen[subj]; ok { - continue - } - seen[subj] = struct{}{} - result = append(result, subj) - } - return result + return js, nil } diff --git a/internal/infra/nats/message_handler_test.go b/internal/infra/nats/message_handler_test.go index d5a411e..a937e14 100644 --- a/internal/infra/nats/message_handler_test.go +++ b/internal/infra/nats/message_handler_test.go @@ -1,9 +1,9 @@ package nats import ( + "github.com/nats-io/nats.go" . "github.com/onsi/ginkgo/v2" . "github.com/onsi/gomega" - "github.com/nats-io/nats.go" "go.uber.org/zap/zaptest" ) @@ -14,10 +14,12 @@ var _ = Describe("MessageHandler", func() { BeforeEach(func() { c = &Consumer{ - mode: "jetstream", - streamName: "TEST_STREAM", - consumerName: "test-consumer", - logger: zaptest.NewLogger(GinkgoT()), + config: consumerConfig{ + mode: "jetstream", + streamName: "TEST_STREAM", + consumerName: "test-consumer", + }, + logger: zaptest.NewLogger(GinkgoT()), } }) @@ -35,7 +37,7 @@ var _ = Describe("MessageHandler", func() { }) It("generates UUID for core mode when header is missing", func() { - c.mode = "core" + c.config.mode = "core" msg := &nats.Msg{ Header: nats.Header{}, } @@ -47,7 +49,7 @@ var _ = Describe("MessageHandler", func() { }) It("returns error for JetStream mode when header and metadata are missing", func() { - c.mode = "jetstream" + c.config.mode = "jetstream" msg := &nats.Msg{ Header: nats.Header{}, } @@ -59,4 +61,3 @@ var _ = Describe("MessageHandler", func() { }) }) }) - diff --git a/internal/infra/nats/metrics_test.go b/internal/infra/nats/metrics_test.go index 9508e98..a1d6b4f 100644 --- a/internal/infra/nats/metrics_test.go +++ b/internal/infra/nats/metrics_test.go @@ -3,9 +3,9 @@ package nats import ( "context" + "github.com/nats-io/nats.go" . "github.com/onsi/ginkgo/v2" . "github.com/onsi/gomega" - "github.com/nats-io/nats.go" "go.uber.org/zap/zaptest" ) @@ -18,9 +18,11 @@ var _ = Describe("Metrics", func() { BeforeEach(func() { ctx = context.Background() c = &Consumer{ - streamName: "TEST_STREAM", - consumerName: "test-consumer", - logger: zaptest.NewLogger(GinkgoT()), + config: consumerConfig{ + streamName: "TEST_STREAM", + consumerName: "test-consumer", + }, + logger: zaptest.NewLogger(GinkgoT()), } }) @@ -54,4 +56,3 @@ var _ = Describe("Metrics", func() { }) }) }) - diff --git a/internal/infra/nats/publisher.go b/internal/infra/nats/publisher.go index d8608fd..c6afe2a 100644 --- a/internal/infra/nats/publisher.go +++ b/internal/infra/nats/publisher.go @@ -1,8 +1,8 @@ package nats import ( - "caatsm/internal/infra/config" "caatsm/internal/adapter/dto" + "caatsm/internal/infra/config" "caatsm/internal/port" "encoding/json" "errors" @@ -59,7 +59,7 @@ func ProvideCorePublisher( } // Publish publishes a message using plain NATS -func (p *CorePublisher) Publish(message interface{}) error { +func (p *CorePublisher) Publish(message any) error { topic := p.cfg.Publisher.Topic if topic == "" { p.logger.Error("publisher topic is not configured") @@ -96,7 +96,7 @@ func (p *CorePublisher) Publish(message interface{}) error { } // Publish publishes a message -func (p *Publisher) Publish(message interface{}) error { +func (p *Publisher) Publish(message any) error { topic := p.cfg.Publisher.Topic if topic == "" { p.logger.Error("publisher topic is not configured") diff --git a/internal/infra/nats/stream_manager.go b/internal/infra/nats/stream_manager.go index b6ca813..258dcd3 100644 --- a/internal/infra/nats/stream_manager.go +++ b/internal/infra/nats/stream_manager.go @@ -1,8 +1,10 @@ package nats import ( + "caatsm/internal/infra/config" "errors" "fmt" + "strings" "github.com/nats-io/nats.go" "go.uber.org/zap" @@ -14,6 +16,7 @@ type StreamManager struct { streamName string subjects []string logger *zap.Logger + cfg *config.StreamLimitsConfig } // NewStreamManager creates a new stream manager @@ -26,14 +29,21 @@ func NewStreamManager(js nats.JetStreamContext, streamName string, subjects []st } } +// NewStreamManagerWithConfig creates a new stream manager with full stream configuration +func NewStreamManagerWithConfig(js nats.JetStreamContext, streamName string, subjects []string, streamLimits *config.StreamLimitsConfig, logger *zap.Logger) *StreamManager { + return &StreamManager{ + js: js, + streamName: streamName, + subjects: subjects, + logger: logger, + cfg: streamLimits, + } +} + // EnsureStream ensures that the configured JetStream stream exists func (sm *StreamManager) EnsureStream() error { - streamConfig := &nats.StreamConfig{ - Name: sm.streamName, - Subjects: sm.subjects, - Retention: nats.LimitsPolicy, - Storage: nats.FileStorage, - } + // Build stream configuration + streamConfig := sm.buildStreamConfig() info, err := sm.js.StreamInfo(sm.streamName) if err != nil { @@ -71,6 +81,41 @@ func (sm *StreamManager) EnsureStream() error { return nil } +// buildStreamConfig builds the stream configuration from manager settings +func (sm *StreamManager) buildStreamConfig() *nats.StreamConfig { + config := &nats.StreamConfig{ + Name: sm.streamName, + Subjects: sm.subjects, + Retention: nats.LimitsPolicy, + Storage: nats.FileStorage, + } + + // Apply stream limits configuration if provided + if sm.cfg != nil { + config.MaxMsgs = sm.cfg.MaxMsgs + config.MaxBytes = sm.cfg.MaxBytes + config.MaxAge = sm.cfg.MaxAge + config.Replicas = sm.cfg.Replicas + + // Map storage type + switch strings.ToLower(sm.cfg.Storage) { + case "memory": + config.Storage = nats.MemoryStorage + case "file": + config.Storage = nats.FileStorage + } + + // Map discard policy + if strings.EqualFold(sm.cfg.Discard, "new") { + config.Discard = nats.DiscardNew + } else { + config.Discard = nats.DiscardOld + } + } + + return config +} + // validateStreamConfig validates the stream configuration func (sm *StreamManager) validateStreamConfig(info *nats.StreamInfo) { if info == nil { diff --git a/internal/infra/nats/utils.go b/internal/infra/nats/utils.go index 3c7da7e..2c16eab 100644 --- a/internal/infra/nats/utils.go +++ b/internal/infra/nats/utils.go @@ -78,3 +78,41 @@ func mapReplayPolicy(value string) nats.ReplayPolicy { return nats.ReplayInstantPolicy } } + +// shouldBootstrapStream checks if streams should be auto-created based on environment. +func shouldBootstrapStream() bool { + switch strings.ToLower(os.Getenv("GO_ENV")) { + case "", "dev", "development", "test", "testing": + return true + default: + return false + } +} + +// containsSubject checks if a subject exists in a list of subjects. +func containsSubject(subjects []string, target string) bool { + for _, s := range subjects { + if s == target { + return true + } + } + return false +} + +// dedupeSubjects removes duplicate and empty subjects from a list. +func dedupeSubjects(subjects []string) []string { + seen := make(map[string]struct{}) + result := make([]string, 0, len(subjects)) + for _, subj := range subjects { + subj = strings.TrimSpace(subj) + if subj == "" { + continue + } + if _, ok := seen[subj]; ok { + continue + } + seen[subj] = struct{}{} + result = append(result, subj) + } + return result +} diff --git a/internal/infra/nats/utils_test.go b/internal/infra/nats/utils_test.go index 50f12e7..799cfe4 100644 --- a/internal/infra/nats/utils_test.go +++ b/internal/infra/nats/utils_test.go @@ -84,4 +84,3 @@ var _ = Describe("Utils", func() { }) }) }) - diff --git a/internal/infra/postgres/repository.go b/internal/infra/postgres/repository.go index cfa2c35..c2a24b7 100644 --- a/internal/infra/postgres/repository.go +++ b/internal/infra/postgres/repository.go @@ -39,7 +39,16 @@ func ProvideRepository(pool *pgxpool.Pool, logger *zap.Logger) (port.Repository, func (r *Repository) InsertOne(ctx context.Context, msg *dto.ParsedTelegram) error { ctx, span := otel.Tracer("caatsm/postgres").Start(ctx, "Repository.InsertOne") defer span.End() - span.SetAttributes(attribute.String("db.table", "aviation.telegrams")) + + // Set semantic database attributes + span.SetAttributes( + attribute.String("db.system", "postgresql"), + attribute.String("db.operation", "insert"), + attribute.String("db.name", "aviation"), + attribute.String("db.table", "telegrams"), + attribute.String("caatsm.message.id", msg.MessageID), + attribute.String("caatsm.message.category", msg.Category), + ) // Optional idempotency check based on business message identity. If we have a // non-empty message ID and date/time, we can cheaply skip duplicates here to diff --git a/internal/infra/telemetry/otel.go b/internal/infra/telemetry/otel.go new file mode 100644 index 0000000..92480fb --- /dev/null +++ b/internal/infra/telemetry/otel.go @@ -0,0 +1,248 @@ +package telemetry + +import ( + "caatsm/internal/infra/buildinfo" + "caatsm/internal/infra/config" + "context" + "crypto/tls" + "fmt" + "os" + "sync" + "time" + + "github.com/google/uuid" + "go.opentelemetry.io/otel" + "go.opentelemetry.io/otel/attribute" + "go.opentelemetry.io/otel/exporters/otlp/otlpmetric/otlpmetrichttp" + "go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp" + sdkmetric "go.opentelemetry.io/otel/sdk/metric" + "go.opentelemetry.io/otel/sdk/resource" + sdktrace "go.opentelemetry.io/otel/sdk/trace" + semconv "go.opentelemetry.io/otel/semconv/v1.21.0" + "go.uber.org/zap" +) + +var ( + // instanceID stores a unique identifier for this running process instance. + // It is initialized once at startup and remains stable for the lifetime of the process. + instanceID string + instanceIDOnce sync.Once +) + +// InitOTEL initializes the OpenTelemetry SDK with proper resource attributes, +// sampling configuration, and exporters. This should be called once at application startup. +// logger is optional; if provided, sensitive telemetry configuration will be logged at debug level. +func InitOTEL(ctx context.Context, cfg *config.Config, logger *zap.Logger) error { + if !cfg.Telemetry.Enabled { + return nil + } + + // Log sensitive telemetry configuration to internal debug logs only + if logger != nil { + logger.Debug("Initializing OpenTelemetry", + zap.String("telemetry.endpoint", cfg.Telemetry.Endpoint), + zap.Bool("telemetry.insecure", cfg.Telemetry.Insecure), + ) + } + + // Create resource with comprehensive service information + res, err := createResource(ctx, cfg) + if err != nil { + return fmt.Errorf("failed to create OTEL resource: %w", err) + } + + // Initialize tracing + if err := initTracing(ctx, cfg, res); err != nil { + return fmt.Errorf("failed to initialize tracing: %w", err) + } + + // Initialize metrics + if err := initMetrics(ctx, cfg, res); err != nil { + return fmt.Errorf("failed to initialize metrics: %w", err) + } + + return nil +} + +// getInstanceID returns a unique identifier for this running process instance. +// It checks environment variables (POD_NAME, CONTAINER_ID, HOSTNAME) first, then generates +// a UUID if no environment variable is available. The ID is initialized once and remains +// stable for the lifetime of the process. +func getInstanceID() string { + instanceIDOnce.Do(func() { + // Check for Kubernetes pod name first (most common in containerized deployments) + if podName := os.Getenv("POD_NAME"); podName != "" { + instanceID = podName + return + } + + // Check for container ID (Docker, containerd, etc.) + if containerID := os.Getenv("CONTAINER_ID"); containerID != "" { + instanceID = containerID + return + } + + // Check for HOSTNAME (often set in containers) + if hostname := os.Getenv("HOSTNAME"); hostname != "" { + // Use hostname if it's not a generic default + if hostname != "localhost" && hostname != "localhost.localdomain" { + instanceID = hostname + return + } + } + + // Generate a UUID for this process instance + instanceID = uuid.NewString() + }) + + return instanceID +} + +// createResource creates a resource with standard and custom attributes. +// Sensitive infrastructure details (endpoint, insecure flag) are excluded from resource +// attributes to prevent leakage. These values are logged internally at debug level if +// a logger is provided to InitOTEL. +func createResource(ctx context.Context, cfg *config.Config) (*resource.Resource, error) { + // Get the runtime instance ID (falls back to buildinfo.Commit if needed) + runtimeInstanceID := getInstanceID() + if runtimeInstanceID == "" { + // Final fallback to build commit if somehow instance ID is empty + runtimeInstanceID = buildinfo.Commit + } + + attrs := []attribute.KeyValue{ + // Standard semantic conventions + semconv.ServiceName("caatsm"), + semconv.ServiceVersion(buildinfo.Version), + semconv.ServiceInstanceID(runtimeInstanceID), + semconv.ServiceNamespace("airport"), + + // Custom attributes + attribute.String("service.component", "receiver"), + attribute.String("service.environment", getEnvironment()), + attribute.String("build.commit", buildinfo.Commit), + attribute.String("build.built_at", buildinfo.BuiltAt), + } + + // Add non-sensitive indicator for telemetry endpoint configuration + // (without exposing the actual endpoint value) + if cfg.Telemetry.Endpoint != "" { + attrs = append(attrs, attribute.Bool("telemetry.endpoint.configured", true)) + } else { + attrs = append(attrs, attribute.Bool("telemetry.endpoint.configured", false)) + } + + return resource.New(ctx, resource.WithAttributes(attrs...)) +} + +// 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)) + + if cfg.Telemetry.Insecure { + traceExporterOptions = append(traceExporterOptions, otlptracehttp.WithTLSClientConfig(&tls.Config{ + InsecureSkipVerify: true, + MinVersion: tls.VersionTLS13, + })) + } + + traceExporter, err := otlptracehttp.New(ctx, traceExporterOptions...) + if err != nil { + return fmt.Errorf("failed to create trace exporter: %w", err) + } + + // Configure sampling based on environment + sampler := getSamplerForEnvironment(getEnvironment()) + + tracerProvider := sdktrace.NewTracerProvider( + sdktrace.WithBatcher(traceExporter, + sdktrace.WithBatchTimeout(1*time.Second), + sdktrace.WithMaxExportBatchSize(512), + sdktrace.WithMaxQueueSize(2048), + ), + sdktrace.WithResource(res), + sdktrace.WithSampler(sdktrace.ParentBased(sampler)), + ) + + otel.SetTracerProvider(tracerProvider) + return nil +} + +// initMetrics sets up the meter provider +func initMetrics(ctx context.Context, cfg *config.Config, res *resource.Resource) error { + var metricExporterOptions []otlpmetrichttp.Option + + metricExporterOptions = append(metricExporterOptions, otlpmetrichttp.WithEndpoint(cfg.Telemetry.Endpoint)) + + if cfg.Telemetry.Insecure { + metricExporterOptions = append(metricExporterOptions, otlpmetrichttp.WithTLSClientConfig(&tls.Config{ + InsecureSkipVerify: true, + MinVersion: tls.VersionTLS12, + })) + } + + metricExporter, err := otlpmetrichttp.New(ctx, metricExporterOptions...) + if err != nil { + return fmt.Errorf("failed to create metric exporter: %w", err) + } + + meterProvider := sdkmetric.NewMeterProvider( + sdkmetric.WithReader(sdkmetric.NewPeriodicReader(metricExporter, + sdkmetric.WithInterval(30*time.Second), + )), + sdkmetric.WithResource(res), + ) + + otel.SetMeterProvider(meterProvider) + return nil +} + +// getSamplerForEnvironment returns appropriate sampling strategy for each environment +func getSamplerForEnvironment(env string) sdktrace.Sampler { + switch env { + case "prod", "production": + // 1% sampling in production to control costs and performance + return sdktrace.TraceIDRatioBased(0.01) + case "staging": + // 10% sampling in staging for better observability + return sdktrace.TraceIDRatioBased(0.1) + case "test", "testing": + // Always sample in testing for complete coverage + return sdktrace.AlwaysSample() + default: + // 100% sampling in development for debugging + return sdktrace.AlwaysSample() + } +} + +// ShutdownOTEL gracefully shuts down the OTEL providers +func ShutdownOTEL(ctx context.Context) error { + var errs []error + + if tracerProvider, ok := otel.GetTracerProvider().(*sdktrace.TracerProvider); ok { + if err := tracerProvider.Shutdown(ctx); err != nil { + errs = append(errs, fmt.Errorf("failed to shutdown tracer provider: %w", err)) + } + } + + if meterProvider, ok := otel.GetMeterProvider().(*sdkmetric.MeterProvider); ok { + if err := meterProvider.Shutdown(ctx); err != nil { + errs = append(errs, fmt.Errorf("failed to shutdown meter provider: %w", err)) + } + } + + if len(errs) > 0 { + return fmt.Errorf("OTEL shutdown errors: %v", errs) + } + return nil +} + +// getEnvironment returns the current environment from GO_ENV +func getEnvironment() string { + if env := os.Getenv("GO_ENV"); env != "" { + return env + } + return "dev" +} diff --git a/test/integration/advisory_dlq_test.go b/test/integration/advisory_dlq_test.go new file mode 100644 index 0000000..15c8aae --- /dev/null +++ b/test/integration/advisory_dlq_test.go @@ -0,0 +1,196 @@ +//go:build integration + +package integration + +import ( + "context" + "encoding/json" + "sync" + "testing" + "time" + + "caatsm/internal/infra/config" + loginfra "caatsm/internal/infra/log" + natsinfra "caatsm/internal/infra/nats" + + "github.com/nats-io/nats.go" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestAdvisoryDLQHandler(t *testing.T) { + if testing.Short() { + t.Skip("skipping integration test in short mode") + } + + ctx, cancel := context.WithTimeout(context.Background(), 3*time.Minute) + defer cancel() + + // Start NATS container + natsContainer, natsURL := startNATS(ctx, t) + defer func() { + _ = natsContainer.Terminate(context.Background()) + }() + + // Connect to NATS + nc, err := nats.Connect(natsURL) + require.NoError(t, err) + defer nc.Close() + + js, err := nc.JetStream() + require.NoError(t, err) + + streamName := "TEST_ADVISORY_STREAM" + consumerName := "test-advisory-consumer" + dlqSubject := "test.dlq" + + // Create stream + streamConfig := &nats.StreamConfig{ + Name: streamName, + Subjects: []string{"test.orders.*"}, + Storage: nats.FileStorage, + } + _, err = js.AddStream(streamConfig) + require.NoError(t, err) + defer func() { + _ = js.DeleteStream(streamName) + }() + + // Create DLQ stream + dlqStreamConfig := &nats.StreamConfig{ + Name: "TEST_DLQ_STREAM", + Subjects: []string{dlqSubject}, + Storage: nats.FileStorage, + } + _, err = js.AddStream(dlqStreamConfig) + require.NoError(t, err) + defer func() { + _ = js.DeleteStream("TEST_DLQ_STREAM") + }() + + // Create consumer with MaxDeliver = 2 for testing + consumerConfig := &nats.ConsumerConfig{ + Durable: consumerName, + AckPolicy: nats.AckExplicitPolicy, + MaxDeliver: 2, // Low value for testing + AckWait: 5 * time.Second, + } + _, err = js.AddConsumer(streamName, consumerConfig) + require.NoError(t, err) + + // Create a mock telemetry recorder + telemetry := &mockTelemetryRecorder{} + + // Create a simple logger for the test + logger, err := loginfra.ProvideLogger(&config.Config{ + Log: config.LogConfig{ + Level: "info", + Format: "console", + }, + }) + require.NoError(t, err) + defer logger.Sync() + + // Create advisory DLQ handler + handler, err := natsinfra.NewAdvisoryDLQHandler( + js, + nc, + streamName, + consumerName, + dlqSubject, + logger, + telemetry, + ) + require.NoError(t, err) + + // Start handler in background + handlerCtx, handlerCancel := context.WithCancel(ctx) + defer handlerCancel() + go func() { + _ = handler.Start(handlerCtx) + }() + + // Give handler time to subscribe + time.Sleep(100 * time.Millisecond) + + // Publish a message that will fail processing + testSubject := "test.orders.1" + testData := []byte("test message data") + _, err = js.Publish(testSubject, testData) + require.NoError(t, err) + + // Create pull subscription and fetch message + sub, err := js.PullSubscribe(testSubject, consumerName, nats.Bind(streamName, consumerName)) + require.NoError(t, err) + defer sub.Unsubscribe() + + // Fetch and NAK the message multiple times to exhaust MaxDeliver + msgs, err := sub.Fetch(1, nats.MaxWait(2*time.Second)) + require.NoError(t, err) + require.Len(t, msgs, 1) + + msg := msgs[0] + // NAK first time + err = msg.Nak() + require.NoError(t, err) + + // Wait for redelivery and NAK again to exhaust MaxDeliver + time.Sleep(6 * time.Second) // Wait for ack_wait + some buffer + + msgs, err = sub.Fetch(1, nats.MaxWait(2*time.Second)) + if err == nil && len(msgs) > 0 { + // NAK second time to exhaust MaxDeliver + err = msgs[0].Nak() + require.NoError(t, err) + } + + // Wait for advisory message to be processed + time.Sleep(2 * time.Second) + + // Verify message was published to DLQ + dlqSub, err := js.SubscribeSync(dlqSubject) + require.NoError(t, err) + defer dlqSub.Unsubscribe() + + dlqMsg, err := dlqSub.NextMsg(5 * time.Second) + if assert.NoError(t, err, "Expected message in DLQ") { + var payload map[string]interface{} + err = json.Unmarshal(dlqMsg.Data, &payload) + require.NoError(t, err) + + // Verify payload structure + assert.Equal(t, streamName, payload["stream"]) + assert.Equal(t, consumerName, payload["consumer"]) + assert.True(t, payload["advisory_source"].(bool)) + assert.Contains(t, payload["error"].(string), "exhausted max_deliver") + assert.Equal(t, string(testData), payload["body"]) + } + + // Verify telemetry was called + assert.Greater(t, telemetry.getDLQMessages(), 0, "Expected DLQ message to be recorded") +} + +// mockTelemetryRecorder implements TelemetryRecorder for testing +type mockTelemetryRecorder struct { + mu sync.RWMutex + dlqMessages int + dlqFailures int +} + +func (m *mockTelemetryRecorder) RecordDLQMessage(ctx context.Context, stream, consumer string) { + m.mu.Lock() + defer m.mu.Unlock() + m.dlqMessages++ +} + +func (m *mockTelemetryRecorder) RecordDLQPublishFailure(ctx context.Context, stream, consumer string) { + m.mu.Lock() + defer m.mu.Unlock() + m.dlqFailures++ +} + +func (m *mockTelemetryRecorder) getDLQMessages() int { + m.mu.RLock() + defer m.mu.RUnlock() + return m.dlqMessages +}