feat(go): expose kafka consumer lag metrics

docs/ops/go-service-observability.md

@@ -40,6 +40,9 @@ curl -fsS http://127.0.0.1:20200/readyz
 | `vehicle_history_writes_total` | TDengine history writes. Labels: `topic`, `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. |
+| `vehicle_stat_kafka_lag` | Estimated Kafka lag for stat writer by topic and partition. |
+| `vehicle_realtime_kafka_lag` | Estimated Kafka lag for realtime projector by topic and partition. |
 
 ## First Principle
 

go/vehicle-gateway/cmd/history-writer/main.go

@@ -94,6 +94,7 @@ func processHistoryMessage(ctx context.Context, logger interface {
 	defer cancel()
 
 	addWriterMetric(registry, "vehicle_history_kafka_messages_total", message, "received")
+	addWriterLagMetric(registry, message)
 	var env envelope.FrameEnvelope
 	if err := json.Unmarshal(message.Value, &env); err != nil {
 		addWriterMetric(registry, "vehicle_history_kafka_messages_total", message, "invalid_json")
@@ -122,6 +123,13 @@ func addWriterMetric(registry *metrics.Registry, name string, message kafka.Mess
 	registry.IncCounter(name, metrics.Labels{"topic": message.Topic, "status": status})
 }
 
+func addWriterLagMetric(registry *metrics.Registry, message kafka.Message) {
+	if registry == nil {
+		return
+	}
+	registry.SetKafkaLag("vehicle_history_kafka_lag", message.Topic, message.Partition, message.Offset, message.HighWaterMark)
+}
+
 type config struct {
 	KafkaBrokers     []string
 	KafkaTopics      []string

go/vehicle-gateway/cmd/history-writer/main_test.go

@@ -58,7 +58,7 @@ func TestProcessHistoryMessageRecordsMetrics(t *testing.T) {
 		registry,
 		&contextCheckingHistoryAppender{},
 		&contextCheckingHistoryCommitter{},
-		kafka.Message{Topic: "vehicle.raw.gb32960.v1", Value: payload},
+		kafka.Message{Topic: "vehicle.raw.gb32960.v1", Partition: 1, Offset: 20, HighWaterMark: 23, Value: payload},
 	)
 
 	text := registry.Render()
@@ -66,6 +66,7 @@ func TestProcessHistoryMessageRecordsMetrics(t *testing.T) {
 		`vehicle_history_kafka_messages_total{status="received",topic="vehicle.raw.gb32960.v1"} 1`,
 		`vehicle_history_writes_total{status="ok",topic="vehicle.raw.gb32960.v1"} 1`,
 		`vehicle_history_kafka_commits_total{status="ok",topic="vehicle.raw.gb32960.v1"} 1`,
+		`vehicle_history_kafka_lag{partition="1",topic="vehicle.raw.gb32960.v1"} 2`,
 	} {
 		if !strings.Contains(text, want) {
 			t.Fatalf("metrics missing %s:\n%s", want, text)

go/vehicle-gateway/cmd/realtime-api/main.go

@@ -218,6 +218,7 @@ func processRealtimeMessage(ctx context.Context, logger interface {
 	defer cancel()
 
 	addRealtimeMetric(registry, "vehicle_realtime_kafka_messages_total", message, "received")
+	addRealtimeLagMetric(registry, message)
 	var env envelope.FrameEnvelope
 	if err := json.Unmarshal(message.Value, &env); err != nil {
 		addRealtimeMetric(registry, "vehicle_realtime_kafka_messages_total", message, "invalid_json")
@@ -249,6 +250,13 @@ func addRealtimeMetric(registry *metrics.Registry, name string, message kafka.Me
 	})
 }
 
+func addRealtimeLagMetric(registry *metrics.Registry, message kafka.Message) {
+	if registry == nil {
+		return
+	}
+	registry.SetKafkaLag("vehicle_realtime_kafka_lag", message.Topic, message.Partition, message.Offset, message.HighWaterMark)
+}
+
 func env(key string, fallback string) string {
 	value := strings.TrimSpace(os.Getenv(key))
 	if value == "" {

go/vehicle-gateway/cmd/realtime-api/main_test.go

@@ -65,7 +65,7 @@ func TestProcessRealtimeMessageRecordsMetrics(t *testing.T) {
 		registry,
 		&contextCheckingRealtimeUpdater{},
 		&contextCheckingMessageCommitter{},
-		kafka.Message{Topic: "vehicle.event.go.unified.v1", Value: payload},
+		kafka.Message{Topic: "vehicle.event.go.unified.v1", Partition: 2, Offset: 10, HighWaterMark: 16, Value: payload},
 	)
 
 	text := registry.Render()
@@ -73,6 +73,7 @@ func TestProcessRealtimeMessageRecordsMetrics(t *testing.T) {
 		`vehicle_realtime_kafka_messages_total{status="received",topic="vehicle.event.go.unified.v1"} 1`,
 		`vehicle_realtime_updates_total{status="ok",topic="vehicle.event.go.unified.v1"} 1`,
 		`vehicle_realtime_kafka_commits_total{status="ok",topic="vehicle.event.go.unified.v1"} 1`,
+		`vehicle_realtime_kafka_lag{partition="2",topic="vehicle.event.go.unified.v1"} 5`,
 	} {
 		if !strings.Contains(text, want) {
 			t.Fatalf("metrics missing %s:\n%s", want, text)

go/vehicle-gateway/cmd/stat-writer/main.go

@@ -90,6 +90,7 @@ func processStatMessage(ctx context.Context, logger interface {
 	defer cancel()
 
 	addStatMetric(registry, "vehicle_stat_kafka_messages_total", message, "received")
+	addStatLagMetric(registry, message)
 	var env envelope.FrameEnvelope
 	if err := json.Unmarshal(message.Value, &env); err != nil {
 		addStatMetric(registry, "vehicle_stat_kafka_messages_total", message, "invalid_json")
@@ -118,6 +119,13 @@ func addStatMetric(registry *metrics.Registry, name string, message kafka.Messag
 	registry.IncCounter(name, metrics.Labels{"topic": message.Topic, "status": status})
 }
 
+func addStatLagMetric(registry *metrics.Registry, message kafka.Message) {
+	if registry == nil {
+		return
+	}
+	registry.SetKafkaLag("vehicle_stat_kafka_lag", message.Topic, message.Partition, message.Offset, message.HighWaterMark)
+}
+
 type config struct {
 	KafkaBrokers []string
 	KafkaTopics  []string

go/vehicle-gateway/cmd/stat-writer/main_test.go

@@ -58,7 +58,7 @@ func TestProcessStatMessageRecordsMetrics(t *testing.T) {
 		registry,
 		&contextCheckingStatAppender{},
 		&contextCheckingStatCommitter{},
-		kafka.Message{Topic: "vehicle.raw.jt808.v1", Value: payload},
+		kafka.Message{Topic: "vehicle.raw.jt808.v1", Partition: 4, Offset: 30, HighWaterMark: 41, Value: payload},
 	)
 
 	text := registry.Render()
@@ -66,6 +66,7 @@ func TestProcessStatMessageRecordsMetrics(t *testing.T) {
 		`vehicle_stat_kafka_messages_total{status="received",topic="vehicle.raw.jt808.v1"} 1`,
 		`vehicle_stat_writes_total{status="ok",topic="vehicle.raw.jt808.v1"} 1`,
 		`vehicle_stat_kafka_commits_total{status="ok",topic="vehicle.raw.jt808.v1"} 1`,
+		`vehicle_stat_kafka_lag{partition="4",topic="vehicle.raw.jt808.v1"} 10`,
 	} {
 		if !strings.Contains(text, want) {
 			t.Fatalf("metrics missing %s:\n%s", want, text)

go/vehicle-gateway/internal/metrics/metrics.go

@@ -4,6 +4,7 @@ import (
 	"fmt"
 	"net/http"
 	"sort"
+	"strconv"
 	"strings"
 	"sync"
 )
@@ -69,6 +70,17 @@ func (r *Registry) AddGauge(name string, labels Labels, value float64) {
 	r.gauges[name][key] += value
 }
 
+func (r *Registry) SetKafkaLag(name string, topic string, partition int, offset int64, highWaterMark int64) {
+	lag := highWaterMark - offset - 1
+	if lag < 0 {
+		lag = 0
+	}
+	r.SetGauge(name, Labels{
+		"topic":     topic,
+		"partition": strconv.Itoa(partition),
+	}, float64(lag))
+}
+
 func (r *Registry) Render() string {
 	if r == nil {
 		return ""

go/vehicle-gateway/internal/metrics/metrics_test.go

@@ -46,6 +46,24 @@ func TestRegistryRendersGaugesWithLabels(t *testing.T) {
 	}
 }
 
+func TestRegistryRecordsKafkaLagGauge(t *testing.T) {
+	registry := NewRegistry()
+
+	registry.SetKafkaLag("vehicle_history_kafka_lag", "vehicle.raw.go.jt808.v1", 3, 101, 130)
+	registry.SetKafkaLag("vehicle_history_kafka_lag", "vehicle.raw.go.gb32960.v1", 0, 9, 9)
+
+	text := registry.Render()
+	for _, want := range []string{
+		`# TYPE vehicle_history_kafka_lag gauge`,
+		`vehicle_history_kafka_lag{partition="0",topic="vehicle.raw.go.gb32960.v1"} 0`,
+		`vehicle_history_kafka_lag{partition="3",topic="vehicle.raw.go.jt808.v1"} 28`,
+	} {
+		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"})