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

fix(platform): degrade vehicles with missing sources

Browse Source
This commit is contained in:
lingniu
2026-07-04 05:46:56 +08:00
parent 5e7c4b56ef
commit 6229f837a0
4 changed files with 55 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files

32
vehicle-data-platform/apps/api/internal/platform/handler_test.go
View File

@@ -117,8 +117,8 @@ func TestHandlerVehicleCoverageFiltersServiceStatus(t *testing.T) {
if !strings.Contains(rec.Body.String(), "LB9A32A24R0LS1426") { if !strings.Contains(rec.Body.String(), "LB9A32A24R0LS1426") {
t.Fatalf("degraded coverage should include partially online multi-source vehicle: %s", rec.Body.String()) t.Fatalf("degraded coverage should include partially online multi-source vehicle: %s", rec.Body.String())
} }
if strings.Contains(rec.Body.String(), "LMRKH9AC2R1004087") { if !strings.Contains(rec.Body.String(), "LMRKH9AC2R1004087") {
t.Fatalf("degraded coverage should exclude healthy single-source vehicle: %s", rec.Body.String()) t.Fatalf("degraded coverage should include online single-source vehicle that misses canonical sources: %s", rec.Body.String())
} }
} }
@@ -138,6 +138,30 @@ func TestHandlerVehicleCoverageFiltersMissingProtocol(t *testing.T) {
} }
} }
func TestHandlerVehicleCoverageTreatsMissingCanonicalSourceAsDegraded(t *testing.T) {
handler := NewHandler(NewService(NewMockStore()))
rec := httptest.NewRecorder()
req := httptest.NewRequest(http.MethodGet, "/api/vehicles/coverage?limit=10&missingProtocol=YUTONG_MQTT&online=online", nil)
handler.ServeHTTP(rec, req)
if rec.Code != http.StatusOK {
t.Fatalf("status = %d body=%s", rec.Code, rec.Body.String())
}
var body struct {
Data Page[VehicleCoverageRow] `json:"data"`
}
if err := json.Unmarshal(rec.Body.Bytes(), &body); err != nil {
t.Fatalf("response JSON should decode: %v body=%s", err, rec.Body.String())
}
if len(body.Data.Items) == 0 {
t.Fatalf("missing source query should return online vehicles with missing canonical source: %s", rec.Body.String())
}
for _, row := range body.Data.Items {
if row.ServiceStatus == nil || row.ServiceStatus.Status != "degraded" {
t.Fatalf("missing canonical source should degrade vehicle service, got row=%+v body=%s", row, rec.Body.String())
}
}
}
func TestHandlerVehicleCoverageSummary(t *testing.T) { func TestHandlerVehicleCoverageSummary(t *testing.T) {
handler := NewHandler(NewService(NewMockStore())) handler := NewHandler(NewService(NewMockStore()))
rec := httptest.NewRecorder() rec := httptest.NewRecorder()
@@ -607,8 +631,8 @@ func TestHandlerVehicleRealtimeFiltersServiceStatus(t *testing.T) {
if !strings.Contains(rec.Body.String(), "LB9A32A24R0LS1426") { if !strings.Contains(rec.Body.String(), "LB9A32A24R0LS1426") {
t.Fatalf("degraded realtime should include partially online vehicle: %s", rec.Body.String()) t.Fatalf("degraded realtime should include partially online vehicle: %s", rec.Body.String())
} }
if strings.Contains(rec.Body.String(), "LNXNEGRR7SR318212") { if !strings.Contains(rec.Body.String(), "LNXNEGRR7SR318212") {
t.Fatalf("degraded realtime should exclude healthy single-source vehicle: %s", rec.Body.String()) t.Fatalf("degraded realtime should include online single-source vehicle that misses canonical sources: %s", rec.Body.String())
} }
} }

20
vehicle-data-platform/apps/api/internal/platform/mysql_queries.go
View File

@@ -16,6 +16,7 @@ type SQLQuery struct {
func buildVehicleListSQL(query url.Values) SQLQuery { func buildVehicleListSQL(query url.Values) SQLQuery {
limit := parsePositive(query.Get("limit"), 20) limit := parsePositive(query.Get("limit"), 20)
offset := parsePositive(query.Get("offset"), 0) offset := parsePositive(query.Get("offset"), 0)
canonicalSourceCount := strconv.Itoa(len(canonicalVehicleProtocols))
args := []any{} args := []any{}
where := []string{"1 = 1"} where := []string{"1 = 1"}
if keyword := strings.TrimSpace(query.Get("keyword")); keyword != "" { if keyword := strings.TrimSpace(query.Get("keyword")); keyword != "" {
@@ -33,9 +34,9 @@ func buildVehicleListSQL(query url.Values) SQLQuery {
case "offline": case "offline":
where = append(where, "b.vin IS NOT NULL", "b.vin <> ''", "COALESCE(vs.source_count, 0) > 0", "COALESCE(vs.online_source_count, 0) = 0") where = append(where, "b.vin IS NOT NULL", "b.vin <> ''", "COALESCE(vs.source_count, 0) > 0", "COALESCE(vs.online_source_count, 0) = 0")
case "degraded": case "degraded":
where = append(where, "b.vin IS NOT NULL", "b.vin <> ''", "vs.source_count > 0", "vs.online_source_count > 0", "vs.online_source_count < vs.source_count") where = append(where, "b.vin IS NOT NULL", "b.vin <> ''", "vs.source_count > 0", "vs.online_source_count > 0", "(vs.online_source_count < vs.source_count OR vs.source_count < "+canonicalSourceCount+")")
case "healthy": case "healthy":
where = append(where, "b.vin IS NOT NULL", "b.vin <> ''", "vs.source_count > 0", "vs.online_source_count = vs.source_count") where = append(where, "b.vin IS NOT NULL", "b.vin <> ''", "vs.source_count = "+canonicalSourceCount, "vs.online_source_count = vs.source_count")
} }
countArgs := append([]any(nil), args...) countArgs := append([]any(nil), args...)
args = append(args, limit, offset) args = append(args, limit, offset)
@@ -66,6 +67,7 @@ func buildVehicleListSQL(query url.Values) SQLQuery {
func buildVehicleCoverageSQL(query url.Values) SQLQuery { func buildVehicleCoverageSQL(query url.Values) SQLQuery {
limit := parsePositive(query.Get("limit"), 20) limit := parsePositive(query.Get("limit"), 20)
offset := parsePositive(query.Get("offset"), 0) offset := parsePositive(query.Get("offset"), 0)
canonicalSourceCount := strconv.Itoa(len(canonicalVehicleProtocols))
args := []any{} args := []any{}
where := []string{"v.vin IS NOT NULL", "v.vin <> ''"} where := []string{"v.vin IS NOT NULL", "v.vin <> ''"}
having := []string{} having := []string{}
@@ -114,10 +116,10 @@ func buildVehicleCoverageSQL(query url.Values) SQLQuery {
having = append(having, "MAX(CASE WHEN b.vin IS NOT NULL AND b.vin <> '' THEN 1 ELSE 0 END) = 1") having = append(having, "MAX(CASE WHEN b.vin IS NOT NULL AND b.vin <> '' THEN 1 ELSE 0 END) = 1")
having = append(having, "COUNT(DISTINCT s.protocol) > 0") having = append(having, "COUNT(DISTINCT s.protocol) > 0")
having = append(having, "COUNT(DISTINCT CASE WHEN s.updated_at >= DATE_SUB(NOW(), INTERVAL 1 MINUTE) THEN s.protocol END) > 0") having = append(having, "COUNT(DISTINCT CASE WHEN s.updated_at >= DATE_SUB(NOW(), INTERVAL 1 MINUTE) THEN s.protocol END) > 0")
having = append(having, "COUNT(DISTINCT CASE WHEN s.updated_at >= DATE_SUB(NOW(), INTERVAL 1 MINUTE) THEN s.protocol END) < COUNT(DISTINCT s.protocol)") having = append(having, "(COUNT(DISTINCT CASE WHEN s.updated_at >= DATE_SUB(NOW(), INTERVAL 1 MINUTE) THEN s.protocol END) < COUNT(DISTINCT s.protocol) OR COUNT(DISTINCT s.protocol) < "+canonicalSourceCount+")")
case "healthy": case "healthy":
having = append(having, "MAX(CASE WHEN b.vin IS NOT NULL AND b.vin <> '' THEN 1 ELSE 0 END) = 1") having = append(having, "MAX(CASE WHEN b.vin IS NOT NULL AND b.vin <> '' THEN 1 ELSE 0 END) = 1")
having = append(having, "COUNT(DISTINCT s.protocol) > 0") having = append(having, "COUNT(DISTINCT s.protocol) = "+canonicalSourceCount)
having = append(having, "COUNT(DISTINCT CASE WHEN s.updated_at >= DATE_SUB(NOW(), INTERVAL 1 MINUTE) THEN s.protocol END) = COUNT(DISTINCT s.protocol)") having = append(having, "COUNT(DISTINCT CASE WHEN s.updated_at >= DATE_SUB(NOW(), INTERVAL 1 MINUTE) THEN s.protocol END) = COUNT(DISTINCT s.protocol)")
} }
countArgs := append([]any(nil), args...) countArgs := append([]any(nil), args...)
@@ -153,6 +155,7 @@ func buildVehicleCoverageSQL(query url.Values) SQLQuery {
} }
func buildVehicleCoverageSummarySQL(query url.Values) SQLQuery { func buildVehicleCoverageSummarySQL(query url.Values) SQLQuery {
canonicalSourceCount := strconv.Itoa(len(canonicalVehicleProtocols))
args := []any{} args := []any{}
where := []string{"v.vin IS NOT NULL", "v.vin <> ''"} where := []string{"v.vin IS NOT NULL", "v.vin <> ''"}
having := []string{} having := []string{}
@@ -201,10 +204,10 @@ func buildVehicleCoverageSummarySQL(query url.Values) SQLQuery {
having = append(having, "MAX(CASE WHEN b.vin IS NOT NULL AND b.vin <> '' THEN 1 ELSE 0 END) = 1") having = append(having, "MAX(CASE WHEN b.vin IS NOT NULL AND b.vin <> '' THEN 1 ELSE 0 END) = 1")
having = append(having, "COUNT(DISTINCT s.protocol) > 0") having = append(having, "COUNT(DISTINCT s.protocol) > 0")
having = append(having, "COUNT(DISTINCT CASE WHEN s.updated_at >= DATE_SUB(NOW(), INTERVAL 1 MINUTE) THEN s.protocol END) > 0") having = append(having, "COUNT(DISTINCT CASE WHEN s.updated_at >= DATE_SUB(NOW(), INTERVAL 1 MINUTE) THEN s.protocol END) > 0")
having = append(having, "COUNT(DISTINCT CASE WHEN s.updated_at >= DATE_SUB(NOW(), INTERVAL 1 MINUTE) THEN s.protocol END) < COUNT(DISTINCT s.protocol)") having = append(having, "(COUNT(DISTINCT CASE WHEN s.updated_at >= DATE_SUB(NOW(), INTERVAL 1 MINUTE) THEN s.protocol END) < COUNT(DISTINCT s.protocol) OR COUNT(DISTINCT s.protocol) < "+canonicalSourceCount+")")
case "healthy": case "healthy":
having = append(having, "MAX(CASE WHEN b.vin IS NOT NULL AND b.vin <> '' THEN 1 ELSE 0 END) = 1") having = append(having, "MAX(CASE WHEN b.vin IS NOT NULL AND b.vin <> '' THEN 1 ELSE 0 END) = 1")
having = append(having, "COUNT(DISTINCT s.protocol) > 0") having = append(having, "COUNT(DISTINCT s.protocol) = "+canonicalSourceCount)
having = append(having, "COUNT(DISTINCT CASE WHEN s.updated_at >= DATE_SUB(NOW(), INTERVAL 1 MINUTE) THEN s.protocol END) = COUNT(DISTINCT s.protocol)") having = append(having, "COUNT(DISTINCT CASE WHEN s.updated_at >= DATE_SUB(NOW(), INTERVAL 1 MINUTE) THEN s.protocol END) = COUNT(DISTINCT s.protocol)")
} }
havingSQL := "" havingSQL := ""
@@ -265,6 +268,7 @@ func buildRealtimeLocationSQL(query url.Values) SQLQuery {
func buildVehicleRealtimeSQL(query url.Values) SQLQuery { func buildVehicleRealtimeSQL(query url.Values) SQLQuery {
limit := parsePositive(query.Get("limit"), 20) limit := parsePositive(query.Get("limit"), 20)
offset := parsePositive(query.Get("offset"), 0) offset := parsePositive(query.Get("offset"), 0)
canonicalSourceCount := strconv.Itoa(len(canonicalVehicleProtocols))
args := []any{} args := []any{}
where := []string{"l.vin IS NOT NULL", "l.vin <> ''"} where := []string{"l.vin IS NOT NULL", "l.vin <> ''"}
having := []string{} having := []string{}
@@ -287,7 +291,7 @@ func buildVehicleRealtimeSQL(query url.Values) SQLQuery {
case "healthy": case "healthy":
having = append(having, having = append(having,
"MAX(CASE WHEN b.vin IS NOT NULL THEN 1 ELSE 0 END) = 1", "MAX(CASE WHEN b.vin IS NOT NULL THEN 1 ELSE 0 END) = 1",
"COUNT(DISTINCT l.protocol) > 0", "COUNT(DISTINCT l.protocol) = "+canonicalSourceCount,
"COUNT(DISTINCT CASE WHEN l.updated_at >= DATE_SUB(NOW(), INTERVAL 1 MINUTE) THEN l.protocol END) = COUNT(DISTINCT l.protocol)", "COUNT(DISTINCT CASE WHEN l.updated_at >= DATE_SUB(NOW(), INTERVAL 1 MINUTE) THEN l.protocol END) = COUNT(DISTINCT l.protocol)",
) )
case "degraded": case "degraded":
@@ -295,7 +299,7 @@ func buildVehicleRealtimeSQL(query url.Values) SQLQuery {
"MAX(CASE WHEN b.vin IS NOT NULL THEN 1 ELSE 0 END) = 1", "MAX(CASE WHEN b.vin IS NOT NULL THEN 1 ELSE 0 END) = 1",
"COUNT(DISTINCT l.protocol) > 0", "COUNT(DISTINCT l.protocol) > 0",
"COUNT(DISTINCT CASE WHEN l.updated_at >= DATE_SUB(NOW(), INTERVAL 1 MINUTE) THEN l.protocol END) > 0", "COUNT(DISTINCT CASE WHEN l.updated_at >= DATE_SUB(NOW(), INTERVAL 1 MINUTE) THEN l.protocol END) > 0",
"COUNT(DISTINCT CASE WHEN l.updated_at >= DATE_SUB(NOW(), INTERVAL 1 MINUTE) THEN l.protocol END) < COUNT(DISTINCT l.protocol)", "(COUNT(DISTINCT CASE WHEN l.updated_at >= DATE_SUB(NOW(), INTERVAL 1 MINUTE) THEN l.protocol END) < COUNT(DISTINCT l.protocol) OR COUNT(DISTINCT l.protocol) < "+canonicalSourceCount+")",
) )
case "offline": case "offline":
having = append(having, having = append(having,

5
vehicle-data-platform/apps/api/internal/platform/production_store.go
View File

@@ -96,6 +96,7 @@ func (s *ProductionStore) serviceStatusStats(ctx context.Context) ([]ServiceStat
} }
func (s *ProductionStore) VehicleServiceSummary(ctx context.Context) (VehicleServiceSummary, error) { func (s *ProductionStore) VehicleServiceSummary(ctx context.Context) (VehicleServiceSummary, error) {
canonicalSourceCount := strconv.Itoa(len(canonicalVehicleProtocols))
rows, err := s.db.QueryContext(ctx, `SELECT rows, err := s.db.QueryContext(ctx, `SELECT
COUNT(*) AS total_vehicles, COUNT(*) AS total_vehicles,
COALESCE(SUM(CASE WHEN bound = 1 THEN 1 ELSE 0 END), 0) AS bound_vehicles, COALESCE(SUM(CASE WHEN bound = 1 THEN 1 ELSE 0 END), 0) AS bound_vehicles,
@@ -104,8 +105,8 @@ COALESCE(SUM(CASE WHEN source_count = 1 THEN 1 ELSE 0 END), 0) AS single_source_
COALESCE(SUM(CASE WHEN source_count > 1 THEN 1 ELSE 0 END), 0) AS multi_source_vehicles, COALESCE(SUM(CASE WHEN source_count > 1 THEN 1 ELSE 0 END), 0) AS multi_source_vehicles,
COALESCE(SUM(CASE WHEN source_count = 0 THEN 1 ELSE 0 END), 0) AS no_data_vehicles, COALESCE(SUM(CASE WHEN source_count = 0 THEN 1 ELSE 0 END), 0) AS no_data_vehicles,
COALESCE(SUM(CASE WHEN bound = 0 THEN 1 ELSE 0 END), 0) AS identity_required_vehicles, COALESCE(SUM(CASE WHEN bound = 0 THEN 1 ELSE 0 END), 0) AS identity_required_vehicles,
COALESCE(SUM(CASE WHEN bound = 1 AND source_count > 0 AND online_source_count = source_count THEN 1 ELSE 0 END), 0) AS healthy_vehicles, COALESCE(SUM(CASE WHEN bound = 1 AND source_count = `+canonicalSourceCount+` AND online_source_count = source_count THEN 1 ELSE 0 END), 0) AS healthy_vehicles,
COALESCE(SUM(CASE WHEN bound = 1 AND source_count > 0 AND online_source_count > 0 AND online_source_count < source_count THEN 1 ELSE 0 END), 0) AS degraded_vehicles, COALESCE(SUM(CASE WHEN bound = 1 AND source_count > 0 AND online_source_count > 0 AND (online_source_count < source_count OR source_count < `+canonicalSourceCount+`) THEN 1 ELSE 0 END), 0) AS degraded_vehicles,
COALESCE(SUM(CASE WHEN bound = 1 AND source_count > 0 AND online_source_count = 0 THEN 1 ELSE 0 END), 0) AS offline_vehicles COALESCE(SUM(CASE WHEN bound = 1 AND source_count > 0 AND online_source_count = 0 THEN 1 ELSE 0 END), 0) AS offline_vehicles
FROM ( FROM (
SELECT v.vin, SELECT v.vin,

13
vehicle-data-platform/apps/api/internal/platform/service.go
View File

@@ -980,11 +980,21 @@ func buildVehicleCoverageServiceStatus(row VehicleCoverageRow) *VehicleServiceSt
OnlineSourceCount: row.OnlineSourceCount, OnlineSourceCount: row.OnlineSourceCount,
} }
} }
if len(row.MissingProtocols) > 0 {
return &VehicleServiceStatus{
Status: "degraded",
Severity: "warning",
Title: "部分来源离线",
Detail: sourceCoverageDetail(row.SourceCount, row.OnlineSourceCount, "车辆服务可用但缺少 "+strings.Join(row.MissingProtocols, "、")+" 来源。"),
SourceCount: row.SourceCount,
OnlineSourceCount: row.OnlineSourceCount,
}
}
return &VehicleServiceStatus{ return &VehicleServiceStatus{
Status: "healthy", Status: "healthy",
Severity: "ok", Severity: "ok",
Title: "服务正常", Title: "服务正常",
Detail: sourceCoverageDetail(row.SourceCount, row.OnlineSourceCount, "全部已知来源在线。"), Detail: sourceCoverageDetail(row.SourceCount, row.OnlineSourceCount, "全部来源在线。"),
SourceCount: row.SourceCount, SourceCount: row.SourceCount,
OnlineSourceCount: row.OnlineSourceCount, OnlineSourceCount: row.OnlineSourceCount,
} }
@@ -997,6 +1007,7 @@ func buildRealtimeServiceStatus(row VehicleRealtimeRow) *VehicleServiceStatus {
Phone: row.Phone, Phone: row.Phone,
OEM: row.OEM, OEM: row.OEM,
Protocols: row.Protocols, Protocols: row.Protocols,
MissingProtocols: missingCanonicalProtocols(row.Protocols),
SourceCount: row.SourceCount, SourceCount: row.SourceCount,
OnlineSourceCount: row.OnlineSourceCount, OnlineSourceCount: row.OnlineSourceCount,
Online: row.Online, Online: row.Online,