Compare commits

..

3 Commits

Author SHA1 Message Date
lingniu
dd95101499 docs: record mileage metric guard verification 2026-07-01 23:11:35 +08:00
lingniu
d347525a75 fix: keep valid realtime mileage 2026-07-01 23:08:31 +08:00
lingniu
19008e840c fix: ignore invalid zero mileage metrics 2026-07-01 23:06:06 +08:00
5 changed files with 173 additions and 5 deletions

View File

@@ -381,3 +381,53 @@ Kafka spool 目录确认:
```
说明:生产 808 中仍有部分 phone 未解析出 VIN后续需要继续维护 `vehicle_identity_binding` 的 phone/device_id/plate 到 VIN 映射,或者增强 808 注册/鉴权帧中的 identity 回写。
## 2026-07-01 23:09 808 里程统计防 0 污染复验
部署版本已更新:
- Git commit`d347525`
- 镜像:`crpi-85r4m0ackrm3qpje.cn-shanghai.personal.cr.aliyuncs.com/oneos/vehicle-gateway-go:go-d347525-20260701230845`
本轮修复内容:
- `stat-writer` 不再把 `total_mileage_km <= 0` 的采样写入 MySQL 每日指标。
- MySQL upsert 对历史 `first_total_mileage_km=0` / `latest_total_mileage_km=0` 做纠偏;后续首次有效总里程会替换掉历史 0。
- `realtime-api` 合并实时快照时,非正 `total_mileage_km` 不再覆盖已有正值;速度、位置、状态等其他字段仍按事件时间更新。
测试验证:
```text
go test ./...
GOOS=linux GOARCH=amd64 CGO_ENABLED=0 go build ./cmd/gateway
GOOS=linux GOARCH=amd64 CGO_ENABLED=0 go build ./cmd/history-writer
GOOS=linux GOARCH=amd64 CGO_ENABLED=0 go build ./cmd/stat-writer
GOOS=linux GOARCH=amd64 CGO_ENABLED=0 go build ./cmd/realtime-api
```
生产回放验证:
- 第一条 808 帧:`device_time=2026-07-01T23:09:35+08:00``total_mileage_km=12345.6`
- 第二条 808 帧:`device_time=2026-07-01T23:09:36+08:00``total_mileage_km=0`
- 两条帧使用同一 VIN`LKLG7C4E3NA774736`
Redis merged 快照确认:第二条 0 里程帧没有覆盖已有有效总里程。
```text
event_time_ms=1782918576000
source_endpoint=172.20.0.1:40946
speed_kmh=0
longitude=119.557997
latitude=29.049092
total_mileage_km=12345.6
field_times_ms.total_mileage_km=1782918575000
```
MySQL `vehicle_daily_metric` 确认0 里程采样未进入统计sample_count 只因有效总里程采样增加一次。
```text
LKLG7C4E3NA774736 | JT808 | 2026-07-01 | daily_mileage_km | 0.000 | sample_count=13 | first=12345.600 | latest=12345.600
LKLG7C4E3NA774736 | JT808 | 2026-07-01 | daily_total_mileage_km | 12345.600 | sample_count=13 | first=12345.600 | latest=12345.600
```
说明:生产 TDengine 中已经存在大量 808 非零 `total_mileage_km` 采样,但许多 phone 尚未映射到 VIN因此 MySQL 按 VIN 的统计只会覆盖已解析 VIN 的车辆。下一步应继续补齐 `vehicle_identity_binding`,让更多 808 车辆进入统计。

View File

@@ -5,6 +5,7 @@ import (
"encoding/json"
"errors"
"sort"
"strconv"
"strings"
"time"
@@ -155,6 +156,9 @@ func mergeFields(snapshot *Snapshot, fields map[string]any, eventMS int64) {
snapshot.FieldTimesMS = map[string]int64{}
}
for key, value := range fields {
if key == envelope.FieldTotalMileageKM && !positiveNumber(value) {
continue
}
if eventMS >= snapshot.FieldTimesMS[key] {
snapshot.Fields[key] = value
snapshot.FieldTimesMS[key] = eventMS
@@ -162,6 +166,28 @@ func mergeFields(snapshot *Snapshot, fields map[string]any, eventMS int64) {
}
}
func positiveNumber(value any) bool {
switch typed := value.(type) {
case float64:
return typed > 0
case float32:
return typed > 0
case int:
return typed > 0
case int64:
return typed > 0
case uint16:
return typed > 0
case uint32:
return typed > 0
case string:
parsed, err := strconv.ParseFloat(strings.TrimSpace(typed), 64)
return err == nil && parsed > 0
default:
return false
}
}
func cloneFields(fields map[string]any) map[string]any {
if len(fields) == 0 {
return map[string]any{}

View File

@@ -94,6 +94,47 @@ func TestRepositoryOnlineStatus(t *testing.T) {
}
}
func TestRepositoryDoesNotOverwritePositiveMileageWithZero(t *testing.T) {
repo, closeFn := newTestRepository(t)
defer closeFn()
ctx := context.Background()
if err := repo.Update(ctx, envelope.FrameEnvelope{
Protocol: envelope.ProtocolJT808,
VIN: "VIN001",
EventTimeMS: 1000,
ReceivedAtMS: 1100,
Fields: map[string]any{
envelope.FieldTotalMileageKM: 12345.6,
},
}); err != nil {
t.Fatalf("Update() error = %v", err)
}
if err := repo.Update(ctx, envelope.FrameEnvelope{
Protocol: envelope.ProtocolJT808,
VIN: "VIN001",
EventTimeMS: 2000,
ReceivedAtMS: 2100,
Fields: map[string]any{
envelope.FieldSpeedKMH: 22.0,
envelope.FieldTotalMileageKM: 0,
},
}); err != nil {
t.Fatalf("Update() error = %v", err)
}
merged, err := repo.GetMerged(ctx, "VIN001")
if err != nil {
t.Fatalf("GetMerged() error = %v", err)
}
if merged.Fields[envelope.FieldTotalMileageKM] != 12345.6 {
t.Fatalf("zero mileage overwrote positive value: %#v", merged.Fields)
}
if merged.Fields[envelope.FieldSpeedKMH] != 22.0 {
t.Fatalf("new speed should still merge: %#v", merged.Fields)
}
}
func TestHandlerReturnsMergedSnapshot(t *testing.T) {
repo, closeFn := newTestRepository(t)
defer closeFn()

View File

@@ -78,6 +78,9 @@ func SamplesFromEnvelope(env envelope.FrameEnvelope, loc *time.Location) ([]Metr
if !ok {
return nil, nil
}
if totalMileage <= 0 {
return nil, nil
}
if loc == nil {
loc = time.FixedZone("Asia/Shanghai", 8*3600)
}
@@ -115,14 +118,40 @@ INSERT INTO vehicle_daily_metric
first_total_mileage_km, latest_total_mileage_km, sample_count, calculation_method)
VALUES (?, ?, ?, ?, ?, 'km', ?, ?, 1, 'TOTAL_MILEAGE_DIFF')
ON DUPLICATE KEY UPDATE
first_total_mileage_km = LEAST(first_total_mileage_km, VALUES(first_total_mileage_km)),
latest_total_mileage_km = GREATEST(latest_total_mileage_km, VALUES(latest_total_mileage_km)),
first_total_mileage_km = CASE
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 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,
metric_value = CASE
WHEN metric_key = 'daily_mileage_km'
THEN GREATEST(latest_total_mileage_km, VALUES(latest_total_mileage_km))
- LEAST(first_total_mileage_km, VALUES(first_total_mileage_km))
THEN GREATEST(
CASE
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 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
WHEN metric_key = 'daily_total_mileage_km'
THEN GREATEST(latest_total_mileage_km, VALUES(latest_total_mileage_km))
THEN GREATEST(
CASE
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)
)
ELSE VALUES(metric_value)
END,
sample_count = sample_count + 1,

View File

@@ -64,6 +64,25 @@ func TestSamplesFromEnvelopeSkipsMissingVINOrMileage(t *testing.T) {
}
}
func TestSamplesFromEnvelopeSkipsNonPositiveMileage(t *testing.T) {
for _, value := range []any{0, 0.0, -1.0, "0"} {
samples, err := SamplesFromEnvelope(envelope.FrameEnvelope{
Protocol: envelope.ProtocolJT808,
VIN: "LNBVIN00000000001",
EventTimeMS: time.Date(2026, 7, 1, 9, 0, 0, 0, time.FixedZone("Asia/Shanghai", 8*3600)).UnixMilli(),
Fields: map[string]any{
envelope.FieldTotalMileageKM: value,
},
}, nil)
if err != nil {
t.Fatalf("SamplesFromEnvelope(%#v) error = %v", value, err)
}
if len(samples) != 0 {
t.Fatalf("expected no samples for non-positive mileage %#v, got %#v", value, samples)
}
}
}
func TestWriterEnsuresSchemaAndUpsertsTwoMetrics(t *testing.T) {
exec := &recordingExec{}
writer := NewWriter(exec, time.FixedZone("Asia/Shanghai", 8*3600))
@@ -90,6 +109,9 @@ func TestWriterEnsuresSchemaAndUpsertsTwoMetrics(t *testing.T) {
if !strings.Contains(exec.calls[1].query, "ON DUPLICATE KEY UPDATE") {
t.Fatalf("unexpected upsert sql: %s", exec.calls[1].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)
}
}
type execCall struct {