feat(platform): add source readiness operations view
This commit is contained in:
@@ -51,6 +51,7 @@ func (h *Handler) routes() {
|
||||
h.mux.HandleFunc("GET /api/alert-events", h.handleQualityIssues)
|
||||
h.mux.HandleFunc("GET /api/alert-events/notification-plan", h.handleQualityNotificationPlan)
|
||||
h.mux.HandleFunc("GET /api/ops/health", h.handleOpsHealth)
|
||||
h.mux.HandleFunc("GET /api/ops/source-readiness", h.handleSourceReadiness)
|
||||
}
|
||||
|
||||
func (h *Handler) handleDashboardSummary(w http.ResponseWriter, r *http.Request) {
|
||||
@@ -205,6 +206,11 @@ func (h *Handler) handleOpsHealth(w http.ResponseWriter, r *http.Request) {
|
||||
h.write(w, r, data, err)
|
||||
}
|
||||
|
||||
func (h *Handler) handleSourceReadiness(w http.ResponseWriter, r *http.Request) {
|
||||
data, err := h.service.SourceReadiness(r.Context())
|
||||
h.write(w, r, data, err)
|
||||
}
|
||||
|
||||
func (h *Handler) write(w http.ResponseWriter, r *http.Request, data any, err error) {
|
||||
if err != nil {
|
||||
if clientErr, ok := asClientError(err); ok {
|
||||
|
||||
@@ -939,6 +939,44 @@ func TestHandlerOpsHealthIncludesVehicleServiceRuntime(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestHandlerSourceReadiness(t *testing.T) {
|
||||
handler := NewHandler(NewServiceWithRuntime(NewMockStore(), RuntimeInfo{PlatformRelease: "platform-source-test"}))
|
||||
rec := httptest.NewRecorder()
|
||||
req := httptest.NewRequest(http.MethodGet, "/api/ops/source-readiness", 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 SourceReadinessPlan `json:"data"`
|
||||
}
|
||||
if err := json.Unmarshal(rec.Body.Bytes(), &body); err != nil {
|
||||
t.Fatalf("response JSON should decode source readiness: %v body=%s", err, rec.Body.String())
|
||||
}
|
||||
if body.Data.PlatformRelease != "platform-source-test" {
|
||||
t.Fatalf("source readiness should expose runtime release, got %+v", body.Data)
|
||||
}
|
||||
if len(body.Data.Sources) < len(canonicalVehicleProtocols) {
|
||||
t.Fatalf("source readiness should expose canonical protocol slots, got %+v", body.Data.Sources)
|
||||
}
|
||||
byProtocol := map[string]SourceReadinessRow{}
|
||||
for _, source := range body.Data.Sources {
|
||||
byProtocol[source.Protocol] = source
|
||||
}
|
||||
for _, protocol := range canonicalVehicleProtocols {
|
||||
row, ok := byProtocol[protocol]
|
||||
if !ok {
|
||||
t.Fatalf("source readiness missing protocol %s: %+v", protocol, body.Data.Sources)
|
||||
}
|
||||
if row.Role == "" || row.Evidence == "" || row.Action == "" || row.Acceptance == "" {
|
||||
t.Fatalf("source readiness row should be operationally actionable: %+v", row)
|
||||
}
|
||||
if !strings.Contains(row.VehiclesHash, "missingProtocol="+protocol) || !strings.Contains(row.RealtimeHash, "protocol="+protocol) {
|
||||
t.Fatalf("source readiness should include workflow hashes, got %+v", row)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestHandlerVehicleDataAPIsResolveVehicleKeyword(t *testing.T) {
|
||||
handler := NewHandler(NewService(NewMockStore()))
|
||||
cases := []struct {
|
||||
|
||||
@@ -173,6 +173,34 @@ type MissingSourceStat struct {
|
||||
Count int `json:"count"`
|
||||
}
|
||||
|
||||
type SourceReadinessPlan struct {
|
||||
TotalVehicles int `json:"totalVehicles"`
|
||||
OnlineVehicles int `json:"onlineVehicles"`
|
||||
KafkaLag *int `json:"kafkaLag"`
|
||||
ActiveConnections *int `json:"activeConnections"`
|
||||
RedisOnlineKeys *int `json:"redisOnlineKeys"`
|
||||
PlatformRelease string `json:"platformRelease"`
|
||||
Sources []SourceReadinessRow `json:"sources"`
|
||||
}
|
||||
|
||||
type SourceReadinessRow struct {
|
||||
Protocol string `json:"protocol"`
|
||||
Role string `json:"role"`
|
||||
Online int `json:"online"`
|
||||
Total int `json:"total"`
|
||||
OnlineRate float64 `json:"onlineRate"`
|
||||
MissingVehicles int `json:"missingVehicles"`
|
||||
Severity string `json:"severity"`
|
||||
Status string `json:"status"`
|
||||
Evidence string `json:"evidence"`
|
||||
Action string `json:"action"`
|
||||
Acceptance string `json:"acceptance"`
|
||||
VehiclesHash string `json:"vehiclesHash"`
|
||||
RealtimeHash string `json:"realtimeHash"`
|
||||
HistoryHash string `json:"historyHash"`
|
||||
AlertHash string `json:"alertHash"`
|
||||
}
|
||||
|
||||
type ArchiveMissingFieldStat struct {
|
||||
Field string `json:"field"`
|
||||
Title string `json:"title"`
|
||||
|
||||
@@ -180,6 +180,119 @@ func completeProtocolStats(stats []ProtocolStat) []ProtocolStat {
|
||||
return out
|
||||
}
|
||||
|
||||
func buildSourceReadinessPlan(summary VehicleServiceSummary, health OpsHealth, quality QualitySummary) SourceReadinessPlan {
|
||||
missingByProtocol := make(map[string]int, len(summary.MissingSources))
|
||||
for _, item := range summary.MissingSources {
|
||||
missingByProtocol[strings.TrimSpace(item.Protocol)] = item.Count
|
||||
}
|
||||
issueByType := make(map[string]int, len(quality.IssueTypes))
|
||||
for _, item := range quality.IssueTypes {
|
||||
issueByType[strings.TrimSpace(item.Name)] = item.Count
|
||||
}
|
||||
protocolStats := completeProtocolStats(summary.Protocols)
|
||||
rows := make([]SourceReadinessRow, 0, len(protocolStats))
|
||||
for _, stat := range protocolStats {
|
||||
protocol := strings.TrimSpace(stat.Protocol)
|
||||
if protocol == "" {
|
||||
continue
|
||||
}
|
||||
missing := missingByProtocol[protocol]
|
||||
row := SourceReadinessRow{
|
||||
Protocol: protocol,
|
||||
Role: sourceReadinessRole(protocol),
|
||||
Online: stat.Online,
|
||||
Total: stat.Total,
|
||||
OnlineRate: sourceOnlineRate(stat),
|
||||
MissingVehicles: missing,
|
||||
Severity: "ok",
|
||||
Status: "生产可用",
|
||||
Evidence: sourceReadinessEvidence(stat, missing, health),
|
||||
Action: "保持实时、轨迹、RAW 和统计链路巡检。",
|
||||
Acceptance: "该来源有在线车辆、无消息积压,车辆服务可回查实时、轨迹和 RAW 证据。",
|
||||
VehiclesHash: buildHash("vehicles", "", protocol, map[string]string{"missingProtocol": protocol}),
|
||||
RealtimeHash: buildHash("realtime", "", protocol, map[string]string{"online": "online"}),
|
||||
HistoryHash: buildHash("history-query", "", protocol, map[string]string{"tab": "raw", "includeFields": "true"}),
|
||||
AlertHash: buildHash("alert-events", "", protocol, nil),
|
||||
}
|
||||
if stat.Total <= 0 {
|
||||
row.Severity = "error"
|
||||
row.Status = "未形成来源证据"
|
||||
row.Action = "确认平台账号、端口监听、订阅配置和协议接入服务是否启动。"
|
||||
row.Acceptance = "该协议至少出现绑定车辆和实时 RAW 证据。"
|
||||
} else if stat.Online <= 0 {
|
||||
row.Severity = "error"
|
||||
row.Status = "来源全离线"
|
||||
row.Action = "优先检查上游平台转发、网关连接、登录/鉴权回复和 Redis 在线 TTL。"
|
||||
row.Acceptance = "该协议恢复在线车辆,Redis 在线状态持续续期。"
|
||||
} else if missing > summary.TotalVehicles/2 && summary.TotalVehicles > 0 {
|
||||
row.Severity = "warning"
|
||||
row.Status = "覆盖不足"
|
||||
row.Action = "核对该协议车辆范围、绑定表和平台转发清单,确认缺失车辆是否应接入。"
|
||||
row.Acceptance = "缺失来源车辆下降,单源车辆可解释。"
|
||||
}
|
||||
if health.KafkaLag != nil && *health.KafkaLag > 0 {
|
||||
row.Severity = maxSourceSeverity(row.Severity, "warning")
|
||||
row.Status = row.Status + " / Kafka积压"
|
||||
row.Action = row.Action + " 同时检查 Kafka 消费 Lag,避免历史和统计滞后。"
|
||||
}
|
||||
if issueByType["NO_SOURCE"] > 0 && stat.Total <= 0 {
|
||||
row.Severity = "error"
|
||||
}
|
||||
rows = append(rows, row)
|
||||
}
|
||||
return SourceReadinessPlan{
|
||||
TotalVehicles: summary.TotalVehicles,
|
||||
OnlineVehicles: summary.OnlineVehicles,
|
||||
KafkaLag: health.KafkaLag,
|
||||
ActiveConnections: health.ActiveConnections,
|
||||
RedisOnlineKeys: health.RedisOnlineKeys,
|
||||
PlatformRelease: health.Runtime.PlatformRelease,
|
||||
Sources: rows,
|
||||
}
|
||||
}
|
||||
|
||||
func sourceReadinessRole(protocol string) string {
|
||||
switch strings.ToUpper(strings.TrimSpace(protocol)) {
|
||||
case "GB32960":
|
||||
return "整车与氢能实时数据主来源"
|
||||
case "JT808":
|
||||
return "位置、总里程和设备在线主来源"
|
||||
case "YUTONG_MQTT":
|
||||
return "宇通派生工况与燃料电池补充来源"
|
||||
default:
|
||||
return "扩展接入来源"
|
||||
}
|
||||
}
|
||||
|
||||
func sourceOnlineRate(stat ProtocolStat) float64 {
|
||||
if stat.Total <= 0 {
|
||||
return 0
|
||||
}
|
||||
return float64(stat.Online) / float64(stat.Total) * 100
|
||||
}
|
||||
|
||||
func sourceReadinessEvidence(stat ProtocolStat, missing int, health OpsHealth) string {
|
||||
parts := []string{
|
||||
"在线 " + strconv.Itoa(stat.Online) + "/" + strconv.Itoa(stat.Total),
|
||||
"缺失车辆 " + strconv.Itoa(missing),
|
||||
}
|
||||
if health.KafkaLag != nil {
|
||||
parts = append(parts, "Kafka Lag "+strconv.Itoa(*health.KafkaLag))
|
||||
}
|
||||
if health.RedisOnlineKeys != nil {
|
||||
parts = append(parts, "Redis在线Key "+strconv.Itoa(*health.RedisOnlineKeys))
|
||||
}
|
||||
return strings.Join(parts, ",")
|
||||
}
|
||||
|
||||
func maxSourceSeverity(left string, right string) string {
|
||||
weight := map[string]int{"ok": 0, "warning": 1, "error": 2}
|
||||
if weight[right] > weight[left] {
|
||||
return right
|
||||
}
|
||||
return left
|
||||
}
|
||||
|
||||
func missingCanonicalProtocols(protocols []string) []string {
|
||||
missing := make([]string, 0, len(canonicalVehicleProtocols))
|
||||
for _, protocol := range canonicalVehicleProtocols {
|
||||
@@ -251,6 +364,23 @@ func (s *Service) VehicleServiceSummary(ctx context.Context) (VehicleServiceSumm
|
||||
return s.store.VehicleServiceSummary(ctx)
|
||||
}
|
||||
|
||||
func (s *Service) SourceReadiness(ctx context.Context) (SourceReadinessPlan, error) {
|
||||
summary, err := s.store.VehicleServiceSummary(ctx)
|
||||
if err != nil {
|
||||
return SourceReadinessPlan{}, err
|
||||
}
|
||||
health, err := s.store.OpsHealth(ctx)
|
||||
if err != nil {
|
||||
return SourceReadinessPlan{}, err
|
||||
}
|
||||
quality, err := s.store.QualitySummary(ctx, url.Values{})
|
||||
if err != nil {
|
||||
return SourceReadinessPlan{}, err
|
||||
}
|
||||
health.Runtime = s.runtime
|
||||
return buildSourceReadinessPlan(summary, health, quality), nil
|
||||
}
|
||||
|
||||
func (s *Service) VehicleRealtime(ctx context.Context, query url.Values) (Page[VehicleRealtimeRow], error) {
|
||||
resolvedQuery, err := s.resolveVehicleQuery(ctx, query)
|
||||
if err != nil {
|
||||
|
||||
Reference in New Issue
Block a user