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

fix: tighten source-aware mileage projection

Browse Source
This commit is contained in:
lingniu
2026-07-08 15:41:59 +08:00
parent 800ca1a8b1
commit e0ae57828e
7 changed files with 500 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -579,7 +579,7 @@ func loadConfig() (config, error) {
DateFrom: dateFrom,
DateTo: dateTo,
Protocols: protocols,
Method: env("BACKFILL_METHOD", "scan"),
Method: env("BACKFILL_METHOD", "last_diff"),
Limit: envInt("BACKFILL_LIMIT", 0),
DryRun: envBool("BACKFILL_DRY_RUN", true),
Reset: envBool("BACKFILL_RESET", false),

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

@@ -160,6 +160,16 @@ func TestWriteAggregatesClearsTargetRowsBeforeStaleCandidateUpsert(t *testing.T)
"2026-07-08",
"JT808",
).
WillReturnResult(sqlmock.NewResult(0, 0))
mock.ExpectExec(`DELETE FROM vehicle_daily_mileage`).
WithArgs(
"LA9GG64L7PBAF4001",
"2026-07-08",
"JT808",
"LA9GG64L7PBAF4001",
"2026-07-08",
"JT808",
).
WillReturnResult(sqlmock.NewResult(0, 1))
written, err := writeAggregates(context.Background(), db, aggregates, 500)
@@ -173,3 +183,19 @@ func TestWriteAggregatesClearsTargetRowsBeforeStaleCandidateUpsert(t *testing.T)
t.Fatalf("sql expectations: %v", err)
}
}
func TestLoadConfigDefaultsBackfillMethodToLastDiff(t *testing.T) {
t.Setenv("BACKFILL_METHOD", "")
t.Setenv("BACKFILL_DATE_FROM", "2026-07-08")
t.Setenv("BACKFILL_DATE_TO", "2026-07-08")
t.Setenv("BACKFILL_PROTOCOLS", "JT808")
t.Setenv("LOCAL_TZ", "Asia/Shanghai")
cfg, err := loadConfig()
if err != nil {
t.Fatalf("loadConfig() error = %v", err)
}
if cfg.Method != "last_diff" {
t.Fatalf("method = %q, want last_diff", cfg.Method)
}
}

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

