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

fix(platform): verify storage health by reads

Browse Source
This commit is contained in:
lingniu
2026-07-03 22:43:36 +08:00
parent 2c24d4734c
commit 354c146506
1 changed files with 39 additions and 19 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -377,20 +377,54 @@ func (s *ProductionStore) OpsHealth(ctx context.Context) (OpsHealth, error) {
mysqlStatus = "error"
mysqlDetail = err.Error()
}
snapshotHealth := s.mysqlTableReadHealth(ctx, "vehicle_realtime_snapshot", "读取实时快照表正常")
locationHealth := s.mysqlTableReadHealth(ctx, "vehicle_realtime_location", "读取实时位置表正常")
tdengineHealth := s.tdengineRawFrameHealth(ctx)
mysqlWritable := mysqlStatus == "ok" && snapshotHealth.Status == "ok" && locationHealth.Status == "ok"
return OpsHealth{
LinkHealth: []LinkHealth{
{Name: "MySQL realtime", Status: mysqlStatus, Detail: mysqlDetail},
{Name: "vehicle_realtime_snapshot", Status: "ok", Detail: "读取实时快照表"},
{Name: "vehicle_realtime_location", Status: "ok", Detail: "读取实时位置表"},
{Name: "TDengine raw_frames", Status: tdengineStatus(s.tdengine), Detail: tdengineDetail(s.tdengine)},
snapshotHealth,
locationHealth,
tdengineHealth,
{Name: "Kafka lag", Status: "warning", Detail: "平台暂未接入 Kafka consumer lag 监控"},
{Name: "Redis online keys", Status: "warning", Detail: "平台暂未接入 Redis 在线 key 读取"},
},
TDengineWritable: s.tdengine != nil,
MySQLWritable: mysqlStatus == "ok",
TDengineWritable: tdengineHealth.Status == "ok",
MySQLWritable: mysqlWritable,
}, nil
}
func (s *ProductionStore) mysqlTableReadHealth(ctx context.Context, table string, okDetail string) LinkHealth {
health := LinkHealth{Name: table, Status: "ok", Detail: okDetail}
query := "SELECT 1 FROM " + table + " LIMIT 1"
rows, err := s.db.QueryContext(ctx, query)
if err != nil {
health.Status = "error"
health.Detail = err.Error()
return health
}
_ = rows.Close()
return health
}
func (s *ProductionStore) tdengineRawFrameHealth(ctx context.Context) LinkHealth {
health := LinkHealth{Name: "TDengine raw_frames", Status: "ok", Detail: "读取 TDengine raw_frames 正常"}
if s.tdengine == nil {
health.Status = "warning"
health.Detail = "TDengine 未配置或连接失败，历史查询降级"
return health
}
rows, err := s.tdengine.QueryContext(ctx, "SELECT ts FROM "+qualifyTDengine(s.tdDatabase, "raw_frames")+" LIMIT 1")
if err != nil {
health.Status = "error"
health.Detail = err.Error()
return health
}
_ = rows.Close()
return health
}
func (s *ProductionStore) protocolStats(ctx context.Context) ([]ProtocolStat, error) {
rows, err := s.db.QueryContext(ctx, `SELECT protocol, SUM(CASE WHEN updated_at >= DATE_SUB(NOW(), INTERVAL 1 MINUTE) THEN 1 ELSE 0 END) AS online_count, COUNT(*) AS total_count FROM vehicle_realtime_snapshot GROUP BY protocol ORDER BY protocol`)
if err != nil {
@@ -451,17 +485,3 @@ func firstNonEmpty(values ...string) string {
}
return ""
}
func tdengineStatus(db *sql.DB) string {
if db == nil {
return "warning"
}
return "ok"
}
func tdengineDetail(db *sql.DB) string {
if db == nil {
return "TDengine 未配置或连接失败，历史查询降级"
}
return "读取 TDengine raw_frames / vehicle_locations"
}