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

fix(stats): elect trusted mileage source

Browse Source
This commit is contained in:
lingniu
2026-07-08 13:31:56 +08:00
parent abfed27846
commit a60a628d25
5 changed files with 284 additions and 40 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -49,12 +49,24 @@ type rawFrameRow struct {
}
type metricAgg struct {
VIN string
Date string
Protocol envelope.Protocol
FirstKM float64
LatestKM float64
Count int64
VIN string
Date string
Protocol envelope.Protocol
FirstKM float64
LatestKM float64
Count int64
SourceKey string
Phone string
SourceEndpoint string
}
type dailySourceLast struct {
VIN string
SourceKey string
Phone string
SourceEndpoint string
TS time.Time
TotalKM float64
}
func main() {
@@ -249,7 +261,7 @@ func writeAggregateBatch(ctx context.Context, db *sql.DB, rows []*metricAgg) err
if daily < 0 {
daily = 0
}
placeholders = append(placeholders, "(?,?,?,?,?,?,?)")
placeholders = append(placeholders, "(?,?,?,?,?,?,?,?,?,?)")
args = append(args,
row.VIN,
row.Date,
@@ -257,16 +269,23 @@ func writeAggregateBatch(ctx context.Context, db *sql.DB, rows []*metricAgg) err
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, sample_count)
(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...)
@@ -278,11 +297,11 @@ func buildLastDiffAggregates(ctx context.Context, db *sql.DB, cfg config) (map[s
if err != nil {
return nil, err
}
lastByProtocolDate := map[envelope.Protocol]map[string]map[string]float64{}
lastByProtocolDate := map[envelope.Protocol]map[string]map[string][]dailySourceLast{}
for _, protocol := range cfg.Protocols {
lastByProtocolDate[protocol] = map[string]map[string]float64{}
lastByProtocolDate[protocol] = map[string]map[string][]dailySourceLast{}
for _, date := range dates {
rows, err := queryDailyLastRows(ctx, db, cfg, protocol, date)
rows, err := queryDailyLastSourceRows(ctx, db, cfg, protocol, date)
if err != nil {
return nil, err
}
@@ -300,19 +319,22 @@ func buildLastDiffAggregates(ctx context.Context, db *sql.DB, cfg config) (map[s
prevDate := previousDate(date)
current := lastByProtocolDate[protocol][date]
previous := lastByProtocolDate[protocol][prevDate]
for vin, latest := range current {
first, ok := previous[vin]
if !ok || latest < first {
for vin, currentSources := range current {
chosen, ok := chooseTrustedSource(currentSources, previous[vin])
if !ok {
continue
}
key := vin + "|" + date + "|" + string(protocol)
aggregates[key] = &metricAgg{
VIN: vin,
Date: date,
Protocol: protocol,
FirstKM: first,
LatestKM: latest,
Count: 1,
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,
}
}
}
@@ -320,7 +342,38 @@ func buildLastDiffAggregates(ctx context.Context, db *sql.DB, cfg config) (map[s
return aggregates, nil
}
func queryDailyLastRows(ctx context.Context, db *sql.DB, cfg config, protocol envelope.Protocol, date string) (map[string]float64, error) {
type trustedChoice struct {
current dailySourceLast
previous dailySourceLast
}
const maxTrustedDailyMileageKM = 1000
func chooseTrustedSource(current []dailySourceLast, previous []dailySourceLast) (trustedChoice, bool) {
previousBySource := map[string]dailySourceLast{}
for _, row := range previous {
previousBySource[row.SourceKey] = row
}
var chosen trustedChoice
var chosenDelta float64
for _, currentRow := range current {
previousRow, ok := previousBySource[currentRow.SourceKey]
if !ok {
continue
}
delta := currentRow.TotalKM - previousRow.TotalKM
if delta < 0 || delta > maxTrustedDailyMileageKM {
continue
}
if chosen.current.SourceKey == "" || delta < chosenDelta {
chosen = trustedChoice{current: currentRow, previous: previousRow}
chosenDelta = delta
}
}
return chosen, chosen.current.SourceKey != ""
}
func queryDailyLastSourceRows(ctx context.Context, db *sql.DB, cfg config, protocol envelope.Protocol, date string) (map[string][]dailySourceLast, error) {
where := []string{
fmt.Sprintf("ts >= '%s 00:00:00'", quote(date)),
fmt.Sprintf("ts < '%s 00:00:00'", quote(nextDate(date))),
@@ -330,20 +383,23 @@ func queryDailyLastRows(ctx context.Context, db *sql.DB, cfg config, protocol en
fmt.Sprintf("protocol = '%s'", quote(string(protocol))),
realtimeMileageFramePredicate(),
}
sqlText := fmt.Sprintf(`SELECT vin, LAST(parsed_json)
sqlText := fmt.Sprintf(`SELECT vin, phone, source_endpoint, LAST(ts), LAST(parsed_json)
FROM %s.raw_frames
WHERE %s
GROUP BY vin`, ident(cfg.TDengineDatabase), strings.Join(where, " AND "))
GROUP BY vin, phone, source_endpoint`, 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{}
latestBySource := map[string]dailySourceLast{}
for rows.Next() {
var vin string
var phone string
var sourceEndpoint string
var ts time.Time
var parsedJSON string
if err := rows.Scan(&vin, &parsedJSON); err != nil {
if err := rows.Scan(&vin, &phone, &sourceEndpoint, &ts, &parsedJSON); err != nil {
return nil, err
}
fields := fieldsForStats(protocol, vin, parsedJSON)
@@ -360,9 +416,53 @@ GROUP BY vin`, ident(cfg.TDengineDatabase), strings.Join(where, " AND "))
if err != nil || len(samples) == 0 {
continue
}
out[strings.TrimSpace(vin)] = samples[0].TotalMileageKM
sourceKey := normalizedSourceKey(phone, sourceEndpoint)
if sourceKey == "" {
continue
}
row := dailySourceLast{
VIN: strings.TrimSpace(vin),
SourceKey: sourceKey,
Phone: strings.TrimSpace(phone),
SourceEndpoint: strings.TrimSpace(sourceEndpoint),
TS: ts,
TotalKM: samples[0].TotalMileageKM,
}
key := row.VIN + "|" + row.SourceKey
if existing, ok := latestBySource[key]; !ok || row.TS.After(existing.TS) {
latestBySource[key] = row
}
}
return out, rows.Err()
if err := rows.Err(); err != nil {
return nil, err
}
out := map[string][]dailySourceLast{}
for _, row := range latestBySource {
out[row.VIN] = append(out[row.VIN], row)
}
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 fieldsForStats(protocol envelope.Protocol, vin string, text string) map[string]any {

40
go/vehicle-gateway/cmd/stats-backfill/main_test.go Normal file
View File

@@ -0,0 +1,40 @@
package main
import "testing"
func TestChooseTrustedSourceKeepsContinuingSourceAndRejectsNewJump(t *testing.T) {
previous := []dailySourceLast{{
VIN: "LA9GG64L7PBAF4001",
SourceKey: normalizedSourceKey("13307765812", "115.231.168.135:42630"),
Phone: "13307765812",
SourceEndpoint: "115.231.168.135:42630",
TotalKM: 4100.8,
}}
current := []dailySourceLast{
{
VIN: "LA9GG64L7PBAF4001",
SourceKey: normalizedSourceKey("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"),
Phone: "13307765812",
SourceEndpoint: "115.231.168.135:20215",
TotalKM: 4123.9,
},
}
chosen, ok := chooseTrustedSource(current, previous)
if !ok {
t.Fatal("chooseTrustedSource() did not choose a source")
}
if chosen.current.SourceKey != normalizedSourceKey("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)
}
}

101
go/vehicle-gateway/internal/stats/daily_metric.go
View File

@@ -29,6 +29,9 @@ type MetricSample struct {
Protocol envelope.Protocol
StatDate string
TotalMileageKM float64
SourceKey string
Phone string
SourceEndpoint string
}
func NewWriter(exec Execer, loc *time.Location) *Writer {
@@ -45,6 +48,11 @@ func (w *Writer) EnsureSchema(ctx context.Context) error {
if _, err := w.exec.ExecContext(ctx, DailyMileageTableSQL); err != nil {
return err
}
for _, statement := range DailyMileageAlterSQL {
if _, err := w.exec.ExecContext(ctx, statement); err != nil && !isDuplicateColumnError(err) {
return err
}
}
return nil
}
@@ -62,7 +70,10 @@ func (w *Writer) Append(ctx context.Context, env envelope.FrameEnvelope) error {
sample.StatDate,
string(sample.Protocol),
sample.TotalMileageKM,
sample.TotalMileageKM); err != nil {
sample.TotalMileageKM,
sample.SourceKey,
sample.Phone,
sample.SourceEndpoint); err != nil {
return err
}
w.markMileageWritten(sample)
@@ -122,9 +133,35 @@ func SamplesFromEnvelope(env envelope.FrameEnvelope, loc *time.Location) ([]Metr
Protocol: env.Protocol,
StatDate: statDate,
TotalMileageKM: totalMileage,
SourceKey: sourceKey(env),
Phone: strings.TrimSpace(env.Phone),
SourceEndpoint: strings.TrimSpace(env.SourceEndpoint),
}}, nil
}
func sourceKey(env envelope.FrameEnvelope) string {
parts := []string{
strings.TrimSpace(env.Phone),
strings.TrimSpace(env.DeviceID),
strings.TrimSpace(env.SourceEndpoint),
}
var kept []string
for _, part := range parts {
if part != "" {
kept = append(kept, part)
}
}
return strings.Join(kept, "@")
}
func isDuplicateColumnError(err error) bool {
if err == nil {
return false
}
text := strings.ToLower(err.Error())
return strings.Contains(text, "duplicate column") || strings.Contains(text, "1060")
}
type mileageFieldMapping struct {
key string
scale float64
@@ -163,32 +200,86 @@ func mileageMappingsByProtocol(protocol envelope.Protocol) []mileageFieldMapping
const upsertDailyMileageSQL = `
INSERT INTO vehicle_daily_mileage
(vin, stat_date, protocol, daily_mileage_km,
first_total_mileage_km, latest_total_mileage_km, sample_count)
VALUES (?, ?, ?, 0, ?, ?, 1)
first_total_mileage_km, latest_total_mileage_km,
trusted_source_key, trusted_phone, trusted_source_endpoint, sample_count)
VALUES (?, ?, ?, 0, ?, ?, ?, ?, ?, 1)
ON DUPLICATE KEY UPDATE
first_total_mileage_km = CASE
WHEN trusted_source_key IS NOT NULL
AND trusted_source_key <> ''
AND trusted_source_key <> VALUES(trusted_source_key)
AND ABS(COALESCE(latest_total_mileage_km, VALUES(latest_total_mileage_km)) - VALUES(latest_total_mileage_km)) > 50
THEN first_total_mileage_km
WHEN first_total_mileage_km IS NULL OR first_total_mileage_km <= 0
THEN VALUES(first_total_mileage_km)
ELSE LEAST(first_total_mileage_km, VALUES(first_total_mileage_km))
END,
latest_total_mileage_km = CASE
WHEN trusted_source_key IS NOT NULL
AND trusted_source_key <> ''
AND trusted_source_key <> VALUES(trusted_source_key)
AND ABS(COALESCE(latest_total_mileage_km, VALUES(latest_total_mileage_km)) - VALUES(latest_total_mileage_km)) > 50
THEN latest_total_mileage_km
WHEN latest_total_mileage_km IS NULL OR latest_total_mileage_km <= 0
THEN VALUES(latest_total_mileage_km)
ELSE GREATEST(latest_total_mileage_km, VALUES(latest_total_mileage_km))
END,
daily_mileage_km = GREATEST(
CASE
WHEN trusted_source_key IS NOT NULL
AND trusted_source_key <> ''
AND trusted_source_key <> VALUES(trusted_source_key)
AND ABS(COALESCE(latest_total_mileage_km, VALUES(latest_total_mileage_km)) - VALUES(latest_total_mileage_km)) > 50
THEN latest_total_mileage_km
WHEN latest_total_mileage_km IS NULL OR latest_total_mileage_km <= 0
THEN VALUES(latest_total_mileage_km)
ELSE latest_total_mileage_km
END,
VALUES(latest_total_mileage_km)
CASE
WHEN trusted_source_key IS NOT NULL
AND trusted_source_key <> ''
AND trusted_source_key <> VALUES(trusted_source_key)
AND ABS(COALESCE(latest_total_mileage_km, VALUES(latest_total_mileage_km)) - VALUES(latest_total_mileage_km)) > 50
THEN latest_total_mileage_km
ELSE VALUES(latest_total_mileage_km)
END
) - CASE
WHEN trusted_source_key IS NOT NULL
AND trusted_source_key <> ''
AND trusted_source_key <> VALUES(trusted_source_key)
AND ABS(COALESCE(latest_total_mileage_km, VALUES(latest_total_mileage_km)) - VALUES(latest_total_mileage_km)) > 50
THEN first_total_mileage_km
WHEN first_total_mileage_km IS NULL OR first_total_mileage_km <= 0
THEN VALUES(first_total_mileage_km)
ELSE LEAST(first_total_mileage_km, VALUES(first_total_mileage_km))
END,
sample_count = sample_count + 1,
trusted_source_key = CASE
WHEN trusted_source_key IS NULL OR trusted_source_key = ''
THEN VALUES(trusted_source_key)
WHEN trusted_source_key = VALUES(trusted_source_key)
THEN trusted_source_key
WHEN ABS(COALESCE(latest_total_mileage_km, VALUES(latest_total_mileage_km)) - VALUES(latest_total_mileage_km)) <= 50
THEN VALUES(trusted_source_key)
ELSE trusted_source_key
END,
trusted_phone = CASE
WHEN trusted_source_key IS NULL OR trusted_source_key = '' OR trusted_source_key = VALUES(trusted_source_key)
THEN VALUES(trusted_phone)
ELSE trusted_phone
END,
trusted_source_endpoint = CASE
WHEN trusted_source_key IS NULL OR trusted_source_key = '' OR trusted_source_key = VALUES(trusted_source_key)
THEN VALUES(trusted_source_endpoint)
ELSE trusted_source_endpoint
END,
sample_count = CASE
WHEN trusted_source_key IS NOT NULL
AND trusted_source_key <> ''
AND trusted_source_key <> VALUES(trusted_source_key)
AND ABS(COALESCE(latest_total_mileage_km, VALUES(latest_total_mileage_km)) - VALUES(latest_total_mileage_km)) > 50
THEN sample_count
ELSE sample_count + 1
END,
updated_at = CURRENT_TIMESTAMP
`

20
go/vehicle-gateway/internal/stats/daily_metric_test.go
View File

@@ -187,19 +187,23 @@ func TestWriterEnsuresSchemaAndUpsertsDailyMileage(t *testing.T) {
if strings.Contains(exec.calls[0].query, "KEY idx_vin (vin)") {
t.Fatalf("daily mileage table should not keep redundant vin index covered by the primary key: %s", exec.calls[0].query)
}
if len(exec.calls) != 2 {
if len(exec.calls) != 5 {
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)
upsertCall := exec.calls[4]
if !strings.Contains(upsertCall.query, "ON DUPLICATE KEY UPDATE") {
t.Fatalf("unexpected upsert sql: %s", upsertCall.query)
}
if strings.Contains(exec.calls[1].query, "metric_key") || strings.Contains(exec.calls[1].query, "metric_unit") {
t.Fatalf("daily mileage upsert should not use generic metric columns: %s", exec.calls[1].query)
if strings.Contains(upsertCall.query, "metric_key") || strings.Contains(upsertCall.query, "metric_unit") {
t.Fatalf("daily mileage upsert should not use generic metric columns: %s", upsertCall.query)
}
if !strings.Contains(exec.calls[1].query, "first_total_mileage_km <= 0") {
t.Fatalf("upsert should ignore legacy zero first mileage: %s", exec.calls[1].query)
if !strings.Contains(upsertCall.query, "first_total_mileage_km <= 0") {
t.Fatalf("upsert should ignore legacy zero first mileage: %s", upsertCall.query)
}
if got := exec.calls[1].args[0]; got != "LNBVIN00000000002" {
if !strings.Contains(upsertCall.query, "trusted_source_key") {
t.Fatalf("upsert should track trusted source: %s", upsertCall.query)
}
if got := upsertCall.args[0]; got != "LNBVIN00000000002" {
t.Fatalf("first upsert arg should be vin, got %#v", got)
}
}

9
go/vehicle-gateway/internal/stats/schema.go
View File

@@ -7,9 +7,18 @@ const DailyMileageTableSQL = `CREATE TABLE IF NOT EXISTS vehicle_daily_mileage (
daily_mileage_km DECIMAL(18,3) NOT NULL DEFAULT 0,
first_total_mileage_km DECIMAL(18,3) NULL,
latest_total_mileage_km DECIMAL(18,3) NULL,
trusted_source_key VARCHAR(256) NULL,
trusted_phone VARCHAR(32) NULL,
trusted_source_endpoint VARCHAR(128) NULL,
sample_count BIGINT NOT NULL DEFAULT 0,
updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
PRIMARY KEY (vin, stat_date, protocol),
KEY idx_stat_date (stat_date),
KEY idx_protocol_date (protocol, stat_date)
)`
var DailyMileageAlterSQL = []string{
"ALTER TABLE vehicle_daily_mileage ADD COLUMN trusted_source_key VARCHAR(256) NULL",
"ALTER TABLE vehicle_daily_mileage ADD COLUMN trusted_phone VARCHAR(32) NULL",
"ALTER TABLE vehicle_daily_mileage ADD COLUMN trusted_source_endpoint VARCHAR(128) NULL",
}