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

fix(stats): skip realtime mileage without source identity

Browse Source
This commit is contained in:
lingniu
2026-07-08 15:44:31 +08:00
parent e0ae57828e
commit 17937a1ad8
2 changed files with 50 additions and 25 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -85,17 +85,18 @@ func (w *Writer) Append(ctx context.Context, env envelope.FrameEnvelope) error {
if w.seenSameMileage(sample) { if w.seenSameMileage(sample) {
continue continue
} }
if hasSource { if !hasSource {
if err := UpsertDataSource(ctx, w.exec, identity, sample.EventTime); err != nil { continue
return err }
} if err := UpsertDataSource(ctx, w.exec, identity, sample.EventTime); err != nil {
candidate := SourceMileageSampleFromMetric(sample, identity) return err
if err := w.applyRealtimeBaseline(ctx, &candidate); err != nil { }
return err candidate := SourceMileageSampleFromMetric(sample, identity)
} if err := w.applyRealtimeBaseline(ctx, &candidate); err != nil {
if err := UpsertSourceMileage(ctx, w.exec, candidate); err != nil { return err
return err }
} if err := UpsertSourceMileage(ctx, w.exec, candidate); err != nil {
return err
} }
if err := ProjectDailyMileage(ctx, w.exec, sample.VIN, sample.StatDate, sample.Protocol); err != nil { if err := ProjectDailyMileage(ctx, w.exec, sample.VIN, sample.StatDate, sample.Protocol); err != nil {
return err return err

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

@@ -190,6 +190,27 @@ func TestWriterAppendWritesSourceCandidateAndProjection(t *testing.T) {
} }
} }
func TestWriterAppendSkipsMileageWithoutSourceIdentity(t *testing.T) {
exec := &recordingExec{}
writer := NewWriter(exec, time.FixedZone("Asia/Shanghai", 8*3600))
event := 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,
},
}
if err := writer.Append(context.Background(), event); err != nil {
t.Fatalf("Append() error = %v", err)
}
if len(exec.calls) != 0 {
t.Fatalf("exec calls = %d, want 0 without source identity", len(exec.calls))
}
}
func TestWriterAppendDedupesRealtimeMileagePerSource(t *testing.T) { func TestWriterAppendDedupesRealtimeMileagePerSource(t *testing.T) {
exec := &recordingExec{} exec := &recordingExec{}
writer := NewWriter(exec, time.FixedZone("Asia/Shanghai", 8*3600)) writer := NewWriter(exec, time.FixedZone("Asia/Shanghai", 8*3600))
@@ -474,9 +495,10 @@ func TestWriterSkipsConsecutiveDuplicateMileageSamples(t *testing.T) {
exec := &recordingExec{} exec := &recordingExec{}
writer := NewWriter(exec, time.FixedZone("Asia/Shanghai", 8*3600)) writer := NewWriter(exec, time.FixedZone("Asia/Shanghai", 8*3600))
event := envelope.FrameEnvelope{ event := envelope.FrameEnvelope{
Protocol: envelope.ProtocolJT808, Protocol: envelope.ProtocolJT808,
VIN: "LNBVIN00000000001", VIN: "LNBVIN00000000001",
EventTimeMS: time.Date(2026, 7, 1, 9, 0, 0, 0, time.FixedZone("Asia/Shanghai", 8*3600)).UnixMilli(), SourceEndpoint: "115.231.168.135:20215",
EventTimeMS: time.Date(2026, 7, 1, 9, 0, 0, 0, time.FixedZone("Asia/Shanghai", 8*3600)).UnixMilli(),
Fields: map[string]any{ Fields: map[string]any{
envelope.FieldTotalMileageKM: 10241.2, envelope.FieldTotalMileageKM: 10241.2,
}, },
@@ -490,8 +512,8 @@ func TestWriterSkipsConsecutiveDuplicateMileageSamples(t *testing.T) {
t.Fatalf("duplicate Append() error = %v", err) t.Fatalf("duplicate Append() error = %v", err)
} }
if len(exec.calls) != 4 { if len(exec.calls) != 6 {
t.Fatalf("exec calls = %d, want 4", len(exec.calls)) t.Fatalf("exec calls = %d, want 6", len(exec.calls))
} }
} }
@@ -499,9 +521,10 @@ func TestWriterDoesNotCacheMileageWhenUpsertFails(t *testing.T) {
exec := &recordingExec{errs: []error{errors.New("mysql unavailable"), nil}} exec := &recordingExec{errs: []error{errors.New("mysql unavailable"), nil}}
writer := NewWriter(exec, time.FixedZone("Asia/Shanghai", 8*3600)) writer := NewWriter(exec, time.FixedZone("Asia/Shanghai", 8*3600))
event := envelope.FrameEnvelope{ event := envelope.FrameEnvelope{
Protocol: envelope.ProtocolJT808, Protocol: envelope.ProtocolJT808,
VIN: "LNBVIN00000000001", VIN: "LNBVIN00000000001",
EventTimeMS: time.Date(2026, 7, 1, 9, 0, 0, 0, time.FixedZone("Asia/Shanghai", 8*3600)).UnixMilli(), SourceEndpoint: "115.231.168.135:20215",
EventTimeMS: time.Date(2026, 7, 1, 9, 0, 0, 0, time.FixedZone("Asia/Shanghai", 8*3600)).UnixMilli(),
Fields: map[string]any{ Fields: map[string]any{
envelope.FieldTotalMileageKM: 10241.2, envelope.FieldTotalMileageKM: 10241.2,
}, },
@@ -514,8 +537,8 @@ func TestWriterDoesNotCacheMileageWhenUpsertFails(t *testing.T) {
t.Fatalf("retry Append() error = %v", err) t.Fatalf("retry Append() error = %v", err)
} }
if len(exec.calls) != 5 { if len(exec.calls) != 7 {
t.Fatalf("exec calls = %d, want 5", len(exec.calls)) t.Fatalf("exec calls = %d, want 7", len(exec.calls))
} }
} }
@@ -523,8 +546,9 @@ func TestWriterKeepsOnlyLatestDateInMileageCache(t *testing.T) {
exec := &recordingExec{} exec := &recordingExec{}
writer := NewWriter(exec, time.FixedZone("Asia/Shanghai", 8*3600)) writer := NewWriter(exec, time.FixedZone("Asia/Shanghai", 8*3600))
event := envelope.FrameEnvelope{ event := envelope.FrameEnvelope{
Protocol: envelope.ProtocolJT808, Protocol: envelope.ProtocolJT808,
VIN: "LNBVIN00000000001", VIN: "LNBVIN00000000001",
SourceEndpoint: "115.231.168.135:20215",
Fields: map[string]any{ Fields: map[string]any{
envelope.FieldTotalMileageKM: 10241.2, envelope.FieldTotalMileageKM: 10241.2,
}, },
@@ -539,8 +563,8 @@ func TestWriterKeepsOnlyLatestDateInMileageCache(t *testing.T) {
t.Fatalf("next-day Append() error = %v", err) t.Fatalf("next-day Append() error = %v", err)
} }
if len(exec.calls) != 8 { if len(exec.calls) != 12 {
t.Fatalf("exec calls = %d, want 8", len(exec.calls)) t.Fatalf("exec calls = %d, want 12", len(exec.calls))
} }
if len(writer.lastTotalMileage) != 1 { if len(writer.lastTotalMileage) != 1 {
t.Fatalf("cache entries = %d, want 1", len(writer.lastTotalMileage)) t.Fatalf("cache entries = %d, want 1", len(writer.lastTotalMileage))