From abfed27846421819db4df20bd2e5ff77baec34b0 Mon Sep 17 00:00:00 2001 From: lingniu Date: Wed, 8 Jul 2026 12:01:18 +0800 Subject: [PATCH] feat(stats): add daily last mileage backfill --- go/vehicle-gateway/cmd/stats-backfill/main.go | 693 ++++++++++++++++++ 1 file changed, 693 insertions(+) create mode 100644 go/vehicle-gateway/cmd/stats-backfill/main.go diff --git a/go/vehicle-gateway/cmd/stats-backfill/main.go b/go/vehicle-gateway/cmd/stats-backfill/main.go new file mode 100644 index 00000000..1bcb785a --- /dev/null +++ b/go/vehicle-gateway/cmd/stats-backfill/main.go @@ -0,0 +1,693 @@ +package main + +import ( + "context" + "database/sql" + "encoding/json" + "fmt" + "log/slog" + "os" + "regexp" + "strconv" + "strings" + "time" + + _ "github.com/go-sql-driver/mysql" + _ "github.com/taosdata/driver-go/v3/taosWS" + + "lingniu-vehicle-ingest/go/vehicle-gateway/internal/envelope" + "lingniu-vehicle-ingest/go/vehicle-gateway/internal/history" + "lingniu-vehicle-ingest/go/vehicle-gateway/internal/realtime" + "lingniu-vehicle-ingest/go/vehicle-gateway/internal/stats" +) + +type config struct { + MySQLDSN string + TDengineDriver string + TDengineDSN string + TDengineDatabase string + DateFrom string + DateTo string + Protocols []envelope.Protocol + Method string + Limit int + DryRun bool + Reset bool + Debug bool + ProgressEvery int64 + Location *time.Location +} + +type rawFrameRow struct { + Protocol envelope.Protocol + VIN string + EventID string + MessageID string + EventTimeMS int64 + ReceivedAtMS int64 + ParsedJSON string +} + +type metricAgg struct { + VIN string + Date string + Protocol envelope.Protocol + FirstKM float64 + LatestKM float64 + Count int64 +} + +func main() { + if err := loadEnvFiles(env("BACKFILL_ENV_FILES", os.Getenv("BACKFILL_ENV_FILE"))); err != nil { + fail("load env file", err) + } + cfg, err := loadConfig() + if err != nil { + fail("load config", err) + } + ctx := context.Background() + + td, err := sql.Open(cfg.TDengineDriver, cfg.TDengineDSN) + if err != nil { + fail("open tdengine", err) + } + defer td.Close() + if err := td.PingContext(ctx); err != nil { + fail("ping tdengine", err) + } + + mysqlDB, err := sql.Open("mysql", cfg.MySQLDSN) + if err != nil { + fail("open mysql", err) + } + defer mysqlDB.Close() + if err := mysqlDB.PingContext(ctx); err != nil { + fail("ping mysql", err) + } + + schemaWriter := stats.NewWriter(mysqlDB, cfg.Location) + if err := schemaWriter.EnsureSchema(ctx); err != nil { + fail("ensure schema", err) + } + if cfg.Reset && !cfg.DryRun { + deleted, err := resetStats(ctx, mysqlDB, cfg) + if err != nil { + fail("reset stats", err) + } + slog.Info("reset stats rows", "deleted", deleted) + } + if cfg.Method == "last_diff" { + aggregates, err := buildLastDiffAggregates(ctx, td, cfg) + if err != nil { + fail("build last-diff aggregates", err) + } + var written int64 + if !cfg.DryRun { + written, err = writeAggregates(ctx, mysqlDB, aggregates, 500) + if err != nil { + fail("write aggregates", err) + } + } + slog.Info("stats backfill complete", "method", cfg.Method, "dry_run", cfg.DryRun, "aggregates", len(aggregates), "written", written) + return + } + + rows, err := queryRawFrames(ctx, td, cfg) + if err != nil { + fail("query raw frames", err) + } + defer rows.Close() + + aggregates := map[string]*metricAgg{} + var scanned, chunked, parsed, sampled, written, skipped int64 + for rows.Next() { + scanned++ + row, err := scanRawFrame(rows) + if err != nil { + fail("scan raw frame", err) + } + text := strings.TrimSpace(row.ParsedJSON) + if isChunkManifest(text) { + chunked++ + text, err = readChunkedPayload(ctx, td, cfg.TDengineDatabase, row.EventID, "parsed_fields") + if err != nil { + slog.Warn("skip chunked parsed fields", "event_id", row.EventID, "error", err) + skipped++ + continue + } + } + fields := fieldsForStats(row.Protocol, row.VIN, text) + if cfg.Debug && scanned <= 5 { + slog.Info("debug raw frame", + "protocol", row.Protocol, + "vin", row.VIN, + "message_id", row.MessageID, + "parsed_len", len(text), + "mileage_keys", mileageKeys(row.Protocol), + "extracted_fields", fields) + } + if len(fields) == 0 { + skipped++ + continue + } + parsed++ + env := envelope.FrameEnvelope{ + EventID: row.EventID, + Protocol: row.Protocol, + MessageID: row.MessageID, + VIN: row.VIN, + EventTimeMS: row.EventTimeMS, + ReceivedAtMS: row.ReceivedAtMS, + Fields: fields, + ParsedFields: fields, + ParseStatus: envelope.ParseOK, + } + samples, err := stats.SamplesFromEnvelope(env, cfg.Location) + if err != nil { + slog.Warn("skip invalid sample", "event_id", row.EventID, "error", err) + skipped++ + continue + } + if len(samples) == 0 { + skipped++ + continue + } + sampled += int64(len(samples)) + addSamples(aggregates, samples) + if cfg.ProgressEvery > 0 && scanned%cfg.ProgressEvery == 0 { + slog.Info("stats backfill progress", "scanned", scanned, "sampled", sampled, "aggregates", len(aggregates), "skipped", skipped) + } + } + if err := rows.Err(); err != nil { + fail("iterate raw frames", err) + } + if !cfg.DryRun { + var err error + written, err = writeAggregates(ctx, mysqlDB, aggregates, 500) + if err != nil { + fail("write aggregates", err) + } + } + slog.Info("stats backfill complete", "dry_run", cfg.DryRun, "scanned", scanned, "parsed", parsed, "chunked", chunked, "sampled", sampled, "aggregates", len(aggregates), "written", written, "skipped", skipped) +} + +func addSamples(aggregates map[string]*metricAgg, samples []stats.MetricSample) { + for _, sample := range samples { + key := sample.VIN + "|" + sample.StatDate + "|" + string(sample.Protocol) + agg, ok := aggregates[key] + if !ok { + aggregates[key] = &metricAgg{ + VIN: sample.VIN, + Date: sample.StatDate, + Protocol: sample.Protocol, + FirstKM: sample.TotalMileageKM, + LatestKM: sample.TotalMileageKM, + Count: 1, + } + continue + } + if sample.TotalMileageKM < agg.FirstKM { + agg.FirstKM = sample.TotalMileageKM + } + if sample.TotalMileageKM > agg.LatestKM { + agg.LatestKM = sample.TotalMileageKM + } + agg.Count++ + } +} + +func writeAggregates(ctx context.Context, db *sql.DB, aggregates map[string]*metricAgg, batchSize int) (int64, error) { + if len(aggregates) == 0 { + return 0, nil + } + if batchSize <= 0 { + batchSize = 500 + } + rows := make([]*metricAgg, 0, len(aggregates)) + for _, agg := range aggregates { + rows = append(rows, agg) + } + var written int64 + for start := 0; start < len(rows); start += batchSize { + end := start + batchSize + if end > len(rows) { + end = len(rows) + } + if err := writeAggregateBatch(ctx, db, rows[start:end]); err != nil { + return written, err + } + written += int64(end - start) + } + return written, nil +} + +func writeAggregateBatch(ctx context.Context, db *sql.DB, rows []*metricAgg) error { + placeholders := make([]string, 0, len(rows)) + args := make([]any, 0, len(rows)*7) + for _, row := range rows { + daily := row.LatestKM - row.FirstKM + if daily < 0 { + daily = 0 + } + placeholders = append(placeholders, "(?,?,?,?,?,?,?)") + args = append(args, + row.VIN, + row.Date, + string(row.Protocol), + daily, + row.FirstKM, + row.LatestKM, + row.Count, + ) + } + sqlText := `INSERT INTO vehicle_daily_mileage + (vin, stat_date, protocol, daily_mileage_km, first_total_mileage_km, latest_total_mileage_km, sample_count) +VALUES ` + strings.Join(placeholders, ",") + ` +ON DUPLICATE KEY UPDATE + daily_mileage_km = VALUES(daily_mileage_km), + first_total_mileage_km = VALUES(first_total_mileage_km), + latest_total_mileage_km = VALUES(latest_total_mileage_km), + sample_count = VALUES(sample_count), + updated_at = CURRENT_TIMESTAMP` + _, err := db.ExecContext(ctx, sqlText, args...) + return err +} + +func buildLastDiffAggregates(ctx context.Context, db *sql.DB, cfg config) (map[string]*metricAgg, error) { + dates, err := dateRangeWithPrevious(cfg.DateFrom, cfg.DateTo) + if err != nil { + return nil, err + } + lastByProtocolDate := map[envelope.Protocol]map[string]map[string]float64{} + for _, protocol := range cfg.Protocols { + lastByProtocolDate[protocol] = map[string]map[string]float64{} + for _, date := range dates { + rows, err := queryDailyLastRows(ctx, db, cfg, protocol, date) + if err != nil { + return nil, err + } + lastByProtocolDate[protocol][date] = rows + slog.Info("daily last loaded", "protocol", protocol, "date", date, "vehicles", len(rows)) + } + } + targetDates, err := dateRange(cfg.DateFrom, cfg.DateTo) + if err != nil { + return nil, err + } + aggregates := map[string]*metricAgg{} + for _, protocol := range cfg.Protocols { + for _, date := range targetDates { + prevDate := previousDate(date) + current := lastByProtocolDate[protocol][date] + previous := lastByProtocolDate[protocol][prevDate] + for vin, latest := range current { + first, ok := previous[vin] + if !ok || latest < first { + continue + } + key := vin + "|" + date + "|" + string(protocol) + aggregates[key] = &metricAgg{ + VIN: vin, + Date: date, + Protocol: protocol, + FirstKM: first, + LatestKM: latest, + Count: 1, + } + } + } + } + return aggregates, nil +} + +func queryDailyLastRows(ctx context.Context, db *sql.DB, cfg config, protocol envelope.Protocol, date string) (map[string]float64, error) { + where := []string{ + fmt.Sprintf("ts >= '%s 00:00:00'", quote(date)), + fmt.Sprintf("ts < '%s 00:00:00'", quote(nextDate(date))), + "parse_status = 'OK'", + "vin IS NOT NULL", + "vin <> ''", + fmt.Sprintf("protocol = '%s'", quote(string(protocol))), + realtimeMileageFramePredicate(), + } + sqlText := fmt.Sprintf(`SELECT vin, LAST(parsed_json) +FROM %s.raw_frames +WHERE %s +GROUP BY vin`, ident(cfg.TDengineDatabase), strings.Join(where, " AND ")) + rows, err := db.QueryContext(ctx, sqlText) + if err != nil { + return nil, err + } + defer rows.Close() + out := map[string]float64{} + for rows.Next() { + var vin string + var parsedJSON string + if err := rows.Scan(&vin, &parsedJSON); err != nil { + return nil, err + } + fields := fieldsForStats(protocol, vin, parsedJSON) + env := envelope.FrameEnvelope{ + Protocol: protocol, + VIN: strings.TrimSpace(vin), + EventTimeMS: time.Now().UnixMilli(), + ReceivedAtMS: time.Now().UnixMilli(), + Fields: fields, + ParsedFields: fields, + ParseStatus: envelope.ParseOK, + } + samples, err := stats.SamplesFromEnvelope(env, cfg.Location) + if err != nil || len(samples) == 0 { + continue + } + out[strings.TrimSpace(vin)] = samples[0].TotalMileageKM + } + return out, rows.Err() +} + +func fieldsForStats(protocol envelope.Protocol, vin string, text string) map[string]any { + fields := map[string]any{} + text = strings.TrimSpace(text) + if text == "" { + return fields + } + parsed := map[string]any{} + if err := json.Unmarshal([]byte(text), &parsed); err == nil && len(parsed) > 0 { + if flattened, _, ok := realtime.ParsedFieldsForEnvelope(envelope.FrameEnvelope{ + Protocol: protocol, + VIN: vin, + Parsed: parsed, + }); ok { + fields = flattened + } else { + fields = parsed + } + } + for _, key := range mileageKeys(protocol) { + if _, exists := fields[key]; exists { + continue + } + value, ok := extractJSONStringField(text, key) + if !ok { + continue + } + fields[key] = value + } + return fields +} + +func mileageKeys(protocol envelope.Protocol) []string { + switch protocol { + case envelope.ProtocolGB32960: + return []string{"gb32960.vehicle.total_mileage_km"} + case envelope.ProtocolJT808: + return []string{"jt808.location.total_mileage_km"} + case envelope.ProtocolYutongMQTT: + return []string{"yutong_mqtt.data.total_mileage", "yutong_mqtt.root.data.total_mileage"} + default: + return nil + } +} + +func extractJSONStringField(text string, key string) (string, bool) { + pattern := regexp.QuoteMeta(`"`+key+`"`) + `\s*:\s*"?([^",}]+)"?` + match := regexp.MustCompile(pattern).FindStringSubmatch(text) + if len(match) != 2 { + return "", false + } + return strings.TrimSpace(match[1]), true +} + +func loadConfig() (config, error) { + loc, err := time.LoadLocation(env("LOCAL_TZ", "Asia/Shanghai")) + if err != nil { + loc = time.FixedZone("Asia/Shanghai", 8*3600) + } + dateTo := env("BACKFILL_DATE_TO", time.Now().In(loc).Format("2006-01-02")) + dateFrom := env("BACKFILL_DATE_FROM", dateTo) + protocols, err := parseProtocols(env("BACKFILL_PROTOCOLS", "GB32960,JT808,YUTONG_MQTT")) + if err != nil { + return config{}, err + } + return config{ + MySQLDSN: env("MYSQL_DSN", ""), + TDengineDriver: env("TDENGINE_DRIVER", "taosWS"), + TDengineDSN: env("TDENGINE_DSN", ""), + TDengineDatabase: env("TDENGINE_DATABASE", history.DefaultDatabase), + DateFrom: dateFrom, + DateTo: dateTo, + Protocols: protocols, + Method: env("BACKFILL_METHOD", "scan"), + Limit: envInt("BACKFILL_LIMIT", 0), + DryRun: envBool("BACKFILL_DRY_RUN", true), + Reset: envBool("BACKFILL_RESET", false), + Debug: envBool("BACKFILL_DEBUG", false), + ProgressEvery: int64(envInt("BACKFILL_PROGRESS_EVERY", 100000)), + Location: loc, + }, nil +} + +func queryRawFrames(ctx context.Context, db *sql.DB, cfg config) (*sql.Rows, error) { + where := []string{ + fmt.Sprintf("ts >= '%s 00:00:00'", quote(cfg.DateFrom)), + fmt.Sprintf("ts < '%s 00:00:00'", quote(nextDate(cfg.DateTo))), + "parse_status = 'OK'", + "vin IS NOT NULL", + "vin <> ''", + } + if len(cfg.Protocols) > 0 { + quoted := make([]string, 0, len(cfg.Protocols)) + for _, protocol := range cfg.Protocols { + quoted = append(quoted, "'"+quote(string(protocol))+"'") + } + where = append(where, "protocol IN ("+strings.Join(quoted, ",")+")") + } + where = append(where, realtimeMileageFramePredicate()) + sqlText := fmt.Sprintf(`SELECT protocol, vin, event_id, message_id, event_time, received_at, parsed_json +FROM %s.raw_frames +WHERE %s +ORDER BY ts ASC`, ident(cfg.TDengineDatabase), strings.Join(where, " AND ")) + if cfg.Limit > 0 { + sqlText += fmt.Sprintf(" LIMIT %d", cfg.Limit) + } + slog.Info("query raw frames", "date_from", cfg.DateFrom, "date_to", cfg.DateTo, "protocols", cfg.Protocols, "limit", cfg.Limit) + return db.QueryContext(ctx, sqlText) +} + +func realtimeMileageFramePredicate() string { + return `( + (protocol = 'GB32960' AND message_id IN (2,3)) + OR (protocol = 'JT808' AND message_id = 512) + OR (protocol = 'YUTONG_MQTT') +)` +} + +func scanRawFrame(rows *sql.Rows) (rawFrameRow, error) { + var protocol, vin, eventID, parsed string + var messageID int64 + var eventTime, receivedAt time.Time + if err := rows.Scan(&protocol, &vin, &eventID, &messageID, &eventTime, &receivedAt, &parsed); err != nil { + return rawFrameRow{}, err + } + return rawFrameRow{ + Protocol: envelope.Protocol(strings.TrimSpace(protocol)), + VIN: strings.TrimSpace(vin), + EventID: strings.TrimSpace(eventID), + MessageID: fmt.Sprintf("0x%04X", messageID), + EventTimeMS: eventTime.UnixMilli(), + ReceivedAtMS: receivedAt.UnixMilli(), + ParsedJSON: parsed, + }, nil +} + +func isChunkManifest(text string) bool { + if !strings.Contains(text, "chunked") || !strings.Contains(text, "payload_kind") { + return false + } + var manifest struct { + Chunked bool `json:"chunked"` + } + return json.Unmarshal([]byte(text), &manifest) == nil && manifest.Chunked +} + +func readChunkedPayload(ctx context.Context, db *sql.DB, database string, eventID string, kind string) (string, error) { + sqlText := fmt.Sprintf(`SELECT chunk_text FROM %s.raw_frame_payload_chunks +WHERE event_id = '%s' AND payload_kind = '%s' +ORDER BY chunk_index ASC`, ident(database), quote(eventID), quote(kind)) + rows, err := db.QueryContext(ctx, sqlText) + if err != nil { + return "", err + } + defer rows.Close() + var b strings.Builder + for rows.Next() { + var chunk string + if err := rows.Scan(&chunk); err != nil { + return "", err + } + b.WriteString(chunk) + } + if err := rows.Err(); err != nil { + return "", err + } + if b.Len() == 0 { + return "", fmt.Errorf("no chunks found") + } + return b.String(), nil +} + +func resetStats(ctx context.Context, db *sql.DB, cfg config) (int64, error) { + clauses := []string{"stat_date >= ?", "stat_date <= ?"} + args := []any{cfg.DateFrom, cfg.DateTo} + if len(cfg.Protocols) > 0 { + placeholders := make([]string, 0, len(cfg.Protocols)) + for _, protocol := range cfg.Protocols { + placeholders = append(placeholders, "?") + args = append(args, string(protocol)) + } + clauses = append(clauses, "protocol IN ("+strings.Join(placeholders, ",")+")") + } + result, err := db.ExecContext(ctx, "DELETE FROM vehicle_daily_mileage WHERE "+strings.Join(clauses, " AND "), args...) + if err != nil { + return 0, err + } + return result.RowsAffected() +} + +func parseProtocols(value string) ([]envelope.Protocol, error) { + var protocols []envelope.Protocol + for _, part := range strings.Split(value, ",") { + part = strings.ToUpper(strings.TrimSpace(part)) + if part == "" { + continue + } + switch envelope.Protocol(part) { + case envelope.ProtocolGB32960, envelope.ProtocolJT808, envelope.ProtocolYutongMQTT: + protocols = append(protocols, envelope.Protocol(part)) + default: + return nil, fmt.Errorf("unsupported protocol %q", part) + } + } + return protocols, nil +} + +func loadEnvFiles(paths string) error { + for _, path := range strings.Split(paths, ",") { + if err := loadEnvFile(path); err != nil { + return err + } + } + return nil +} + +func loadEnvFile(path string) error { + path = strings.TrimSpace(path) + if path == "" { + return nil + } + data, err := os.ReadFile(path) + if err != nil { + return err + } + for _, line := range strings.Split(string(data), "\n") { + line = strings.TrimSpace(line) + if line == "" || strings.HasPrefix(line, "#") || !strings.Contains(line, "=") { + continue + } + key, value, _ := strings.Cut(line, "=") + key = strings.TrimSpace(key) + value = strings.Trim(strings.TrimSpace(value), "\"'") + if key != "" && os.Getenv(key) == "" { + _ = os.Setenv(key, value) + } + } + return nil +} + +func env(key string, fallback string) string { + value := strings.TrimSpace(os.Getenv(key)) + if value == "" { + return fallback + } + return value +} + +func envInt(key string, fallback int) int { + value := strings.TrimSpace(os.Getenv(key)) + if value == "" { + return fallback + } + parsed, err := strconv.Atoi(value) + if err != nil { + return fallback + } + return parsed +} + +func envBool(key string, fallback bool) bool { + value := strings.ToLower(strings.TrimSpace(os.Getenv(key))) + if value == "" { + return fallback + } + return value == "1" || value == "true" || value == "yes" || value == "y" +} + +func nextDate(date string) string { + parsed, err := time.Parse("2006-01-02", date) + if err != nil { + return date + } + return parsed.AddDate(0, 0, 1).Format("2006-01-02") +} + +func previousDate(date string) string { + parsed, err := time.Parse("2006-01-02", date) + if err != nil { + return date + } + return parsed.AddDate(0, 0, -1).Format("2006-01-02") +} + +func dateRangeWithPrevious(from string, to string) ([]string, error) { + dates, err := dateRange(from, to) + if err != nil { + return nil, err + } + return append([]string{previousDate(from)}, dates...), nil +} + +func dateRange(from string, to string) ([]string, error) { + start, err := time.Parse("2006-01-02", from) + if err != nil { + return nil, err + } + end, err := time.Parse("2006-01-02", to) + if err != nil { + return nil, err + } + if end.Before(start) { + return nil, fmt.Errorf("dateTo before dateFrom") + } + var dates []string + for day := start; !day.After(end); day = day.AddDate(0, 0, 1) { + dates = append(dates, day.Format("2006-01-02")) + } + return dates, nil +} + +func ident(value string) string { + value = strings.TrimSpace(strings.Trim(value, "`")) + if value == "" { + return history.DefaultDatabase + } + return value +} + +func quote(value string) string { + return strings.ReplaceAll(value, "'", "''") +} + +func fail(stage string, err error) { + slog.Error(stage, "error", err) + os.Exit(1) +}