diff --git a/Taskfile.yml b/Taskfile.yml index eb74482..f5e6342 100644 --- a/Taskfile.yml +++ b/Taskfile.yml @@ -147,6 +147,13 @@ tasks: - echo "Stopping dev infrastructure..." - docker compose -f docker-compose.dev.yml down -v + dev-run-compose: + desc: Run receiver inside docker-compose dev stack + cmds: + - echo "Starting dev stack including caatsm-receiver..." + - docker compose -f docker-compose.dev.yml up -d postgres nats nats-box nats-exporter + - docker compose -f docker-compose.dev.yml up -d otel-collector jaeger prometheus grafana caatsm-receiver + dev-run: desc: Run receiver locally against dev stack deps: @@ -158,6 +165,9 @@ tasks: CAATSM_TELEMETRY_ENABLED: "true" CAATSM_TELEMETRY_ENDPOINT: localhost:4318 CAATSM_TELEMETRY_INSECURE: "true" + CAATSM_MONITORING_ADDR: 0.0.0.0:2112 + CAATSM_MONITORING_ENABLE_METRICS: "true" + CAATSM_MONITORING_ENABLE_HEALTH: "true" GO_ENV: dev cmds: - | @@ -174,9 +184,9 @@ tasks: GO_ENV=dev \ NATS_URL=${CAATSM_NATS_URL:-nats://localhost:4222} \ SUBJECT=${CAATSM_NATS_SUBJECT:-telegram.serial} \ - COUNT=${COUNT:-10} \ - CATEGORY=${CATEGORY:-mixed} \ - STATUS=${STATUS:-random} \ + COUNT={{.COUNT | default "10"}} \ + CATEGORY={{.CATEGORY | default "mixed"}} \ + STATUS={{.STATUS | default "random"}} \ bash -c ' set -euo pipefail cmd=(go run ./cmd/seed-telegrams) diff --git a/configs/config.dev.toml b/configs/config.dev.toml index 1432001..1985196 100644 --- a/configs/config.dev.toml +++ b/configs/config.dev.toml @@ -49,7 +49,7 @@ monitor_interval = "30s" [log] level = "info" -format = "json" +format = "console" [telemetry] enabled = true diff --git a/configs/prometheus.dev.yml b/configs/prometheus.dev.yml index 705dbe5..858947f 100644 --- a/configs/prometheus.dev.yml +++ b/configs/prometheus.dev.yml @@ -15,7 +15,5 @@ scrape_configs: - job_name: "caatsm-receiver" static_configs: - targets: - # When go-caatsm runs on the host/WSL and Prometheus runs in Docker, - # use host.docker.internal to reach the host monitoring server. - - "host.docker.internal:2112" + - 172.23.189.12:2112 diff --git a/internal/infra/nats/consumer.go b/internal/infra/nats/consumer.go index 21a7189..e68ed0b 100644 --- a/internal/infra/nats/consumer.go +++ b/internal/infra/nats/consumer.go @@ -241,6 +241,17 @@ func (c *Consumer) startJetStream(ctx context.Context) error { // Timeout is expected when no messages are available continue } + if errors.Is(err, nats.ErrNoResponders) { + // JetStream API is currently unavailable (e.g., NATS just restarted or JetStream not ready). + // Back off a bit to avoid log spam while allowing the system to recover. + c.logger.Warn("JetStream not available, will retry", + zap.Error(err), + zap.String("stream", c.streamName), + zap.String("consumer", c.consumerName), + ) + time.Sleep(5 * time.Second) + continue + } c.logger.Error("Failed to fetch messages", zap.Error(err)) time.Sleep(time.Second) continue @@ -377,7 +388,7 @@ func (c *Consumer) emitConsumerStats(ctx context.Context) { continue } - c.logger.Info("JetStream consumer metrics", + c.logger.Debug("JetStream consumer metrics", zap.String("stream", c.streamName), zap.String("consumer", c.consumerName), zap.Uint64("num_ack_pending", uint64(info.NumAckPending)), diff --git a/internal/infra/nats/jetstream.go b/internal/infra/nats/jetstream.go index b03d2a3..0e732a2 100644 --- a/internal/infra/nats/jetstream.go +++ b/internal/infra/nats/jetstream.go @@ -18,6 +18,8 @@ func ProvideNATSConn(cfg *config.Config, logger *zap.Logger) (*nats.Conn, error) nats.RetryOnFailedConnect(true), nats.Timeout(cfg.Timeouts.Server), nats.ReconnectWait(cfg.Timeouts.ReconnectWait), + // Use infinite reconnects so the app survives long NATS outages (e.g. docker compose down/up). + nats.MaxReconnects(-1), nats.DisconnectErrHandler(func(nc *nats.Conn, err error) { if err != nil { logger.Warn("NATS disconnected", zap.Error(err))