From 71905aca3dd0938700ea050de2fb867cfb49b236 Mon Sep 17 00:00:00 2001 From: lingniu Date: Fri, 3 Jul 2026 18:46:40 +0800 Subject: [PATCH] feat(go): expose history batch pending metrics --- docs/ops/100k-capacity-baseline.md | 4 +- docs/ops/go-service-observability.md | 3 + docs/ops/go-vehicle-ingest-memory.md | 4 +- docs/ops/vehicle-ingest-runbook.md | 2 + go/vehicle-gateway/cmd/history-writer/main.go | 10 +++ .../cmd/history-writer/main_test.go | 63 +++++++++++++++++-- 6 files changed, 79 insertions(+), 7 deletions(-) diff --git a/docs/ops/100k-capacity-baseline.md b/docs/ops/100k-capacity-baseline.md index 4a141486..8ff05529 100644 --- a/docs/ops/100k-capacity-baseline.md +++ b/docs/ops/100k-capacity-baseline.md @@ -68,6 +68,8 @@ systemd gateway 已配置 `LimitNOFILE=1048576`，需要持续保持。 | Gateway | parse/publish duration | p99 小于容量目标 | | History writer | `vehicle_history_batch_flush_total{status}` | `error` 不增长 | | History writer | `vehicle_history_batch_rows_total{status}` | batch 行数持续增长 | +| History writer | `vehicle_history_batch_pending_messages` | 稳态接近 0，burst 后下降 | +| History writer | `vehicle_history_batch_pending_rows` | 稳态接近 0，burst 后下降 | | History writer | `vehicle_history_batch_flush_duration_ms{status}` | flush 延迟不持续上升 | | NATS bridge | `vehicle_bridge_nats_consumer_ack_pending` | 稳态为 0 | | NATS bridge | `vehicle_bridge_nats_consumer_pending` | burst 后下降 | @@ -112,7 +114,7 @@ systemd gateway 已配置 `LimitNOFILE=1048576`，需要持续保持。 - gateway frame/publish counters - NATS pending 和 ack pending - Kafka lag -- Redis/MySQL/TDengine 写入耗时 +- Redis/MySQL/TDengine 写入耗时和 pending depth - CPU/load/memory/file descriptors - 错误日志 diff --git a/docs/ops/go-service-observability.md b/docs/ops/go-service-observability.md index 5320f897..b6220be4 100644 --- a/docs/ops/go-service-observability.md +++ b/docs/ops/go-service-observability.md @@ -43,6 +43,9 @@ curl -fsS http://127.0.0.1:20200/readyz | `vehicle_bridge_nats_consumer_ack_pending` | JetStream messages delivered to bridge but not yet acked. | | `vehicle_bridge_nats_consumer_waiting` | Pull requests waiting on the bridge durable consumer. | | `vehicle_history_writes_total` | TDengine history writes. Labels: `topic`, `status`. | +| `vehicle_history_batch_pending_messages` | Messages fetched by history writer but not yet flushed and committed. | +| `vehicle_history_batch_pending_rows` | Parsed raw envelopes waiting in the current TDengine batch. | +| `vehicle_history_batch_flush_duration_ms` | Last TDengine batch flush duration. Labels: `status`. | | `vehicle_stat_writes_total` | MySQL metric writes. Labels: `topic`, `status`. | | `vehicle_realtime_updates_total` | Redis/MySQL realtime projector updates. Labels: `topic`, `status`. | | `vehicle_history_kafka_lag` | Estimated Kafka lag for history writer by topic and partition. | diff --git a/docs/ops/go-vehicle-ingest-memory.md b/docs/ops/go-vehicle-ingest-memory.md index 79a86bd3..c8228f7a 100644 --- a/docs/ops/go-vehicle-ingest-memory.md +++ b/docs/ops/go-vehicle-ingest-memory.md @@ -420,10 +420,12 @@ vehicle_async_sink_queue_depth{sink} ```text vehicle_history_batch_flush_total{status} vehicle_history_batch_rows_total{status} +vehicle_history_batch_pending_messages +vehicle_history_batch_pending_rows vehicle_history_batch_flush_duration_ms{status} ``` -进入带帧率压测时，必须同时观察 `vehicle_history_kafka_lag` 和 `vehicle_history_batch_*`。 +进入带帧率压测时，必须同时观察 `vehicle_history_kafka_lag`、`vehicle_history_batch_pending_*` 和 `vehicle_history_batch_*`。如果 pending 持续非 0 但 Kafka lag 尚未放大，说明 history-writer 已经在本地批写或提交阶段形成早期积压。 ## 下次恢复上下文时先做 diff --git a/docs/ops/vehicle-ingest-runbook.md b/docs/ops/vehicle-ingest-runbook.md index 96db5240..b79fa765 100644 --- a/docs/ops/vehicle-ingest-runbook.md +++ b/docs/ops/vehicle-ingest-runbook.md @@ -120,6 +120,8 @@ go run ./cmd/load-sim \ | `vehicle_async_sink_enqueue_total{status="timeout"}` | 任意增长 | Gateway publish 队列已满或 worker 长时间阻塞，入口可能开始丢实时性。 | | `vehicle_async_sink_publish_total{status="error"}` | 连续增长 | NATS/Kafka publish 失败，需要先查中间件连接和日志。 | | `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 放大。 | | `vehicle_history_batch_flush_duration_ms{status="ok"}` | 持续上升 | TDengine 写入延迟增加，可能需要降低 batch size 或扩容 TDengine。 | | `vehicle_bridge_nats_consumer_ack_pending` | 连续 2 分钟 `> 0` | 消息已投递给 bridge，但 Kafka 写入后未完成 ack。 | | `vehicle_bridge_nats_consumer_pending` | 持续增长且 `> 10000` | bridge 消费 NATS 的速度跟不上生产速度。 | diff --git a/go/vehicle-gateway/cmd/history-writer/main.go b/go/vehicle-gateway/cmd/history-writer/main.go index 35dc7a8e..c77696a5 100644 --- a/go/vehicle-gateway/cmd/history-writer/main.go +++ b/go/vehicle-gateway/cmd/history-writer/main.go @@ -174,6 +174,8 @@ func processHistoryBatch(ctx context.Context, logger interface { envelopes = append(envelopes, env) } if len(envelopes) > 0 { + setBatchPending(registry, len(messages), len(envelopes)) + defer setBatchPending(registry, 0, 0) started := time.Now() err := appender.AppendAllBatch(messageCtx, envelopes) elapsed := time.Since(started) @@ -230,6 +232,14 @@ func setBatchDuration(registry *metrics.Registry, status string, elapsed time.Du registry.SetGauge("vehicle_history_batch_flush_duration_ms", metrics.Labels{"status": status}, float64(elapsed.Milliseconds())) } +func setBatchPending(registry *metrics.Registry, messages int, rows int) { + if registry == nil { + return + } + registry.SetGauge("vehicle_history_batch_pending_messages", nil, float64(messages)) + registry.SetGauge("vehicle_history_batch_pending_rows", nil, float64(rows)) +} + func addWriterLagMetric(registry *metrics.Registry, message kafka.Message) { if registry == nil { return diff --git a/go/vehicle-gateway/cmd/history-writer/main_test.go b/go/vehicle-gateway/cmd/history-writer/main_test.go index c911df31..ab80de64 100644 --- a/go/vehicle-gateway/cmd/history-writer/main_test.go +++ b/go/vehicle-gateway/cmd/history-writer/main_test.go @@ -125,6 +125,55 @@ func TestProcessHistoryBatchAppendsAllBeforeCommit(t *testing.T) { } } +func TestProcessHistoryBatchExposesPendingBatchDepthDuringAppend(t *testing.T) { + first := envelope.FrameEnvelope{Protocol: envelope.ProtocolGB32960, VIN: "VIN001", MessageID: "0x02"} + second := envelope.FrameEnvelope{Protocol: envelope.ProtocolJT808, Phone: "13307795425", MessageID: "0x0200"} + firstPayload, err := json.Marshal(first) + if err != nil { + t.Fatal(err) + } + secondPayload, err := json.Marshal(second) + if err != nil { + t.Fatal(err) + } + registry := metrics.NewRegistry() + appender := &contextCheckingHistoryAppender{ + onAppendBatch: func() { + text := registry.Render() + for _, want := range []string{ + `vehicle_history_batch_pending_messages 2`, + `vehicle_history_batch_pending_rows 2`, + } { + if !strings.Contains(text, want) { + t.Fatalf("pending batch metric missing %s while appending:\n%s", want, text) + } + } + }, + } + + processHistoryBatch( + context.Background(), + discardHistoryLogger{}, + registry, + appender, + &contextCheckingHistoryCommitter{}, + []kafka.Message{ + {Topic: "vehicle.raw.go.gb32960.v1", Partition: 1, Offset: 10, HighWaterMark: 13, Value: firstPayload}, + {Topic: "vehicle.raw.go.jt808.v1", Partition: 2, Offset: 20, HighWaterMark: 21, Value: secondPayload}, + }, + ) + + text := registry.Render() + for _, want := range []string{ + `vehicle_history_batch_pending_messages 0`, + `vehicle_history_batch_pending_rows 0`, + } { + if !strings.Contains(text, want) { + t.Fatalf("pending batch metric should reset after append, missing %s:\n%s", want, text) + } + } +} + func TestProcessHistoryBatchDoesNotCommitWhenAppendFails(t *testing.T) { env := envelope.FrameEnvelope{Protocol: envelope.ProtocolGB32960, VIN: "VIN001", MessageID: "0x02"} payload, err := json.Marshal(env) @@ -189,11 +238,12 @@ func TestLoadConfigReadsBatchSettings(t *testing.T) { } type contextCheckingHistoryAppender struct { - ctxErr error - count int - batchCount int - batch []envelope.FrameEnvelope - err error + ctxErr error + count int + batchCount int + batch []envelope.FrameEnvelope + err error + onAppendBatch func() } func (a *contextCheckingHistoryAppender) AppendAll(ctx context.Context, _ envelope.FrameEnvelope) error { @@ -209,6 +259,9 @@ func (a *contextCheckingHistoryAppender) AppendAllBatch(ctx context.Context, env a.ctxErr = ctx.Err() a.batchCount++ a.batch = append([]envelope.FrameEnvelope(nil), envs...) + if a.onAppendBatch != nil { + a.onAppendBatch() + } if a.ctxErr != nil { return a.ctxErr }