@@ -19,9 +19,11 @@ type Execer interface {
type Writer struct {
exec Execer
query Queryer
loc *time.Location
mu sync.Mutex
lastTotalMileage map[string]float64
baselineCache map[string]sourceBaselineCacheEntry
}
type MetricSample struct {
@@ -43,7 +45,16 @@ func NewWriter(exec Execer, loc *time.Location) *Writer {
if loc == nil {
loc = time.FixedZone("Asia/Shanghai", 8*3600)
}
return &Writer{exec: exec, loc: loc, lastTotalMileage: map[string]float64{}}
writer := &Writer{
exec: exec,
loc: loc,
lastTotalMileage: map[string]float64{},
baselineCache: map[string]sourceBaselineCacheEntry{},
}
if query, ok := exec.(Queryer); ok {
writer.query = query
}
return writer
}
func (w *Writer) EnsureSchema(ctx context.Context) error {
@@ -79,6 +90,9 @@ func (w *Writer) Append(ctx context.Context, env envelope.FrameEnvelope) error {
return err
}
candidate := SourceMileageSampleFromMetric(sample, identity)
if err := w.applyRealtimeBaseline(ctx, &candidate); err != nil {
return err
}
if err := UpsertSourceMileage(ctx, w.exec, candidate); err != nil {
return err
}
@@ -92,8 +106,7 @@ func (w *Writer) Append(ctx context.Context, env envelope.FrameEnvelope) error {
}
func (w *Writer) seenSameMileage(sample MetricSample) bool {
prefix := fmt.Sprintf("%s|%s|", sample.VIN, sample.Protocol)
key := prefix + sample.StatDate
key := mileageCacheKey(sample)
w.mu.Lock()
defer w.mu.Unlock()
if last, ok := w.lastTotalMileage[key]; ok && last == sample.TotalMileageKM {
@@ -103,8 +116,8 @@ func (w *Writer) seenSameMileage(sample MetricSample) bool {
}
func (w *Writer) markMileageWritten(sample MetricSample) {
prefix := fmt.Sprintf("%s|%s|", sample.VIN, sample.Protocol)
key := prefix + sample.StatDate
prefix := mileageCachePrefix(sample)
key := mileageCacheKey(sample)
w.mu.Lock()
defer w.mu.Unlock()
for existing := range w.lastTotalMileage {
@@ -112,9 +125,77 @@ func (w *Writer) markMileageWritten(sample MetricSample) {
delete(w.lastTotalMileage, existing)
}
}
for existing := range w.baselineCache {
if strings.HasPrefix(existing, prefix) && existing != key {
delete(w.baselineCache, existing)
}
}
w.lastTotalMileage[key] = sample.TotalMileageKM
}
func (w *Writer) applyRealtimeBaseline(ctx context.Context, candidate *SourceMileageSample) error {
if candidate == nil {
return nil
}
baseline, found, err := w.previousBaseline(ctx, *candidate)
if err != nil {
return err
}
if !found {
candidate.QualityStatus = QualityNoPreviousBaseline
candidate.QualityReason = "missing_previous_source"
return nil
}
candidate.FirstTotalKM = baseline.LatestTotalKM
candidate.FirstEventTime = baseline.LatestEventTime
candidate.DailyKM = candidate.LatestTotalKM - baseline.LatestTotalKM
candidate.QualityStatus = QualityOK
candidate.QualityReason = "same_source_previous_day"
if candidate.DailyKM < 0 || candidate.DailyKM > maxSelectedDailyMileageKM {
candidate.QualityStatus = QualityInvalidDelta
candidate.QualityReason = "outside_daily_range"
}
return nil
}
func (w *Writer) previousBaseline(ctx context.Context, candidate SourceMileageSample) (sourceBaseline, bool, error) {
cacheKey := mileageCacheKey(MetricSample{
VIN: candidate.VIN,
Protocol: candidate.Protocol,
StatDate: candidate.StatDate,
SourceKey: candidate.SourceKey,
})
w.mu.Lock()
if cached, ok := w.baselineCache[cacheKey]; ok {
w.mu.Unlock()
return cached.baseline, cached.found, nil
}
w.mu.Unlock()
baseline, found, err := lookupPreviousSourceBaseline(ctx, w.query, candidate.VIN, candidate.StatDate, candidate.Protocol, candidate.SourceKey)
if err != nil {
return sourceBaseline{}, false, err
}
w.mu.Lock()
w.baselineCache[cacheKey] = sourceBaselineCacheEntry{baseline: baseline, found: found}
w.mu.Unlock()
return baseline, found, nil
}
func mileageCachePrefix(sample MetricSample) string {
return fmt.Sprintf("%s|%s|%s|", sample.VIN, sample.Protocol, sample.SourceKey)
}
func mileageCacheKey(sample MetricSample) string {
return mileageCachePrefix(sample) + sample.StatDate
}
type sourceBaselineCacheEntry struct {
baseline sourceBaseline
found bool
}
func SamplesFromEnvelope(env envelope.FrameEnvelope, loc *time.Location) ([]MetricSample, error) {
vin := strings.TrimSpace(env.VIN)
if vin == "" {

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

@@ -3,11 +3,14 @@ package stats
import (
"context"
"database/sql"
"database/sql/driver"
"errors"
"math"
"strings"
"testing"
"time"
"github.com/DATA-DOG/go-sqlmock"
"lingniu-vehicle-ingest/go/vehicle-gateway/internal/envelope"
)
@@ -173,8 +176,8 @@ func TestWriterAppendWritesSourceCandidateAndProjection(t *testing.T) {
if err := writer.Append(context.Background(), event); err != nil {
t.Fatalf("Append() error = %v", err)
}
if len(exec.calls) != 5 {
t.Fatalf("exec calls = %d, want source + candidate + clear + project + mark", len(exec.calls))
if len(exec.calls) != 6 {
t.Fatalf("exec calls = %d, want source + candidate + clear + project + mark + cleanup", len(exec.calls))
}
if !strings.Contains(exec.calls[0].query, "INSERT INTO vehicle_data_source") {
t.Fatalf("first call should upsert source: %s", exec.calls[0].query)
@@ -187,6 +190,216 @@ func TestWriterAppendWritesSourceCandidateAndProjection(t *testing.T) {
}
}
func TestWriterAppendDedupesRealtimeMileagePerSource(t *testing.T) {
exec := &recordingExec{}
writer := NewWriter(exec, time.FixedZone("Asia/Shanghai", 8*3600))
base := envelope.FrameEnvelope{
Protocol: envelope.ProtocolJT808,
VIN: "LA9GG64L7PBAF4001",
Phone: "13307765812",
EventTimeMS: time.Date(2026, 7, 8, 13, 20, 0, 0, time.FixedZone("Asia/Shanghai", 8*3600)).UnixMilli(),
Fields: map[string]any{
"jt808.location.total_mileage_km": 4123.9,
},
}
sourceA := base
sourceA.SourceEndpoint = "115.231.168.135:20215"
if err := writer.Append(context.Background(), sourceA); err != nil {
t.Fatalf("Append(sourceA) error = %v", err)
}
sourceB := base
sourceB.SourceEndpoint = "115.159.85.149:28316"
sourceB.ReceivedAtMS = sourceB.EventTimeMS + 1000
if err := writer.Append(context.Background(), sourceB); err != nil {
t.Fatalf("Append(sourceB) error = %v", err)
}
sourceADuplicate := sourceA
sourceADuplicate.ReceivedAtMS = sourceA.EventTimeMS + 2000
if err := writer.Append(context.Background(), sourceADuplicate); err != nil {
t.Fatalf("Append(sourceADuplicate) error = %v", err)
}
if got := countExecQueries(exec.calls, "INSERT INTO vehicle_daily_mileage_source"); got != 2 {
t.Fatalf("candidate upserts = %d, want 2", got)
}
if got := countExecQueries(exec.calls, "INSERT INTO vehicle_data_source"); got != 2 {
t.Fatalf("source upserts = %d, want 2", got)
}
}
func TestWriterAppendWritesNoPreviousBaselineCandidateAndCleansProjection(t *testing.T) {
db, mock, err := sqlmock.New()
if err != nil {
t.Fatalf("sqlmock.New() error = %v", err)
}
defer db.Close()
loc := time.FixedZone("Asia/Shanghai", 8*3600)
writer := NewWriter(db, loc)
eventTime := time.Date(2026, 7, 8, 13, 20, 0, 0, loc)
event := envelope.FrameEnvelope{
Protocol: envelope.ProtocolJT808,
VIN: "LA9GG64L7PBAF4001",
Phone: "13307765812",
SourceEndpoint: "115.231.168.135:20215",
EventTimeMS: eventTime.UnixMilli(),
Fields: map[string]any{
"jt808.location.total_mileage_km": 4123.9,
},
}
mock.ExpectExec(`INSERT INTO vehicle_data_source`).
WithArgs("JT808", "115.231.168.135", "115.231.168.135:20215", sqlmock.AnyArg(), sqlmock.AnyArg()).
WillReturnResult(sqlmock.NewResult(0, 1))
mock.ExpectQuery(`SELECT latest_total_mileage_km, latest_event_time FROM vehicle_daily_mileage_source`).
WithArgs("LA9GG64L7PBAF4001", "2026-07-07", "JT808", "JT808:13307765812@115.231.168.135").
WillReturnRows(sqlmock.NewRows([]string{"latest_total_mileage_km", "latest_event_time"}))
mock.ExpectExec(`INSERT INTO vehicle_daily_mileage_source`).
WithArgs(
"LA9GG64L7PBAF4001",
"2026-07-08",
"JT808",
"JT808:13307765812@115.231.168.135",
"115.231.168.135",
"115.231.168.135:20215",
"13307765812",
"",
"",
float64(4123.9),
float64(4123.9),
float64(0),
int64(1),
eventTime,
eventTime,
QualityNoPreviousBaseline,
"missing_previous_source",
).
WillReturnResult(sqlmock.NewResult(0, 1))
mock.ExpectExec(`UPDATE vehicle_daily_mileage_source`).
WithArgs("LA9GG64L7PBAF4001", "2026-07-08", "JT808").
WillReturnResult(sqlmock.NewResult(0, 0))
mock.ExpectExec(`INSERT INTO vehicle_daily_mileage`).
WithArgs("LA9GG64L7PBAF4001", "2026-07-08", "JT808", int64(1000)).
WillReturnResult(sqlmock.NewResult(0, 0))
mock.ExpectExec(`UPDATE vehicle_daily_mileage_source s`).
WithArgs(
"LA9GG64L7PBAF4001",
"2026-07-08",
"JT808",
int64(1000),
"LA9GG64L7PBAF4001",
"2026-07-08",
"JT808",
).
WillReturnResult(sqlmock.NewResult(0, 0))
mock.ExpectExec(`DELETE FROM vehicle_daily_mileage`).
WithArgs(
"LA9GG64L7PBAF4001",
"2026-07-08",
"JT808",
"LA9GG64L7PBAF4001",
"2026-07-08",
"JT808",
).
WillReturnResult(sqlmock.NewResult(0, 1))
if err := writer.Append(context.Background(), event); err != nil {
t.Fatalf("Append() error = %v", err)
}
if err := mock.ExpectationsWereMet(); err != nil {
t.Fatalf("sql expectations: %v", err)
}
}
func TestWriterAppendUsesPreviousSourceBaselineForRealtimeCandidate(t *testing.T) {
db, mock, err := sqlmock.New()
if err != nil {
t.Fatalf("sqlmock.New() error = %v", err)
}
defer db.Close()
loc := time.FixedZone("Asia/Shanghai", 8*3600)
writer := NewWriter(db, loc)
previousTime := time.Date(2026, 7, 7, 23, 58, 0, 0, loc)
currentTime := time.Date(2026, 7, 8, 13, 20, 0, 0, loc)
event := envelope.FrameEnvelope{
Protocol: envelope.ProtocolJT808,
VIN: "LA9GG64L7PBAF4001",
Phone: "13307765812",
SourceEndpoint: "115.231.168.135:20215",
EventTimeMS: currentTime.UnixMilli(),
Fields: map[string]any{
"jt808.location.total_mileage_km": 4123.9,
},
}
mock.ExpectExec(`INSERT INTO vehicle_data_source`).
WithArgs("JT808", "115.231.168.135", "115.231.168.135:20215", sqlmock.AnyArg(), sqlmock.AnyArg()).
WillReturnResult(sqlmock.NewResult(0, 1))
mock.ExpectQuery(`SELECT latest_total_mileage_km, latest_event_time FROM vehicle_daily_mileage_source`).
WithArgs("LA9GG64L7PBAF4001", "2026-07-07", "JT808", "JT808:13307765812@115.231.168.135").
WillReturnRows(sqlmock.NewRows([]string{"latest_total_mileage_km", "latest_event_time"}).
AddRow(4100.8, previousTime))
mock.ExpectExec(`INSERT INTO vehicle_daily_mileage_source`).
WithArgs(
"LA9GG64L7PBAF4001",
"2026-07-08",
"JT808",
"JT808:13307765812@115.231.168.135",
"115.231.168.135",
"115.231.168.135:20215",
"13307765812",
"",
"",
float64(4100.8),
float64(4123.9),
approxFloat64{want: 23.1, tolerance: 0.000001},
int64(1),
previousTime,
currentTime,
QualityOK,
"same_source_previous_day",
).
WillReturnResult(sqlmock.NewResult(0, 1))
mock.ExpectExec(`UPDATE vehicle_daily_mileage_source`).
WithArgs("LA9GG64L7PBAF4001", "2026-07-08", "JT808").
WillReturnResult(sqlmock.NewResult(0, 0))
mock.ExpectExec(`INSERT INTO vehicle_daily_mileage`).
WithArgs("LA9GG64L7PBAF4001", "2026-07-08", "JT808", int64(1000)).
WillReturnResult(sqlmock.NewResult(0, 1))
mock.ExpectExec(`UPDATE vehicle_daily_mileage_source s`).
WithArgs(
"LA9GG64L7PBAF4001",
"2026-07-08",
"JT808",
int64(1000),
"LA9GG64L7PBAF4001",
"2026-07-08",
"JT808",
).
WillReturnResult(sqlmock.NewResult(0, 1))
mock.ExpectExec(`DELETE FROM vehicle_daily_mileage`).
WithArgs(
"LA9GG64L7PBAF4001",
"2026-07-08",
"JT808",
"LA9GG64L7PBAF4001",
"2026-07-08",
"JT808",
).
WillReturnResult(sqlmock.NewResult(0, 0))
if err := writer.Append(context.Background(), event); err != nil {
t.Fatalf("Append() error = %v", err)
}
if err := mock.ExpectationsWereMet(); err != nil {
t.Fatalf("sql expectations: %v", err)
}
}
func TestWriterEnsuresSchemaAndUpsertsDailyMileage(t *testing.T) {
exec := &recordingExec{}
writer := NewWriter(exec, time.FixedZone("Asia/Shanghai", 8*3600))
@@ -206,7 +419,7 @@ func TestWriterEnsuresSchemaAndUpsertsDailyMileage(t *testing.T) {
t.Fatalf("Append() error = %v", err)
}
if len(exec.calls) != 11 {
if len(exec.calls) != 12 {
t.Fatalf("exec calls = %d", len(exec.calls))
}
for i, want := range []string{
@@ -277,8 +490,8 @@ func TestWriterSkipsConsecutiveDuplicateMileageSamples(t *testing.T) {
t.Fatalf("duplicate Append() error = %v", err)
}
if len(exec.calls) != 3 {
t.Fatalf("exec calls = %d, want 3", len(exec.calls))
if len(exec.calls) != 4 {
t.Fatalf("exec calls = %d, want 4", len(exec.calls))
}
}
@@ -301,8 +514,8 @@ func TestWriterDoesNotCacheMileageWhenUpsertFails(t *testing.T) {
t.Fatalf("retry Append() error = %v", err)
}
if len(exec.calls) != 4 {
t.Fatalf("exec calls = %d, want 4", len(exec.calls))
if len(exec.calls) != 5 {
t.Fatalf("exec calls = %d, want 5", len(exec.calls))
}
}
@@ -326,14 +539,24 @@ func TestWriterKeepsOnlyLatestDateInMileageCache(t *testing.T) {
t.Fatalf("next-day Append() error = %v", err)
}
if len(exec.calls) != 6 {
t.Fatalf("exec calls = %d, want 6", len(exec.calls))
if len(exec.calls) != 8 {
t.Fatalf("exec calls = %d, want 8", len(exec.calls))
}
if len(writer.lastTotalMileage) != 1 {
t.Fatalf("cache entries = %d, want 1", len(writer.lastTotalMileage))
}
}
func countExecQueries(calls []execCall, fragment string) int {
count := 0
for _, call := range calls {
if strings.Contains(call.query, fragment) {
count++
}
}
return count
}
type execCall struct {
query string
args []any
@@ -353,3 +576,16 @@ func (e *recordingExec) ExecContext(_ context.Context, query string, args ...any
}
return nil, nil
}
type approxFloat64 struct {
want float64
tolerance float64
}
func (m approxFloat64) Match(value driver.Value) bool {
got, ok := value.(float64)
if !ok {
return false
}
return math.Abs(got-m.want) <= m.tolerance
}

79
go/vehicle-gateway/internal/stats/source_mileage.go
View File

@@ -2,6 +2,7 @@ package stats
import (
"context"
"database/sql"
"strings"
"time"
@@ -127,6 +128,17 @@ func ProjectDailyMileage(ctx context.Context, exec Execer, vin string, statDate
statDate,
string(protocol),
)
if err != nil {
return err
}
_, err = exec.ExecContext(ctx, cleanupProjectedDailyMileageSQL,
vin,
statDate,
string(protocol),
vin,
statDate,
string(protocol),
)
return err
}
@@ -257,3 +269,70 @@ JOIN (
SET s.is_selected = 1
WHERE s.vin = ? AND s.stat_date = ? AND s.protocol = ?
`
const cleanupProjectedDailyMileageSQL = `
DELETE FROM vehicle_daily_mileage
WHERE vin = ? AND stat_date = ? AND protocol = ?
AND NOT EXISTS (
SELECT 1
FROM vehicle_daily_mileage_source
WHERE vin = ? AND stat_date = ? AND protocol = ? AND is_selected = 1
)
`
type sourceBaseline struct {
LatestTotalKM float64
LatestEventTime time.Time
}
func lookupPreviousSourceBaseline(ctx context.Context, query Queryer, vin string, statDate string, protocol envelope.Protocol, sourceKey string) (sourceBaseline, bool, error) {
if query == nil || strings.TrimSpace(vin) == "" || strings.TrimSpace(statDate) == "" || strings.TrimSpace(sourceKey) == "" {
return sourceBaseline{}, false, nil
}
previousDate, ok := previousStatDate(statDate)
if !ok {
return sourceBaseline{}, false, nil
}
rows, err := query.QueryContext(ctx, previousSourceBaselineSQL, vin, previousDate, string(protocol), sourceKey)
if err != nil {
return sourceBaseline{}, false, err
}
defer rows.Close()
if !rows.Next() {
if err := rows.Err(); err != nil {
return sourceBaseline{}, false, err
}
return sourceBaseline{}, false, nil
}
var latestTotal sql.NullFloat64
var latestEvent sql.NullTime
if err := rows.Scan(&latestTotal, &latestEvent); err != nil {
return sourceBaseline{}, false, err
}
if err := rows.Err(); err != nil {
return sourceBaseline{}, false, err
}
return sourceBaseline{
LatestTotalKM: latestTotal.Float64,
LatestEventTime: latestEvent.Time,
}, latestTotal.Valid, nil
}
func previousStatDate(statDate string) (string, bool) {
day, err := time.Parse("2006-01-02", strings.TrimSpace(statDate))
if err != nil {
return "", false
}
return day.AddDate(0, 0, -1).Format("2006-01-02"), true
}
const previousSourceBaselineSQL = `
SELECT latest_total_mileage_km, latest_event_time
FROM vehicle_daily_mileage_source
WHERE vin = ?
AND stat_date = ?
AND protocol = ?
AND source_key = ?
ORDER BY latest_event_time DESC
LIMIT 1
`

18
go/vehicle-gateway/internal/stats/source_mileage_test.go
View File

@@ -88,12 +88,13 @@ func TestProjectDailyMileageSelectsCandidateAndMarksSource(t *testing.T) {
if err != nil {
t.Fatalf("ProjectDailyMileage() error = %v", err)
}
if len(exec.calls) != 3 {
t.Fatalf("exec calls = %d, want 3", len(exec.calls))
if len(exec.calls) != 4 {
t.Fatalf("exec calls = %d, want 4", len(exec.calls))
}
updateSources := exec.calls[0].query
projectFinal := exec.calls[1].query
markSelected := exec.calls[2].query
cleanupFinal := exec.calls[3].query
if !strings.Contains(updateSources, "UPDATE vehicle_daily_mileage_source") || !strings.Contains(updateSources, "is_selected = 0") {
t.Fatalf("first query should clear selected candidates: %s", updateSources)
}
@@ -142,4 +143,17 @@ func TestProjectDailyMileageSelectsCandidateAndMarksSource(t *testing.T) {
if got := exec.calls[2].args[3]; got != maxSelectedDailyMileageKM {
t.Fatalf("mark query max mileage arg = %#v, want %d", got, maxSelectedDailyMileageKM)
}
for _, want := range []string{
"DELETE FROM vehicle_daily_mileage",
"NOT EXISTS (",
"FROM vehicle_daily_mileage_source",
"is_selected = 1",
} {
if !strings.Contains(cleanupFinal, want) {
t.Fatalf("cleanup query missing %q: %s", want, cleanupFinal)
}
}
if len(exec.calls[3].args) != 6 {
t.Fatalf("cleanup query args = %d, want 6", len(exec.calls[3].args))
}
}