diff --git a/go/vehicle-gateway/cmd/realtime-api/main.go b/go/vehicle-gateway/cmd/realtime-api/main.go index bab0911b..0671ecea 100644 --- a/go/vehicle-gateway/cmd/realtime-api/main.go +++ b/go/vehicle-gateway/cmd/realtime-api/main.go @@ -90,14 +90,20 @@ func main() { logger.Error("realtime snapshot mysql schema bootstrap failed", "error", err) os.Exit(1) } - mysqlUpdater := realtimeUpdater(storeMetricUpdater{store: "mysql", delegate: snapshotWriter, registry: registry}) + mysqlDelegate := realtimeUpdater(retryRealtimeUpdater{ + delegate: snapshotWriter, + attempts: envInt("MYSQL_REALTIME_RETRY_ATTEMPTS", 3), + delay: time.Duration(envInt("MYSQL_REALTIME_RETRY_DELAY_MS", 20)) * time.Millisecond, + }) + mysqlUpdater := realtimeUpdater(storeMetricUpdater{store: "mysql", delegate: mysqlDelegate, registry: registry}) if env("MYSQL_REALTIME_ASYNC_ENABLED", "true") != "false" { mysqlUpdater = newAsyncSecondaryRealtimeUpdater( nil, - storeMetricUpdater{store: "mysql", delegate: snapshotWriter, registry: registry}, + storeMetricUpdater{store: "mysql", delegate: mysqlDelegate, registry: registry}, envInt("MYSQL_REALTIME_ASYNC_QUEUE_SIZE", 20000), envInt("MYSQL_REALTIME_ASYNC_WORKERS", 4), registry, + logger, ) } updater = compositeRealtimeUpdater{ @@ -185,11 +191,13 @@ func main() { } } -func consumeKafka(ctx context.Context, logger interface { +type realtimeLogger interface { Info(string, ...any) Error(string, ...any) Warn(string, ...any) -}, registry *metrics.Registry, updater realtimeUpdater, brokers []string) { +} + +func consumeKafka(ctx context.Context, logger realtimeLogger, registry *metrics.Registry, updater realtimeUpdater, brokers []string) { kafkaTopics := kafkaTopicsFromEnv() reader := kafka.NewReader(realtimeReaderConfig(brokers, kafkaTopics)) defer reader.Close() @@ -247,7 +255,9 @@ func (u storeMetricUpdater) Update(ctx context.Context, env envelope.FrameEnvelo if u.delegate == nil { return nil } + started := time.Now() err := u.delegate.Update(ctx, env) + elapsed := time.Since(started) status := "ok" if err != nil { status = "error" @@ -258,10 +268,58 @@ func (u storeMetricUpdater) Update(ctx context.Context, env envelope.FrameEnvelo "protocol": string(env.Protocol), "status": status, }) + u.registry.SetGauge("vehicle_realtime_store_update_duration_ms", metrics.Labels{ + "store": u.store, + "protocol": string(env.Protocol), + "status": status, + }, float64(elapsed.Milliseconds())) } return err } +type retryRealtimeUpdater struct { + delegate realtimeUpdater + attempts int + delay time.Duration +} + +func (u retryRealtimeUpdater) Update(ctx context.Context, env envelope.FrameEnvelope) error { + if u.delegate == nil { + return nil + } + attempts := u.attempts + if attempts <= 0 { + attempts = 1 + } + var err error + for attempt := 1; attempt <= attempts; attempt++ { + err = u.delegate.Update(ctx, env) + if err == nil || !isTransientMySQLWriteError(err) || attempt == attempts { + return err + } + if u.delay <= 0 { + continue + } + timer := time.NewTimer(u.delay) + select { + case <-ctx.Done(): + timer.Stop() + return ctx.Err() + case <-timer.C: + } + } + return err +} + +func isTransientMySQLWriteError(err error) bool { + text := strings.ToLower(strings.TrimSpace(err.Error())) + return strings.Contains(text, "deadlock") || + strings.Contains(text, "error 1213") || + strings.Contains(text, "40001") || + strings.Contains(text, "lock wait timeout") || + strings.Contains(text, "error 1205") +} + type compositeRealtimeUpdater struct { primary realtimeUpdater secondary realtimeUpdater @@ -283,9 +341,10 @@ type asyncSecondaryRealtimeUpdater struct { queue chan envelope.FrameEnvelope cancel context.CancelFunc registry *metrics.Registry + logger realtimeLogger } -func newAsyncSecondaryRealtimeUpdater(primary realtimeUpdater, secondary realtimeUpdater, queueSize int, workers int, registry *metrics.Registry) *asyncSecondaryRealtimeUpdater { +func newAsyncSecondaryRealtimeUpdater(primary realtimeUpdater, secondary realtimeUpdater, queueSize int, workers int, registry *metrics.Registry, logger realtimeLogger) *asyncSecondaryRealtimeUpdater { if queueSize <= 0 { queueSize = 20000 } @@ -299,6 +358,7 @@ func newAsyncSecondaryRealtimeUpdater(primary realtimeUpdater, secondary realtim queue: make(chan envelope.FrameEnvelope, queueSize), cancel: cancel, registry: registry, + logger: logger, } for i := 0; i < workers; i++ { go updater.run(ctx) @@ -318,9 +378,11 @@ func (u *asyncSecondaryRealtimeUpdater) Update(ctx context.Context, env envelope select { case u.queue <- env: u.recordQueueMetric(env, "queued") + u.recordQueueDepth(env) return nil default: u.recordQueueMetric(env, "dropped") + u.recordQueueDepth(env) return nil } } @@ -331,9 +393,13 @@ func (u *asyncSecondaryRealtimeUpdater) run(ctx context.Context) { case <-ctx.Done(): return case env := <-u.queue: + u.recordQueueDepth(env) secondaryCtx, cancel := context.WithTimeout(context.Background(), kafkaMessageOperationTimeout) - _ = u.secondary.Update(secondaryCtx, env) + if err := u.secondary.Update(secondaryCtx, env); err != nil && u.logger != nil { + u.logger.Error("async realtime secondary update failed", "protocol", env.Protocol, "vin", env.VIN, "event_id", env.StableEventID(), "error", err) + } cancel() + u.recordQueueDepth(env) } } } @@ -349,6 +415,16 @@ func (u *asyncSecondaryRealtimeUpdater) recordQueueMetric(env envelope.FrameEnve }) } +func (u *asyncSecondaryRealtimeUpdater) recordQueueDepth(env envelope.FrameEnvelope) { + if u.registry == nil { + return + } + u.registry.SetGauge("vehicle_realtime_async_queue_depth", metrics.Labels{ + "store": "mysql", + "protocol": string(env.Protocol), + }, float64(len(u.queue))) +} + 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 20f5c065..7807505a 100644 --- a/go/vehicle-gateway/cmd/realtime-api/main_test.go +++ b/go/vehicle-gateway/cmd/realtime-api/main_test.go @@ -3,6 +3,7 @@ package main import ( "context" "encoding/json" + "errors" "os" "strings" "sync" @@ -103,7 +104,7 @@ func TestAsyncSecondaryRealtimeUpdaterDoesNotBlockPrimaryPath(t *testing.T) { primary := &contextCheckingRealtimeUpdater{} secondary := &blockingRealtimeUpdater{started: make(chan struct{}), release: make(chan struct{})} registry := metrics.NewRegistry() - updater := newAsyncSecondaryRealtimeUpdater(primary, secondary, 8, 1, registry) + updater := newAsyncSecondaryRealtimeUpdater(primary, secondary, 8, 1, registry, nil) defer updater.Close() start := time.Now() @@ -122,8 +123,14 @@ 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) + text := registry.Render() + for _, want := range []string{ + `vehicle_realtime_async_queue_total{protocol="GB32960",status="queued",store="mysql"} 1`, + `vehicle_realtime_async_queue_depth{protocol="GB32960",store="mysql"}`, + } { + if !strings.Contains(text, want) { + t.Fatalf("async queue metric missing %s:\n%s", want, text) + } } } @@ -139,11 +146,49 @@ func TestStoreMetricUpdaterRecordsStoreUpdateResults(t *testing.T) { 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) + text := registry.Render() + for _, want := range []string{ + `vehicle_realtime_store_updates_total{protocol="JT808",status="ok",store="redis"} 1`, + `vehicle_realtime_store_update_duration_ms{protocol="JT808",status="ok",store="redis"}`, + } { + if !strings.Contains(text, want) { + t.Fatalf("store update metric missing %s:\n%s", want, text) + } } } +func TestRetryRealtimeUpdaterRetriesTransientMySQLDeadlock(t *testing.T) { + delegate := &retryOnceRealtimeUpdater{err: errors.New("Error 1213 (40001): Deadlock found when trying to get lock")} + updater := retryRealtimeUpdater{delegate: delegate, attempts: 3} + + if err := updater.Update(context.Background(), envelope.FrameEnvelope{Protocol: envelope.ProtocolGB32960, VIN: "VIN001"}); err != nil { + t.Fatalf("Update() error = %v", err) + } + if delegate.count != 2 { + t.Fatalf("delegate updates = %d, want 2", delegate.count) + } +} + +func TestAsyncSecondaryRealtimeUpdaterLogsSecondaryErrors(t *testing.T) { + env := envelope.FrameEnvelope{Protocol: envelope.ProtocolGB32960, VIN: "VIN001", EventID: "event-1"} + logger := &recordingRealtimeLogger{} + updater := newAsyncSecondaryRealtimeUpdater(nil, failingRealtimeUpdater{}, 8, 1, nil, logger) + defer updater.Close() + + if err := updater.Update(context.Background(), env); err != nil { + t.Fatalf("Update() error = %v", err) + } + + deadline := time.Now().Add(time.Second) + for time.Now().Before(deadline) { + if logger.containsError("async realtime secondary update failed") { + return + } + time.Sleep(10 * time.Millisecond) + } + t.Fatalf("secondary error was not logged: %#v", logger.errors) +} + func TestKafkaTopicsFromEnvDefaultsToGoRawTopics(t *testing.T) { got := strings.Join(kafkaTopicsFromEnv(), ",") want := "vehicle.raw.go.gb32960.v1,vehicle.raw.go.jt808.v1,vehicle.raw.go.yutong-mqtt.v1" @@ -235,6 +280,30 @@ func (discardRealtimeLogger) Info(string, ...any) {} func (discardRealtimeLogger) Error(string, ...any) {} func (discardRealtimeLogger) Warn(string, ...any) {} +type recordingRealtimeLogger struct { + mu sync.Mutex + errors []string +} + +func (l *recordingRealtimeLogger) Info(string, ...any) {} +func (l *recordingRealtimeLogger) Warn(string, ...any) {} +func (l *recordingRealtimeLogger) Error(message string, _ ...any) { + l.mu.Lock() + defer l.mu.Unlock() + l.errors = append(l.errors, message) +} + +func (l *recordingRealtimeLogger) containsError(message string) bool { + l.mu.Lock() + defer l.mu.Unlock() + for _, got := range l.errors { + if got == message { + return true + } + } + return false +} + type blockingRealtimeUpdater struct { started chan struct{} release chan struct{} @@ -246,3 +315,24 @@ func (u *blockingRealtimeUpdater) Update(context.Context, envelope.FrameEnvelope <-u.release return nil } + +type failingRealtimeUpdater struct{} + +func (failingRealtimeUpdater) Update(context.Context, envelope.FrameEnvelope) error { + return errTestRealtimeUpdate +} + +var errTestRealtimeUpdate = errors.New("test realtime update failed") + +type retryOnceRealtimeUpdater struct { + err error + count int +} + +func (u *retryOnceRealtimeUpdater) Update(context.Context, envelope.FrameEnvelope) error { + u.count++ + if u.count == 1 { + return u.err + } + return nil +} diff --git a/go/vehicle-gateway/internal/gateway/mqtt_client.go b/go/vehicle-gateway/internal/gateway/mqtt_client.go index e67acd29..84621cd0 100644 --- a/go/vehicle-gateway/internal/gateway/mqtt_client.go +++ b/go/vehicle-gateway/internal/gateway/mqtt_client.go @@ -179,6 +179,12 @@ func (c *MQTTClient) buildTLSConfig() (*tls.Config, error) { } func (c *MQTTClient) handleMessage(ctx context.Context, topic string, payload []byte) { + started := time.Now() + frameStatus := envelope.ParseBadFrame + defer func() { + c.recordFrameDuration(frameStatus, time.Since(started)) + }() + messageCtx, cancelMessage := context.WithTimeout(context.WithoutCancel(ctx), frameOperationTimeout) defer cancelMessage() @@ -214,6 +220,7 @@ func (c *MQTTClient) handleMessage(ctx context.Context, topic string, payload [] c.recordIdentityMetric(identityStatus(env)) } } + frameStatus = env.ParseStatus c.recordFrameMetric(env.ParseStatus) if err := c.cfg.Sink.PublishRaw(messageCtx, env); err != nil { c.recordPublishMetric("raw", "error") @@ -252,6 +259,16 @@ func (c *MQTTClient) recordFrameMetric(status envelope.ParseStatus) { }) } +func (c *MQTTClient) recordFrameDuration(status envelope.ParseStatus, elapsed time.Duration) { + if c.cfg.Metrics == nil { + return + } + c.cfg.Metrics.SetGauge("vehicle_gateway_frame_duration_ms", metrics.Labels{ + "protocol": string(envelope.ProtocolYutongMQTT), + "status": string(status), + }, float64(elapsed.Milliseconds())) +} + func (c *MQTTClient) recordPublishMetric(kind string, status string) { if c.cfg.Metrics == nil { return diff --git a/go/vehicle-gateway/internal/gateway/mqtt_client_test.go b/go/vehicle-gateway/internal/gateway/mqtt_client_test.go index 7b18bc8a..a901d838 100644 --- a/go/vehicle-gateway/internal/gateway/mqtt_client_test.go +++ b/go/vehicle-gateway/internal/gateway/mqtt_client_test.go @@ -79,6 +79,7 @@ func TestMQTTClientRecordsMessageMetrics(t *testing.T) { `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`, + `vehicle_gateway_frame_duration_ms{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 f74fa63e..58056670 100644 --- a/go/vehicle-gateway/internal/gateway/tcp_server.go +++ b/go/vehicle-gateway/internal/gateway/tcp_server.go @@ -195,6 +195,12 @@ func (s *TCPServer) handleConnection(ctx context.Context, conn net.Conn) { } func (s *TCPServer) handleFrame(ctx context.Context, conn net.Conn, raw []byte, source string, state *connectionState) { + started := time.Now() + frameStatus := envelope.ParseBadFrame + defer func() { + s.recordFrameDuration(frameStatus, time.Since(started)) + }() + frameCtx, cancelFrame := context.WithTimeout(context.WithoutCancel(ctx), frameOperationTimeout) defer cancelFrame() @@ -229,6 +235,7 @@ func (s *TCPServer) handleFrame(ctx context.Context, conn net.Conn, raw []byte, } enrichConnectionPlatform(&env, state) } + frameStatus = env.ParseStatus s.recordFrameMetric(env.ParseStatus) if err := s.sink.PublishRaw(frameCtx, env); err != nil { @@ -307,6 +314,16 @@ func (s *TCPServer) recordFrameMetric(status envelope.ParseStatus) { }) } +func (s *TCPServer) recordFrameDuration(status envelope.ParseStatus, elapsed time.Duration) { + if s.metrics == nil { + return + } + s.metrics.SetGauge("vehicle_gateway_frame_duration_ms", metrics.Labels{ + "protocol": string(s.protocol.Protocol), + "status": string(status), + }, float64(elapsed.Milliseconds())) +} + func (s *TCPServer) recordPublishMetric(kind string, status string) { if s.metrics == nil { return diff --git a/go/vehicle-gateway/internal/gateway/tcp_server_test.go b/go/vehicle-gateway/internal/gateway/tcp_server_test.go index 57f5cd68..90d1f256 100644 --- a/go/vehicle-gateway/internal/gateway/tcp_server_test.go +++ b/go/vehicle-gateway/internal/gateway/tcp_server_test.go @@ -81,6 +81,7 @@ func TestTCPServerRecordsFrameMetrics(t *testing.T) { `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`, + `vehicle_gateway_frame_duration_ms{protocol="JT808",status="OK"}`, } { if !strings.Contains(text, want) { t.Fatalf("metrics missing %s:\n%s", want, text)