From afb301cb7a6ad0626c2e3b22933ff75e2d4e5648 Mon Sep 17 00:00:00 2001 From: lingniu Date: Fri, 3 Jul 2026 19:02:24 +0800 Subject: [PATCH] feat(go): expose async sink publish latency histogram --- docs/ops/100k-capacity-baseline.md | 1 + docs/ops/go-service-observability.md | 1 + docs/ops/go-vehicle-ingest-memory.md | 5 ++++- docs/ops/vehicle-ingest-runbook.md | 1 + go/vehicle-gateway/internal/eventbus/async_sink.go | 6 +++++- go/vehicle-gateway/internal/eventbus/async_sink_test.go | 3 +++ 6 files changed, 15 insertions(+), 2 deletions(-) diff --git a/docs/ops/100k-capacity-baseline.md b/docs/ops/100k-capacity-baseline.md index e35cfb3a..7949e09a 100644 --- a/docs/ops/100k-capacity-baseline.md +++ b/docs/ops/100k-capacity-baseline.md @@ -64,6 +64,7 @@ systemd gateway 已配置 `LimitNOFILE=1048576`，需要持续保持。 | 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 async sink | `vehicle_async_sink_publish_duration_ms_histogram_bucket` | p99 不持续上升 | | Gateway | `vehicle_gateway_publish_total{status="ok"}` | 持续增长 | | Gateway | `vehicle_gateway_frame_duration_ms_histogram_bucket` | p99 小于容量目标 | | History writer | `vehicle_history_batch_flush_total{status}` | `error` 不增长 | diff --git a/docs/ops/go-service-observability.md b/docs/ops/go-service-observability.md index 34ca8313..d8da6877 100644 --- a/docs/ops/go-service-observability.md +++ b/docs/ops/go-service-observability.md @@ -37,6 +37,7 @@ curl -fsS http://127.0.0.1:20200/readyz | `vehicle_gateway_frames_total` | Protocol frames received and parsed by the gateway. Labels: `protocol`, `status`. | | `vehicle_gateway_frame_duration_ms` | Last observed Gateway frame handling duration and histogram buckets for p95/p99. Labels: `protocol`, `status`. | | `vehicle_gateway_publish_total` | Gateway publish result for raw events. Unified appears only when the compatibility stream is explicitly enabled. Labels: `protocol`, `kind`, `status`. | +| `vehicle_async_sink_publish_duration_ms_histogram` | Async sink publish duration histogram covering worker publish calls to the delegate sink. Labels: `sink`, `kind`, `status`. | | `vehicle_bridge_messages_total` | NATS messages fetched by the bridge. Labels: `subject`, `status`. | | `vehicle_bridge_kafka_writes_total` | Bridge writes to Kafka. Labels: `topic`, `status`. | | `vehicle_bridge_nats_acks_total` | NATS ack results after Kafka write. Labels: `subject`, `status`. | diff --git a/docs/ops/go-vehicle-ingest-memory.md b/docs/ops/go-vehicle-ingest-memory.md index 0efe2ec6..17880f79 100644 --- a/docs/ops/go-vehicle-ingest-memory.md +++ b/docs/ops/go-vehicle-ingest-memory.md @@ -393,6 +393,9 @@ net.ipv4.tcp_tw_reuse = 1 vehicle_async_sink_enqueue_total{sink,kind,status} vehicle_async_sink_publish_total{sink,kind,status} vehicle_async_sink_publish_duration_ms{sink,kind,status} +vehicle_async_sink_publish_duration_ms_histogram_bucket{sink,kind,status,le} +vehicle_async_sink_publish_duration_ms_histogram_count{sink,kind,status} +vehicle_async_sink_publish_duration_ms_histogram_sum{sink,kind,status} vehicle_async_sink_queue_depth{sink} ``` @@ -404,7 +407,7 @@ vehicle_async_sink_queue_depth{sink} - NATS bridge ack pending 为 `0` - Gateway 启动后窗口无 `error|failed|panic|fatal` 日志 -这些指标是进入带帧率压测前的关键保护栏：如果 `queue_depth` 持续增长或 `enqueue_total{status="timeout"}` 增长，说明入口 publish 队列已成为瓶颈。 +这些指标是进入带帧率压测前的关键保护栏：如果 `queue_depth` 持续增长、`enqueue_total{status="timeout"}` 增长，或 `vehicle_async_sink_publish_duration_ms_histogram_bucket` 的 p99 持续上升，说明入口 publish 队列或下游 NATS/Kafka 写入已成为瓶颈。 ### NATS Kafka bridge batch 指标 diff --git a/docs/ops/vehicle-ingest-runbook.md b/docs/ops/vehicle-ingest-runbook.md index b3bf3629..7560169f 100644 --- a/docs/ops/vehicle-ingest-runbook.md +++ b/docs/ops/vehicle-ingest-runbook.md @@ -120,6 +120,7 @@ go run ./cmd/load-sim \ | `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_async_sink_publish_duration_ms_histogram_bucket` | p99 连续 5 分钟上升 | Gateway async worker 写 NATS/Kafka 变慢，通常会带动 queue depth 增长。 | | `vehicle_history_batch_flush_total{status="error"}` | 任意增长 | TDengine 批写失败；Kafka offset 不会提交，应先查 TDengine 和 SQL 错误。 | | `vehicle_history_batch_pending_messages` | 持续非 0 或 burst 后不回落 | history-writer 已拉取但未完成写入/提交，可能卡在 TDengine 或 Kafka commit。 | | `vehicle_history_batch_pending_rows` | 持续非 0 或 burst 后不回落 | TDengine 有批量写入积压，通常早于 Kafka lag 放大。 | diff --git a/go/vehicle-gateway/internal/eventbus/async_sink.go b/go/vehicle-gateway/internal/eventbus/async_sink.go index dac18f42..faa0605f 100644 --- a/go/vehicle-gateway/internal/eventbus/async_sink.go +++ b/go/vehicle-gateway/internal/eventbus/async_sink.go @@ -40,6 +40,8 @@ type asyncJob struct { var ErrAsyncSinkClosed = errors.New("async sink is closed") +var asyncSinkPublishDurationBucketsMS = []float64{1, 5, 10, 25, 50, 100, 250, 500, 1000, 5000} + func NewAsyncSink(delegate Sink, cfg AsyncConfig) *AsyncSink { if delegate == nil { panic("async delegate sink must not be nil") @@ -169,7 +171,9 @@ func (s *AsyncSink) recordPublish(kind string, status string, elapsed time.Durat "status": status, } s.metrics.IncCounter("vehicle_async_sink_publish_total", labels) - s.metrics.SetGauge("vehicle_async_sink_publish_duration_ms", labels, float64(elapsed.Milliseconds())) + elapsedMS := float64(elapsed.Milliseconds()) + s.metrics.SetGauge("vehicle_async_sink_publish_duration_ms", labels, elapsedMS) + s.metrics.ObserveHistogram("vehicle_async_sink_publish_duration_ms_histogram", labels, asyncSinkPublishDurationBucketsMS, elapsedMS) } func (s *AsyncSink) recordQueueDepth() { diff --git a/go/vehicle-gateway/internal/eventbus/async_sink_test.go b/go/vehicle-gateway/internal/eventbus/async_sink_test.go index 8c331499..05af6aec 100644 --- a/go/vehicle-gateway/internal/eventbus/async_sink_test.go +++ b/go/vehicle-gateway/internal/eventbus/async_sink_test.go @@ -83,6 +83,9 @@ func TestAsyncSinkRecordsQueueMetrics(t *testing.T) { `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`, + `vehicle_async_sink_publish_duration_ms_histogram_bucket{le="+Inf",kind="raw",sink="nats",status="ok"} 1`, + `vehicle_async_sink_publish_duration_ms_histogram_count{kind="raw",sink="nats",status="ok"} 1`, + `vehicle_async_sink_publish_duration_ms_histogram_sum{kind="raw",sink="nats",status="ok"}`, } { if !strings.Contains(text, want) { t.Fatalf("async sink metric missing %s:\n%s", want, text)