From 16ee09180ccdbc9d3afc8e182be876a66271ae3a Mon Sep 17 00:00:00 2001 From: lingniu Date: Fri, 3 Jul 2026 18:54:00 +0800 Subject: [PATCH] feat(go): expose gateway frame duration histogram --- docs/ops/100k-capacity-baseline.md | 4 +- docs/ops/go-service-observability.md | 1 + docs/ops/go-vehicle-ingest-memory.md | 12 ++ docs/ops/vehicle-ingest-runbook.md | 1 + .../internal/gateway/mqtt_client.go | 8 +- .../internal/gateway/mqtt_client_test.go | 3 + .../internal/gateway/tcp_server.go | 10 +- .../internal/gateway/tcp_server_test.go | 3 + .../internal/metrics/metrics.go | 128 +++++++++++++++++- .../internal/metrics/metrics_test.go | 21 +++ 10 files changed, 182 insertions(+), 9 deletions(-) diff --git a/docs/ops/100k-capacity-baseline.md b/docs/ops/100k-capacity-baseline.md index 8ff05529..8bc34105 100644 --- a/docs/ops/100k-capacity-baseline.md +++ b/docs/ops/100k-capacity-baseline.md @@ -38,7 +38,7 @@ - `net.core.somaxconn=128`，无法作为 100K 连接生产基线。 - Gateway 默认 `TCP_MAX_CONNECTIONS` 已调整为 `120000`，生产仍可通过环境变量覆盖。 - 已新增可重复的 TCP 连接压测工具，后续需要用它跑 10K/50K/100K 阶段测试并记录结果。 -- 仍缺按协议维度的连接生命周期、读超时、parse/publish p95/p99 直方图。 +- 仍缺读超时按协议维度计数；Gateway frame duration 已提供 histogram，可用于 parse + enqueue + response 的 p95/p99 估算。 - TDengine history writer 当前逐消息插入，后续需要批量写入。 - Kafka topic 当前 12 分区，100K 目标下需要结合实际 FPS 再评估分区数。 @@ -65,7 +65,7 @@ systemd gateway 已配置 `LimitNOFILE=1048576`，需要持续保持。 | 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 小于容量目标 | +| Gateway | `vehicle_gateway_frame_duration_ms_histogram_bucket` | 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 后下降 | diff --git a/docs/ops/go-service-observability.md b/docs/ops/go-service-observability.md index b6220be4..7f553d56 100644 --- a/docs/ops/go-service-observability.md +++ b/docs/ops/go-service-observability.md @@ -35,6 +35,7 @@ curl -fsS http://127.0.0.1:20200/readyz | --- | --- | | `vehicle_gateway_active_connections` | Current TCP connections by protocol. Labels: `protocol`. | | `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_bridge_messages_total` | NATS messages fetched by the bridge. Labels: `subject`, `status`. | | `vehicle_bridge_kafka_writes_total` | Bridge writes to Kafka. Labels: `topic`, `status`. | diff --git a/docs/ops/go-vehicle-ingest-memory.md b/docs/ops/go-vehicle-ingest-memory.md index c8228f7a..1f14c31b 100644 --- a/docs/ops/go-vehicle-ingest-memory.md +++ b/docs/ops/go-vehicle-ingest-memory.md @@ -406,6 +406,18 @@ vehicle_async_sink_queue_depth{sink} 这些指标是进入带帧率压测前的关键保护栏：如果 `queue_depth` 持续增长或 `enqueue_total{status="timeout"}` 增长，说明入口 publish 队列已成为瓶颈。 +### Gateway frame duration histogram + +2026-07-03 已为 Gateway 帧处理耗时增加 histogram： + +```text +vehicle_gateway_frame_duration_ms_histogram_bucket{protocol,status,le} +vehicle_gateway_frame_duration_ms_histogram_count{protocol,status} +vehicle_gateway_frame_duration_ms_histogram_sum{protocol,status} +``` + +该耗时覆盖单帧从解析、identity resolve、parsed fields 生成、publish enqueue 到协议响应写入的整体入口处理路径。带帧率压测时用它按协议观察 p95/p99，不能只看 `vehicle_gateway_frame_duration_ms` 最后一条 gauge。 + ### History writer TDengine 批写 2026-07-03 已实现 history-writer 第一阶段批写： diff --git a/docs/ops/vehicle-ingest-runbook.md b/docs/ops/vehicle-ingest-runbook.md index b79fa765..0f487987 100644 --- a/docs/ops/vehicle-ingest-runbook.md +++ b/docs/ops/vehicle-ingest-runbook.md @@ -116,6 +116,7 @@ go run ./cmd/load-sim \ | `/readyz` 非 ok | 立即处理 | 服务或依赖不可用。 | | Gateway 活跃连接 | 预期有流量时某协议降为 `0` | 上游网络、监听端口或进程可能异常。 | | `vehicle_gateway_frames_total{status!="OK"}` | 连续 2 分钟增长 | 解析器或上游报文质量异常。 | +| `vehicle_gateway_frame_duration_ms_histogram_bucket` | p99 连续 5 分钟超过容量目标 | Gateway parse、identity resolve、publish enqueue 或响应链路变慢。 | | `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 失败，需要先查中间件连接和日志。 | diff --git a/go/vehicle-gateway/internal/gateway/mqtt_client.go b/go/vehicle-gateway/internal/gateway/mqtt_client.go index 81a6fa74..6b19e805 100644 --- a/go/vehicle-gateway/internal/gateway/mqtt_client.go +++ b/go/vehicle-gateway/internal/gateway/mqtt_client.go @@ -266,10 +266,16 @@ func (c *MQTTClient) recordFrameDuration(status envelope.ParseStatus, elapsed ti if c.cfg.Metrics == nil { return } + elapsedMS := float64(elapsed.Milliseconds()) + labels := metrics.Labels{ + "protocol": string(envelope.ProtocolYutongMQTT), + "status": string(status), + } c.cfg.Metrics.SetGauge("vehicle_gateway_frame_duration_ms", metrics.Labels{ "protocol": string(envelope.ProtocolYutongMQTT), "status": string(status), - }, float64(elapsed.Milliseconds())) + }, elapsedMS) + c.cfg.Metrics.ObserveHistogram("vehicle_gateway_frame_duration_ms_histogram", labels, gatewayFrameDurationBucketsMS, elapsedMS) } func (c *MQTTClient) recordPublishMetric(kind string, status string) { diff --git a/go/vehicle-gateway/internal/gateway/mqtt_client_test.go b/go/vehicle-gateway/internal/gateway/mqtt_client_test.go index a901d838..8ab059af 100644 --- a/go/vehicle-gateway/internal/gateway/mqtt_client_test.go +++ b/go/vehicle-gateway/internal/gateway/mqtt_client_test.go @@ -80,6 +80,9 @@ func TestMQTTClientRecordsMessageMetrics(t *testing.T) { `vehicle_gateway_identity_total{protocol="YUTONG_MQTT",status="resolved"} 1`, `vehicle_gateway_publish_total{kind="raw",protocol="YUTONG_MQTT",status="ok"} 1`, `vehicle_gateway_frame_duration_ms{protocol="YUTONG_MQTT",status="OK"}`, + `vehicle_gateway_frame_duration_ms_histogram_bucket{le="+Inf",protocol="YUTONG_MQTT",status="OK"} 1`, + `vehicle_gateway_frame_duration_ms_histogram_count{protocol="YUTONG_MQTT",status="OK"} 1`, + `vehicle_gateway_frame_duration_ms_histogram_sum{protocol="YUTONG_MQTT",status="OK"}`, } { if !strings.Contains(text, want) { t.Fatalf("metrics missing %s:\n%s", want, text) diff --git a/go/vehicle-gateway/internal/gateway/tcp_server.go b/go/vehicle-gateway/internal/gateway/tcp_server.go index d6817cfc..eadd6d47 100644 --- a/go/vehicle-gateway/internal/gateway/tcp_server.go +++ b/go/vehicle-gateway/internal/gateway/tcp_server.go @@ -63,6 +63,8 @@ type TCPServerConfig struct { const frameOperationTimeout = 30 * time.Second +var gatewayFrameDurationBucketsMS = []float64{1, 5, 10, 25, 50, 100, 250, 500, 1000, 5000} + func NewTCPServer(cfg TCPServerConfig) (*TCPServer, error) { if cfg.Protocol.Protocol == "" { return nil, errors.New("protocol is required") @@ -322,10 +324,16 @@ func (s *TCPServer) recordFrameDuration(status envelope.ParseStatus, elapsed tim if s.metrics == nil { return } + elapsedMS := float64(elapsed.Milliseconds()) + labels := metrics.Labels{ + "protocol": string(s.protocol.Protocol), + "status": string(status), + } s.metrics.SetGauge("vehicle_gateway_frame_duration_ms", metrics.Labels{ "protocol": string(s.protocol.Protocol), "status": string(status), - }, float64(elapsed.Milliseconds())) + }, elapsedMS) + s.metrics.ObserveHistogram("vehicle_gateway_frame_duration_ms_histogram", labels, gatewayFrameDurationBucketsMS, elapsedMS) } func (s *TCPServer) recordPublishMetric(kind string, status string) { diff --git a/go/vehicle-gateway/internal/gateway/tcp_server_test.go b/go/vehicle-gateway/internal/gateway/tcp_server_test.go index dc0fe5da..3220e007 100644 --- a/go/vehicle-gateway/internal/gateway/tcp_server_test.go +++ b/go/vehicle-gateway/internal/gateway/tcp_server_test.go @@ -82,6 +82,9 @@ func TestTCPServerRecordsFrameMetrics(t *testing.T) { `vehicle_gateway_identity_total{protocol="JT808",status="unresolved"} 1`, `vehicle_gateway_publish_total{kind="raw",protocol="JT808",status="ok"} 1`, `vehicle_gateway_frame_duration_ms{protocol="JT808",status="OK"}`, + `vehicle_gateway_frame_duration_ms_histogram_bucket{le="+Inf",protocol="JT808",status="OK"} 1`, + `vehicle_gateway_frame_duration_ms_histogram_count{protocol="JT808",status="OK"} 1`, + `vehicle_gateway_frame_duration_ms_histogram_sum{protocol="JT808",status="OK"}`, } { if !strings.Contains(text, want) { t.Fatalf("metrics missing %s:\n%s", want, text) diff --git a/go/vehicle-gateway/internal/metrics/metrics.go b/go/vehicle-gateway/internal/metrics/metrics.go index f3ab6789..b5507288 100644 --- a/go/vehicle-gateway/internal/metrics/metrics.go +++ b/go/vehicle-gateway/internal/metrics/metrics.go @@ -2,6 +2,7 @@ package metrics import ( "fmt" + "math" "net/http" "sort" "strconv" @@ -12,15 +13,23 @@ import ( type Labels map[string]string type Registry struct { - mu sync.RWMutex - counters map[string]map[string]float64 - gauges map[string]map[string]float64 + mu sync.RWMutex + counters map[string]map[string]float64 + gauges map[string]map[string]float64 + histograms map[string]*histogramFamily +} + +type histogramFamily struct { + buckets map[string]map[float64]float64 + counts map[string]float64 + sums map[string]float64 } func NewRegistry() *Registry { return &Registry{ - counters: map[string]map[string]float64{}, - gauges: map[string]map[string]float64{}, + counters: map[string]map[string]float64{}, + gauges: map[string]map[string]float64{}, + histograms: map[string]*histogramFamily{}, } } @@ -81,6 +90,40 @@ func (r *Registry) SetKafkaLag(name string, topic string, partition int, offset }, float64(lag)) } +func (r *Registry) ObserveHistogram(name string, labels Labels, buckets []float64, value float64) { + name = strings.TrimSpace(name) + if r == nil || name == "" { + return + } + key := labelsKey(labels) + r.mu.Lock() + defer r.mu.Unlock() + family := r.histograms[name] + if family == nil { + family = &histogramFamily{ + buckets: map[string]map[float64]float64{}, + counts: map[string]float64{}, + sums: map[string]float64{}, + } + r.histograms[name] = family + } + if family.buckets[key] == nil { + family.buckets[key] = map[float64]float64{} + } + normalizedBuckets := append([]float64(nil), buckets...) + sort.Float64s(normalizedBuckets) + for _, bucket := range normalizedBuckets { + if value <= bucket { + family.buckets[key][bucket]++ + } else if _, ok := family.buckets[key][bucket]; !ok { + family.buckets[key][bucket] = 0 + } + } + family.buckets[key][math.Inf(1)]++ + family.counts[key]++ + family.sums[key] += value +} + func (r *Registry) Render() string { if r == nil { return "" @@ -107,6 +150,15 @@ func (r *Registry) Render() string { for _, name := range names { renderMetricFamily(&b, name, "gauge", r.gauges[name]) } + + names = names[:0] + for name := range r.histograms { + names = append(names, name) + } + sort.Strings(names) + for _, name := range names { + renderHistogramFamily(&b, name, r.histograms[name]) + } return b.String() } @@ -134,6 +186,72 @@ func renderMetricFamily(b *strings.Builder, name string, metricType string, valu } } +func renderHistogramFamily(b *strings.Builder, name string, family *histogramFamily) { + if family == nil { + return + } + b.WriteString("# TYPE ") + b.WriteString(name) + b.WriteString(" histogram\n") + var keys []string + for key := range family.counts { + keys = append(keys, key) + } + sort.Strings(keys) + for _, key := range keys { + var buckets []float64 + for bucket := range family.buckets[key] { + buckets = append(buckets, bucket) + } + sort.Float64s(buckets) + for _, bucket := range buckets { + renderHistogramMetric(b, name+"_bucket", key, "le", bucketLabel(bucket), family.buckets[key][bucket]) + } + renderMetricSample(b, name+"_count", key, family.counts[key]) + renderMetricSample(b, name+"_sum", key, family.sums[key]) + } +} + +func renderHistogramMetric(b *strings.Builder, name string, key string, extraLabel string, extraValue string, value float64) { + b.WriteString(name) + labels := mergeLabelKey(key, extraLabel, extraValue) + if labels != "" { + b.WriteString("{") + b.WriteString(labels) + b.WriteString("}") + } + b.WriteString(" ") + b.WriteString(formatNumber(value)) + b.WriteString("\n") +} + +func renderMetricSample(b *strings.Builder, name string, key string, value float64) { + b.WriteString(name) + if key != "" { + b.WriteString("{") + b.WriteString(key) + b.WriteString("}") + } + b.WriteString(" ") + b.WriteString(formatNumber(value)) + b.WriteString("\n") +} + +func mergeLabelKey(key string, extraLabel string, extraValue string) string { + extra := fmt.Sprintf(`%s="%s"`, sanitizeLabelName(extraLabel), escapeLabelValue(extraValue)) + if key == "" { + return extra + } + return extra + "," + key +} + +func bucketLabel(value float64) string { + if math.IsInf(value, 1) { + return "+Inf" + } + return formatNumber(value) +} + func NewHandler(registry *Registry) http.Handler { if registry == nil { registry = NewRegistry() diff --git a/go/vehicle-gateway/internal/metrics/metrics_test.go b/go/vehicle-gateway/internal/metrics/metrics_test.go index 88209a1b..f9b92ad7 100644 --- a/go/vehicle-gateway/internal/metrics/metrics_test.go +++ b/go/vehicle-gateway/internal/metrics/metrics_test.go @@ -46,6 +46,27 @@ func TestRegistryRendersGaugesWithLabels(t *testing.T) { } } +func TestRegistryObservesHistogramBuckets(t *testing.T) { + registry := NewRegistry() + + registry.ObserveHistogram("vehicle_gateway_frame_duration_ms", Labels{"protocol": "JT808"}, []float64{1, 5, 10}, 7) + + text := registry.Render() + for _, want := range []string{ + `# TYPE vehicle_gateway_frame_duration_ms histogram`, + `vehicle_gateway_frame_duration_ms_bucket{le="1",protocol="JT808"} 0`, + `vehicle_gateway_frame_duration_ms_bucket{le="5",protocol="JT808"} 0`, + `vehicle_gateway_frame_duration_ms_bucket{le="10",protocol="JT808"} 1`, + `vehicle_gateway_frame_duration_ms_bucket{le="+Inf",protocol="JT808"} 1`, + `vehicle_gateway_frame_duration_ms_count{protocol="JT808"} 1`, + `vehicle_gateway_frame_duration_ms_sum{protocol="JT808"} 7`, + } { + if !strings.Contains(text, want) { + t.Fatalf("histogram missing %s:\n%s", want, text) + } + } +} + func TestRegistryRecordsKafkaLagGauge(t *testing.T) { registry := NewRegistry()