From b8299faefa369faebb8f42ed7c9b52d2d5ef71a2 Mon Sep 17 00:00:00 2001 From: lingniu Date: Fri, 3 Jul 2026 15:06:27 +0800 Subject: [PATCH] feat(go): add realtime pipeline observability --- go/vehicle-gateway/cmd/realtime-api/main.go | 60 +++++- .../cmd/realtime-api/main_test.go | 35 +++- .../internal/gateway/mqtt_client.go | 24 +++ .../internal/gateway/mqtt_client_test.go | 6 + .../internal/gateway/tcp_server.go | 68 +++++++ .../internal/gateway/tcp_server_test.go | 38 ++++ go/vehicle-gateway/internal/realtime/http.go | 94 +++++++++ go/vehicle-gateway/internal/realtime/model.go | 18 ++ .../internal/realtime/repository.go | 180 ++++++++++++++++++ .../internal/realtime/repository_test.go | 119 ++++++++++++ 10 files changed, 636 insertions(+), 6 deletions(-) diff --git a/go/vehicle-gateway/cmd/realtime-api/main.go b/go/vehicle-gateway/cmd/realtime-api/main.go index b43efb04..bab0911b 100644 --- a/go/vehicle-gateway/cmd/realtime-api/main.go +++ b/go/vehicle-gateway/cmd/realtime-api/main.go @@ -57,8 +57,14 @@ func main() { mux.Handle("/openapi.json", realtime.NewOpenAPIHandler()) mux.Handle("/swagger-ui/", realtime.NewSwaggerUIHandler()) mux.Handle("/api/realtime/vehicles/", realtime.NewHandler(repository)) + mux.Handle("/api/realtime/online", realtime.NewOnlineIndexHandler(repository)) + mux.Handle("/api/debug/pipeline", realtime.NewPipelineDebugHandler(repository)) closeStats := func() {} - var updater realtimeUpdater = fastRealtimeUpdater{repository: repository} + var updater realtimeUpdater = storeMetricUpdater{ + store: "redis", + delegate: fastRealtimeUpdater{repository: repository}, + registry: registry, + } if dsn := strings.TrimSpace(os.Getenv("MYSQL_DSN")); dsn != "" { db, err := sql.Open("mysql", dsn) if err != nil { @@ -84,16 +90,20 @@ func main() { logger.Error("realtime snapshot mysql schema bootstrap failed", "error", err) os.Exit(1) } - mysqlUpdater := realtimeUpdater(snapshotWriter) + mysqlUpdater := realtimeUpdater(storeMetricUpdater{store: "mysql", delegate: snapshotWriter, registry: registry}) if env("MYSQL_REALTIME_ASYNC_ENABLED", "true") != "false" { mysqlUpdater = newAsyncSecondaryRealtimeUpdater( nil, - snapshotWriter, + storeMetricUpdater{store: "mysql", delegate: snapshotWriter, registry: registry}, envInt("MYSQL_REALTIME_ASYNC_QUEUE_SIZE", 20000), envInt("MYSQL_REALTIME_ASYNC_WORKERS", 4), + registry, ) } - updater = compositeRealtimeUpdater{primary: fastRealtimeUpdater{repository: repository}, secondary: mysqlUpdater} + updater = compositeRealtimeUpdater{ + primary: storeMetricUpdater{store: "redis", delegate: fastRealtimeUpdater{repository: repository}, registry: registry}, + secondary: mysqlUpdater, + } logger.Info("realtime mysql snapshot enabled", "table", "vehicle_realtime_snapshot", "plate_binding_table", bindingTable, "plate_cache_ttl_seconds", envInt("PLATE_CACHE_TTL_SECONDS", 600)) } mux.Handle("/api/stats/daily-metrics", stats.NewMetricHandler(stats.NewMetricRepository(db))) @@ -227,6 +237,31 @@ func (u fastRealtimeUpdater) Update(ctx context.Context, env envelope.FrameEnvel return u.repository.FastUpdate(ctx, env) } +type storeMetricUpdater struct { + store string + delegate realtimeUpdater + registry *metrics.Registry +} + +func (u storeMetricUpdater) Update(ctx context.Context, env envelope.FrameEnvelope) error { + if u.delegate == nil { + return nil + } + err := u.delegate.Update(ctx, env) + status := "ok" + if err != nil { + status = "error" + } + if u.registry != nil { + u.registry.IncCounter("vehicle_realtime_store_updates_total", metrics.Labels{ + "store": u.store, + "protocol": string(env.Protocol), + "status": status, + }) + } + return err +} + type compositeRealtimeUpdater struct { primary realtimeUpdater secondary realtimeUpdater @@ -247,9 +282,10 @@ type asyncSecondaryRealtimeUpdater struct { secondary realtimeUpdater queue chan envelope.FrameEnvelope cancel context.CancelFunc + registry *metrics.Registry } -func newAsyncSecondaryRealtimeUpdater(primary realtimeUpdater, secondary realtimeUpdater, queueSize int, workers int) *asyncSecondaryRealtimeUpdater { +func newAsyncSecondaryRealtimeUpdater(primary realtimeUpdater, secondary realtimeUpdater, queueSize int, workers int, registry *metrics.Registry) *asyncSecondaryRealtimeUpdater { if queueSize <= 0 { queueSize = 20000 } @@ -262,6 +298,7 @@ func newAsyncSecondaryRealtimeUpdater(primary realtimeUpdater, secondary realtim secondary: secondary, queue: make(chan envelope.FrameEnvelope, queueSize), cancel: cancel, + registry: registry, } for i := 0; i < workers; i++ { go updater.run(ctx) @@ -280,8 +317,10 @@ func (u *asyncSecondaryRealtimeUpdater) Update(ctx context.Context, env envelope } select { case u.queue <- env: + u.recordQueueMetric(env, "queued") return nil default: + u.recordQueueMetric(env, "dropped") return nil } } @@ -299,6 +338,17 @@ func (u *asyncSecondaryRealtimeUpdater) run(ctx context.Context) { } } +func (u *asyncSecondaryRealtimeUpdater) recordQueueMetric(env envelope.FrameEnvelope, status string) { + if u.registry == nil { + return + } + u.registry.IncCounter("vehicle_realtime_async_queue_total", metrics.Labels{ + "store": "mysql", + "protocol": string(env.Protocol), + "status": status, + }) +} + func (u *asyncSecondaryRealtimeUpdater) Close() { if u.cancel != nil { u.cancel() diff --git a/go/vehicle-gateway/cmd/realtime-api/main_test.go b/go/vehicle-gateway/cmd/realtime-api/main_test.go index 085d5c31..20f5c065 100644 --- a/go/vehicle-gateway/cmd/realtime-api/main_test.go +++ b/go/vehicle-gateway/cmd/realtime-api/main_test.go @@ -102,7 +102,8 @@ func TestAsyncSecondaryRealtimeUpdaterDoesNotBlockPrimaryPath(t *testing.T) { env := envelope.FrameEnvelope{Protocol: envelope.ProtocolGB32960, VIN: "VIN001"} primary := &contextCheckingRealtimeUpdater{} secondary := &blockingRealtimeUpdater{started: make(chan struct{}), release: make(chan struct{})} - updater := newAsyncSecondaryRealtimeUpdater(primary, secondary, 8, 1) + registry := metrics.NewRegistry() + updater := newAsyncSecondaryRealtimeUpdater(primary, secondary, 8, 1, registry) defer updater.Close() start := time.Now() @@ -121,6 +122,26 @@ func TestAsyncSecondaryRealtimeUpdaterDoesNotBlockPrimaryPath(t *testing.T) { t.Fatal("secondary update was not started asynchronously") } close(secondary.release) + if text := registry.Render(); !strings.Contains(text, `vehicle_realtime_async_queue_total{protocol="GB32960",status="queued",store="mysql"} 1`) { + t.Fatalf("async queue metric missing:\n%s", text) + } +} + +func TestStoreMetricUpdaterRecordsStoreUpdateResults(t *testing.T) { + registry := metrics.NewRegistry() + env := envelope.FrameEnvelope{Protocol: envelope.ProtocolJT808, VIN: "VIN001"} + updater := storeMetricUpdater{ + store: "redis", + delegate: &contextCheckingRealtimeUpdater{}, + registry: registry, + } + + if err := updater.Update(context.Background(), env); err != nil { + t.Fatalf("Update() error = %v", err) + } + if text := registry.Render(); !strings.Contains(text, `vehicle_realtime_store_updates_total{protocol="JT808",status="ok",store="redis"} 1`) { + t.Fatalf("store update metric missing:\n%s", text) + } } func TestKafkaTopicsFromEnvDefaultsToGoRawTopics(t *testing.T) { @@ -174,6 +195,18 @@ func TestRealtimeAPIExposesPostRawFrameQueryRoute(t *testing.T) { } } +func TestRealtimeAPIExposesOperationalRealtimeRoutes(t *testing.T) { + source, err := os.ReadFile("main.go") + if err != nil { + t.Fatalf("read main.go: %v", err) + } + for _, want := range []string{`"/api/realtime/online"`, `"/api/debug/pipeline"`} { + if !strings.Contains(string(source), want) { + t.Fatalf("realtime api should expose operational route %s", want) + } + } +} + type contextCheckingRealtimeUpdater struct { ctxErr error count int diff --git a/go/vehicle-gateway/internal/gateway/mqtt_client.go b/go/vehicle-gateway/internal/gateway/mqtt_client.go index 355dea7c..e67acd29 100644 --- a/go/vehicle-gateway/internal/gateway/mqtt_client.go +++ b/go/vehicle-gateway/internal/gateway/mqtt_client.go @@ -197,6 +197,7 @@ func (c *MQTTClient) handleMessage(ctx context.Context, topic string, payload [] ParseError: err.Error(), } env.EventID = env.StableEventID() + c.recordParseErrorMetric(err) } else { resolved, resolveErr := c.cfg.Resolver.Resolve(messageCtx, env) if resolveErr != nil { @@ -206,8 +207,11 @@ func (c *MQTTClient) handleMessage(ctx context.Context, topic string, payload [] } env.Parsed["identity"] = map[string]any{"resolved": false, "error": resolveErr.Error()} env.ParseStatus = envelope.ParsePartial + c.recordIdentityMetric("error") } else { env = resolved + annotateIdentityUnresolved(&env) + c.recordIdentityMetric(identityStatus(env)) } } c.recordFrameMetric(env.ParseStatus) @@ -258,3 +262,23 @@ func (c *MQTTClient) recordPublishMetric(kind string, status string) { "status": status, }) } + +func (c *MQTTClient) recordParseErrorMetric(err error) { + if c.cfg.Metrics == nil { + return + } + c.cfg.Metrics.IncCounter("vehicle_gateway_parse_errors_total", metrics.Labels{ + "protocol": string(envelope.ProtocolYutongMQTT), + "reason": classifyError(err), + }) +} + +func (c *MQTTClient) recordIdentityMetric(status string) { + if c.cfg.Metrics == nil { + return + } + c.cfg.Metrics.IncCounter("vehicle_gateway_identity_total", metrics.Labels{ + "protocol": string(envelope.ProtocolYutongMQTT), + "status": status, + }) +} diff --git a/go/vehicle-gateway/internal/gateway/mqtt_client_test.go b/go/vehicle-gateway/internal/gateway/mqtt_client_test.go index 072b884c..7b18bc8a 100644 --- a/go/vehicle-gateway/internal/gateway/mqtt_client_test.go +++ b/go/vehicle-gateway/internal/gateway/mqtt_client_test.go @@ -77,6 +77,7 @@ func TestMQTTClientRecordsMessageMetrics(t *testing.T) { text := registry.Render() for _, want := range []string{ `vehicle_gateway_frames_total{protocol="YUTONG_MQTT",status="OK"} 1`, + `vehicle_gateway_identity_total{protocol="YUTONG_MQTT",status="resolved"} 1`, `vehicle_gateway_publish_total{kind="raw",protocol="YUTONG_MQTT",status="ok"} 1`, } { if !strings.Contains(text, want) { @@ -90,6 +91,7 @@ func TestMQTTClientRecordsMessageMetrics(t *testing.T) { func TestMQTTClientHandleBadPayloadPublishesOnlyRaw(t *testing.T) { sink := &recordingSink{} + registry := metrics.NewRegistry() client, err := NewMQTTClient(MQTTClientConfig{ EndpointName: "endpoint-a", Broker: "tcp://127.0.0.1:1883", @@ -97,6 +99,7 @@ func TestMQTTClientHandleBadPayloadPublishesOnlyRaw(t *testing.T) { Topics: []string{"/ytforward/shln/+"}, Sink: sink, Logger: slog.New(slog.NewTextHandler(testWriter{t: t}, nil)), + Metrics: registry, }) if err != nil { t.Fatalf("NewMQTTClient() error = %v", err) @@ -116,6 +119,9 @@ func TestMQTTClientHandleBadPayloadPublishesOnlyRaw(t *testing.T) { if sink.raw[0].RawHex != "" { t.Fatalf("bad mqtt raw envelope should not duplicate text payload as hex: %q", sink.raw[0].RawHex) } + if text := registry.Render(); !strings.Contains(text, `vehicle_gateway_parse_errors_total{protocol="YUTONG_MQTT",reason="json"} 1`) { + t.Fatalf("parse error metric missing:\n%s", text) + } } func TestMQTTClientUsesUncancelledMessageContextForReceivedMessage(t *testing.T) { diff --git a/go/vehicle-gateway/internal/gateway/tcp_server.go b/go/vehicle-gateway/internal/gateway/tcp_server.go index c3f6cd7a..f74fa63e 100644 --- a/go/vehicle-gateway/internal/gateway/tcp_server.go +++ b/go/vehicle-gateway/internal/gateway/tcp_server.go @@ -211,6 +211,7 @@ func (s *TCPServer) handleFrame(ctx context.Context, conn net.Conn, raw []byte, ParseError: err.Error(), } env.EventID = env.StableEventID() + s.recordParseErrorMetric(err) } else { resolved, resolveErr := s.resolver.Resolve(frameCtx, env) if resolveErr != nil { @@ -220,8 +221,11 @@ func (s *TCPServer) handleFrame(ctx context.Context, conn net.Conn, raw []byte, } env.Parsed["identity"] = map[string]any{"resolved": false, "error": resolveErr.Error()} env.ParseStatus = envelope.ParsePartial + s.recordIdentityMetric("error") } else { env = resolved + annotateIdentityUnresolved(&env) + s.recordIdentityMetric(identityStatus(env)) } enrichConnectionPlatform(&env, state) } @@ -314,6 +318,26 @@ func (s *TCPServer) recordPublishMetric(kind string, status string) { }) } +func (s *TCPServer) recordParseErrorMetric(err error) { + if s.metrics == nil { + return + } + s.metrics.IncCounter("vehicle_gateway_parse_errors_total", metrics.Labels{ + "protocol": string(s.protocol.Protocol), + "reason": classifyError(err), + }) +} + +func (s *TCPServer) recordIdentityMetric(status string) { + if s.metrics == nil { + return + } + s.metrics.IncCounter("vehicle_gateway_identity_total", metrics.Labels{ + "protocol": string(s.protocol.Protocol), + "status": status, + }) +} + func (s *TCPServer) recordConnectionMetric(delta float64) { if s.metrics == nil { return @@ -323,6 +347,50 @@ func (s *TCPServer) recordConnectionMetric(delta float64) { }, delta) } +func identityStatus(env envelope.FrameEnvelope) string { + if strings.TrimSpace(env.VIN) != "" { + return "resolved" + } + if env.ParseStatus == envelope.ParsePartial { + return "error" + } + return "unresolved" +} + +func annotateIdentityUnresolved(env *envelope.FrameEnvelope) { + if env == nil || env.ParseStatus == envelope.ParseBadFrame || strings.TrimSpace(env.VIN) != "" { + return + } + if env.Parsed == nil { + env.Parsed = map[string]any{} + } + if _, exists := env.Parsed["identity"]; exists { + return + } + env.Parsed["identity"] = map[string]any{ + "resolved": false, + "reason": "no_binding", + } +} + +func classifyError(err error) string { + text := strings.ToLower(strings.TrimSpace(fmt.Sprint(err))) + switch { + case text == "": + return "unknown" + case strings.Contains(text, "bcc") || strings.Contains(text, "checksum"): + return "checksum" + case strings.Contains(text, "short") || strings.Contains(text, "truncated") || strings.Contains(text, "length"): + return "length" + case strings.Contains(text, "json"): + return "json" + case strings.Contains(text, "start"): + return "start_symbol" + default: + return "parse" + } +} + 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 d7fcf6d9..57f5cd68 100644 --- a/go/vehicle-gateway/internal/gateway/tcp_server_test.go +++ b/go/vehicle-gateway/internal/gateway/tcp_server_test.go @@ -79,6 +79,7 @@ func TestTCPServerRecordsFrameMetrics(t *testing.T) { text := registry.Render() for _, want := range []string{ `vehicle_gateway_frames_total{protocol="JT808",status="OK"} 1`, + `vehicle_gateway_identity_total{protocol="JT808",status="unresolved"} 1`, `vehicle_gateway_publish_total{kind="raw",protocol="JT808",status="ok"} 1`, } { if !strings.Contains(text, want) { @@ -139,12 +140,14 @@ func TestTCPServerPublishesBadFrameOnlyToRaw(t *testing.T) { good := buildGBFrame(0x02, 0xfe, "LNBSCB3D4R1234567", nil) good[len(good)-1] ^= 0xff sink := &recordingSink{} + registry := metrics.NewRegistry() server := newTestServer(t, TCPProtocol{ Protocol: envelope.ProtocolGB32960, Addr: ":0", Extract: gb32960.ExtractFrames, Parse: gb32960.ParseFrame, }, sink) + server.metrics = registry client, done := runPipe(t, server) if _, err := client.Write(good); err != nil { @@ -162,6 +165,41 @@ func TestTCPServerPublishesBadFrameOnlyToRaw(t *testing.T) { if sink.raw[0].ParseError == "" { t.Fatal("parse error should be recorded") } + if text := registry.Render(); !strings.Contains(text, `vehicle_gateway_parse_errors_total{protocol="GB32960",reason="checksum"} 1`) { + t.Fatalf("parse error metric missing:\n%s", text) + } +} + +func TestTCPServerAnnotatesUnresolvedIdentity(t *testing.T) { + sink := &recordingSink{} + server := newTestServer(t, TCPProtocol{ + Protocol: envelope.ProtocolJT808, + Addr: ":0", + Extract: func(raw []byte) ([][]byte, []byte, error) { + return [][]byte{raw}, nil, nil + }, + Parse: func(_ []byte, receivedAtMS int64, sourceEndpoint string) (envelope.FrameEnvelope, error) { + return envelope.FrameEnvelope{ + Protocol: envelope.ProtocolJT808, + MessageID: "0x0200", + Phone: "13307795425", + SourceEndpoint: sourceEndpoint, + ReceivedAtMS: receivedAtMS, + EventTimeMS: receivedAtMS, + ParseStatus: envelope.ParseOK, + }, nil + }, + }, sink) + + server.handleFrame(context.Background(), nil, []byte{0x01}, "127.0.0.1:808", &connectionState{}) + + if len(sink.raw) != 1 { + t.Fatalf("raw count = %d", len(sink.raw)) + } + identity, ok := sink.raw[0].Parsed["identity"].(map[string]any) + if !ok || identity["resolved"] != false || identity["reason"] != "no_binding" { + t.Fatalf("identity metadata = %#v", sink.raw[0].Parsed["identity"]) + } } func TestTCPServerWritesProtocolResponseAfterPublish(t *testing.T) { diff --git a/go/vehicle-gateway/internal/realtime/http.go b/go/vehicle-gateway/internal/realtime/http.go index 5c710131..a37d6cfb 100644 --- a/go/vehicle-gateway/internal/realtime/http.go +++ b/go/vehicle-gateway/internal/realtime/http.go @@ -3,7 +3,9 @@ package realtime import ( "encoding/json" "errors" + "fmt" "net/http" + "strconv" "strings" "github.com/redis/go-redis/v9" @@ -15,6 +17,14 @@ type Handler struct { repository *Repository } +type OnlineIndexHandler struct { + repository *Repository +} + +type PipelineDebugHandler struct { + repository *Repository +} + func NewHandler(repository *Repository) *Handler { if repository == nil { panic("realtime repository must not be nil") @@ -22,6 +32,20 @@ func NewHandler(repository *Repository) *Handler { return &Handler{repository: repository} } +func NewOnlineIndexHandler(repository *Repository) *OnlineIndexHandler { + if repository == nil { + panic("online index repository must not be nil") + } + return &OnlineIndexHandler{repository: repository} +} + +func NewPipelineDebugHandler(repository *Repository) *PipelineDebugHandler { + if repository == nil { + panic("pipeline debug repository must not be nil") + } + return &PipelineDebugHandler{repository: repository} +} + func (h *Handler) ServeHTTP(w http.ResponseWriter, r *http.Request) { if r.Method != http.MethodGet { writeError(w, http.StatusMethodNotAllowed, "method not allowed") @@ -52,6 +76,76 @@ func (h *Handler) ServeHTTP(w http.ResponseWriter, r *http.Request) { } } +func (h *OnlineIndexHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodGet { + writeError(w, http.StatusMethodNotAllowed, "method not allowed") + return + } + if strings.Trim(r.URL.Path, "/") != "api/realtime/online" { + writeError(w, http.StatusNotFound, "route not found") + return + } + query, err := parseOnlineListQuery(r) + if err != nil { + writeError(w, http.StatusBadRequest, err.Error()) + return + } + items, total, err := h.repository.ListOnline(r.Context(), query) + if err != nil { + writeError(w, http.StatusInternalServerError, err.Error()) + return + } + w.Header().Set("Content-Type", "application/json") + _ = json.NewEncoder(w).Encode(map[string]any{ + "items": items, + "total": total, + "limit": query.Limit, + "offset": query.Offset, + }) +} + +func (h *PipelineDebugHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodGet { + writeError(w, http.StatusMethodNotAllowed, "method not allowed") + return + } + if strings.Trim(r.URL.Path, "/") != "api/debug/pipeline" { + writeError(w, http.StatusNotFound, "route not found") + return + } + summary, err := h.repository.PipelineSummary(r.Context()) + writeResult(w, summary, err) +} + +func parseOnlineListQuery(r *http.Request) (OnlineListQuery, error) { + values := r.URL.Query() + limit, err := parseBoundedInt(values.Get("limit"), 100, 1, 1000, "limit") + if err != nil { + return OnlineListQuery{}, err + } + offset, err := parseBoundedInt(values.Get("offset"), 0, 0, 1_000_000, "offset") + if err != nil { + return OnlineListQuery{}, err + } + return normalizeOnlineListQuery(OnlineListQuery{ + Protocol: envelope.Protocol(values.Get("protocol")), + Limit: limit, + Offset: offset, + }), nil +} + +func parseBoundedInt(raw string, fallback int, min int, max int, name string) (int, error) { + raw = strings.TrimSpace(raw) + if raw == "" { + return fallback, nil + } + value, err := strconv.Atoi(raw) + if err != nil || value < min || value > max { + return 0, fmt.Errorf("%s must be between %d and %d", name, min, max) + } + return value, nil +} + func writeResult(w http.ResponseWriter, value any, err error) { if err != nil { if errors.Is(err, redis.Nil) { diff --git a/go/vehicle-gateway/internal/realtime/model.go b/go/vehicle-gateway/internal/realtime/model.go index 57268766..b646c073 100644 --- a/go/vehicle-gateway/internal/realtime/model.go +++ b/go/vehicle-gateway/internal/realtime/model.go @@ -37,6 +37,24 @@ type OnlineStatus struct { TTLSeconds int64 `json:"ttl_seconds"` } +type OnlineListQuery struct { + Protocol envelope.Protocol `json:"protocol,omitempty"` + Limit int `json:"limit"` + Offset int `json:"offset"` +} + +type ProtocolPipelineSummary struct { + Protocol envelope.Protocol `json:"protocol"` + IndexedCount int64 `json:"indexed_count"` + OnlineCount int64 `json:"online_count"` + LatestSeenMS int64 `json:"latest_seen_ms"` +} + +type PipelineSummary struct { + Protocols []ProtocolPipelineSummary `json:"protocols"` + UpdatedMS int64 `json:"updated_ms"` +} + type Config struct { OnlineTTL time.Duration } diff --git a/go/vehicle-gateway/internal/realtime/repository.go b/go/vehicle-gateway/internal/realtime/repository.go index 52c6b8e1..2f8f4f9c 100644 --- a/go/vehicle-gateway/internal/realtime/repository.go +++ b/go/vehicle-gateway/internal/realtime/repository.go @@ -195,6 +195,141 @@ func (r *Repository) IsOnline(ctx context.Context, vehicleKey string) (OnlineSta return best, nil } +func (r *Repository) ListOnline(ctx context.Context, query OnlineListQuery) ([]OnlineStatus, int64, error) { + query = normalizeOnlineListQuery(query) + total, err := r.onlineTotal(ctx, query.Protocol) + if err != nil { + return nil, 0, err + } + items := make([]OnlineStatus, 0, query.Limit) + start := int64(query.Offset) + if query.Protocol != "" { + start = 0 + } + skipped := 0 + batchSize := int64(query.Limit * 5) + if batchSize < 100 { + batchSize = 100 + } + for len(items) < query.Limit { + stop := start + batchSize - 1 + members, err := r.client.ZRevRangeWithScores(ctx, "vehicle:last_seen", start, stop).Result() + if err != nil { + return nil, 0, err + } + if len(members) == 0 { + break + } + for _, member := range members { + protocol, vin, ok := splitOnlineMember(fmt.Sprint(member.Member)) + if !ok { + continue + } + if query.Protocol != "" && protocol != query.Protocol { + continue + } + if query.Protocol != "" && skipped < query.Offset { + skipped++ + continue + } + status, err := r.onlineStatusFor(ctx, protocol, vin, int64(member.Score)) + if err != nil { + return nil, 0, err + } + items = append(items, status) + if len(items) >= query.Limit { + break + } + } + if int64(len(members)) < batchSize { + break + } + start += batchSize + } + return items, total, nil +} + +func (r *Repository) PipelineSummary(ctx context.Context) (PipelineSummary, error) { + out := PipelineSummary{ + Protocols: make([]ProtocolPipelineSummary, 0, len(knownRealtimeProtocols())), + UpdatedMS: time.Now().UnixMilli(), + } + for _, protocol := range knownRealtimeProtocols() { + indexed, err := r.client.SCard(ctx, realtimeIndexKey(protocol)).Result() + if err != nil { + return PipelineSummary{}, err + } + onlineCount, latestSeen, err := r.protocolOnlineStats(ctx, protocol) + if err != nil { + return PipelineSummary{}, err + } + out.Protocols = append(out.Protocols, ProtocolPipelineSummary{ + Protocol: protocol, + IndexedCount: indexed, + OnlineCount: onlineCount, + LatestSeenMS: latestSeen, + }) + } + return out, nil +} + +func (r *Repository) onlineTotal(ctx context.Context, protocol envelope.Protocol) (int64, error) { + if protocol != "" { + return r.client.SCard(ctx, realtimeIndexKey(protocol)).Result() + } + return r.client.ZCard(ctx, "vehicle:last_seen").Result() +} + +func (r *Repository) protocolOnlineStats(ctx context.Context, protocol envelope.Protocol) (int64, int64, error) { + members, err := r.client.SMembers(ctx, realtimeIndexKey(protocol)).Result() + if err != nil { + return 0, 0, err + } + var onlineCount int64 + var latestSeen int64 + for _, vin := range members { + status, err := r.onlineStatusFor(ctx, protocol, vin, 0) + if err != nil { + return 0, 0, err + } + if status.Online { + onlineCount++ + } + if status.LastSeenMS > latestSeen { + latestSeen = status.LastSeenMS + } + } + return onlineCount, latestSeen, nil +} + +func (r *Repository) onlineStatusFor(ctx context.Context, protocol envelope.Protocol, vin string, scoreMS int64) (OnlineStatus, error) { + key := onlineKey(protocol, vin) + stateKey := onlineStateKey(protocol, vin) + state, err := r.client.HGetAll(ctx, stateKey).Result() + if err != nil { + return OnlineStatus{}, err + } + ttl := r.client.TTL(ctx, key).Val() + status := OnlineStatus{ + VehicleKey: firstNonEmptyString(state["vehicle_key"], vin), + VIN: firstNonEmptyString(state["vin"], vin), + Protocol: protocol, + Online: ttl > 0, + LastSeenMS: firstPositiveInt64(parseInt64(state["last_seen_ms"]), scoreMS), + OfflineAfterMS: parseInt64(state["offline_after_ms"]), + SourceEndpoint: state["source_endpoint"], + Protocols: []envelope.Protocol{protocol}, + TTLSeconds: int64(ttl.Seconds()), + } + if status.TTLSeconds < 0 { + status.TTLSeconds = 0 + } + if status.OfflineAfterMS <= 0 && status.LastSeenMS > 0 { + status.OfflineAfterMS = status.LastSeenMS + r.cfg.ttl().Milliseconds() + } + return status, nil +} + func (r *Repository) getSnapshot(ctx context.Context, key string) (Snapshot, error) { var snapshot Snapshot payload, err := r.client.Get(ctx, key).Bytes() @@ -741,3 +876,48 @@ func knownRealtimeProtocols() []envelope.Protocol { func protocolsKey(vehicleKey string) string { return "vehicle:protocols:" + strings.TrimSpace(vehicleKey) } + +func normalizeOnlineListQuery(query OnlineListQuery) OnlineListQuery { + query.Protocol = envelope.Protocol(strings.ToUpper(strings.TrimSpace(string(query.Protocol)))) + if query.Limit <= 0 { + query.Limit = 100 + } + if query.Limit > 1000 { + query.Limit = 1000 + } + if query.Offset < 0 { + query.Offset = 0 + } + return query +} + +func splitOnlineMember(member string) (envelope.Protocol, string, bool) { + protocolText, vin, ok := strings.Cut(strings.TrimSpace(member), ":") + if !ok || strings.TrimSpace(protocolText) == "" || strings.TrimSpace(vin) == "" { + return "", "", false + } + return envelope.Protocol(protocolText), strings.TrimSpace(vin), true +} + +func parseInt64(value string) int64 { + parsed, _ := strconv.ParseInt(strings.TrimSpace(value), 10, 64) + return parsed +} + +func firstPositiveInt64(values ...int64) int64 { + for _, value := range values { + if value > 0 { + return value + } + } + return 0 +} + +func firstNonEmptyString(values ...string) string { + for _, value := range values { + if strings.TrimSpace(value) != "" { + return strings.TrimSpace(value) + } + } + return "" +} diff --git a/go/vehicle-gateway/internal/realtime/repository_test.go b/go/vehicle-gateway/internal/realtime/repository_test.go index 793585e7..1f3a16cd 100644 --- a/go/vehicle-gateway/internal/realtime/repository_test.go +++ b/go/vehicle-gateway/internal/realtime/repository_test.go @@ -430,6 +430,125 @@ func TestRepositoryFastUpdateOnlyWritesPermanentKVAndMinuteOnline(t *testing.T) } } +func TestRepositoryListsOnlineStatusesAndPipelineSummary(t *testing.T) { + repo, closeFn := newTestRepository(t) + defer closeFn() + ctx := context.Background() + + for _, env := range []envelope.FrameEnvelope{ + { + Protocol: envelope.ProtocolGB32960, + VIN: "VIN001", + EventTimeMS: 1000, + ReceivedAtMS: 1100, + Parsed: map[string]any{ + "data_units": []any{map[string]any{"type": "0x01", "name": "vehicle", "value": map[string]any{"soc_percent": 88.0}}}, + }, + }, + { + Protocol: envelope.ProtocolJT808, + VIN: "VIN002", + EventTimeMS: 2000, + ReceivedAtMS: 2200, + MessageID: "0x0200", + Parsed: map[string]any{"location": map[string]any{"longitude": 121.1, "latitude": 30.2}}, + Fields: map[string]any{envelope.FieldLongitude: 121.1, envelope.FieldLatitude: 30.2}, + }, + } { + if err := repo.FastUpdate(ctx, env); err != nil { + t.Fatalf("FastUpdate() error = %v", err) + } + } + + items, total, err := repo.ListOnline(ctx, OnlineListQuery{Protocol: envelope.ProtocolJT808, Limit: 10}) + if err != nil { + t.Fatalf("ListOnline() error = %v", err) + } + if total != 1 || len(items) != 1 || items[0].VIN != "VIN002" || !items[0].Online { + t.Fatalf("jt808 online list total=%d items=%#v", total, items) + } + + summary, err := repo.PipelineSummary(ctx) + if err != nil { + t.Fatalf("PipelineSummary() error = %v", err) + } + got := map[envelope.Protocol]ProtocolPipelineSummary{} + for _, item := range summary.Protocols { + got[item.Protocol] = item + } + if got[envelope.ProtocolGB32960].IndexedCount != 1 || got[envelope.ProtocolGB32960].OnlineCount != 1 || got[envelope.ProtocolGB32960].LatestSeenMS != 1100 { + t.Fatalf("gb32960 summary = %#v", got[envelope.ProtocolGB32960]) + } + if got[envelope.ProtocolJT808].IndexedCount != 1 || got[envelope.ProtocolJT808].OnlineCount != 1 || got[envelope.ProtocolJT808].LatestSeenMS != 2200 { + t.Fatalf("jt808 summary = %#v", got[envelope.ProtocolJT808]) + } +} + +func TestOnlineIndexHandlerReturnsPagedOnlineStatuses(t *testing.T) { + repo, closeFn := newTestRepository(t) + defer closeFn() + ctx := context.Background() + + if err := repo.FastUpdate(ctx, envelope.FrameEnvelope{ + Protocol: envelope.ProtocolGB32960, + VIN: "VIN001", + EventTimeMS: 1000, + ReceivedAtMS: 1100, + Parsed: map[string]any{ + "data_units": []any{map[string]any{"type": "0x01", "name": "vehicle", "value": map[string]any{"soc_percent": 88.0}}}, + }, + }); err != nil { + t.Fatalf("FastUpdate() error = %v", err) + } + + handler := NewOnlineIndexHandler(repo) + request := httptest.NewRequest(http.MethodGet, "/api/realtime/online?protocol=gb32960&limit=5", nil) + response := httptest.NewRecorder() + handler.ServeHTTP(response, request) + + if response.Code != http.StatusOK { + t.Fatalf("status = %d body=%s", response.Code, response.Body.String()) + } + body := response.Body.String() + for _, want := range []string{`"protocol":"GB32960"`, `"vin":"VIN001"`, `"online":true`, `"total":1`} { + if !strings.Contains(body, want) { + t.Fatalf("response missing %s: %s", want, body) + } + } +} + +func TestPipelineDebugHandlerReturnsProtocolSummary(t *testing.T) { + repo, closeFn := newTestRepository(t) + defer closeFn() + ctx := context.Background() + + if err := repo.FastUpdate(ctx, envelope.FrameEnvelope{ + Protocol: envelope.ProtocolYutongMQTT, + VIN: "VIN003", + EventTimeMS: 3000, + ReceivedAtMS: 3300, + Parsed: map[string]any{"data": map[string]any{"TOTAL_MILEAGE": 1}}, + Fields: map[string]any{envelope.FieldTotalMileageKM: 1}, + }); err != nil { + t.Fatalf("FastUpdate() error = %v", err) + } + + handler := NewPipelineDebugHandler(repo) + request := httptest.NewRequest(http.MethodGet, "/api/debug/pipeline", nil) + response := httptest.NewRecorder() + handler.ServeHTTP(response, request) + + if response.Code != http.StatusOK { + t.Fatalf("status = %d body=%s", response.Code, response.Body.String()) + } + body := response.Body.String() + for _, want := range []string{`"protocol":"YUTONG_MQTT"`, `"indexed_count":1`, `"online_count":1`, `"latest_seen_ms":3300`} { + if !strings.Contains(body, want) { + t.Fatalf("response missing %s: %s", want, body) + } + } +} + func TestRepositoryReplacesJT808LocationWhenUpdatingRealtimeRaw(t *testing.T) { repo, closeFn := newTestRepository(t) defer closeFn()