From 53208997e8f7cf053d9bde6be88a2959129bb91e Mon Sep 17 00:00:00 2001 From: windyboy Date: Sat, 15 Nov 2025 16:30:29 +0800 Subject: [PATCH] =?UTF-8?q?=E2=9C=A8=20Integrate=20monitoring=20server=20f?= =?UTF-8?q?or=20observability,=20adding=20health=20and=20metrics=20endpoin?= =?UTF-8?q?ts.=20Update=20configuration=20to=20enable=20monitoring=20featu?= =?UTF-8?q?res=20and=20enhance=20README=20with=20deployment=20examples=20f?= =?UTF-8?q?or=20Kubernetes=20and=20systemd.=20Refactor=20application=20ini?= =?UTF-8?q?tialization=20to=20include=20monitoring=20server=20setup=20and?= =?UTF-8?q?=20improve=20error=20handling=20in=20message=20processing=20met?= =?UTF-8?q?rics.?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- Makefile | 106 +++---- README.md | 92 ++++-- Taskfile.yml | 153 +++++----- cmd/main/main.go | 10 +- configs/config.dev.toml | 10 +- configs/config.testdefaults.toml | 11 + docs/deploy-k8s.md | 109 +++++++ docs/deploy-systemd.md | 64 ++++ go.mod | 42 +++ internal/app/processor.go | 30 +- internal/domain/pln_test.go | 51 ++++ internal/infra/config/config.go | 48 ++- internal/infra/monitoring/server.go | 144 +++++++++ internal/observability/metrics/metrics.go | 71 +++++ pkg/di/wire.go | 17 +- pkg/di/wire_gen.go | 38 ++- .../jetstream_to_timescale_test.go | 278 ++++++++++++++++++ 17 files changed, 1092 insertions(+), 182 deletions(-) create mode 100644 docs/deploy-k8s.md create mode 100644 docs/deploy-systemd.md create mode 100644 internal/domain/pln_test.go create mode 100644 internal/infra/monitoring/server.go create mode 100644 internal/observability/metrics/metrics.go create mode 100644 test/integration/jetstream_to_timescale_test.go diff --git a/Makefile b/Makefile index 525c88d..c5e7eb5 100644 --- a/Makefile +++ b/Makefile @@ -1,84 +1,86 @@ -# Variables -APP_NAME = tele-proc -GO_FILES = $(shell find . -name '*.go' -type f) -CONFIG_DIR = configs -BUILD_DIR = build -MAIN_RECEIVER = ./cmd/main/main.go +# Build variables +BUILD_DIR ?= bin +BINARY := $(BUILD_DIR)/receiver +CMD := ./cmd/main +GO_ENV ?= dev # Default target .PHONY: all -all: build +all: build ## Build the application -# Build the receiver application .PHONY: build -build: build-receiver - -.PHONY: build-receiver -build-receiver: +build: ## Build the receiver binary + @mkdir -p $(BUILD_DIR) @echo "Building receiver..." - @go build -o $(BUILD_DIR)/receiver $(MAIN_RECEIVER) + @go build -o $(BINARY) $(CMD) -# Run the receiver application with different configurations .PHONY: run -run: run-dev +run: run-dev ## Alias for run-dev .PHONY: run-dev -run-dev: +run-dev: build ## Run the receiver in development mode @echo "Running receiver in development mode..." - @GO_ENV=development $(BUILD_DIR)/receiver & + @GO_ENV=dev $(BINARY) listen .PHONY: run-prod -run-prod: +run-prod: build ## Run the receiver in production mode @echo "Running receiver in production mode..." - @GO_ENV=production $(BUILD_DIR)/receiver & + @GO_ENV=prod $(BINARY) listen .PHONY: run-test -run-test: +run-test: build ## Run the receiver in test mode @echo "Running receiver in test mode..." - @GO_ENV=test $(BUILD_DIR)/receiver & + @GO_ENV=test $(BINARY) listen + +.PHONY: run-local +run-local: ## Run receiver directly via go run + @echo "Running receiver via go run..." + @GO_ENV=$(GO_ENV) go run $(CMD) listen -# Test the application .PHONY: test -test: - @echo "Running tests..." - @ginkgo -r -v +test: ## Run unit tests + @echo "Running go test..." + @go test ./... -# Clean build artifacts -.PHONY: clean -clean: - @echo "Cleaning build artifacts..." - @rm -rf $(BUILD_DIR) +.PHONY: test-int +test-int: ## Run integration tests (requires Docker) + @echo "Running integration tests..." + @GO_ENV=$(GO_ENV) go test -tags=integration ./test/integration/... + +.PHONY: test-ginkgo +test-ginkgo: ## Run ginkgo test suites + @command -v ginkgo >/dev/null || (echo "Please install ginkgo (go install github.com/onsi/ginkgo/v2/ginkgo@latest)"; exit 1) + @ginkgo -r -v + +.PHONY: coverage +coverage: ## Run coverage and generate report + @mkdir -p coverage + @echo "Generating coverage report..." + @go test ./... -coverprofile=coverage/coverage.out + @go tool cover -html=coverage/coverage.out -o coverage/coverage.html -# Format the code .PHONY: fmt -fmt: +fmt: ## Format Go code @echo "Formatting code..." @go fmt ./... -# Install dependencies .PHONY: deps -deps: - @echo "Installing dependencies..." +deps: ## Sync go.mod / go.sum + @echo "Tidying go modules..." @go mod tidy -# Lint the code .PHONY: lint -lint: +lint: ## Run golangci-lint + @command -v golangci-lint >/dev/null || (echo "Please install golangci-lint (https://golangci-lint.run/)"; exit 1) @echo "Linting code..." - @golangci-lint run + @golangci-lint run ./... + +.PHONY: clean +clean: ## Clean build artifacts and coverage files + @echo "Cleaning build artifacts..." + @rm -rf $(BUILD_DIR) coverage -# Help .PHONY: help -help: - @echo "Makefile usage:" - @echo " make build - Build the application" - @echo " make run - Run the receiver in development mode" - @echo " make run-dev - Run the receiver in development mode" - @echo " make run-prod - Run the receiver in production mode" - @echo " make run-test - Run the receiver in test mode" - @echo " make test - Run tests" - @echo " make clean - Clean build artifacts" - @echo " make fmt - Format the code" - @echo " make deps - Install dependencies" - @echo " make lint - Lint the code" - @echo " make help - Show this help message" +help: ## Show this help + @printf "Makefile targets:\n" + @grep -E '^[a-zA-Z0-9_-]+:.*##' $(MAKEFILE_LIST) | sort | awk 'BEGIN {FS = ":.*##"} {printf " %-15s %s\n", $$1, $$2}' diff --git a/README.md b/README.md index 5a4d9bb..01d9cef 100644 --- a/README.md +++ b/README.md @@ -149,30 +149,43 @@ Environment variable names are converted from `CAATSM_NATS_URL` to `nats.url` in ### Build +Using Make (writes `bin/receiver`): + ```bash -go build -o bin/receiver ./cmd/main +make build ``` -Or using Task: +Using Task: ```bash task build ``` -### Run +Or directly with Go: ```bash -# Development mode -GO_ENV=dev ./bin/receiver listen - -# Production mode -GO_ENV=prod ./bin/receiver listen +go build -o bin/receiver ./cmd/main ``` -Or using Task: +### Run + +Use Make targets (binary mode): + +```bash +make run-dev # GO_ENV=dev +make run-prod # GO_ENV=prod +make run-test # GO_ENV=test +make run-local # go run ./cmd/main listen (honors GO_ENV) +``` + +Task equivalents: ```bash task run-dev +task run-prod +task run-test +task run-local # go run ./cmd/main listen +task dev-run # boots docker-compose dev stack + go run ``` ### Command Line Options @@ -227,6 +240,26 @@ Critical overrides stay available through CLI flags; advanced tuning such as str - `caatsm_parse_duration_ms` (histogram) These flow through the collector → Prometheus → Grafana dashboards in the dev stack. +### Observability & Health + +A lightweight monitoring server exposes both readiness information and Prometheus-friendly metrics: + +- `GET /healthz` probes PostgreSQL (connection ping) and NATS (connection status). It returns HTTP 200 when both dependencies respond within `monitoring.health_timeout`, otherwise 503. +- `GET /metrics` streams `caatsm_processed_total`, `caatsm_failures_total`, and `caatsm_parse_latency_seconds` counters/histograms from the built-in Prometheus registry. +- Configure the server via the `[monitoring]` block (defaults shown): + +```toml +[monitoring] +addr = ":2112" +enable_metrics = true +enable_health = true +read_timeout = "5s" +write_timeout = "5s" +health_timeout = "2s" +``` + +Set `monitoring.disabled = true` (or `addr = ""`) if you need to turn the HTTP server off, e.g., during certain integration tests. + ## Development See `docs/dev-guide.md` for the full development workflow, including Docker Compose instructions, observability tooling, and troubleshooting tips. @@ -241,7 +274,7 @@ docker compose -f docker-compose.dev.yml up -d postgres nats nats-box docker compose -f docker-compose.dev.yml up -d otel-collector jaeger prometheus grafana ``` -Run the processor locally while the infra runs in Docker (dev config defaults to `nats.mode = "core"` so the consumer reads from plain NATS subjects): +Run the processor locally while the infra runs in Docker (default mode is JetStream; switch to core only if you explicitly set `CAATSM_NATS_MODE=core`): ```bash GO_ENV=dev \ @@ -252,6 +285,11 @@ CAATSM_POSTGRES_URL=postgres://caatsm:caatsm@localhost:5432/aviation?sslmode=dis Tear everything down with `docker compose -f docker-compose.dev.yml down -v`. +## Deployment Examples + +- `docs/deploy-systemd.md` shows a minimal systemd unit that wires configuration via environment files and restarts on failure. +- `docs/deploy-k8s.md` provides a reference Deployment + ConfigMap/Secret with liveness/readiness probes hitting `/healthz` and `/metrics`. + ### Project Structure - **Domain Layer** (`internal/domain`): Pure business logic and domain models @@ -278,20 +316,25 @@ Dependencies are managed using Google Wire. To add a new dependency: ### Testing -```bash -# Run all tests -go test ./... +The project keeps tests close to the code that they exercise: -# Run tests with coverage -task coverage +- **Domain/adapter/app unit tests** live under `internal/**` and cover parsing, validation, orchestration, and adapters. Run them all with `task test` (or `make test`), which is just `go test ./...`. +- **Integration tests** under `test/integration` spin up disposable TimescaleDB and NATS JetStream instances (via `testcontainers-go`) and execute a full ingestion flow. Use `task test-int` after ensuring Docker is running. +- **Coverage goals** are tracked via `task coverage`, which produces both a coverage profile and an HTML report under `coverage/coverage.html`. -# Run all Ginkgo suites (requires go install github.com/onsi/ginkgo/v2/ginkgo@latest) -ginkgo -r -``` +| Purpose | Make command | Task command | +|------------------------|---------------------|---------------------| +| Run unit tests | `make test` | `task test` | +| Run integration tests | `make test-int` | `task test-int` | +| Run Ginkgo suites | `make test-ginkgo` | `task test-ginkgo` | +| Generate coverage html | `make coverage` | `task coverage` | +| Lint (golangci-lint) | `make lint` | `task lint` | + +> Integration tests need Docker available on the host. Ginkgo or lint targets require the respective binaries (`go install github.com/onsi/ginkgo/v2/ginkgo@latest`, [golangci-lint install guide](https://golangci-lint.run/)). Use `task install-test` to bootstrap Ginkgo tooling. ## Message Flow -1. **NATS Consumer** receives raw telegram messages from NATS (JetStream durable pull in production; plain `nc.Subscribe` in dev when `nats.mode=core`) +1. **NATS Consumer** receives raw telegram messages from NATS (JetStream durable pull by default; plain `nc.Subscribe` only when you opt into `nats.mode=core`) 2. **MessageProcessor** orchestrates the processing: - Parses the message using the Parser adapter - Stores the parsed message in PostgreSQL via Repository @@ -299,6 +342,13 @@ ginkgo -r 3. **ACK/NAK** is sent based on processing success/failure 4. **Retry Logic** handles transient failures automatically +### Error Handling & Retries + +- **Parser failures** (invalid headers/body) are treated as permanent: the raw payload is stored in `aviation.telegrams_raw`, the message is ACKed, and no JetStream retries are attempted. +- **Repository failures** are transient: the consumer returns an error, the message is `NAK`ed, and JetStream redelivers it using `[nats.consumer_rules.backoff]` and `ack_wait` to space retries. +- **Publisher failures** are logged and persisted as raw records, but they are marked permanent to avoid hammering downstream topics; the deduplicated output can be replayed from the raw table later. +- Tune JetStream retry behavior via `[nats.consumer_rules.max_deliver]`, `[nats.consumer_rules.backoff]`, and CLI overrides like `--ack-wait`. The monitoring server plus Prometheus counters provide visibility into each failure bucket. + ### Failure Buckets Messages that cannot be parsed or fail to publish are written to `aviation.telegrams_raw` with a status: @@ -612,9 +662,9 @@ The legacy code has been removed. See the project history for migration details. ## License -[Add your license here] +This repository has not declared a public license yet. ## Contributing -[Add contributing guidelines here] +Contribution guidelines are not published; please coordinate changes via pull requests or direct maintainers. diff --git a/Taskfile.yml b/Taskfile.yml index 268ac84..d925918 100644 --- a/Taskfile.yml +++ b/Taskfile.yml @@ -1,14 +1,9 @@ version: '3' + vars: - app_name: tele-proc - go_files: - sh: find . -name '*.go' -type f - config_dir: configs - build_dir: build - main_receiver: ./cmd/main/main.go - hasura_endpoint: http://localhost:8080/v1/graphql - schema_file: ./internal/repository/schema.graphql - hasura_secret: # Set your Hasura admin secret here + build_dir: bin + binary: '{{.build_dir}}/receiver' + cmd: ./cmd/main tasks: all: @@ -17,92 +12,117 @@ tasks: - task: build build: - desc: Build the receiver application - cmds: - - task: build-receiver - - build-receiver: - desc: Build the receiver application + desc: Build the receiver binary cmds: + - mkdir -p {{.build_dir}} - echo "Building receiver..." - - go build -o {{.build_dir}}/receiver {{.main_receiver}} + - go build -o {{.binary}} {{.cmd}} run: - desc: Run the receiver application with different configurations + desc: Alias for run-dev cmds: - task: run-dev + run-dev: + desc: Run the receiver in development mode + deps: + - build + cmds: + - echo "Running receiver in development mode..." + - GO_ENV=dev {{.binary}} listen + run-prod: desc: Run the receiver in production mode + deps: + - build cmds: - - task: build-receiver - echo "Running receiver in production mode..." - - GO_ENV=prod {{.build_dir}}/receiver + - GO_ENV=prod {{.binary}} listen run-test: desc: Run the receiver in test mode + deps: + - build cmds: - - task: build-receiver - echo "Running receiver in test mode..." - - GO_ENV=test {{.build_dir}}/receiver + - GO_ENV=test {{.binary}} listen - upgrade: - desc: Upgrade go dependencies + run-local: + desc: Run receiver via go run (default GO_ENV=dev) cmds: - - echo "Upgrading go dependencies..." - - go get -u -v ./... - - go mod tidy - - install-test: - desc: Install ginkgo for testing - cmds: - - echo "Installing ginkgo..." - - go install github.com/onsi/ginkgo/v2/ginkgo - - go get github.com/onsi/gomega/... + - | + echo "Running receiver via go run (GO_ENV=${GO_ENV:-dev})..." + GO_ENV=${GO_ENV:-dev} go run {{.cmd}} listen test: - desc: Test the application + desc: Run go test ./... cmds: - - echo "Running tests..." + - echo "Running go test..." + - go test ./... + + test-int: + desc: Run integration tests (requires Docker) + cmds: + - | + echo "Running integration tests (Docker required)..." + GO_ENV=${GO_ENV:-test} go test -tags=integration ./test/integration/... + + test-ginkgo: + desc: Run ginkgo suites + cmds: + - command -v ginkgo >/dev/null || { echo "Install ginkgo (go install github.com/onsi/ginkgo/v2/ginkgo@latest)"; exit 1; } - ginkgo -r -v coverage: - desc: Generate test coverage report + desc: Generate coverage report (profile + HTML) cmds: + - mkdir -p coverage - echo "Generating test coverage report..." - - ginkgo --json-report ./ginkgo.report -coverpkg=./... -coverprofile=./coverage/coverage.out -r - - go tool cover -html=./coverage/coverage.out -o ./coverage/coverage.html - - clean: - desc: Clean build artifacts - cmds: - - echo "Cleaning build artifacts..." - - rm -rf {{.build_dir}} + - go test ./... -coverprofile=coverage/coverage.out + - go tool cover -html=coverage/coverage.out -o coverage/coverage.html fmt: - desc: Format the code + desc: Format Go code cmds: - echo "Formatting code..." - go fmt ./... - deps: - desc: Install dependencies + lint: + desc: Run golangci-lint cmds: - - echo "Installing dependencies..." + - | + if ! command -v golangci-lint >/dev/null; then + echo "Install golangci-lint: https://golangci-lint.run/" + exit 1 + fi + - echo "Linting code..." + - golangci-lint run ./... + + deps: + desc: Sync go.mod / go.sum + cmds: + - echo "Tidying go modules..." - go mod tidy - lint: - desc: Lint the code + upgrade: + desc: Upgrade Go dependencies cmds: - - echo "Linting code..." - - golangci-lint run + - echo "Upgrading go modules..." + - go get -u -v ./... + - go mod tidy - run-dev: - desc: Run the receiver in development mode (binary) + install-test: + desc: Install ginkgo/gomega tooling cmds: - - task: build-receiver - - echo "Running receiver in development mode..." - - GO_ENV=dev {{.build_dir}}/receiver listen + - echo "Installing ginkgo tooling..." + - go install github.com/onsi/ginkgo/v2/ginkgo@latest + - go get github.com/onsi/gomega/... + + clean: + desc: Clean build + coverage artifacts + cmds: + - echo "Cleaning build artifacts..." + - rm -rf {{.build_dir}} coverage up: desc: Start TimescaleDB + NATS dev stack (docker compose) @@ -161,21 +181,10 @@ tasks: ) exec "${cmd[@]}" ' + help: desc: Show this help message cmds: - - echo "Taskfile usage:" - - echo " task build - Build the application" - - echo " task run - Run the receiver in development mode" - - echo " task run-dev - Run the receiver in development mode" - - echo " task run-prod - Run the receiver in production mode" - - echo " task run-test - Run the receiver in test mode" - - echo " task test - Run tests" - - echo " task coverage - Generate test coverage report" - - echo " task install-test - Install ginkgo for testing" - - echo " task clean - Clean build artifacts" - - echo " task fmt - Format the code" - - echo " task deps - Install dependencies" - - echo " task lint - Lint the code" - - echo " task upgrade - Upgrade go dependencies" - - echo " task help - Show this help message" + - | + printf "Task targets:\n" + task --list diff --git a/cmd/main/main.go b/cmd/main/main.go index 0ed854d..a806fef 100644 --- a/cmd/main/main.go +++ b/cmd/main/main.go @@ -127,15 +127,21 @@ func executeListen(c *cli.Context) error { } // Initialize dependencies using Wire - processor, consumer, err := di.InitializeAppWithConfig(cfg) + processor, consumer, monitorServer, err := di.InitializeAppWithConfig(cfg) if err != nil { return fmt.Errorf("failed to initialize app: %w", err) } - // Create context with cancellation ctx, cancel := context.WithCancel(context.Background()) defer cancel() + if monitorServer != nil { + if err := monitorServer.Start(ctx); err != nil { + return fmt.Errorf("failed to start monitoring server: %w", err) + } + defer monitorServer.Shutdown(context.Background()) + } + // Handle graceful shutdown sigChan := make(chan os.Signal, 1) signal.Notify(sigChan, os.Interrupt, syscall.SIGTERM) diff --git a/configs/config.dev.toml b/configs/config.dev.toml index 31878d3..ac2d030 100644 --- a/configs/config.dev.toml +++ b/configs/config.dev.toml @@ -54,4 +54,12 @@ format = "json" [telemetry] enabled = true endpoint = "localhost:4318" -insecure = true \ No newline at end of file +insecure = true + +[monitoring] +addr = ":2112" +enable_metrics = true +enable_health = true +read_timeout = "5s" +write_timeout = "5s" +health_timeout = "2s" \ No newline at end of file diff --git a/configs/config.testdefaults.toml b/configs/config.testdefaults.toml index 8ce2e73..3c198a4 100644 --- a/configs/config.testdefaults.toml +++ b/configs/config.testdefaults.toml @@ -19,3 +19,14 @@ monitor_interval = "1s" level = "info" format = "json" +[telemetry] +enabled = false + +[monitoring] +addr = ":0" +enable_metrics = false +enable_health = false +read_timeout = "0s" +write_timeout = "0s" +health_timeout = "0s" + diff --git a/docs/deploy-k8s.md b/docs/deploy-k8s.md new file mode 100644 index 0000000..2eec729 --- /dev/null +++ b/docs/deploy-k8s.md @@ -0,0 +1,109 @@ +# Kubernetes Deployment Example + +The following manifest shows the essential pieces needed to run `caatsm` on Kubernetes with native config management and probes. + +## 1. ConfigMap and Secret + +```yaml +apiVersion: v1 +kind: ConfigMap +metadata: + name: caatsm-config +data: + config.toml: | + # trimmed for brevity; see configs/config.prod.toml + [nats] + url = "nats://nats.jetstream.svc:4222" + stream = "TELEGRAM" + consumer = "telegram-consumer" + + [postgres] + # default overridden by CAATSM_POSTGRES_URL in env + url = "postgres://caatsm:password@timescale.svc:5432/aviation?sslmode=disable" + + [monitoring] + addr = ":2112" + enable_metrics = true + enable_health = true +--- +apiVersion: v1 +kind: Secret +metadata: + name: caatsm-secrets +type: Opaque +stringData: + CAATSM_POSTGRES_URL: postgres://caatsm:super-secret@timescale.svc:5432/aviation?sslmode=disable +``` + +## 2. Deployment + +```yaml +apiVersion: apps/v1 +kind: Deployment +metadata: + name: caatsm +spec: + replicas: 1 + selector: + matchLabels: + app: caatsm + template: + metadata: + labels: + app: caatsm + spec: + containers: + - name: caatsm + image: ghcr.io//caatsm:latest + args: ["listen", "--config", "/etc/caatsm/config.toml"] + envFrom: + - secretRef: + name: caatsm-secrets + env: + - name: GO_ENV + value: prod + ports: + - name: monitoring + containerPort: 2112 + volumeMounts: + - name: config + mountPath: /etc/caatsm + livenessProbe: + httpGet: + path: /healthz + port: monitoring + initialDelaySeconds: 10 + periodSeconds: 15 + readinessProbe: + httpGet: + path: /healthz + port: monitoring + initialDelaySeconds: 5 + periodSeconds: 15 + volumes: + - name: config + configMap: + name: caatsm-config +``` + +## 3. Service and Scraping + +```yaml +apiVersion: v1 +kind: Service +metadata: + name: caatsm-metrics + labels: + app: caatsm +spec: + selector: + app: caatsm + ports: + - name: http + port: 2112 + targetPort: monitoring + protocol: TCP +``` + +Point Prometheus at the service above (or annotate it if you use `prometheus-operator`). The `/healthz` probe doubles as a readiness check and quickly surfaces upstream connectivity issues. + diff --git a/docs/deploy-systemd.md b/docs/deploy-systemd.md new file mode 100644 index 0000000..dd6bf4e --- /dev/null +++ b/docs/deploy-systemd.md @@ -0,0 +1,64 @@ +# Systemd Deployment Example + +This example targets a single host running the `caatsm` binary under systemd with minimal moving parts. + +## 1. Install the Binary + +``` +sudo install -m 755 bin/receiver /usr/local/bin/caatsm +sudo install -d /etc/caatsm/configs +sudo cp configs/config.dev.toml /etc/caatsm/configs/config.prod.toml +``` + +Adjust the config file to point at your production NATS cluster, TimescaleDB endpoint, and telemetry collector. + +## 2. Environment File + +Create `/etc/caatsm/caatsm.env` to hold secrets or overrides (systemd keeps file permissions intact): + +``` +CAATSM_NATS_URL=nats://nats.prod.svc.cluster.local:4222 +CAATSM_POSTGRES_URL=postgres://caatsm:***@tsdb.prod:5432/aviation?sslmode=require +CAATSM_LOG_LEVEL=info +GO_ENV=prod +``` + +## 3. systemd Unit + +`/etc/systemd/system/caatsm.service` + +``` +[Unit] +Description=CAATSM Telegram Processor +After=network-online.target +Wants=network-online.target + +[Service] +Type=simple +EnvironmentFile=/etc/caatsm/caatsm.env +WorkingDirectory=/etc/caatsm +ExecStart=/usr/local/bin/caatsm listen +Restart=on-failure +RestartSec=5 +StandardOutput=journal +StandardError=journal +LimitNOFILE=65535 + +[Install] +WantedBy=multi-user.target +``` + +Reload and start: + +``` +sudo systemctl daemon-reload +sudo systemctl enable --now caatsm +``` + +## 4. Observability Hooks + +- Expose `monitoring.addr = ":2112"` (default) and add firewall rules so Prometheus can scrape `http://host:2112/metrics`. +- systemd watchdogs can use `curl -sf http://127.0.0.1:2112/healthz`. + +With these three files (binary, config, env) the service becomes repeatable and easy to operate. + diff --git a/go.mod b/go.mod index 60a0e6e..2987b00 100644 --- a/go.mod +++ b/go.mod @@ -13,6 +13,8 @@ require ( github.com/nats-io/nats.go v1.47.0 github.com/onsi/ginkgo/v2 v2.27.2 github.com/onsi/gomega v1.38.2 + github.com/prometheus/client_golang v1.20.3 + github.com/testcontainers/testcontainers-go v0.30.0 github.com/urfave/cli/v2 v2.27.7 go.opentelemetry.io/otel v1.38.0 go.opentelemetry.io/otel/exporters/otlp/otlpmetric/otlpmetrichttp v1.38.0 @@ -26,14 +28,31 @@ require ( ) require ( + dario.cat/mergo v1.0.0 // indirect + github.com/Azure/go-ansiterm v0.0.0-20210617225240-d185dfc1b5a1 // indirect github.com/Masterminds/semver/v3 v3.4.0 // indirect + github.com/Microsoft/go-winio v0.6.1 // indirect + github.com/Microsoft/hcsshim v0.11.4 // indirect + github.com/beorn7/perks v1.0.1 // indirect + github.com/cenkalti/backoff/v4 v4.2.1 // indirect github.com/cenkalti/backoff/v5 v5.0.3 // indirect + github.com/cespare/xxhash/v2 v2.3.0 // indirect + github.com/containerd/containerd v1.7.12 // indirect + github.com/containerd/log v0.1.0 // indirect + github.com/cpuguy83/dockercfg v0.3.1 // indirect github.com/cpuguy83/go-md2man/v2 v2.0.7 // indirect + github.com/distribution/reference v0.5.0 // indirect + github.com/docker/docker v25.0.5+incompatible // indirect + github.com/docker/go-connections v0.5.0 // indirect + github.com/docker/go-units v0.5.0 // indirect + github.com/felixge/httpsnoop v1.0.4 // indirect github.com/fsnotify/fsnotify v1.9.0 // indirect github.com/go-logr/logr v1.4.3 // indirect github.com/go-logr/stdr v1.2.2 // indirect + github.com/go-ole/go-ole v1.2.6 // indirect github.com/go-task/slim-sprig/v3 v3.0.0 // indirect github.com/go-viper/mapstructure/v2 v2.4.0 // indirect + github.com/gogo/protobuf v1.3.2 // indirect github.com/google/go-cmp v0.7.0 // indirect github.com/google/pprof v0.0.0-20251114195745-4902fdda35c8 // indirect github.com/grpc-ecosystem/grpc-gateway/v2 v2.27.3 // indirect @@ -42,19 +61,42 @@ require ( github.com/jackc/puddle/v2 v2.2.2 // indirect github.com/klauspost/compress v1.18.1 // indirect github.com/knadh/koanf/maps v0.1.2 // indirect + github.com/lufia/plan9stats v0.0.0-20211012122336-39d0f177ccd0 // indirect + github.com/magiconair/properties v1.8.7 // indirect github.com/mitchellh/copystructure v1.2.0 // indirect github.com/mitchellh/reflectwalk v1.0.2 // indirect + github.com/moby/patternmatcher v0.6.0 // indirect + github.com/moby/sys/sequential v0.5.0 // indirect + github.com/moby/sys/user v0.1.0 // indirect + github.com/moby/term v0.5.0 // indirect + github.com/morikuni/aec v1.0.0 // indirect + github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect github.com/nats-io/nkeys v0.4.11 // indirect github.com/nats-io/nuid v1.0.1 // indirect + github.com/opencontainers/go-digest v1.0.0 // indirect + github.com/opencontainers/image-spec v1.1.0 // indirect github.com/pelletier/go-toml v1.9.5 // indirect + github.com/pkg/errors v0.9.1 // indirect + github.com/power-devops/perfstat v0.0.0-20210106213030-5aafc221ea8c // indirect + github.com/prometheus/client_model v0.6.1 // indirect + github.com/prometheus/common v0.55.0 // indirect + github.com/prometheus/procfs v0.15.1 // indirect github.com/russross/blackfriday/v2 v2.1.0 // indirect + github.com/shirou/gopsutil/v3 v3.23.12 // indirect + github.com/shoenig/go-m1cpu v0.1.6 // indirect + github.com/sirupsen/logrus v1.9.3 // indirect + github.com/tklauser/go-sysconf v0.3.12 // indirect + github.com/tklauser/numcpus v0.6.1 // indirect github.com/xrash/smetrics v0.0.0-20250705151800-55b8f293f342 // indirect + github.com/yusufpapurcu/wmi v1.2.3 // indirect go.opentelemetry.io/auto/sdk v1.2.1 // indirect + go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.49.0 // indirect go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.38.0 // indirect go.opentelemetry.io/proto/otlp v1.9.0 // indirect go.uber.org/multierr v1.11.0 // indirect go.yaml.in/yaml/v3 v3.0.4 // indirect golang.org/x/crypto v0.44.0 // indirect + golang.org/x/exp v0.0.0-20230510235704-dd950f8aeaea // indirect golang.org/x/mod v0.30.0 // indirect golang.org/x/net v0.47.0 // indirect golang.org/x/sync v0.18.0 // indirect diff --git a/internal/app/processor.go b/internal/app/processor.go index bd702f5..1f77396 100644 --- a/internal/app/processor.go +++ b/internal/app/processor.go @@ -4,6 +4,7 @@ import ( "caatsm/internal/adapter" "caatsm/internal/adapter/parser" "caatsm/internal/domain" + obsmetrics "caatsm/internal/observability/metrics" "context" "fmt" "strings" @@ -122,13 +123,15 @@ func (p *MessageProcessor) Handle(ctx context.Context, raw []byte, msgID string) zap.String("content_preview", truncateContent(parsed.Content, 256)), zap.Error(parseErr), ) - parseLatencyHistogram.Record(ctx, float64(parsed.ParsedAt.Sub(receivedAt).Milliseconds()), + latency := parsed.ParsedAt.Sub(receivedAt) + parseLatencyHistogram.Record(ctx, float64(latency.Milliseconds()), metric.WithAttributes( messageStatusAttrKey.String(string(parsed.Status)), messageCategoryAttrKey.String(parsed.Category), ), ) - recordProcessedMetric(ctx, parsed) + obsmetrics.RecordFailure("parser") + recordProcessedMetric(ctx, parsed, latency) return Permanent(fmt.Errorf("parser error: %w", parseErr)) } parsed.ErrorReason = "" @@ -153,13 +156,15 @@ func (p *MessageProcessor) Handle(ctx context.Context, raw []byte, msgID string) span.RecordError(err) span.SetStatus(codes.Error, err.Error()) parsed.Status = domain.MessageStatusRepositoryFail - parseLatencyHistogram.Record(ctx, float64(parsed.ParsedAt.Sub(receivedAt).Milliseconds()), + latency := parsed.ParsedAt.Sub(receivedAt) + parseLatencyHistogram.Record(ctx, float64(latency.Milliseconds()), metric.WithAttributes( messageStatusAttrKey.String(string(parsed.Status)), messageCategoryAttrKey.String(parsed.Category), ), ) - recordProcessedMetric(ctx, parsed) + obsmetrics.RecordFailure("repository") + recordProcessedMetric(ctx, parsed, latency) return fmt.Errorf("failed to insert message: %w", err) } @@ -180,13 +185,15 @@ func (p *MessageProcessor) Handle(ctx context.Context, raw []byte, msgID string) messageCategoryAttrKey.String(parsed.Category), ), ) - parseLatencyHistogram.Record(ctx, float64(parsed.ParsedAt.Sub(receivedAt).Milliseconds()), + latency := parsed.ParsedAt.Sub(receivedAt) + parseLatencyHistogram.Record(ctx, float64(latency.Milliseconds()), metric.WithAttributes( messageStatusAttrKey.String(string(parsed.Status)), messageCategoryAttrKey.String(parsed.Category), ), ) - recordProcessedMetric(ctx, parsed) + obsmetrics.RecordFailure("publisher") + recordProcessedMetric(ctx, parsed, latency) p.persistRaw(ctx, parsed) // Mark as permanent so the consumer will ack instead of retrying pubSpan.End() @@ -194,13 +201,14 @@ func (p *MessageProcessor) Handle(ctx context.Context, raw []byte, msgID string) } pubSpan.End() - parseLatencyHistogram.Record(ctx, float64(parsed.ParsedAt.Sub(receivedAt).Milliseconds()), + latency := parsed.ParsedAt.Sub(receivedAt) + parseLatencyHistogram.Record(ctx, float64(latency.Milliseconds()), metric.WithAttributes( messageStatusAttrKey.String(string(parsed.Status)), messageCategoryAttrKey.String(parsed.Category), ), ) - recordProcessedMetric(ctx, parsed) + recordProcessedMetric(ctx, parsed, latency) return nil } @@ -242,7 +250,7 @@ func truncateContent(content string, limit int) string { return content[:limit-3] + "..." } -func recordProcessedMetric(ctx context.Context, msg *domain.ParsedMessage) { +func recordProcessedMetric(ctx context.Context, msg *domain.ParsedMessage, elapsed time.Duration) { if msg == nil { return } @@ -252,4 +260,8 @@ func recordProcessedMetric(ctx context.Context, msg *domain.ParsedMessage) { messageCategoryAttrKey.String(msg.Category), ), ) + if elapsed < 0 { + elapsed = 0 + } + obsmetrics.RecordProcessed(string(msg.Status), msg.Category, elapsed) } diff --git a/internal/domain/pln_test.go b/internal/domain/pln_test.go new file mode 100644 index 0000000..86471ef --- /dev/null +++ b/internal/domain/pln_test.go @@ -0,0 +1,51 @@ +package domain + +import ( + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" +) + +var _ = Describe("ScheduleLine", func() { + var base ScheduleLine + + BeforeEach(func() { + base = ScheduleLine{ + Index: "001", + Date: "30OCT", + Task: "H/G", + FlightNumber: []string{ + "CA1014", + }, + AircraftReg: "B2458", + PassengerConfig: "1/1", + ILS: "ILS(0)", + Waypoints: []WayPoint{ + {Airport: "ZBTJ", DepartureTime: "0100", ArrivalTime: "0200"}, + }, + Comments: "all green", + } + }) + + It("passes validation when all required fields exist", func() { + Expect(base.Validate()).To(Succeed()) + }) + + DescribeTable("required field validation", + func(mutator func(line *ScheduleLine), expected string) { + line := base + mutator(&line) + err := line.Validate() + Expect(err).To(HaveOccurred()) + Expect(err.Error()).To(ContainSubstring(expected)) + }, + Entry("missing date", func(line *ScheduleLine) { + line.Date = "" + }, "date is required"), + Entry("missing flight number", func(line *ScheduleLine) { + line.FlightNumber = nil + }, "flight number is required"), + Entry("missing aircraft registration", func(line *ScheduleLine) { + line.AircraftReg = "" + }, "aircraft registration is required"), + ) +}) diff --git a/internal/infra/config/config.go b/internal/infra/config/config.go index 250b004..951921c 100644 --- a/internal/infra/config/config.go +++ b/internal/infra/config/config.go @@ -14,12 +14,13 @@ import ( // Config holds all application configuration type Config struct { - NATS NATSConfig `koanf:"nats"` - Postgres PostgresConfig `koanf:"postgres"` - App AppConfig `koanf:"app"` - Log LogConfig `koanf:"log"` - Publisher PublisherConfig `koanf:"publisher"` - Telemetry TelemetryConfig `koanf:"telemetry"` + NATS NATSConfig `koanf:"nats"` + Postgres PostgresConfig `koanf:"postgres"` + App AppConfig `koanf:"app"` + Log LogConfig `koanf:"log"` + Publisher PublisherConfig `koanf:"publisher"` + Telemetry TelemetryConfig `koanf:"telemetry"` + Monitoring MonitoringConfig `koanf:"monitoring"` // Legacy fields for backward compatibility during migration Subscription SubscriptionConfig `koanf:"subscription"` Timeouts TimeoutsConfig `koanf:"timeouts"` @@ -92,6 +93,17 @@ type TelemetryConfig struct { Insecure bool `koanf:"insecure"` } +// MonitoringConfig controls the lightweight HTTP server that exposes health and metrics endpoints. +type MonitoringConfig struct { + Disabled bool `koanf:"disabled"` + Addr string `koanf:"addr"` + EnableMetrics bool `koanf:"enable_metrics"` + EnableHealth bool `koanf:"enable_health"` + ReadTimeout time.Duration `koanf:"read_timeout"` + WriteTimeout time.Duration `koanf:"write_timeout"` + HealthTimeout time.Duration `koanf:"health_timeout"` +} + // SubscriptionConfig holds subscription configuration (legacy) type SubscriptionConfig struct { Topic string `koanf:"topic"` @@ -212,6 +224,21 @@ func LoadConfig() (*Config, error) { cfg.Telemetry.Endpoint = "" } + if !cfg.Monitoring.Disabled && cfg.Monitoring.Addr == "" && !cfg.Monitoring.EnableHealth && !cfg.Monitoring.EnableMetrics { + cfg.Monitoring.Addr = ":2112" + cfg.Monitoring.EnableHealth = true + cfg.Monitoring.EnableMetrics = true + } + if cfg.Monitoring.ReadTimeout == 0 { + cfg.Monitoring.ReadTimeout = 5 * time.Second + } + if cfg.Monitoring.WriteTimeout == 0 { + cfg.Monitoring.WriteTimeout = 5 * time.Second + } + if cfg.Monitoring.HealthTimeout == 0 { + cfg.Monitoring.HealthTimeout = 2 * time.Second + } + // Validate configuration if err := cfg.Validate(); err != nil { return nil, fmt.Errorf("config validation failed: %w", err) @@ -292,6 +319,15 @@ func (c *Config) Validate() error { if c.Telemetry.Endpoint == "" && c.Telemetry.Enabled { return fmt.Errorf("telemetry.endpoint is required when telemetry.enabled=true") } + if c.Monitoring.ReadTimeout < 0 { + return fmt.Errorf("monitoring.read_timeout must be >= 0") + } + if c.Monitoring.WriteTimeout < 0 { + return fmt.Errorf("monitoring.write_timeout must be >= 0") + } + if c.Monitoring.HealthTimeout < 0 { + return fmt.Errorf("monitoring.health_timeout must be >= 0") + } return nil } diff --git a/internal/infra/monitoring/server.go b/internal/infra/monitoring/server.go new file mode 100644 index 0000000..223425c --- /dev/null +++ b/internal/infra/monitoring/server.go @@ -0,0 +1,144 @@ +package monitoring + +import ( + "caatsm/internal/infra/config" + obsmetrics "caatsm/internal/observability/metrics" + "context" + "encoding/json" + "errors" + "net/http" + "time" + + "github.com/jackc/pgx/v5/pgxpool" + "github.com/nats-io/nats.go" + "go.uber.org/zap" +) + +// Server exposes /healthz and /metrics endpoints for basic operations and observability checks. +type Server struct { + cfg config.MonitoringConfig + logger *zap.Logger + pool *pgxpool.Pool + conn *nats.Conn + httpServer *http.Server +} + +// ProvideServer wires a monitoring server if enabled in configuration. +func ProvideServer( + cfg *config.Config, + logger *zap.Logger, + pool *pgxpool.Pool, + conn *nats.Conn, +) (*Server, error) { + if cfg == nil || logger == nil || cfg.Monitoring.Disabled { + return nil, nil + } + if cfg.Monitoring.Addr == "" { + return nil, nil + } + + mux := http.NewServeMux() + server := &Server{ + cfg: cfg.Monitoring, + logger: logger, + pool: pool, + conn: conn, + } + + routes := 0 + if cfg.Monitoring.EnableHealth { + mux.HandleFunc("/healthz", server.handleHealth) + routes++ + } + if cfg.Monitoring.EnableMetrics { + mux.Handle("/metrics", obsmetrics.Handler()) + routes++ + } + + if routes == 0 { + return nil, nil + } + + httpServer := &http.Server{ + Addr: cfg.Monitoring.Addr, + Handler: mux, + ReadHeaderTimeout: 3 * time.Second, + ReadTimeout: cfg.Monitoring.ReadTimeout, + WriteTimeout: cfg.Monitoring.WriteTimeout, + } + server.httpServer = httpServer + + return server, nil +} + +// Start launches the monitoring HTTP server in the background. +func (s *Server) Start(ctx context.Context) error { + if s == nil || s.httpServer == nil { + return nil + } + + go func() { + <-ctx.Done() + _ = s.Shutdown(context.Background()) + }() + + go func() { + s.logger.Info("Monitoring server listening", + zap.String("addr", s.httpServer.Addr), + zap.Bool("metrics", s.cfg.EnableMetrics), + zap.Bool("health", s.cfg.EnableHealth), + ) + if err := s.httpServer.ListenAndServe(); err != nil && !errors.Is(err, http.ErrServerClosed) { + s.logger.Error("Monitoring server exited", zap.Error(err)) + } + }() + + return nil +} + +// Shutdown gracefully stops the HTTP server. +func (s *Server) Shutdown(ctx context.Context) error { + if s == nil || s.httpServer == nil { + return nil + } + return s.httpServer.Shutdown(ctx) +} + +func (s *Server) handleHealth(w http.ResponseWriter, r *http.Request) { + status := http.StatusOK + result := map[string]interface{}{ + "postgres": "ok", + "nats": "ok", + } + + ctx, cancel := context.WithTimeout(r.Context(), s.healthTimeout()) + defer cancel() + + if s.pool == nil { + result["postgres"] = "unconfigured" + status = http.StatusServiceUnavailable + } else if err := s.pool.Ping(ctx); err != nil { + result["postgres"] = err.Error() + status = http.StatusServiceUnavailable + } + + if s.conn == nil { + result["nats"] = "unconfigured" + status = http.StatusServiceUnavailable + } else if s.conn.Status() != nats.CONNECTED { + result["nats"] = s.conn.Status().String() + status = http.StatusServiceUnavailable + } + + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(status) + _ = json.NewEncoder(w).Encode(result) +} + +func (s *Server) healthTimeout() time.Duration { + timeout := s.cfg.HealthTimeout + if timeout <= 0 { + timeout = 2 * time.Second + } + return timeout +} diff --git a/internal/observability/metrics/metrics.go b/internal/observability/metrics/metrics.go new file mode 100644 index 0000000..f8d1341 --- /dev/null +++ b/internal/observability/metrics/metrics.go @@ -0,0 +1,71 @@ +package metrics + +import ( + "math" + "net/http" + "strings" + "sync" + "time" + + "github.com/prometheus/client_golang/prometheus" + "github.com/prometheus/client_golang/prometheus/promhttp" +) + +var ( + once sync.Once + registry *prometheus.Registry + processedCounter *prometheus.CounterVec + failureCounter *prometheus.CounterVec + parseLatency *prometheus.HistogramVec +) + +func initCollectors() { + registry = prometheus.NewRegistry() + processedCounter = prometheus.NewCounterVec(prometheus.CounterOpts{ + Name: "caatsm_processed_total", + Help: "Count of telegrams processed by status and category.", + }, []string{"status", "category"}) + failureCounter = prometheus.NewCounterVec(prometheus.CounterOpts{ + Name: "caatsm_failures_total", + Help: "Count of processor failures by stage (parser, repository, publisher).", + }, []string{"stage"}) + parseLatency = prometheus.NewHistogramVec(prometheus.HistogramOpts{ + Name: "caatsm_parse_latency_seconds", + Help: "Latency between reception and parse completion.", + Buckets: prometheus.DefBuckets, + }, []string{"status", "category"}) + + registry.MustRegister(processedCounter, failureCounter, parseLatency) +} + +func ensureCollectors() { + once.Do(initCollectors) +} + +// Handler exposes the Prometheus metrics registry. +func Handler() http.Handler { + ensureCollectors() + return promhttp.HandlerFor(registry, promhttp.HandlerOpts{}) +} + +// RecordProcessed tracks the final status of a telegram along with the parse latency. +func RecordProcessed(status, category string, elapsed time.Duration) { + ensureCollectors() + processedCounter.WithLabelValues(labelValue(status), labelValue(category)).Inc() + seconds := math.Max(elapsed.Seconds(), 0) + parseLatency.WithLabelValues(labelValue(status), labelValue(category)).Observe(seconds) +} + +// RecordFailure increments the failure counter for the supplied stage. +func RecordFailure(stage string) { + ensureCollectors() + failureCounter.WithLabelValues(labelValue(stage)).Inc() +} + +func labelValue(value string) string { + value = strings.TrimSpace(value) + if value == "" { + return "unknown" + } + return strings.ToLower(value) +} diff --git a/pkg/di/wire.go b/pkg/di/wire.go index 3db4401..e3b2450 100644 --- a/pkg/di/wire.go +++ b/pkg/di/wire.go @@ -8,6 +8,7 @@ import ( "caatsm/internal/app" "caatsm/internal/infra/config" "caatsm/internal/infra/log" + "caatsm/internal/infra/monitoring" "caatsm/internal/infra/nats" "caatsm/internal/infra/postgres" @@ -15,21 +16,21 @@ import ( ) // InitializeApp initializes the application with all dependencies -func InitializeApp() (*app.MessageProcessor, *nats.Consumer, error) { +func InitializeApp() (*app.MessageProcessor, *nats.Consumer, *monitoring.Server, error) { comps, err := buildAppComponents() if err != nil { - return nil, nil, err + return nil, nil, nil, err } - return comps.Processor, comps.Consumer, nil + return comps.Processor, comps.Consumer, comps.Monitoring, nil } // InitializeAppWithConfig wires dependencies using a pre-loaded configuration. -func InitializeAppWithConfig(cfg *config.Config) (*app.MessageProcessor, *nats.Consumer, error) { +func InitializeAppWithConfig(cfg *config.Config) (*app.MessageProcessor, *nats.Consumer, *monitoring.Server, error) { comps, err := buildAppComponentsWithConfig(cfg) if err != nil { - return nil, nil, err + return nil, nil, nil, err } - return comps.Processor, comps.Consumer, nil + return comps.Processor, comps.Consumer, comps.Monitoring, nil } var runtimeSet = wire.NewSet( @@ -53,6 +54,9 @@ var runtimeSet = wire.NewSet( // Consumer nats.ProvideConsumer, + + // Monitoring HTTP server + monitoring.ProvideServer, ) func buildAppComponents() (*appComponents, error) { @@ -75,4 +79,5 @@ func buildAppComponentsWithConfig(cfg *config.Config) (*appComponents, error) { type appComponents struct { Processor *app.MessageProcessor Consumer *nats.Consumer + Monitoring *monitoring.Server } diff --git a/pkg/di/wire_gen.go b/pkg/di/wire_gen.go index ba0796b..01fc818 100644 --- a/pkg/di/wire_gen.go +++ b/pkg/di/wire_gen.go @@ -11,6 +11,7 @@ import ( "caatsm/internal/app" "caatsm/internal/infra/config" "caatsm/internal/infra/log" + "caatsm/internal/infra/monitoring" "caatsm/internal/infra/nats" "caatsm/internal/infra/postgres" "github.com/google/wire" @@ -53,9 +54,14 @@ func buildAppComponents() (*appComponents, error) { if err != nil { return nil, err } + server, err := monitoring.ProvideServer(configConfig, logger, pool, conn) + if err != nil { + return nil, err + } diAppComponents := &appComponents{ - Processor: messageProcessor, - Consumer: consumer, + Processor: messageProcessor, + Consumer: consumer, + Monitoring: server, } return diAppComponents, nil } @@ -91,9 +97,14 @@ func buildAppComponentsWithConfig(cfg *config.Config) (*appComponents, error) { if err != nil { return nil, err } + server, err := monitoring.ProvideServer(cfg, logger, pool, conn) + if err != nil { + return nil, err + } diAppComponents := &appComponents{ - Processor: messageProcessor, - Consumer: consumer, + Processor: messageProcessor, + Consumer: consumer, + Monitoring: server, } return diAppComponents, nil } @@ -101,26 +112,27 @@ func buildAppComponentsWithConfig(cfg *config.Config) (*appComponents, error) { // wire.go: // InitializeApp initializes the application with all dependencies -func InitializeApp() (*app.MessageProcessor, *nats.Consumer, error) { +func InitializeApp() (*app.MessageProcessor, *nats.Consumer, *monitoring.Server, error) { comps, err := buildAppComponents() if err != nil { - return nil, nil, err + return nil, nil, nil, err } - return comps.Processor, comps.Consumer, nil + return comps.Processor, comps.Consumer, comps.Monitoring, nil } // InitializeAppWithConfig wires dependencies using a pre-loaded configuration. -func InitializeAppWithConfig(cfg *config.Config) (*app.MessageProcessor, *nats.Consumer, error) { +func InitializeAppWithConfig(cfg *config.Config) (*app.MessageProcessor, *nats.Consumer, *monitoring.Server, error) { comps, err := buildAppComponentsWithConfig(cfg) if err != nil { - return nil, nil, err + return nil, nil, nil, err } - return comps.Processor, comps.Consumer, nil + return comps.Processor, comps.Consumer, comps.Monitoring, nil } -var runtimeSet = wire.NewSet(log.ProvideLogger, postgres.ProvideDB, postgres.ProvideRepository, nats.ProvideNATSConn, nats.ProvideJetStream, nats.ProvidePublisher, parser.ProvideParser, app.NewMessageProcessor, nats.ProvideConsumer) +var runtimeSet = wire.NewSet(log.ProvideLogger, postgres.ProvideDB, postgres.ProvideRepository, nats.ProvideNATSConn, nats.ProvideJetStream, nats.ProvidePublisher, parser.ProvideParser, app.NewMessageProcessor, nats.ProvideConsumer, monitoring.ProvideServer) type appComponents struct { - Processor *app.MessageProcessor - Consumer *nats.Consumer + Processor *app.MessageProcessor + Consumer *nats.Consumer + Monitoring *monitoring.Server } diff --git a/test/integration/jetstream_to_timescale_test.go b/test/integration/jetstream_to_timescale_test.go new file mode 100644 index 0000000..e3286dd --- /dev/null +++ b/test/integration/jetstream_to_timescale_test.go @@ -0,0 +1,278 @@ +//go:build integration + +package integration + +import ( + "context" + "fmt" + "os" + "path/filepath" + "testing" + "time" + + "caatsm/internal/adapter/parser" + "caatsm/internal/app" + "caatsm/internal/domain" + "caatsm/internal/infra/config" + natsinfra "caatsm/internal/infra/nats" + postgresinfra "caatsm/internal/infra/postgres" + + "github.com/jackc/pgx/v5/pgxpool" + "github.com/nats-io/nats.go" + tc "github.com/testcontainers/testcontainers-go" + "github.com/testcontainers/testcontainers-go/wait" + "go.uber.org/zap" +) + +func TestJetStreamToTimescaleFlow(t *testing.T) { + if testing.Short() { + t.Skip("skipping integration test in short mode") + } + + ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute) + defer cancel() + + pgContainer, pgURL := startPostgres(ctx, t) + defer func() { + _ = pgContainer.Terminate(context.Background()) + }() + + natsContainer, natsURL := startNATS(ctx, t) + defer func() { + _ = natsContainer.Terminate(context.Background()) + }() + + cfg := buildTestConfig(natsURL, pgURL) + logger := zap.NewNop() + + pool, err := postgresinfra.ProvideDB(cfg) + if err != nil { + t.Fatalf("failed to init postgres: %v", err) + } + defer pool.Close() + + if err := applyDDL(ctx, pool); err != nil { + t.Fatalf("failed to apply schema: %v", err) + } + + repo, err := postgresinfra.ProvideRepository(pool, logger) + if err != nil { + t.Fatalf("failed to init repository: %v", err) + } + + conn, err := natsinfra.ProvideNATSConn(cfg, logger) + if err != nil { + t.Fatalf("failed to connect nats: %v", err) + } + defer conn.Drain() + + js, err := natsinfra.ProvideJetStream(conn, cfg, logger) + if err != nil { + t.Fatalf("failed to init jetstream: %v", err) + } + + publisher, err := natsinfra.ProvidePublisher(js, cfg, logger) + if err != nil { + t.Fatalf("failed to init publisher: %v", err) + } + + proc := app.NewMessageProcessor(parser.ProvideParser(), repo, publisher, logger) + consumer, err := natsinfra.ProvideConsumer(conn, js, proc, cfg, logger) + if err != nil { + t.Fatalf("failed to init consumer: %v", err) + } + + runCtx, runCancel := context.WithCancel(ctx) + defer runCancel() + + errCh := make(chan error, 1) + go func() { + errCh <- consumer.Start(runCtx) + }() + defer func() { + runCancel() + select { + case <-errCh: + case <-time.After(2 * time.Second): + } + _ = consumer.Shutdown(context.Background()) + }() + + // Publish a message to the input subject. + payload := []byte(`ZCZC ARR1234 150631 +FF ZBTJZPZX +150630 ZBACZQZX +(ARR-CCA1234-A1234-ZBTJ1500-ZGGG0135) +NNNN`) + + msg := nats.NewMsg(cfg.EffectiveSubscriptionTopic()) + msg.Data = payload + msg.Header.Set("Nats-Msg-Id", "integration-1") + if _, err := js.PublishMsg(msg); err != nil { + t.Fatalf("failed to publish test telegram: %v", err) + } + + waitCtx, waitCancel := context.WithTimeout(ctx, 20*time.Second) + defer waitCancel() + + for { + select { + case <-waitCtx.Done(): + t.Fatalf("telegrams row not persisted: %v", waitCtx.Err()) + default: + } + + var status string + err := pool.QueryRow(ctx, ` + SELECT status FROM aviation.telegrams WHERE message_id = $1 LIMIT 1 + `, "ARR1234").Scan(&status) + if err == nil && status == string(domain.MessageStatusParsed) { + return + } + + time.Sleep(500 * time.Millisecond) + } +} + +func startPostgres(ctx context.Context, t *testing.T) (tc.Container, string) { + t.Helper() + req := tc.ContainerRequest{ + Image: "timescale/timescaledb:2.15.2-pg16", + ExposedPorts: []string{"5432/tcp"}, + Env: map[string]string{ + "POSTGRES_USER": "postgres", + "POSTGRES_PASSWORD": "postgres", + "POSTGRES_DB": "aviation", + }, + WaitingFor: wait.ForAll( + wait.ForListeningPort("5432/tcp"), + wait.ForLog("database system is ready to accept connections"), + ).WithDeadline(2 * time.Minute), + } + + container, err := tc.GenericContainer(ctx, tc.GenericContainerRequest{ + ContainerRequest: req, + Started: true, + }) + if err != nil { + t.Fatalf("failed to start postgres container: %v", err) + } + + host, err := container.Host(ctx) + if err != nil { + t.Fatalf("failed to resolve postgres host: %v", err) + } + port, err := container.MappedPort(ctx, "5432") + if err != nil { + t.Fatalf("failed to resolve postgres port: %v", err) + } + + url := fmt.Sprintf("postgres://postgres:postgres@%s:%s/aviation?sslmode=disable", host, port.Port()) + return container, url +} + +func startNATS(ctx context.Context, t *testing.T) (tc.Container, string) { + t.Helper() + req := tc.ContainerRequest{ + Image: "nats:2.10-alpine", + ExposedPorts: []string{"4222/tcp"}, + Cmd: []string{"-js", "--server_name=integration"}, + WaitingFor: wait.ForAll( + wait.ForListeningPort("4222/tcp"), + wait.ForLog("Server is ready"), + ).WithDeadline(2 * time.Minute), + } + + container, err := tc.GenericContainer(ctx, tc.GenericContainerRequest{ + ContainerRequest: req, + Started: true, + }) + if err != nil { + t.Fatalf("failed to start nats container: %v", err) + } + + host, err := container.Host(ctx) + if err != nil { + t.Fatalf("failed to resolve nats host: %v", err) + } + port, err := container.MappedPort(ctx, "4222") + if err != nil { + t.Fatalf("failed to resolve nats port: %v", err) + } + + return container, fmt.Sprintf("nats://%s:%s", host, port.Port()) +} + +func buildTestConfig(natsURL, pgURL string) *config.Config { + cfg := &config.Config{ + NATS: config.NATSConfig{ + URL: natsURL, + Mode: "jetstream", + Stream: "INTEGRATION_TELEGRAM", + Consumer: "integration-consumer", + StreamLimits: config.StreamLimitsConfig{ + MaxMsgs: 1000, + MaxBytes: 67108864, + MaxAge: time.Hour, + Discard: "old", + Storage: "file", + Replicas: 1, + }, + ConsumerRules: config.ConsumerRulesConfig{ + MaxDeliver: 3, + AckWait: 15 * time.Second, + MaxAckPending: 128, + DeliverPolicy: "all", + ReplayPolicy: "instant", + }, + }, + Postgres: config.PostgresConfig{ + URL: pgURL, + MaxConns: 4, + MinConns: 1, + }, + App: config.AppConfig{ + BatchSize: 1, + BatchTimeout: time.Second, + MonitorInterval: time.Second, + }, + Log: config.LogConfig{ + Level: "error", + Format: "json", + }, + Publisher: config.PublisherConfig{ + Topic: "integration.telegram.json", + }, + Subscription: config.SubscriptionConfig{ + Topic: "integration.telegram.serial", + }, + Telemetry: config.TelemetryConfig{ + Enabled: false, + }, + Timeouts: config.TimeoutsConfig{ + Server: 5 * time.Second, + ReconnectWait: 2 * time.Second, + Close: 5 * time.Second, + AckWait: 15 * time.Second, + }, + } + + // Monitoring server disabled for tests. + cfg.Monitoring.Disabled = true + + if err := cfg.Validate(); err != nil { + panic(fmt.Sprintf("invalid integration config: %v", err)) + } + + return cfg +} + +func applyDDL(ctx context.Context, pool *pgxpool.Pool) error { + ddlPath := filepath.Join("..", "..", "internal", "repository", "telegrams.ddl") + bytes, err := os.ReadFile(ddlPath) + if err != nil { + return fmt.Errorf("read ddl: %w", err) + } + _, err = pool.Exec(ctx, string(bytes)) + return err +}