diff --git a/go/vehicle-gateway/cmd/stats-backfill/main.go b/go/vehicle-gateway/cmd/stats-backfill/main.go index 1d11a3d7..743b2cae 100644 --- a/go/vehicle-gateway/cmd/stats-backfill/main.go +++ b/go/vehicle-gateway/cmd/stats-backfill/main.go @@ -39,31 +39,40 @@ type config struct { } type rawFrameRow struct { - Protocol envelope.Protocol - VIN string - EventID string - MessageID string - EventTimeMS int64 - ReceivedAtMS int64 - ParsedJSON string + Protocol envelope.Protocol + VIN string + Phone string + DeviceID string + SourceEndpoint 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 - SourceKey string - Phone string - SourceEndpoint string + VIN string + Date string + Protocol envelope.Protocol + FirstKM float64 + LatestKM float64 + Count int64 + SourceKey string + Phone string + DeviceID string + SourceEndpoint string + FirstEventTime time.Time + LatestEventTime time.Time + QualityStatus string + QualityReason string } type dailySourceLast struct { VIN string SourceKey string Phone string + DeviceID string SourceEndpoint string TS time.Time TotalKM float64 @@ -164,15 +173,18 @@ func main() { } 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, + EventID: row.EventID, + Protocol: row.Protocol, + MessageID: row.MessageID, + VIN: row.VIN, + Phone: row.Phone, + DeviceID: row.DeviceID, + SourceEndpoint: row.SourceEndpoint, + EventTimeMS: row.EventTimeMS, + ReceivedAtMS: row.ReceivedAtMS, + Fields: fields, + ParsedFields: fields, + ParseStatus: envelope.ParseOK, } samples, err := stats.SamplesFromEnvelope(env, cfg.Location) if err != nil { @@ -205,16 +217,24 @@ func main() { func addSamples(aggregates map[string]*metricAgg, samples []stats.MetricSample) { 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] if !ok { aggregates[key] = &metricAgg{ - VIN: sample.VIN, - Date: sample.StatDate, - Protocol: sample.Protocol, - FirstKM: sample.TotalMileageKM, - LatestKM: sample.TotalMileageKM, - Count: 1, + VIN: sample.VIN, + Date: sample.StatDate, + Protocol: sample.Protocol, + FirstKM: sample.TotalMileageKM, + LatestKM: sample.TotalMileageKM, + 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 } @@ -224,6 +244,19 @@ func addSamples(aggregates map[string]*metricAgg, samples []stats.MetricSample) if sample.TotalMileageKM > agg.LatestKM { 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++ } } @@ -232,66 +265,56 @@ func writeAggregates(ctx context.Context, db *sql.DB, aggregates map[string]*met 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) + for _, agg := range aggregates { + identity := stats.SourceIdentity{ + Protocol: agg.Protocol, + 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 } - 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 } -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) { dates, err := dateRangeWithPrevious(cfg.DateFrom, cfg.DateTo) if err != nil { @@ -320,21 +343,35 @@ func buildLastDiffAggregates(ctx context.Context, db *sql.DB, cfg config) (map[s current := lastByProtocolDate[protocol][date] previous := lastByProtocolDate[protocol][prevDate] for vin, currentSources := range current { - chosen, ok := chooseTrustedSource(currentSources, previous[vin]) - if !ok { - continue + previousBySource := map[string]dailySourceLast{} + for _, previousRow := range previous[vin] { + previousBySource[previousRow.SourceKey] = previousRow } - key := vin + "|" + date + "|" + string(protocol) - aggregates[key] = &metricAgg{ - VIN: vin, - Date: date, - Protocol: protocol, - FirstKM: chosen.previous.TotalKM, - LatestKM: chosen.current.TotalKM, - Count: 1, - SourceKey: chosen.current.SourceKey, - Phone: chosen.current.Phone, - SourceEndpoint: chosen.current.SourceEndpoint, + for _, currentRow := range currentSources { + key := vin + "|" + date + "|" + string(protocol) + "|" + currentRow.SourceKey + agg := &metricAgg{ + VIN: vin, + Date: date, + Protocol: protocol, + FirstKM: currentRow.TotalKM, + LatestKM: currentRow.TotalKM, + Count: 1, + SourceKey: currentRow.SourceKey, + Phone: currentRow.Phone, + 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))), 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 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) if err != nil { return nil, err @@ -396,10 +433,11 @@ GROUP BY vin, phone, source_endpoint`, ident(cfg.TDengineDatabase), strings.Join for rows.Next() { var vin string var phone string + var deviceID string var sourceEndpoint string var ts time.Time 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 } 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 { continue } - sourceKey := normalizedSourceKey(phone, sourceEndpoint) + sourceKey := normalizedSourceKey(string(protocol), phone, deviceID, sourceEndpoint) if sourceKey == "" { continue } @@ -424,6 +462,7 @@ GROUP BY vin, phone, source_endpoint`, ident(cfg.TDengineDatabase), strings.Join VIN: strings.TrimSpace(vin), SourceKey: sourceKey, Phone: strings.TrimSpace(phone), + DeviceID: strings.TrimSpace(deviceID), SourceEndpoint: strings.TrimSpace(sourceEndpoint), TS: ts, TotalKM: samples[0].TotalMileageKM, @@ -443,26 +482,9 @@ GROUP BY vin, phone, source_endpoint`, ident(cfg.TDengineDatabase), strings.Join return out, nil } -func normalizedSourceKey(phone string, endpoint string) string { - var parts []string - if phone = strings.TrimSpace(phone); phone != "" { - 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 normalizedSourceKey(protocol string, phone string, deviceID string, endpoint string) string { + sourceIP := stats.NormalizeSourceIP(endpoint) + return stats.SourceKey(envelope.Protocol(protocol), phone, deviceID, sourceIP) } 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, 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 WHERE %s ORDER BY ts ASC`, ident(cfg.TDengineDatabase), strings.Join(where, " AND ")) @@ -583,20 +605,23 @@ func realtimeMileageFramePredicate() string { } 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 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{ - 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, + Protocol: envelope.Protocol(strings.TrimSpace(protocol)), + VIN: strings.TrimSpace(vin), + Phone: strings.TrimSpace(phone), + DeviceID: strings.TrimSpace(deviceID), + SourceEndpoint: strings.TrimSpace(sourceEndpoint), + EventID: strings.TrimSpace(eventID), + MessageID: fmt.Sprintf("0x%04X", messageID), + EventTimeMS: eventTime.UnixMilli(), + ReceivedAtMS: receivedAt.UnixMilli(), + ParsedJSON: parsed, }, nil } @@ -647,11 +672,19 @@ func resetStats(ctx context.Context, db *sql.DB, cfg config) (int64, error) { } 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 + 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 { + return deleted, err + } + affected, err := result.RowsAffected() + if err != nil { + return deleted, err + } + deleted += affected } - return result.RowsAffected() + return deleted, nil } func parseProtocols(value string) ([]envelope.Protocol, error) { diff --git a/go/vehicle-gateway/cmd/stats-backfill/main_test.go b/go/vehicle-gateway/cmd/stats-backfill/main_test.go index 076aa15c..4a20c67d 100644 --- a/go/vehicle-gateway/cmd/stats-backfill/main_test.go +++ b/go/vehicle-gateway/cmd/stats-backfill/main_test.go @@ -5,7 +5,7 @@ import "testing" func TestChooseTrustedSourceKeepsContinuingSourceAndRejectsNewJump(t *testing.T) { previous := []dailySourceLast{{ VIN: "LA9GG64L7PBAF4001", - SourceKey: normalizedSourceKey("13307765812", "115.231.168.135:42630"), + SourceKey: normalizedSourceKey("JT808", "13307765812", "", "115.231.168.135:42630"), Phone: "13307765812", SourceEndpoint: "115.231.168.135:42630", TotalKM: 4100.8, @@ -13,14 +13,14 @@ func TestChooseTrustedSourceKeepsContinuingSourceAndRejectsNewJump(t *testing.T) current := []dailySourceLast{ { VIN: "LA9GG64L7PBAF4001", - SourceKey: normalizedSourceKey("13307765812", "115.159.85.149:28316"), + SourceKey: normalizedSourceKey("JT808", "13307765812", "", "115.159.85.149:28316"), Phone: "13307765812", SourceEndpoint: "115.159.85.149:28316", TotalKM: 42447.2, }, { VIN: "LA9GG64L7PBAF4001", - SourceKey: normalizedSourceKey("13307765812", "115.231.168.135:20215"), + SourceKey: normalizedSourceKey("JT808", "13307765812", "", "115.231.168.135:20215"), Phone: "13307765812", SourceEndpoint: "115.231.168.135:20215", TotalKM: 4123.9, @@ -31,10 +31,30 @@ func TestChooseTrustedSourceKeepsContinuingSourceAndRejectsNewJump(t *testing.T) if !ok { 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) } if delta := chosen.current.TotalKM - chosen.previous.TotalKM; delta < 23 || delta > 24 { 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) + } +}