diff --git a/go/vehicle-gateway/cmd/stat-writer/main.go b/go/vehicle-gateway/cmd/stat-writer/main.go new file mode 100644 index 00000000..fb6716d4 --- /dev/null +++ b/go/vehicle-gateway/cmd/stat-writer/main.go @@ -0,0 +1,122 @@ +package main + +import ( + "context" + "database/sql" + "encoding/json" + "os" + "os/signal" + "strings" + "syscall" + "time" + + _ "github.com/go-sql-driver/mysql" + "github.com/segmentio/kafka-go" + + "lingniu-vehicle-ingest/go/vehicle-gateway/internal/envelope" + "lingniu-vehicle-ingest/go/vehicle-gateway/internal/observability" + "lingniu-vehicle-ingest/go/vehicle-gateway/internal/stats" +) + +func main() { + logger := observability.NewLogger("vehicle-stat-writer") + ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM) + defer stop() + + cfg := loadConfig() + db, err := sql.Open("mysql", cfg.MySQLDSN) + if err != nil { + logger.Error("mysql open failed", "error", err) + os.Exit(1) + } + defer db.Close() + if err := db.PingContext(ctx); err != nil { + logger.Error("mysql ping failed", "error", err) + os.Exit(1) + } + + writer := stats.NewWriter(db, cfg.Location) + if cfg.EnsureSchema { + if err := writer.EnsureSchema(ctx); err != nil { + logger.Error("mysql schema bootstrap failed", "error", err) + os.Exit(1) + } + } + + reader := kafka.NewReader(kafka.ReaderConfig{ + Brokers: cfg.KafkaBrokers, + GroupID: cfg.KafkaGroup, + GroupTopics: cfg.KafkaTopics, + MinBytes: 1, + MaxBytes: 10e6, + }) + defer reader.Close() + + logger.Info("stat writer started", "group", cfg.KafkaGroup, "topics", strings.Join(cfg.KafkaTopics, ",")) + for { + message, err := reader.FetchMessage(ctx) + if err != nil { + if ctx.Err() != nil { + return + } + logger.Error("kafka fetch failed", "error", err) + continue + } + var env envelope.FrameEnvelope + if err := json.Unmarshal(message.Value, &env); err != nil { + logger.Warn("skip invalid envelope json", "topic", message.Topic, "partition", message.Partition, "offset", message.Offset, "error", err) + _ = reader.CommitMessages(ctx, message) + continue + } + if err := writer.Append(ctx, env); err != nil { + logger.Error("mysql append failed", "topic", message.Topic, "partition", message.Partition, "offset", message.Offset, "event_id", env.StableEventID(), "error", err) + continue + } + if err := reader.CommitMessages(ctx, message); err != nil { + logger.Error("kafka commit failed", "topic", message.Topic, "partition", message.Partition, "offset", message.Offset, "error", err) + } + } +} + +type config struct { + KafkaBrokers []string + KafkaTopics []string + KafkaGroup string + MySQLDSN string + EnsureSchema bool + Location *time.Location +} + +func loadConfig() config { + loc, err := time.LoadLocation(env("LOCAL_TZ", "Asia/Shanghai")) + if err != nil { + loc = time.FixedZone("Asia/Shanghai", 8*3600) + } + return config{ + KafkaBrokers: splitCSV(env("KAFKA_BROKERS", "127.0.0.1:9092")), + KafkaTopics: splitCSV(env("KAFKA_TOPICS", "vehicle.raw.gb32960.v1,vehicle.raw.jt808.v1")), + KafkaGroup: env("KAFKA_GROUP", "go-stat-writer"), + MySQLDSN: env("MYSQL_DSN", ""), + EnsureSchema: env("MYSQL_ENSURE_SCHEMA", "true") != "false", + Location: loc, + } +} + +func env(key string, fallback string) string { + value := strings.TrimSpace(os.Getenv(key)) + if value == "" { + return fallback + } + return value +} + +func splitCSV(value string) []string { + var out []string + for _, item := range strings.Split(value, ",") { + item = strings.TrimSpace(item) + if item != "" { + out = append(out, item) + } + } + return out +} diff --git a/go/vehicle-gateway/go.mod b/go/vehicle-gateway/go.mod index f5c6af4f..f0ce1a80 100644 --- a/go/vehicle-gateway/go.mod +++ b/go/vehicle-gateway/go.mod @@ -3,11 +3,13 @@ module lingniu-vehicle-ingest/go/vehicle-gateway go 1.26 require ( + github.com/go-sql-driver/mysql v1.9.3 github.com/segmentio/kafka-go v0.4.49 github.com/taosdata/driver-go/v3 v3.8.1 ) require ( + filippo.io/edwards25519 v1.1.0 // indirect github.com/google/uuid v1.6.0 // indirect github.com/gorilla/websocket v1.5.0 // indirect github.com/json-iterator/go v1.1.12 // indirect diff --git a/go/vehicle-gateway/go.sum b/go/vehicle-gateway/go.sum index d5ca35d3..a65a9396 100644 --- a/go/vehicle-gateway/go.sum +++ b/go/vehicle-gateway/go.sum @@ -1,6 +1,10 @@ +filippo.io/edwards25519 v1.1.0 h1:FNf4tywRC1HmFuKW5xopWpigGjJKiJSV0Cqo0cJWDaA= +filippo.io/edwards25519 v1.1.0/go.mod h1:BxyFTGdWcka3PhytdK4V28tE5sGfRvvvRV7EaN4VDT4= github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c= github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= +github.com/go-sql-driver/mysql v1.9.3 h1:U/N249h2WzJ3Ukj8SowVFjdtZKfu9vlLZxjPXV1aweo= +github.com/go-sql-driver/mysql v1.9.3/go.mod h1:qn46aNg1333BRMNU69Lq93t8du/dwxI64Gl8i5p1WMU= github.com/google/gofuzz v1.0.0/go.mod h1:dBl0BpW6vV/+mYPU4Po3pmUjxk6FQPldtuIdl/M65Eg= github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0= github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= diff --git a/go/vehicle-gateway/internal/stats/daily_metric.go b/go/vehicle-gateway/internal/stats/daily_metric.go new file mode 100644 index 00000000..468f5daa --- /dev/null +++ b/go/vehicle-gateway/internal/stats/daily_metric.go @@ -0,0 +1,159 @@ +package stats + +import ( + "context" + "database/sql" + "errors" + "strconv" + "strings" + "time" + + "lingniu-vehicle-ingest/go/vehicle-gateway/internal/envelope" +) + +const ( + MetricDailyMileageKM = "daily_mileage_km" + MetricDailyTotalMileageKM = "daily_total_mileage_km" +) + +type Execer interface { + ExecContext(context.Context, string, ...any) (sql.Result, error) +} + +type Writer struct { + exec Execer + loc *time.Location +} + +type MetricSample struct { + VIN string + Protocol envelope.Protocol + StatDate string + MetricKey string + MetricValue float64 + TotalMileageKM float64 +} + +func NewWriter(exec Execer, loc *time.Location) *Writer { + if exec == nil { + panic("stats execer must not be nil") + } + if loc == nil { + loc = time.FixedZone("Asia/Shanghai", 8*3600) + } + return &Writer{exec: exec, loc: loc} +} + +func (w *Writer) EnsureSchema(ctx context.Context) error { + _, err := w.exec.ExecContext(ctx, DailyMetricTableSQL) + return err +} + +func (w *Writer) Append(ctx context.Context, env envelope.FrameEnvelope) error { + samples, err := SamplesFromEnvelope(env, w.loc) + if err != nil { + return err + } + for _, sample := range samples { + if _, err := w.exec.ExecContext(ctx, upsertDailyMetricSQL, + sample.VIN, + sample.StatDate, + string(sample.Protocol), + sample.MetricKey, + sample.MetricValue, + sample.TotalMileageKM, + sample.TotalMileageKM); err != nil { + return err + } + } + return nil +} + +func SamplesFromEnvelope(env envelope.FrameEnvelope, loc *time.Location) ([]MetricSample, error) { + vin := strings.TrimSpace(env.VIN) + if vin == "" { + return nil, nil + } + totalMileage, ok := floatField(env, envelope.FieldTotalMileageKM) + if !ok { + return nil, nil + } + if loc == nil { + loc = time.FixedZone("Asia/Shanghai", 8*3600) + } + eventMS := env.EventTimeMS + if eventMS <= 0 { + eventMS = env.ReceivedAtMS + } + if eventMS <= 0 { + return nil, errors.New("event or received time is required") + } + statDate := time.UnixMilli(eventMS).In(loc).Format("2006-01-02") + return []MetricSample{ + { + VIN: vin, + Protocol: env.Protocol, + StatDate: statDate, + MetricKey: MetricDailyMileageKM, + MetricValue: 0, + TotalMileageKM: totalMileage, + }, + { + VIN: vin, + Protocol: env.Protocol, + StatDate: statDate, + MetricKey: MetricDailyTotalMileageKM, + MetricValue: totalMileage, + TotalMileageKM: totalMileage, + }, + }, nil +} + +const upsertDailyMetricSQL = ` +INSERT INTO vehicle_daily_metric + (vin, stat_date, protocol, metric_key, metric_value, metric_unit, + first_total_mileage_km, latest_total_mileage_km, sample_count, calculation_method) +VALUES (?, ?, ?, ?, ?, 'km', ?, ?, 1, 'TOTAL_MILEAGE_DIFF') +ON DUPLICATE KEY UPDATE + first_total_mileage_km = LEAST(first_total_mileage_km, VALUES(first_total_mileage_km)), + latest_total_mileage_km = GREATEST(latest_total_mileage_km, VALUES(latest_total_mileage_km)), + metric_value = CASE + WHEN metric_key = 'daily_mileage_km' + THEN GREATEST(latest_total_mileage_km, VALUES(latest_total_mileage_km)) + - LEAST(first_total_mileage_km, VALUES(first_total_mileage_km)) + WHEN metric_key = 'daily_total_mileage_km' + THEN GREATEST(latest_total_mileage_km, VALUES(latest_total_mileage_km)) + ELSE VALUES(metric_value) + END, + sample_count = sample_count + 1, + updated_at = CURRENT_TIMESTAMP +` + +func floatField(env envelope.FrameEnvelope, key string) (float64, bool) { + if env.Fields == nil { + return 0, false + } + value, ok := env.Fields[key] + if !ok || value == nil { + return 0, false + } + switch typed := value.(type) { + case float64: + return typed, true + case float32: + return float64(typed), true + case int: + return float64(typed), true + case int64: + return float64(typed), true + case uint16: + return float64(typed), true + case uint32: + return float64(typed), true + case string: + parsed, err := strconv.ParseFloat(strings.TrimSpace(typed), 64) + return parsed, err == nil + default: + return 0, false + } +} diff --git a/go/vehicle-gateway/internal/stats/daily_metric_test.go b/go/vehicle-gateway/internal/stats/daily_metric_test.go new file mode 100644 index 00000000..3ec011d7 --- /dev/null +++ b/go/vehicle-gateway/internal/stats/daily_metric_test.go @@ -0,0 +1,107 @@ +package stats + +import ( + "context" + "database/sql" + "strings" + "testing" + "time" + + "lingniu-vehicle-ingest/go/vehicle-gateway/internal/envelope" +) + +func TestSamplesFromEnvelopeDerivesDailyMileageAndTotal(t *testing.T) { + loc := time.FixedZone("Asia/Shanghai", 8*3600) + env := envelope.FrameEnvelope{ + Protocol: envelope.ProtocolJT808, + VIN: "LNBVIN00000000001", + EventTimeMS: time.Date(2026, 7, 1, 8, 30, 0, 0, loc).UnixMilli(), + Fields: map[string]any{ + envelope.FieldTotalMileageKM: 10241.2, + }, + } + + samples, err := SamplesFromEnvelope(env, loc) + if err != nil { + t.Fatalf("SamplesFromEnvelope() error = %v", err) + } + if len(samples) != 2 { + t.Fatalf("sample count = %d", len(samples)) + } + if samples[0].MetricKey != MetricDailyMileageKM || samples[0].MetricValue != 0 { + t.Fatalf("unexpected daily mileage sample: %#v", samples[0]) + } + if samples[1].MetricKey != MetricDailyTotalMileageKM || samples[1].MetricValue != 10241.2 { + t.Fatalf("unexpected daily total sample: %#v", samples[1]) + } + if samples[0].StatDate != "2026-07-01" { + t.Fatalf("stat date = %q", samples[0].StatDate) + } +} + +func TestSamplesFromEnvelopeSkipsMissingVINOrMileage(t *testing.T) { + samples, err := SamplesFromEnvelope(envelope.FrameEnvelope{ + Protocol: envelope.ProtocolJT808, + Fields: map[string]any{envelope.FieldTotalMileageKM: 1.2}, + }, nil) + if err != nil { + t.Fatalf("SamplesFromEnvelope() error = %v", err) + } + if len(samples) != 0 { + t.Fatalf("expected no samples without vin, got %#v", samples) + } + + samples, err = SamplesFromEnvelope(envelope.FrameEnvelope{ + Protocol: envelope.ProtocolGB32960, + VIN: "LNBVIN00000000002", + Fields: map[string]any{}, + }, nil) + if err != nil { + t.Fatalf("SamplesFromEnvelope() error = %v", err) + } + if len(samples) != 0 { + t.Fatalf("expected no samples without mileage, got %#v", samples) + } +} + +func TestWriterEnsuresSchemaAndUpsertsTwoMetrics(t *testing.T) { + exec := &recordingExec{} + writer := NewWriter(exec, time.FixedZone("Asia/Shanghai", 8*3600)) + if err := writer.EnsureSchema(context.Background()); err != nil { + t.Fatalf("EnsureSchema() error = %v", err) + } + if err := writer.Append(context.Background(), envelope.FrameEnvelope{ + Protocol: envelope.ProtocolGB32960, + VIN: "LNBVIN00000000002", + EventTimeMS: time.Date(2026, 7, 1, 9, 0, 0, 0, time.FixedZone("Asia/Shanghai", 8*3600)).UnixMilli(), + Fields: map[string]any{ + envelope.FieldTotalMileageKM: "10000.0", + }, + }); err != nil { + t.Fatalf("Append() error = %v", err) + } + + if !strings.Contains(exec.calls[0].query, "CREATE TABLE IF NOT EXISTS vehicle_daily_metric") { + t.Fatalf("unexpected schema sql: %s", exec.calls[0].query) + } + if len(exec.calls) != 3 { + t.Fatalf("exec calls = %d", len(exec.calls)) + } + if !strings.Contains(exec.calls[1].query, "ON DUPLICATE KEY UPDATE") { + t.Fatalf("unexpected upsert sql: %s", exec.calls[1].query) + } +} + +type execCall struct { + query string + args []any +} + +type recordingExec struct { + calls []execCall +} + +func (e *recordingExec) ExecContext(_ context.Context, query string, args ...any) (sql.Result, error) { + e.calls = append(e.calls, execCall{query: query, args: args}) + return nil, nil +} diff --git a/go/vehicle-gateway/internal/stats/schema.go b/go/vehicle-gateway/internal/stats/schema.go new file mode 100644 index 00000000..7bf860e8 --- /dev/null +++ b/go/vehicle-gateway/internal/stats/schema.go @@ -0,0 +1,20 @@ +package stats + +const DailyMetricTableSQL = `CREATE TABLE IF NOT EXISTS vehicle_daily_metric ( + id BIGINT PRIMARY KEY AUTO_INCREMENT, + vin VARCHAR(32) NOT NULL, + stat_date DATE NOT NULL, + protocol VARCHAR(32) NOT NULL, + metric_key VARCHAR(64) NOT NULL, + metric_value DECIMAL(18,3) NOT NULL, + metric_unit VARCHAR(16) NOT NULL, + first_total_mileage_km DECIMAL(18,3) NULL, + latest_total_mileage_km DECIMAL(18,3) NULL, + sample_count BIGINT NOT NULL DEFAULT 0, + calculation_method VARCHAR(64) NOT NULL, + created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, + updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, + UNIQUE KEY uk_daily_metric (vin, stat_date, protocol, metric_key), + KEY idx_stat_date (stat_date), + KEY idx_protocol_metric (protocol, metric_key) +)`