windy/go-caatsm
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

✨ Revise observability features by introducing new health and liveness endpoints (/livez and /readyz), updating Prometheus metrics to track NATS consumer pending messages, and enhancing telemetry integration for better monitoring. Update documentation to reflect these changes and ensure consistency in metric naming conventions.

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