From 41ff0bcef494024501708c425b3de0f4bc864e05 Mon Sep 17 00:00:00 2001 From: lingniu Date: Thu, 2 Jul 2026 19:16:25 +0800 Subject: [PATCH] feat(go): track gateway active connections --- docs/ops/go-service-observability.md | 1 + .../internal/gateway/tcp_server.go | 11 +++ .../internal/gateway/tcp_server_test.go | 45 ++++++++++ .../internal/metrics/metrics.go | 87 ++++++++++++++----- .../internal/metrics/metrics_test.go | 19 ++++ 5 files changed, 143 insertions(+), 20 deletions(-) diff --git a/docs/ops/go-service-observability.md b/docs/ops/go-service-observability.md index c48f2130..9315d930 100644 --- a/docs/ops/go-service-observability.md +++ b/docs/ops/go-service-observability.md @@ -31,6 +31,7 @@ curl -fsS http://127.0.0.1:20200/readyz | Metric | Meaning | | --- | --- | +| `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_publish_total` | Gateway publish result for raw and unified events. Labels: `protocol`, `kind`, `status`. | | `vehicle_bridge_messages_total` | NATS messages fetched by the bridge. Labels: `subject`, `status`. | diff --git a/go/vehicle-gateway/internal/gateway/tcp_server.go b/go/vehicle-gateway/internal/gateway/tcp_server.go index 4861cd77..e7587674 100644 --- a/go/vehicle-gateway/internal/gateway/tcp_server.go +++ b/go/vehicle-gateway/internal/gateway/tcp_server.go @@ -145,6 +145,8 @@ func (s *TCPServer) handleConnection(ctx context.Context, conn net.Conn) { defer conn.Close() source := conn.RemoteAddr().String() log := s.logger.With("protocol", s.protocol.Protocol, "remote", source) + s.recordConnectionMetric(1) + defer s.recordConnectionMetric(-1) log.Info("tcp connection opened") defer log.Info("tcp connection closed") @@ -268,6 +270,15 @@ func (s *TCPServer) recordPublishMetric(kind string, status string) { }) } +func (s *TCPServer) recordConnectionMetric(delta float64) { + if s.metrics == nil { + return + } + s.metrics.AddGauge("vehicle_gateway_active_connections", metrics.Labels{ + "protocol": string(s.protocol.Protocol), + }, delta) +} + func (p TCPProtocol) String() string { return fmt.Sprintf("%s@%s", p.Protocol, p.Addr) } diff --git a/go/vehicle-gateway/internal/gateway/tcp_server_test.go b/go/vehicle-gateway/internal/gateway/tcp_server_test.go index d6e308ff..8a3743b0 100644 --- a/go/vehicle-gateway/internal/gateway/tcp_server_test.go +++ b/go/vehicle-gateway/internal/gateway/tcp_server_test.go @@ -88,6 +88,51 @@ func TestTCPServerRecordsFrameMetrics(t *testing.T) { } } +func TestTCPServerRecordsActiveConnectionGauge(t *testing.T) { + registry := metrics.NewRegistry() + server, err := NewTCPServer(TCPServerConfig{ + Protocol: TCPProtocol{ + Protocol: envelope.ProtocolJT808, + Addr: ":0", + Extract: jt808.ExtractFrames, + Parse: jt808.ParseFrame, + }, + Sink: &recordingSink{}, + Logger: slog.New(slog.NewTextHandler(testWriter{t: t}, nil)), + Metrics: registry, + IdleTimeout: time.Minute, + }) + if err != nil { + t.Fatalf("NewTCPServer() error = %v", err) + } + client, srv := net.Pipe() + opened := make(chan struct{}) + done := make(chan struct{}) + go func() { + close(opened) + server.handleConnection(context.Background(), srv) + close(done) + }() + <-opened + deadline := time.Now().Add(2 * time.Second) + for time.Now().Before(deadline) { + if strings.Contains(registry.Render(), `vehicle_gateway_active_connections{protocol="JT808"} 1`) { + break + } + time.Sleep(10 * time.Millisecond) + } + if text := registry.Render(); !strings.Contains(text, `vehicle_gateway_active_connections{protocol="JT808"} 1`) { + t.Fatalf("active connection gauge missing while open:\n%s", text) + } + + _ = client.Close() + <-done + + if text := registry.Render(); !strings.Contains(text, `vehicle_gateway_active_connections{protocol="JT808"} 0`) { + t.Fatalf("active connection gauge should return to zero:\n%s", text) + } +} + func TestTCPServerPublishesBadFrameOnlyToRaw(t *testing.T) { good := buildGBFrame(0x02, 0xfe, "LNBSCB3D4R1234567", nil) good[len(good)-1] ^= 0xff diff --git a/go/vehicle-gateway/internal/metrics/metrics.go b/go/vehicle-gateway/internal/metrics/metrics.go index 2793756a..dfd45d10 100644 --- a/go/vehicle-gateway/internal/metrics/metrics.go +++ b/go/vehicle-gateway/internal/metrics/metrics.go @@ -13,10 +13,14 @@ type Labels map[string]string type Registry struct { mu sync.RWMutex counters map[string]map[string]float64 + gauges map[string]map[string]float64 } func NewRegistry() *Registry { - return &Registry{counters: map[string]map[string]float64{}} + return &Registry{ + counters: map[string]map[string]float64{}, + gauges: map[string]map[string]float64{}, + } } func (r *Registry) IncCounter(name string, labels Labels) { @@ -37,6 +41,34 @@ func (r *Registry) AddCounter(name string, labels Labels, value float64) { r.counters[name][key] += value } +func (r *Registry) SetGauge(name string, labels Labels, value float64) { + name = strings.TrimSpace(name) + if r == nil || name == "" { + return + } + key := labelsKey(labels) + r.mu.Lock() + defer r.mu.Unlock() + if r.gauges[name] == nil { + r.gauges[name] = map[string]float64{} + } + r.gauges[name][key] = value +} + +func (r *Registry) AddGauge(name string, labels Labels, value float64) { + name = strings.TrimSpace(name) + if r == nil || name == "" || value == 0 { + return + } + key := labelsKey(labels) + r.mu.Lock() + defer r.mu.Unlock() + if r.gauges[name] == nil { + r.gauges[name] = map[string]float64{} + } + r.gauges[name][key] += value +} + func (r *Registry) Render() string { if r == nil { return "" @@ -52,29 +84,44 @@ func (r *Registry) Render() string { var b strings.Builder for _, name := range names { - b.WriteString("# TYPE ") - b.WriteString(name) - b.WriteString(" counter\n") - var keys []string - for key := range r.counters[name] { - keys = append(keys, key) - } - sort.Strings(keys) - for _, key := range keys { - b.WriteString(name) - if key != "" { - b.WriteString("{") - b.WriteString(key) - b.WriteString("}") - } - b.WriteString(" ") - b.WriteString(formatNumber(r.counters[name][key])) - b.WriteString("\n") - } + renderMetricFamily(&b, name, "counter", r.counters[name]) + } + + names = names[:0] + for name := range r.gauges { + names = append(names, name) + } + sort.Strings(names) + for _, name := range names { + renderMetricFamily(&b, name, "gauge", r.gauges[name]) } return b.String() } +func renderMetricFamily(b *strings.Builder, name string, metricType string, values map[string]float64) { + b.WriteString("# TYPE ") + b.WriteString(name) + b.WriteString(" ") + b.WriteString(metricType) + b.WriteString("\n") + var keys []string + for key := range values { + keys = append(keys, key) + } + sort.Strings(keys) + for _, key := range keys { + b.WriteString(name) + if key != "" { + b.WriteString("{") + b.WriteString(key) + b.WriteString("}") + } + b.WriteString(" ") + b.WriteString(formatNumber(values[key])) + b.WriteString("\n") + } +} + 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 a8cccaa5..2ab72e74 100644 --- a/go/vehicle-gateway/internal/metrics/metrics_test.go +++ b/go/vehicle-gateway/internal/metrics/metrics_test.go @@ -27,6 +27,25 @@ func TestRegistryRendersCountersWithLabels(t *testing.T) { } } +func TestRegistryRendersGaugesWithLabels(t *testing.T) { + registry := NewRegistry() + registry.SetGauge("vehicle_gateway_connections", Labels{"protocol": "JT808"}, 2) + registry.AddGauge("vehicle_gateway_connections", Labels{"protocol": "JT808"}, -1) + registry.SetGauge("vehicle_gateway_connections", Labels{"protocol": "GB32960"}, 3) + + text := registry.Render() + + for _, want := range []string{ + `# TYPE vehicle_gateway_connections gauge`, + `vehicle_gateway_connections{protocol="GB32960"} 3`, + `vehicle_gateway_connections{protocol="JT808"} 1`, + } { + if !strings.Contains(text, want) { + t.Fatalf("metrics missing %s:\n%s", want, text) + } + } +} + func TestHandlerServesPrometheusText(t *testing.T) { registry := NewRegistry() registry.IncCounter("vehicle_kafka_commits_total", Labels{"service": "history"})