fix(stats): guard backfill source-less mileage
This commit is contained in:
@@ -233,8 +233,8 @@ func addSamples(aggregates map[string]*metricAgg, samples []stats.MetricSample)
|
|||||||
SourceEndpoint: sample.SourceEndpoint,
|
SourceEndpoint: sample.SourceEndpoint,
|
||||||
FirstEventTime: sample.EventTime,
|
FirstEventTime: sample.EventTime,
|
||||||
LatestEventTime: sample.EventTime,
|
LatestEventTime: sample.EventTime,
|
||||||
QualityStatus: stats.QualityOK,
|
QualityStatus: stats.QualityNoPreviousBaseline,
|
||||||
QualityReason: "scan_backfill",
|
QualityReason: "scan_backfill_missing_previous_source",
|
||||||
}
|
}
|
||||||
continue
|
continue
|
||||||
}
|
}
|
||||||
@@ -268,6 +268,14 @@ func writeAggregates(ctx context.Context, db *sql.DB, aggregates map[string]*met
|
|||||||
var written int64
|
var written int64
|
||||||
clearedTargets := map[string]struct{}{}
|
clearedTargets := map[string]struct{}{}
|
||||||
for _, agg := range aggregates {
|
for _, agg := range aggregates {
|
||||||
|
identity := stats.SourceIdentity{
|
||||||
|
Protocol: agg.Protocol,
|
||||||
|
SourceIP: stats.NormalizeSourceIP(agg.SourceEndpoint),
|
||||||
|
SourceEndpoint: agg.SourceEndpoint,
|
||||||
|
}
|
||||||
|
if strings.TrimSpace(identity.SourceIP) == "" {
|
||||||
|
continue
|
||||||
|
}
|
||||||
target := agg.VIN + "|" + agg.Date + "|" + string(agg.Protocol)
|
target := agg.VIN + "|" + agg.Date + "|" + string(agg.Protocol)
|
||||||
if _, ok := clearedTargets[target]; !ok {
|
if _, ok := clearedTargets[target]; !ok {
|
||||||
if err := clearBackfillTargetMileage(ctx, db, agg.VIN, agg.Date, agg.Protocol); err != nil {
|
if err := clearBackfillTargetMileage(ctx, db, agg.VIN, agg.Date, agg.Protocol); err != nil {
|
||||||
@@ -275,11 +283,6 @@ func writeAggregates(ctx context.Context, db *sql.DB, aggregates map[string]*met
|
|||||||
}
|
}
|
||||||
clearedTargets[target] = struct{}{}
|
clearedTargets[target] = struct{}{}
|
||||||
}
|
}
|
||||||
identity := stats.SourceIdentity{
|
|
||||||
Protocol: agg.Protocol,
|
|
||||||
SourceIP: stats.NormalizeSourceIP(agg.SourceEndpoint),
|
|
||||||
SourceEndpoint: agg.SourceEndpoint,
|
|
||||||
}
|
|
||||||
if err := stats.UpsertDataSource(ctx, db, identity, agg.LatestEventTime); err != nil {
|
if err := stats.UpsertDataSource(ctx, db, identity, agg.LatestEventTime); err != nil {
|
||||||
return written, err
|
return written, err
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -184,6 +184,71 @@ func TestWriteAggregatesClearsTargetRowsBeforeStaleCandidateUpsert(t *testing.T)
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func TestWriteAggregatesSkipsBlankSourceBeforeClearingTarget(t *testing.T) {
|
||||||
|
db, mock, err := sqlmock.New()
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("sqlmock.New() error = %v", err)
|
||||||
|
}
|
||||||
|
defer db.Close()
|
||||||
|
|
||||||
|
written, err := writeAggregates(context.Background(), db, map[string]*metricAgg{
|
||||||
|
"LA9GG64L7PBAF4001|2026-07-08|JT808|JT808:13307765812@": {
|
||||||
|
VIN: "LA9GG64L7PBAF4001",
|
||||||
|
Date: "2026-07-08",
|
||||||
|
Protocol: envelope.ProtocolJT808,
|
||||||
|
FirstKM: 4100,
|
||||||
|
LatestKM: 4101,
|
||||||
|
Count: 1,
|
||||||
|
SourceKey: "JT808:13307765812@",
|
||||||
|
Phone: "13307765812",
|
||||||
|
SourceEndpoint: "",
|
||||||
|
FirstEventTime: time.Date(2026, 7, 8, 12, 0, 0, 0, time.FixedZone("Asia/Shanghai", 8*3600)),
|
||||||
|
LatestEventTime: time.Date(2026, 7, 8, 12, 0, 0, 0, time.FixedZone("Asia/Shanghai", 8*3600)),
|
||||||
|
QualityStatus: stats.QualityNoPreviousBaseline,
|
||||||
|
QualityReason: "missing_previous_source",
|
||||||
|
},
|
||||||
|
}, 500)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("writeAggregates() error = %v", err)
|
||||||
|
}
|
||||||
|
if written != 0 {
|
||||||
|
t.Fatalf("written = %d, want 0", written)
|
||||||
|
}
|
||||||
|
if err := mock.ExpectationsWereMet(); err != nil {
|
||||||
|
t.Fatalf("sql expectations: %v", err)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestAddSamplesKeepsScanCandidatesNoPreviousBaseline(t *testing.T) {
|
||||||
|
loc := time.FixedZone("Asia/Shanghai", 8*3600)
|
||||||
|
samples, err := stats.SamplesFromEnvelope(envelope.FrameEnvelope{
|
||||||
|
Protocol: envelope.ProtocolJT808,
|
||||||
|
VIN: "LA9GG64L7PBAF4001",
|
||||||
|
Phone: "13307765812",
|
||||||
|
SourceEndpoint: "115.231.168.135:20215",
|
||||||
|
EventTimeMS: time.Date(2026, 7, 8, 12, 0, 0, 0, loc).UnixMilli(),
|
||||||
|
Fields: map[string]any{
|
||||||
|
"jt808.location.total_mileage_km": 4123.9,
|
||||||
|
},
|
||||||
|
}, loc)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("SamplesFromEnvelope() error = %v", err)
|
||||||
|
}
|
||||||
|
aggregates := map[string]*metricAgg{}
|
||||||
|
addSamples(aggregates, samples)
|
||||||
|
if len(aggregates) != 1 {
|
||||||
|
t.Fatalf("aggregate count = %d, want 1", len(aggregates))
|
||||||
|
}
|
||||||
|
for _, agg := range aggregates {
|
||||||
|
if agg.QualityStatus != stats.QualityNoPreviousBaseline {
|
||||||
|
t.Fatalf("quality = %q, want %q", agg.QualityStatus, stats.QualityNoPreviousBaseline)
|
||||||
|
}
|
||||||
|
if agg.QualityReason != "scan_backfill_missing_previous_source" {
|
||||||
|
t.Fatalf("quality reason = %q", agg.QualityReason)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
func TestLoadConfigDefaultsBackfillMethodToLastDiff(t *testing.T) {
|
func TestLoadConfigDefaultsBackfillMethodToLastDiff(t *testing.T) {
|
||||||
t.Setenv("BACKFILL_METHOD", "")
|
t.Setenv("BACKFILL_METHOD", "")
|
||||||
t.Setenv("BACKFILL_DATE_FROM", "2026-07-08")
|
t.Setenv("BACKFILL_DATE_FROM", "2026-07-08")
|
||||||
|
|||||||
Reference in New Issue
Block a user