From 1dfd540700c6ce7d6f7bab42048eca68d0985e7c Mon Sep 17 00:00:00 2001 From: lingniu Date: Thu, 2 Jul 2026 18:53:02 +0800 Subject: [PATCH] feat(go): add platform principles and health probes --- .../iot-data-platform-principles.md | 81 +++++++++++ go/vehicle-gateway/cmd/gateway/main.go | 2 + go/vehicle-gateway/cmd/history-writer/main.go | 4 + .../cmd/nats-kafka-bridge/main.go | 9 ++ go/vehicle-gateway/cmd/realtime-api/main.go | 9 ++ go/vehicle-gateway/cmd/stat-writer/main.go | 4 + go/vehicle-gateway/internal/health/health.go | 132 ++++++++++++++++++ .../internal/health/health_test.go | 98 +++++++++++++ 8 files changed, 339 insertions(+) create mode 100644 docs/architecture/iot-data-platform-principles.md create mode 100644 go/vehicle-gateway/internal/health/health.go create mode 100644 go/vehicle-gateway/internal/health/health_test.go diff --git a/docs/architecture/iot-data-platform-principles.md b/docs/architecture/iot-data-platform-principles.md new file mode 100644 index 00000000..e887936a --- /dev/null +++ b/docs/architecture/iot-data-platform-principles.md @@ -0,0 +1,81 @@ +# Vehicle IoT Data Platform Principles + +## Goal + +Build a production vehicle data platform that can receive GB32960, JT808, and Yutong MQTT data reliably, keep raw evidence traceable, and expose only simple business tables for realtime, history, identity, and metrics. + +## First Principles + +1. Raw data is the source of truth. +2. Kafka is the durable replay log. +3. Redis is realtime cache, not permanent storage. +4. MySQL stores identity, light realtime business snapshots, and low-cardinality metrics. +5. TDengine stores time-series raw and location history. +6. Protocol-specific fields stay in raw parsed JSON unless they become a stable query or metric requirement. +7. Upper business tables must not duplicate full parsed payloads. + +## Data Flow + +```mermaid +flowchart LR + GB["GB32960 TCP"] --> GW["Go Gateway"] + JT["JT808 TCP"] --> GW + YM["Yutong MQTT"] --> GW + + GW --> NATS["NATS JetStream ingress"] + NATS --> BR["NATS Kafka Bridge"] + BR --> KRAW["Kafka raw topics"] + GW -. fallback .-> KRAW + + KRAW --> HW["history-writer"] + KRAW --> SW["stat-writer"] + KRAW --> RA["realtime-api projector"] + + HW --> TDRAW["TDengine raw_frames"] + HW --> TDLOC["TDengine vehicle_locations"] + SW --> MYMET["MySQL vehicle_daily_metric"] + RA --> REDIS["Redis realtime-raw"] + RA --> MYSNAP["MySQL vehicle_realtime_snapshot"] + RA --> MYLOC["MySQL vehicle_realtime_location"] +``` + +## Minimal Storage Contract + +| Store | Table or key | Purpose | Keep | Avoid | +| --- | --- | --- | --- | --- | +| TDengine | `raw_frames` | Replay and audit evidence | raw hex/text, full parsed JSON, parse status, protocol tags | business-only duplicated columns | +| TDengine | `vehicle_locations` | High-volume location history | time, vin, protocol, longitude, latitude, speed, direction, mileage | full parsed JSON | +| MySQL | `vehicle_realtime_snapshot` | Latest per-protocol vehicle heartbeat | protocol, vin, plate, event time, received time | phone, device, message sequence, parsed JSON | +| MySQL | `vehicle_realtime_location` | Latest business location cache | vin, plate, location, speed, mileage, SOC, event time | full raw payload and protocol internals | +| MySQL | `vehicle_daily_metric` | Queryable daily metrics | vin or vehicle key, date, metric key, value, method, source mileage | per-frame raw details | +| MySQL | `vehicle_identity_binding` | Manual identity mapping | vin, plate, phone, device id | registration history | +| MySQL | `jt808_registration` | JT808 registration and auth trace | phone, device id, plate, auth code, vin match state, first/latest seen | GB32960 or MQTT records | +| Redis | `realtime-raw:{protocol}:{vin}` | Latest full realtime protocol state | merged latest protocol payload | historical data | + +## Event Envelope Rules + +Every received frame should become one `FrameEnvelope`. + +- `protocol`, `message_id`, `event_time_ms`, `received_at_ms`, and `parse_status` are mandatory. +- `vin` is preferred as the vehicle identity. +- JT808 can use `phone` as a temporary key before VIN binding is resolved. +- `raw_hex` or `raw_text` must be retained for replay and parser correction. +- `parsed` stores protocol-specific fields. +- `fields` stores only stable cross-protocol core fields: speed, total mileage, longitude, latitude, SOC. + +## Optimization Order + +1. Stabilize ingress: connection lifecycle, protocol parser correctness, bounded backpressure, durable publish. +2. Stabilize event log: NATS to Kafka bridge, topic names, partition keys, retry and replay. +3. Simplify storage: keep only the minimal tables above, remove duplicated payloads from business tables. +4. Stabilize realtime: Redis full latest raw state, MySQL lightweight snapshot and location. +5. Stabilize history: raw and location query pagination, clear time zone behavior. +6. Stabilize metrics: idempotent daily metrics from mileage differences and raw replay. +7. Add operations: health, readiness, metrics, lag, connection counts, deploy notes. + +## Current Next Steps + +1. Add health and readiness endpoints to each long-running Go service. +2. Add runtime metrics for protocol frame counts, parse failures, Kafka/NATS publish failures, and writer lag. +3. Review TDengine tables and remove unused tables only after export or explicit confirmation. +4. Keep new business tables narrow by default. Add a column only when a query or index proves it is needed. diff --git a/go/vehicle-gateway/cmd/gateway/main.go b/go/vehicle-gateway/cmd/gateway/main.go index 505a3c78..c7f341cc 100644 --- a/go/vehicle-gateway/cmd/gateway/main.go +++ b/go/vehicle-gateway/cmd/gateway/main.go @@ -16,6 +16,7 @@ import ( "lingniu-vehicle-ingest/go/vehicle-gateway/internal/envelope" "lingniu-vehicle-ingest/go/vehicle-gateway/internal/eventbus" "lingniu-vehicle-ingest/go/vehicle-gateway/internal/gateway" + "lingniu-vehicle-ingest/go/vehicle-gateway/internal/health" "lingniu-vehicle-ingest/go/vehicle-gateway/internal/identity" "lingniu-vehicle-ingest/go/vehicle-gateway/internal/observability" "lingniu-vehicle-ingest/go/vehicle-gateway/internal/protocol/gb32960" @@ -39,6 +40,7 @@ func main() { os.Exit(1) } defer closeResolver() + health.Start(ctx, logger, health.NewServer(env("HEALTH_ADDR", ""), "vehicle-gateway", nil)) protocols := []gateway.TCPProtocol{ { diff --git a/go/vehicle-gateway/cmd/history-writer/main.go b/go/vehicle-gateway/cmd/history-writer/main.go index 3017381f..efd66766 100644 --- a/go/vehicle-gateway/cmd/history-writer/main.go +++ b/go/vehicle-gateway/cmd/history-writer/main.go @@ -14,6 +14,7 @@ import ( _ "github.com/taosdata/driver-go/v3/taosWS" "lingniu-vehicle-ingest/go/vehicle-gateway/internal/envelope" + "lingniu-vehicle-ingest/go/vehicle-gateway/internal/health" "lingniu-vehicle-ingest/go/vehicle-gateway/internal/history" "lingniu-vehicle-ingest/go/vehicle-gateway/internal/observability" ) @@ -34,6 +35,9 @@ func main() { logger.Error("tdengine ping failed", "error", err) os.Exit(1) } + health.Start(ctx, logger, health.NewServer(env("HEALTH_ADDR", ""), "vehicle-history-writer", []health.Check{ + {Name: "tdengine", Check: db.PingContext}, + })) writer := history.NewWriter(db) if cfg.EnsureSchema { diff --git a/go/vehicle-gateway/cmd/nats-kafka-bridge/main.go b/go/vehicle-gateway/cmd/nats-kafka-bridge/main.go index a3325499..e4ede85a 100644 --- a/go/vehicle-gateway/cmd/nats-kafka-bridge/main.go +++ b/go/vehicle-gateway/cmd/nats-kafka-bridge/main.go @@ -17,6 +17,7 @@ import ( "github.com/segmentio/kafka-go" "lingniu-vehicle-ingest/go/vehicle-gateway/internal/envelope" + "lingniu-vehicle-ingest/go/vehicle-gateway/internal/health" "lingniu-vehicle-ingest/go/vehicle-gateway/internal/observability" ) @@ -32,6 +33,14 @@ func main() { os.Exit(1) } defer conn.Close() + health.Start(ctx, logger, health.NewServer(env("HEALTH_ADDR", ""), "nats-kafka-bridge", []health.Check{ + {Name: "nats", Check: func(context.Context) error { + if conn.Status() != nats.CONNECTED { + return fmt.Errorf("nats status is %s", conn.Status().String()) + } + return nil + }}, + })) js, err := conn.JetStream() if err != nil { logger.Error("nats jetstream init failed", "error", err) diff --git a/go/vehicle-gateway/cmd/realtime-api/main.go b/go/vehicle-gateway/cmd/realtime-api/main.go index baf8ad14..31b78ab0 100644 --- a/go/vehicle-gateway/cmd/realtime-api/main.go +++ b/go/vehicle-gateway/cmd/realtime-api/main.go @@ -18,6 +18,7 @@ import ( _ "github.com/taosdata/driver-go/v3/taosWS" "lingniu-vehicle-ingest/go/vehicle-gateway/internal/envelope" + "lingniu-vehicle-ingest/go/vehicle-gateway/internal/health" "lingniu-vehicle-ingest/go/vehicle-gateway/internal/history" "lingniu-vehicle-ingest/go/vehicle-gateway/internal/observability" "lingniu-vehicle-ingest/go/vehicle-gateway/internal/realtime" @@ -46,6 +47,9 @@ func main() { }) mux := http.NewServeMux() + healthChecks := []health.Check{ + {Name: "redis", Check: func(ctx context.Context) error { return client.Ping(ctx).Err() }}, + } mux.Handle("/openapi.json", realtime.NewOpenAPIHandler()) mux.Handle("/swagger-ui/", realtime.NewSwaggerUIHandler()) mux.Handle("/api/realtime/vehicles/", realtime.NewHandler(repository)) @@ -62,6 +66,7 @@ func main() { logger.Error("stats mysql ping failed", "error", err) os.Exit(1) } + healthChecks = append(healthChecks, health.Check{Name: "mysql", Check: db.PingContext}) closeStats = func() { _ = db.Close() } bindingTable := env("VEHICLE_IDENTITY_TABLE", "vehicle_identity_binding") snapshotWriter := realtime.NewSnapshotWriterWithPlateResolver(db, realtime.NewBindingPlateResolver(db, bindingTable)) @@ -108,6 +113,7 @@ func main() { logger.Error("history tdengine ping failed", "driver", driver, "error", err) os.Exit(1) } + healthChecks = append(healthChecks, health.Check{Name: "tdengine", Check: db.PingContext}) closeHistory = func() { _ = db.Close() } database := env("TDENGINE_DATABASE", history.DefaultDatabase) mux.Handle("/api/history/raw-frames", history.NewRawFrameHandler(history.NewRawFrameRepository(db, database))) @@ -126,6 +132,9 @@ func main() { logger.Warn("TDENGINE_DSN is empty; history query api disabled") } defer closeHistory() + healthHandler := health.NewHandler("vehicle-realtime-api", healthChecks) + mux.Handle("/healthz", healthHandler) + mux.Handle("/readyz", healthHandler) server := &http.Server{ Addr: env("HTTP_ADDR", ":20210"), diff --git a/go/vehicle-gateway/cmd/stat-writer/main.go b/go/vehicle-gateway/cmd/stat-writer/main.go index b11de4c2..01ff8e5e 100644 --- a/go/vehicle-gateway/cmd/stat-writer/main.go +++ b/go/vehicle-gateway/cmd/stat-writer/main.go @@ -14,6 +14,7 @@ import ( "github.com/segmentio/kafka-go" "lingniu-vehicle-ingest/go/vehicle-gateway/internal/envelope" + "lingniu-vehicle-ingest/go/vehicle-gateway/internal/health" "lingniu-vehicle-ingest/go/vehicle-gateway/internal/observability" "lingniu-vehicle-ingest/go/vehicle-gateway/internal/stats" ) @@ -34,6 +35,9 @@ func main() { logger.Error("mysql ping failed", "error", err) os.Exit(1) } + health.Start(ctx, logger, health.NewServer(env("HEALTH_ADDR", ""), "vehicle-stat-writer", []health.Check{ + {Name: "mysql", Check: db.PingContext}, + })) writer := stats.NewWriter(db, cfg.Location) if cfg.EnsureSchema { diff --git a/go/vehicle-gateway/internal/health/health.go b/go/vehicle-gateway/internal/health/health.go new file mode 100644 index 00000000..d83a5174 --- /dev/null +++ b/go/vehicle-gateway/internal/health/health.go @@ -0,0 +1,132 @@ +package health + +import ( + "context" + "encoding/json" + "errors" + "net/http" + "strings" + "time" +) + +type Check struct { + Name string + Check func(context.Context) error +} + +type Handler struct { + service string + checks []Check +} + +type Logger interface { + Info(string, ...any) + Error(string, ...any) +} + +func NewHandler(service string, checks []Check) *Handler { + service = strings.TrimSpace(service) + if service == "" { + service = "vehicle-service" + } + return &Handler{service: service, checks: checks} +} + +func NewMux(service string, checks []Check) *http.ServeMux { + mux := http.NewServeMux() + handler := NewHandler(service, checks) + mux.Handle("/healthz", handler) + mux.Handle("/readyz", handler) + return mux +} + +func NewServer(addr string, service string, checks []Check) *http.Server { + addr = strings.TrimSpace(addr) + if addr == "" { + return nil + } + return &http.Server{ + Addr: addr, + Handler: NewMux(service, checks), + ReadHeaderTimeout: 5 * time.Second, + } +} + +func Start(ctx context.Context, logger Logger, server *http.Server) { + if server == nil { + return + } + go func() { + <-ctx.Done() + shutdownCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + _ = server.Shutdown(shutdownCtx) + }() + go func() { + if logger != nil { + logger.Info("health server started", "addr", server.Addr) + } + if err := server.ListenAndServe(); err != nil && !errors.Is(err, http.ErrServerClosed) && logger != nil { + logger.Error("health server failed", "error", err) + } + }() +} + +func (h *Handler) ServeHTTP(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodGet { + writeHealth(w, http.StatusMethodNotAllowed, map[string]any{"error": "method not allowed"}) + return + } + path := strings.Trim(r.URL.Path, "/") + if path != "healthz" && path != "readyz" { + writeHealth(w, http.StatusNotFound, map[string]any{"error": "route not found"}) + return + } + if path == "healthz" { + writeHealth(w, http.StatusOK, map[string]any{ + "service": h.service, + "status": "ok", + "time": time.Now().UTC().Format(time.RFC3339), + }) + return + } + h.writeReadiness(w, r) +} + +func (h *Handler) writeReadiness(w http.ResponseWriter, r *http.Request) { + ctx, cancel := context.WithTimeout(r.Context(), 2*time.Second) + defer cancel() + + status := "ok" + code := http.StatusOK + checks := make(map[string]any, len(h.checks)) + for _, check := range h.checks { + name := strings.TrimSpace(check.Name) + if name == "" { + continue + } + if check.Check == nil { + checks[name] = map[string]any{"status": "ok"} + continue + } + if err := check.Check(ctx); err != nil { + status = "degraded" + code = http.StatusServiceUnavailable + checks[name] = map[string]any{"status": "error", "error": err.Error()} + continue + } + checks[name] = map[string]any{"status": "ok"} + } + writeHealth(w, code, map[string]any{ + "service": h.service, + "status": status, + "time": time.Now().UTC().Format(time.RFC3339), + "checks": checks, + }) +} + +func writeHealth(w http.ResponseWriter, status int, payload map[string]any) { + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(status) + _ = json.NewEncoder(w).Encode(payload) +} diff --git a/go/vehicle-gateway/internal/health/health_test.go b/go/vehicle-gateway/internal/health/health_test.go new file mode 100644 index 00000000..ab74a47b --- /dev/null +++ b/go/vehicle-gateway/internal/health/health_test.go @@ -0,0 +1,98 @@ +package health + +import ( + "context" + "errors" + "net/http" + "net/http/httptest" + "strings" + "testing" +) + +func TestHandlerReturnsHealthyWhenAllChecksPass(t *testing.T) { + handler := NewHandler("vehicle-realtime-api", []Check{ + {Name: "redis", Check: func(context.Context) error { return nil }}, + {Name: "mysql", Check: func(context.Context) error { return nil }}, + }) + request := httptest.NewRequest(http.MethodGet, "/readyz", 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{`"service":"vehicle-realtime-api"`, `"status":"ok"`, `"redis":{"status":"ok"`, `"mysql":{"status":"ok"`} { + if !strings.Contains(body, want) { + t.Fatalf("response missing %s: %s", want, body) + } + } +} + +func TestHandlerReturnsServiceUnavailableWhenCheckFails(t *testing.T) { + handler := NewHandler("vehicle-history-writer", []Check{ + {Name: "tdengine", Check: func(context.Context) error { return errors.New("ping failed") }}, + }) + request := httptest.NewRequest(http.MethodGet, "/readyz", nil) + response := httptest.NewRecorder() + + handler.ServeHTTP(response, request) + + if response.Code != http.StatusServiceUnavailable { + t.Fatalf("status = %d body=%s", response.Code, response.Body.String()) + } + body := response.Body.String() + for _, want := range []string{`"service":"vehicle-history-writer"`, `"status":"degraded"`, `"tdengine":`, `"status":"error"`, `"error":"ping failed"`} { + if !strings.Contains(body, want) { + t.Fatalf("response missing %s: %s", want, body) + } + } +} + +func TestHandlerRejectsUnknownPath(t *testing.T) { + handler := NewHandler("vehicle-gateway", nil) + request := httptest.NewRequest(http.MethodGet, "/metrics", nil) + response := httptest.NewRecorder() + + handler.ServeHTTP(response, request) + + if response.Code != http.StatusNotFound { + t.Fatalf("status = %d body=%s", response.Code, response.Body.String()) + } +} + +func TestNewMuxRegistersHealthAndReadinessRoutes(t *testing.T) { + mux := NewMux("vehicle-stat-writer", []Check{ + {Name: "mysql", Check: func(context.Context) error { return nil }}, + }) + for _, path := range []string{"/healthz", "/readyz"} { + request := httptest.NewRequest(http.MethodGet, path, nil) + response := httptest.NewRecorder() + + mux.ServeHTTP(response, request) + + if response.Code != http.StatusOK { + t.Fatalf("%s status = %d body=%s", path, response.Code, response.Body.String()) + } + } +} + +func TestNewServerReturnsNilWhenAddressIsEmpty(t *testing.T) { + if server := NewServer("", "vehicle-gateway", nil); server != nil { + t.Fatalf("server = %#v, want nil", server) + } +} + +func TestNewServerBuildsConfiguredHTTPServer(t *testing.T) { + server := NewServer(":20290", "vehicle-gateway", nil) + if server == nil { + t.Fatal("server is nil") + } + if server.Addr != ":20290" { + t.Fatalf("addr = %q", server.Addr) + } + if server.ReadHeaderTimeout == 0 { + t.Fatal("ReadHeaderTimeout should be configured") + } +}