diff --git a/docs/ops/100k-capacity-baseline.md b/docs/ops/100k-capacity-baseline.md index 81416022..ff552ec6 100644 --- a/docs/ops/100k-capacity-baseline.md +++ b/docs/ops/100k-capacity-baseline.md @@ -61,6 +61,9 @@ systemd gateway 已配置 `LimitNOFILE=1048576`，需要持续保持。 | --- | --- | --- | | Gateway | `vehicle_gateway_active_connections` | 能稳定到阶段目标连接数 | | Gateway | `vehicle_gateway_connection_rejections_total` | 容量目标内不增长 | +| Gateway async sink | `vehicle_async_sink_queue_depth{sink}` | 队列深度不持续增长 | +| Gateway async sink | `vehicle_async_sink_enqueue_total{sink,kind,status}` | `timeout/closed` 不增长 | +| Gateway async sink | `vehicle_async_sink_publish_total{sink,kind,status}` | `error` 不增长 | | Gateway | `vehicle_gateway_publish_total{status="ok"}` | 持续增长 | | Gateway | parse/publish duration | p99 小于容量目标 | | NATS bridge | `vehicle_bridge_nats_consumer_ack_pending` | 稳态为 0 | diff --git a/docs/ops/vehicle-ingest-runbook.md b/docs/ops/vehicle-ingest-runbook.md index c37ac248..f71317f1 100644 --- a/docs/ops/vehicle-ingest-runbook.md +++ b/docs/ops/vehicle-ingest-runbook.md @@ -67,7 +67,7 @@ for port in 20211 20212 20213 20214 20200; do done curl -fsS http://127.0.0.1:20211/metrics \ - | grep -E 'vehicle_gateway_(active_connections|frames_total|publish_total)' + | grep -E 'vehicle_gateway_(active_connections|frames_total|publish_total)|vehicle_async_sink' curl -fsS http://127.0.0.1:20214/metrics \ | grep -E 'vehicle_bridge_(nats_consumer|kafka_writes_total|nats_acks_total)' @@ -115,6 +115,9 @@ go run ./cmd/load-sim \ | `/readyz` 非 ok | 立即处理 | 服务或依赖不可用。 | | Gateway 活跃连接 | 预期有流量时某协议降为 `0` | 上游网络、监听端口或进程可能异常。 | | `vehicle_gateway_frames_total{status!="OK"}` | 连续 2 分钟增长 | 解析器或上游报文质量异常。 | +| `vehicle_async_sink_queue_depth{sink="nats"}` | 持续增长且不回落 | Gateway 到 NATS/Kafka 的异步 publish 队列开始积压。 | +| `vehicle_async_sink_enqueue_total{status="timeout"}` | 任意增长 | Gateway publish 队列已满或 worker 长时间阻塞，入口可能开始丢实时性。 | +| `vehicle_async_sink_publish_total{status="error"}` | 连续增长 | NATS/Kafka publish 失败，需要先查中间件连接和日志。 | | `vehicle_bridge_nats_consumer_ack_pending` | 连续 2 分钟 `> 0` | 消息已投递给 bridge，但 Kafka 写入后未完成 ack。 | | `vehicle_bridge_nats_consumer_pending` | 持续增长且 `> 10000` | bridge 消费 NATS 的速度跟不上生产速度。 | | Kafka lag | 连续 5 分钟增长或 `> 10000` | 下游 consumer 或存储存在瓶颈。 | diff --git a/go/vehicle-gateway/cmd/gateway/main.go b/go/vehicle-gateway/cmd/gateway/main.go index 84e7397c..47bc429a 100644 --- a/go/vehicle-gateway/cmd/gateway/main.go +++ b/go/vehicle-gateway/cmd/gateway/main.go @@ -30,7 +30,8 @@ func main() { ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM) defer stop() - sink, err := buildSink(ctx, logger) + registry := metrics.NewRegistry() + sink, err := buildSink(ctx, logger, registry) if err != nil { logger.Error("build sink failed", "error", err) os.Exit(1) @@ -42,7 +43,6 @@ func main() { os.Exit(1) } defer closeResolver() - registry := metrics.NewRegistry() health.Start(ctx, logger, health.NewServer(env("HEALTH_ADDR", ""), "vehicle-gateway", nil, registry)) publishUnified := envBool("PUBLISH_UNIFIED_ENABLED", false) @@ -161,7 +161,7 @@ func buildIdentityResolver(ctx context.Context, logger *slog.Logger) (identity.R return resolver, func() { _ = db.Close() }, nil } -func buildSink(ctx context.Context, logger *slog.Logger) (eventbus.Sink, error) { +func buildSink(ctx context.Context, logger *slog.Logger, registry *metrics.Registry) (eventbus.Sink, error) { if strings.TrimSpace(os.Getenv("NATS_URL")) != "" { sink, err := eventbus.NewNATSSink(natsSinkConfigFromEnv()) if err != nil { @@ -180,6 +180,8 @@ func buildSink(ctx context.Context, logger *slog.Logger) (eventbus.Sink, error) QueueSize: queueSize, Workers: workers, OperationTimeout: timeout, + Metrics: registry, + Name: "nats", OnError: func(err error) { logger.Warn("nats async publish failed", "error", err) }, @@ -235,6 +237,8 @@ func buildSink(ctx context.Context, logger *slog.Logger) (eventbus.Sink, error) QueueSize: queueSize, Workers: workers, OperationTimeout: timeout, + Metrics: registry, + Name: "kafka", OnError: func(err error) { logger.Warn("kafka async publish failed", "error", err) }, diff --git a/go/vehicle-gateway/cmd/gateway/main_test.go b/go/vehicle-gateway/cmd/gateway/main_test.go index fe18dba6..d9cbf985 100644 --- a/go/vehicle-gateway/cmd/gateway/main_test.go +++ b/go/vehicle-gateway/cmd/gateway/main_test.go @@ -56,6 +56,23 @@ func TestGatewayDefaultsTo100KConnectionCeiling(t *testing.T) { } } +func TestGatewayPassesMetricsRegistryToAsyncSink(t *testing.T) { + source, err := os.ReadFile("main.go") + if err != nil { + t.Fatalf("read main.go: %v", err) + } + for _, want := range []string{ + "buildSink(ctx, logger, registry)", + "Metrics: registry", + `Name: "nats"`, + `Name: "kafka"`, + } { + if !strings.Contains(string(source), want) { + t.Fatalf("gateway async sink metrics wiring missing %s", want) + } + } +} + func TestNATSSinkConfigFromEnvDefaultsToGoSubjects(t *testing.T) { t.Setenv("NATS_URL", "nats://172.17.111.56:4222") diff --git a/go/vehicle-gateway/internal/eventbus/async_sink.go b/go/vehicle-gateway/internal/eventbus/async_sink.go index 8315427f..dac18f42 100644 --- a/go/vehicle-gateway/internal/eventbus/async_sink.go +++ b/go/vehicle-gateway/internal/eventbus/async_sink.go @@ -7,6 +7,7 @@ import ( "time" "lingniu-vehicle-ingest/go/vehicle-gateway/internal/envelope" + "lingniu-vehicle-ingest/go/vehicle-gateway/internal/metrics" ) type AsyncConfig struct { @@ -14,6 +15,8 @@ type AsyncConfig struct { Workers int OperationTimeout time.Duration OnError func(error) + Metrics *metrics.Registry + Name string } type AsyncSink struct { @@ -21,6 +24,8 @@ type AsyncSink struct { jobs chan asyncJob timeout time.Duration onError func(error) + metrics *metrics.Registry + name string closeOnce sync.Once closed chan struct{} @@ -48,11 +53,16 @@ func NewAsyncSink(delegate Sink, cfg AsyncConfig) *AsyncSink { if cfg.OperationTimeout <= 0 { cfg.OperationTimeout = 30 * time.Second } + if cfg.Name == "" { + cfg.Name = "async" + } s := &AsyncSink{ delegate: delegate, jobs: make(chan asyncJob, cfg.QueueSize), timeout: cfg.OperationTimeout, onError: cfg.OnError, + metrics: cfg.Metrics, + name: cfg.Name, closed: make(chan struct{}), done: make(chan struct{}), } @@ -91,15 +101,21 @@ func (s *AsyncSink) Close() error { func (s *AsyncSink) enqueue(ctx context.Context, job asyncJob) error { select { case <-s.closed: + s.recordEnqueue(job.kind, "closed") return ErrAsyncSinkClosed default: } select { case s.jobs <- job: + s.recordEnqueue(job.kind, "queued") + s.recordQueueDepth() return nil case <-s.closed: + s.recordEnqueue(job.kind, "closed") return ErrAsyncSinkClosed case <-ctx.Done(): + s.recordEnqueue(job.kind, "timeout") + s.recordQueueDepth() return ctx.Err() } } @@ -107,7 +123,9 @@ func (s *AsyncSink) enqueue(ctx context.Context, job asyncJob) error { func (s *AsyncSink) worker() { defer s.wg.Done() for job := range s.jobs { + s.recordQueueDepth() ctx, cancel := context.WithTimeout(context.Background(), s.timeout) + started := time.Now() var err error switch job.kind { case "raw": @@ -118,8 +136,47 @@ func (s *AsyncSink) worker() { err = s.delegate.PublishFields(ctx, job.env) } cancel() + status := "ok" + if err != nil { + status = "error" + } + s.recordPublish(job.kind, status, time.Since(started)) if err != nil && s.onError != nil { s.onError(err) } + s.recordQueueDepth() } } + +func (s *AsyncSink) recordEnqueue(kind string, status string) { + if s.metrics == nil { + return + } + s.metrics.IncCounter("vehicle_async_sink_enqueue_total", metrics.Labels{ + "sink": s.name, + "kind": kind, + "status": status, + }) +} + +func (s *AsyncSink) recordPublish(kind string, status string, elapsed time.Duration) { + if s.metrics == nil { + return + } + labels := metrics.Labels{ + "sink": s.name, + "kind": kind, + "status": status, + } + s.metrics.IncCounter("vehicle_async_sink_publish_total", labels) + s.metrics.SetGauge("vehicle_async_sink_publish_duration_ms", labels, float64(elapsed.Milliseconds())) +} + +func (s *AsyncSink) recordQueueDepth() { + if s.metrics == nil { + return + } + s.metrics.SetGauge("vehicle_async_sink_queue_depth", metrics.Labels{ + "sink": s.name, + }, float64(len(s.jobs))) +} diff --git a/go/vehicle-gateway/internal/eventbus/async_sink_test.go b/go/vehicle-gateway/internal/eventbus/async_sink_test.go index 4264806c..8c331499 100644 --- a/go/vehicle-gateway/internal/eventbus/async_sink_test.go +++ b/go/vehicle-gateway/internal/eventbus/async_sink_test.go @@ -2,10 +2,12 @@ package eventbus import ( "context" + "strings" "testing" "time" "lingniu-vehicle-ingest/go/vehicle-gateway/internal/envelope" + "lingniu-vehicle-ingest/go/vehicle-gateway/internal/metrics" ) func TestAsyncSinkPublishRawReturnsAfterQueueing(t *testing.T) { @@ -51,6 +53,85 @@ func TestAsyncSinkPublishesInFIFOOrderWithSingleWorker(t *testing.T) { } } +func TestAsyncSinkRecordsQueueMetrics(t *testing.T) { + registry := metrics.NewRegistry() + delegate := newBlockingSink() + sink := NewAsyncSink(delegate, AsyncConfig{ + QueueSize: 2, + Workers: 1, + OperationTimeout: time.Second, + Metrics: registry, + Name: "nats", + }) + + env := envelope.FrameEnvelope{Protocol: envelope.ProtocolGB32960, VIN: "VIN001"} + if err := sink.PublishRaw(context.Background(), env); err != nil { + t.Fatalf("PublishRaw() error = %v", err) + } + if err := sink.PublishFields(context.Background(), env); err != nil { + t.Fatalf("PublishFields() error = %v", err) + } + delegate.release() + if err := sink.Close(); err != nil { + t.Fatalf("Close() error = %v", err) + } + + text := registry.Render() + for _, want := range []string{ + `vehicle_async_sink_enqueue_total{kind="raw",sink="nats",status="queued"} 1`, + `vehicle_async_sink_enqueue_total{kind="fields",sink="nats",status="queued"} 1`, + `vehicle_async_sink_queue_depth{sink="nats"}`, + `vehicle_async_sink_publish_total{kind="raw",sink="nats",status="ok"} 1`, + `vehicle_async_sink_publish_total{kind="fields",sink="nats",status="ok"} 1`, + } { + if !strings.Contains(text, want) { + t.Fatalf("async sink metric missing %s:\n%s", want, text) + } + } +} + +func TestAsyncSinkRecordsEnqueueTimeoutWhenQueueIsFull(t *testing.T) { + registry := metrics.NewRegistry() + delegate := newBlockingSink() + sink := NewAsyncSink(delegate, AsyncConfig{ + QueueSize: 1, + Workers: 1, + OperationTimeout: time.Second, + Metrics: registry, + Name: "nats", + }) + defer sink.Close() + + env := envelope.FrameEnvelope{Protocol: envelope.ProtocolJT808, Phone: "13307795425"} + if err := sink.PublishRaw(context.Background(), env); err != nil { + t.Fatalf("first PublishRaw() error = %v", err) + } + select { + case <-delegate.rawStarted: + case <-time.After(time.Second): + t.Fatal("delegate raw publish was not started") + } + if err := sink.PublishFields(context.Background(), env); err != nil { + t.Fatalf("second PublishFields() error = %v", err) + } + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Millisecond) + defer cancel() + if err := sink.PublishUnified(ctx, env); err == nil { + t.Fatal("third PublishUnified() error = nil, want context deadline exceeded") + } + + text := registry.Render() + for _, want := range []string{ + `vehicle_async_sink_enqueue_total{kind="unified",sink="nats",status="timeout"} 1`, + `vehicle_async_sink_queue_depth{sink="nats"} 1`, + } { + if !strings.Contains(text, want) { + t.Fatalf("async sink timeout metric missing %s:\n%s", want, text) + } + } + delegate.release() +} + type blockingSink struct { rawStarted chan struct{} releaseRaw chan struct{}