feat(go): track gateway active connections

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)
 }

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