shishengliang/lingniu-vehicle-ingest
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(stats): backfill mileage source candidates

Browse Source
This commit is contained in:
lingniu
2026-07-08 14:58:19 +08:00
parent 762c7265d7
commit 804c238bff
2 changed files with 193 additions and 140 deletions
Download Patch File Download Diff File Expand all files Collapse all files

219
go/vehicle-gateway/cmd/stats-backfill/main.go
View File

@@ -41,6 +41,9 @@ type config struct {
type rawFrameRow struct { type rawFrameRow struct {
Protocol envelope.Protocol Protocol envelope.Protocol
VIN string VIN string
Phone string
DeviceID string
SourceEndpoint string
EventID string EventID string
MessageID string MessageID string
EventTimeMS int64 EventTimeMS int64
@@ -57,13 +60,19 @@ type metricAgg struct {
Count int64 Count int64
SourceKey string SourceKey string
Phone string Phone string
DeviceID string
SourceEndpoint string SourceEndpoint string
FirstEventTime time.Time
LatestEventTime time.Time
QualityStatus string
QualityReason string
} }
type dailySourceLast struct { type dailySourceLast struct {
VIN string VIN string
SourceKey string SourceKey string
Phone string Phone string
DeviceID string
SourceEndpoint string SourceEndpoint string
TS time.Time TS time.Time
TotalKM float64 TotalKM float64
@@ -168,6 +177,9 @@ func main() {
Protocol: row.Protocol, Protocol: row.Protocol,
MessageID: row.MessageID, MessageID: row.MessageID,
VIN: row.VIN, VIN: row.VIN,
Phone: row.Phone,
DeviceID: row.DeviceID,
SourceEndpoint: row.SourceEndpoint,
EventTimeMS: row.EventTimeMS, EventTimeMS: row.EventTimeMS,
ReceivedAtMS: row.ReceivedAtMS, ReceivedAtMS: row.ReceivedAtMS,
Fields: fields, Fields: fields,
@@ -205,7 +217,7 @@ func main() {
func addSamples(aggregates map[string]*metricAgg, samples []stats.MetricSample) { func addSamples(aggregates map[string]*metricAgg, samples []stats.MetricSample) {
for _, sample := range samples { for _, sample := range samples {
key := sample.VIN + "|" + sample.StatDate + "|" + string(sample.Protocol) key := sample.VIN + "|" + sample.StatDate + "|" + string(sample.Protocol) + "|" + sample.SourceKey
agg, ok := aggregates[key] agg, ok := aggregates[key]
if !ok { if !ok {
aggregates[key] = &metricAgg{ aggregates[key] = &metricAgg{
@@ -215,6 +227,14 @@ func addSamples(aggregates map[string]*metricAgg, samples []stats.MetricSample)
FirstKM: sample.TotalMileageKM, FirstKM: sample.TotalMileageKM,
LatestKM: sample.TotalMileageKM, LatestKM: sample.TotalMileageKM,
Count: 1, Count: 1,
SourceKey: sample.SourceKey,
Phone: sample.Phone,
DeviceID: sample.DeviceID,
SourceEndpoint: sample.SourceEndpoint,
FirstEventTime: sample.EventTime,
LatestEventTime: sample.EventTime,
QualityStatus: stats.QualityOK,
QualityReason: "scan_backfill",
} }
continue continue
} }
@@ -224,6 +244,19 @@ func addSamples(aggregates map[string]*metricAgg, samples []stats.MetricSample)
if sample.TotalMileageKM > agg.LatestKM { if sample.TotalMileageKM > agg.LatestKM {
agg.LatestKM = sample.TotalMileageKM agg.LatestKM = sample.TotalMileageKM
} }
if agg.FirstEventTime.IsZero() || sample.EventTime.Before(agg.FirstEventTime) {
agg.FirstEventTime = sample.EventTime
}
if sample.EventTime.After(agg.LatestEventTime) {
agg.LatestEventTime = sample.EventTime
agg.SourceEndpoint = sample.SourceEndpoint
if strings.TrimSpace(sample.Phone) != "" {
agg.Phone = sample.Phone
}
if strings.TrimSpace(sample.DeviceID) != "" {
agg.DeviceID = sample.DeviceID
}
}
agg.Count++ agg.Count++
} }
} }
@@ -232,66 +265,56 @@ func writeAggregates(ctx context.Context, db *sql.DB, aggregates map[string]*met
if len(aggregates) == 0 { if len(aggregates) == 0 {
return 0, nil 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 var written int64
for start := 0; start < len(rows); start += batchSize { for _, agg := range aggregates {
end := start + batchSize identity := stats.SourceIdentity{
if end > len(rows) { Protocol: agg.Protocol,
end = len(rows) SourceIP: stats.NormalizeSourceIP(agg.SourceEndpoint),
SourceEndpoint: agg.SourceEndpoint,
} }
if err := writeAggregateBatch(ctx, db, rows[start:end]); err != nil { if err := stats.UpsertDataSource(ctx, db, identity, agg.LatestEventTime); err != nil {
return written, err return written, err
} }
written += int64(end - start) dailyKM := agg.LatestKM - agg.FirstKM
candidate := stats.SourceMileageSample{
VIN: agg.VIN,
StatDate: agg.Date,
Protocol: agg.Protocol,
SourceKey: agg.SourceKey,
SourceIP: identity.SourceIP,
SourceEndpoint: agg.SourceEndpoint,
Phone: agg.Phone,
DeviceID: agg.DeviceID,
FirstTotalKM: agg.FirstKM,
LatestTotalKM: agg.LatestKM,
DailyKM: dailyKM,
SampleCount: agg.Count,
FirstEventTime: agg.FirstEventTime,
LatestEventTime: agg.LatestEventTime,
QualityStatus: agg.QualityStatus,
QualityReason: agg.QualityReason,
}
if candidate.QualityStatus == "" {
candidate.QualityStatus = stats.QualityOK
}
if candidate.QualityReason == "" {
candidate.QualityReason = "same_source_previous_day"
}
if candidate.QualityStatus == stats.QualityOK && (candidate.DailyKM < 0 || candidate.DailyKM > maxTrustedDailyMileageKM) {
candidate.QualityStatus = stats.QualityInvalidDelta
candidate.QualityReason = "outside_daily_range"
}
if err := stats.UpsertSourceMileage(ctx, db, candidate); err != nil {
return written, err
}
if err := stats.ProjectDailyMileage(ctx, db, agg.VIN, agg.Date, agg.Protocol); err != nil {
return written, err
}
written++
} }
return written, nil 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.SourceKey,
row.Phone,
row.SourceEndpoint,
row.Count,
)
}
sqlText := `INSERT INTO vehicle_daily_mileage
(vin, stat_date, protocol, daily_mileage_km, first_total_mileage_km, latest_total_mileage_km,
trusted_source_key, trusted_phone, trusted_source_endpoint, 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),
trusted_source_key = VALUES(trusted_source_key),
trusted_phone = VALUES(trusted_phone),
trusted_source_endpoint = VALUES(trusted_source_endpoint),
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) { func buildLastDiffAggregates(ctx context.Context, db *sql.DB, cfg config) (map[string]*metricAgg, error) {
dates, err := dateRangeWithPrevious(cfg.DateFrom, cfg.DateTo) dates, err := dateRangeWithPrevious(cfg.DateFrom, cfg.DateTo)
if err != nil { if err != nil {
@@ -320,21 +343,35 @@ func buildLastDiffAggregates(ctx context.Context, db *sql.DB, cfg config) (map[s
current := lastByProtocolDate[protocol][date] current := lastByProtocolDate[protocol][date]
previous := lastByProtocolDate[protocol][prevDate] previous := lastByProtocolDate[protocol][prevDate]
for vin, currentSources := range current { for vin, currentSources := range current {
chosen, ok := chooseTrustedSource(currentSources, previous[vin]) previousBySource := map[string]dailySourceLast{}
if !ok { for _, previousRow := range previous[vin] {
continue previousBySource[previousRow.SourceKey] = previousRow
} }
key := vin + "|" + date + "|" + string(protocol) for _, currentRow := range currentSources {
aggregates[key] = &metricAgg{ key := vin + "|" + date + "|" + string(protocol) + "|" + currentRow.SourceKey
agg := &metricAgg{
VIN: vin, VIN: vin,
Date: date, Date: date,
Protocol: protocol, Protocol: protocol,
FirstKM: chosen.previous.TotalKM, FirstKM: currentRow.TotalKM,
LatestKM: chosen.current.TotalKM, LatestKM: currentRow.TotalKM,
Count: 1, Count: 1,
SourceKey: chosen.current.SourceKey, SourceKey: currentRow.SourceKey,
Phone: chosen.current.Phone, Phone: currentRow.Phone,
SourceEndpoint: chosen.current.SourceEndpoint, DeviceID: currentRow.DeviceID,
SourceEndpoint: currentRow.SourceEndpoint,
FirstEventTime: currentRow.TS,
LatestEventTime: currentRow.TS,
QualityStatus: stats.QualityNoPreviousBaseline,
QualityReason: "missing_previous_source",
}
if previousRow, ok := previousBySource[currentRow.SourceKey]; ok {
agg.FirstKM = previousRow.TotalKM
agg.FirstEventTime = previousRow.TS
agg.QualityStatus = stats.QualityOK
agg.QualityReason = "same_source_previous_day"
}
aggregates[key] = agg
} }
} }
} }
@@ -383,10 +420,10 @@ func queryDailyLastSourceRows(ctx context.Context, db *sql.DB, cfg config, proto
fmt.Sprintf("protocol = '%s'", quote(string(protocol))), fmt.Sprintf("protocol = '%s'", quote(string(protocol))),
realtimeMileageFramePredicate(), realtimeMileageFramePredicate(),
} }
sqlText := fmt.Sprintf(`SELECT vin, phone, source_endpoint, LAST(ts), LAST(parsed_json) sqlText := fmt.Sprintf(`SELECT vin, phone, device_id, source_endpoint, LAST(ts), LAST(parsed_json)
FROM %s.raw_frames FROM %s.raw_frames
WHERE %s WHERE %s
GROUP BY vin, phone, source_endpoint`, ident(cfg.TDengineDatabase), strings.Join(where, " AND ")) GROUP BY vin, phone, device_id, source_endpoint`, ident(cfg.TDengineDatabase), strings.Join(where, " AND "))
rows, err := db.QueryContext(ctx, sqlText) rows, err := db.QueryContext(ctx, sqlText)
if err != nil { if err != nil {
return nil, err return nil, err
@@ -396,10 +433,11 @@ GROUP BY vin, phone, source_endpoint`, ident(cfg.TDengineDatabase), strings.Join
for rows.Next() { for rows.Next() {
var vin string var vin string
var phone string var phone string
var deviceID string
var sourceEndpoint string var sourceEndpoint string
var ts time.Time var ts time.Time
var parsedJSON string var parsedJSON string
if err := rows.Scan(&vin, &phone, &sourceEndpoint, &ts, &parsedJSON); err != nil { if err := rows.Scan(&vin, &phone, &deviceID, &sourceEndpoint, &ts, &parsedJSON); err != nil {
return nil, err return nil, err
} }
fields := fieldsForStats(protocol, vin, parsedJSON) fields := fieldsForStats(protocol, vin, parsedJSON)
@@ -416,7 +454,7 @@ GROUP BY vin, phone, source_endpoint`, ident(cfg.TDengineDatabase), strings.Join
if err != nil || len(samples) == 0 { if err != nil || len(samples) == 0 {
continue continue
} }
sourceKey := normalizedSourceKey(phone, sourceEndpoint) sourceKey := normalizedSourceKey(string(protocol), phone, deviceID, sourceEndpoint)
if sourceKey == "" { if sourceKey == "" {
continue continue
} }
@@ -424,6 +462,7 @@ GROUP BY vin, phone, source_endpoint`, ident(cfg.TDengineDatabase), strings.Join
VIN: strings.TrimSpace(vin), VIN: strings.TrimSpace(vin),
SourceKey: sourceKey, SourceKey: sourceKey,
Phone: strings.TrimSpace(phone), Phone: strings.TrimSpace(phone),
DeviceID: strings.TrimSpace(deviceID),
SourceEndpoint: strings.TrimSpace(sourceEndpoint), SourceEndpoint: strings.TrimSpace(sourceEndpoint),
TS: ts, TS: ts,
TotalKM: samples[0].TotalMileageKM, TotalKM: samples[0].TotalMileageKM,
@@ -443,26 +482,9 @@ GROUP BY vin, phone, source_endpoint`, ident(cfg.TDengineDatabase), strings.Join
return out, nil return out, nil
} }
func normalizedSourceKey(phone string, endpoint string) string { func normalizedSourceKey(protocol string, phone string, deviceID string, endpoint string) string {
var parts []string sourceIP := stats.NormalizeSourceIP(endpoint)
if phone = strings.TrimSpace(phone); phone != "" { return stats.SourceKey(envelope.Protocol(protocol), phone, deviceID, sourceIP)
parts = append(parts, phone)
}
if host := endpointHost(endpoint); host != "" {
parts = append(parts, host)
}
return strings.Join(parts, "@")
}
func endpointHost(endpoint string) string {
endpoint = strings.TrimSpace(endpoint)
if endpoint == "" {
return ""
}
if host, _, ok := strings.Cut(endpoint, ":"); ok {
return strings.TrimSpace(host)
}
return endpoint
} }
func fieldsForStats(protocol envelope.Protocol, vin string, text string) map[string]any { func fieldsForStats(protocol envelope.Protocol, vin string, text string) map[string]any {
@@ -563,7 +585,7 @@ func queryRawFrames(ctx context.Context, db *sql.DB, cfg config) (*sql.Rows, err
where = append(where, "protocol IN ("+strings.Join(quoted, ",")+")") where = append(where, "protocol IN ("+strings.Join(quoted, ",")+")")
} }
where = append(where, realtimeMileageFramePredicate()) where = append(where, realtimeMileageFramePredicate())
sqlText := fmt.Sprintf(`SELECT protocol, vin, event_id, message_id, event_time, received_at, parsed_json sqlText := fmt.Sprintf(`SELECT protocol, vin, phone, device_id, source_endpoint, event_id, message_id, event_time, received_at, parsed_json
FROM %s.raw_frames FROM %s.raw_frames
WHERE %s WHERE %s
ORDER BY ts ASC`, ident(cfg.TDengineDatabase), strings.Join(where, " AND ")) ORDER BY ts ASC`, ident(cfg.TDengineDatabase), strings.Join(where, " AND "))
@@ -583,15 +605,18 @@ func realtimeMileageFramePredicate() string {
} }
func scanRawFrame(rows *sql.Rows) (rawFrameRow, error) { func scanRawFrame(rows *sql.Rows) (rawFrameRow, error) {
var protocol, vin, eventID, parsed string var protocol, vin, phone, deviceID, sourceEndpoint, eventID, parsed string
var messageID int64 var messageID int64
var eventTime, receivedAt time.Time var eventTime, receivedAt time.Time
if err := rows.Scan(&protocol, &vin, &eventID, &messageID, &eventTime, &receivedAt, &parsed); err != nil { if err := rows.Scan(&protocol, &vin, &phone, &deviceID, &sourceEndpoint, &eventID, &messageID, &eventTime, &receivedAt, &parsed); err != nil {
return rawFrameRow{}, err return rawFrameRow{}, err
} }
return rawFrameRow{ return rawFrameRow{
Protocol: envelope.Protocol(strings.TrimSpace(protocol)), Protocol: envelope.Protocol(strings.TrimSpace(protocol)),
VIN: strings.TrimSpace(vin), VIN: strings.TrimSpace(vin),
Phone: strings.TrimSpace(phone),
DeviceID: strings.TrimSpace(deviceID),
SourceEndpoint: strings.TrimSpace(sourceEndpoint),
EventID: strings.TrimSpace(eventID), EventID: strings.TrimSpace(eventID),
MessageID: fmt.Sprintf("0x%04X", messageID), MessageID: fmt.Sprintf("0x%04X", messageID),
EventTimeMS: eventTime.UnixMilli(), EventTimeMS: eventTime.UnixMilli(),
@@ -647,11 +672,19 @@ func resetStats(ctx context.Context, db *sql.DB, cfg config) (int64, error) {
} }
clauses = append(clauses, "protocol IN ("+strings.Join(placeholders, ",")+")") clauses = append(clauses, "protocol IN ("+strings.Join(placeholders, ",")+")")
} }
result, err := db.ExecContext(ctx, "DELETE FROM vehicle_daily_mileage WHERE "+strings.Join(clauses, " AND "), args...) var deleted int64
for _, table := range []string{"vehicle_daily_mileage_source", "vehicle_daily_mileage"} {
result, err := db.ExecContext(ctx, "DELETE FROM "+table+" WHERE "+strings.Join(clauses, " AND "), args...)
if err != nil { if err != nil {
return 0, err return deleted, err
} }
return result.RowsAffected() affected, err := result.RowsAffected()
if err != nil {
return deleted, err
}
deleted += affected
}
return deleted, nil
} }
func parseProtocols(value string) ([]envelope.Protocol, error) { func parseProtocols(value string) ([]envelope.Protocol, error) {

28
go/vehicle-gateway/cmd/stats-backfill/main_test.go
View File

@@ -5,7 +5,7 @@ import "testing"
func TestChooseTrustedSourceKeepsContinuingSourceAndRejectsNewJump(t *testing.T) { func TestChooseTrustedSourceKeepsContinuingSourceAndRejectsNewJump(t *testing.T) {
previous := []dailySourceLast{{ previous := []dailySourceLast{{
VIN: "LA9GG64L7PBAF4001", VIN: "LA9GG64L7PBAF4001",
SourceKey: normalizedSourceKey("13307765812", "115.231.168.135:42630"), SourceKey: normalizedSourceKey("JT808", "13307765812", "", "115.231.168.135:42630"),
Phone: "13307765812", Phone: "13307765812",
SourceEndpoint: "115.231.168.135:42630", SourceEndpoint: "115.231.168.135:42630",
TotalKM: 4100.8, TotalKM: 4100.8,
@@ -13,14 +13,14 @@ func TestChooseTrustedSourceKeepsContinuingSourceAndRejectsNewJump(t *testing.T)
current := []dailySourceLast{ current := []dailySourceLast{
{ {
VIN: "LA9GG64L7PBAF4001", VIN: "LA9GG64L7PBAF4001",
SourceKey: normalizedSourceKey("13307765812", "115.159.85.149:28316"), SourceKey: normalizedSourceKey("JT808", "13307765812", "", "115.159.85.149:28316"),
Phone: "13307765812", Phone: "13307765812",
SourceEndpoint: "115.159.85.149:28316", SourceEndpoint: "115.159.85.149:28316",
TotalKM: 42447.2, TotalKM: 42447.2,
}, },
{ {
VIN: "LA9GG64L7PBAF4001", VIN: "LA9GG64L7PBAF4001",
SourceKey: normalizedSourceKey("13307765812", "115.231.168.135:20215"), SourceKey: normalizedSourceKey("JT808", "13307765812", "", "115.231.168.135:20215"),
Phone: "13307765812", Phone: "13307765812",
SourceEndpoint: "115.231.168.135:20215", SourceEndpoint: "115.231.168.135:20215",
TotalKM: 4123.9, TotalKM: 4123.9,
@@ -31,10 +31,30 @@ func TestChooseTrustedSourceKeepsContinuingSourceAndRejectsNewJump(t *testing.T)
if !ok { if !ok {
t.Fatal("chooseTrustedSource() did not choose a source") t.Fatal("chooseTrustedSource() did not choose a source")
} }
if chosen.current.SourceKey != normalizedSourceKey("13307765812", "115.231.168.135:20215") { if chosen.current.SourceKey != normalizedSourceKey("JT808", "13307765812", "", "115.231.168.135:20215") {
t.Fatalf("chosen source = %q", chosen.current.SourceKey) t.Fatalf("chosen source = %q", chosen.current.SourceKey)
} }
if delta := chosen.current.TotalKM - chosen.previous.TotalKM; delta < 23 || delta > 24 { if delta := chosen.current.TotalKM - chosen.previous.TotalKM; delta < 23 || delta > 24 {
t.Fatalf("delta = %v, want about 23.1", delta) t.Fatalf("delta = %v, want about 23.1", delta)
} }
} }
func TestDailySourceLastBuildsCandidateKeysBySourceIP(t *testing.T) {
sourceA := dailySourceLast{
VIN: "LA9GG64L7PBAF4001",
SourceKey: normalizedSourceKey("JT808", "13307765812", "", "115.231.168.135:20215"),
Phone: "13307765812",
SourceEndpoint: "115.231.168.135:20215",
TotalKM: 4123.9,
}
sourceB := dailySourceLast{
VIN: "LA9GG64L7PBAF4001",
SourceKey: normalizedSourceKey("JT808", "13307765812", "", "115.231.168.135:42630"),
Phone: "13307765812",
SourceEndpoint: "115.231.168.135:42630",
TotalKM: 4100.8,
}
if sourceA.SourceKey != sourceB.SourceKey {
t.Fatalf("same source IP should produce same source key: %q vs %q", sourceA.SourceKey, sourceB.SourceKey)
}
}