Compare commits
3 Commits
c803a19bf5
...
dd95101499
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
dd95101499 | ||
|
|
d347525a75 | ||
|
|
19008e840c |
@@ -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 车辆进入统计。
|
||||
|
||||
@@ -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{}
|
||||
|
||||
@@ -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()
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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 {
|
||||
|
||||
Reference in New Issue
Block a user