From b26a37f6f7533c55b56b9a585d17e92a3914c4db Mon Sep 17 00:00:00 2001 From: lingniu Date: Sat, 4 Jul 2026 08:49:44 +0800 Subject: [PATCH] feat(platform): report capacity check health --- .../apps/api/internal/app/server.go | 4 ++ .../api/internal/platform/capacity_check.go | 53 +++++++++++++++++++ .../api/internal/platform/production_store.go | 47 ++++++++++++++-- .../platform/production_store_test.go | 44 +++++++++++++++ 4 files changed, 143 insertions(+), 5 deletions(-) create mode 100644 vehicle-data-platform/apps/api/internal/platform/capacity_check.go diff --git a/vehicle-data-platform/apps/api/internal/app/server.go b/vehicle-data-platform/apps/api/internal/app/server.go index cbc0e7c3..19190702 100644 --- a/vehicle-data-platform/apps/api/internal/app/server.go +++ b/vehicle-data-platform/apps/api/internal/app/server.go @@ -38,6 +38,10 @@ func NewServer(cfg config.Config) http.Handler { productionStore.WithRedisOnlineKeyCounter(platform.NewRedisOnlineKeyCounter(cfg.RedisAddr, cfg.RedisUsername, cfg.RedisPassword, cfg.RedisDB)) log.Printf("production redis health probe enabled") } + if cfg.CapacityCheckBin != "" { + productionStore.WithCapacityChecker(platform.NewCapacityCheckCommand(cfg.CapacityCheckBin)) + log.Printf("production capacity-check probe enabled") + } store = productionStore log.Printf("production mysql store enabled") } diff --git a/vehicle-data-platform/apps/api/internal/platform/capacity_check.go b/vehicle-data-platform/apps/api/internal/platform/capacity_check.go new file mode 100644 index 00000000..6decbe22 --- /dev/null +++ b/vehicle-data-platform/apps/api/internal/platform/capacity_check.go @@ -0,0 +1,53 @@ +package platform + +import ( + "context" + "encoding/json" + "errors" + "os/exec" + "strings" +) + +type CapacityCheckResult struct { + Status string `json:"status"` + ActiveConnections int64 `json:"active_connections"` + KafkaLag float64 `json:"kafka_lag"` + Findings []string `json:"findings"` +} + +type capacityCheckReport struct { + Status string `json:"status"` + Totals struct { + ActiveConnections int64 `json:"active_connections"` + KafkaLag float64 `json:"kafka_lag"` + } `json:"totals"` + Findings []string `json:"findings"` +} + +type CapacityCheckCommand struct { + bin string +} + +func NewCapacityCheckCommand(bin string) *CapacityCheckCommand { + return &CapacityCheckCommand{bin: strings.TrimSpace(bin)} +} + +func (c *CapacityCheckCommand) CheckCapacity(ctx context.Context) (CapacityCheckResult, error) { + if c.bin == "" { + return CapacityCheckResult{}, errors.New("capacity-check bin is empty") + } + output, err := exec.CommandContext(ctx, c.bin).CombinedOutput() + var report capacityCheckReport + if jsonErr := json.Unmarshal(output, &report); jsonErr != nil { + if err != nil { + return CapacityCheckResult{}, errors.New(err.Error() + ": " + strings.TrimSpace(string(output))) + } + return CapacityCheckResult{}, jsonErr + } + return CapacityCheckResult{ + Status: report.Status, + ActiveConnections: report.Totals.ActiveConnections, + KafkaLag: report.Totals.KafkaLag, + Findings: report.Findings, + }, nil +} diff --git a/vehicle-data-platform/apps/api/internal/platform/production_store.go b/vehicle-data-platform/apps/api/internal/platform/production_store.go index 8a4d797c..84e73d22 100644 --- a/vehicle-data-platform/apps/api/internal/platform/production_store.go +++ b/vehicle-data-platform/apps/api/internal/platform/production_store.go @@ -11,16 +11,21 @@ import ( ) type ProductionStore struct { - db *sql.DB - tdengine *sql.DB - tdDatabase string - redisOnline redisOnlineKeyCounter + db *sql.DB + tdengine *sql.DB + tdDatabase string + redisOnline redisOnlineKeyCounter + capacityCheck capacityChecker } type redisOnlineKeyCounter interface { CountOnlineKeys(context.Context) (int, error) } +type capacityChecker interface { + CheckCapacity(context.Context) (CapacityCheckResult, error) +} + func NewProductionStore(db *sql.DB, tdengine *sql.DB, tdengineDatabase string) *ProductionStore { if db == nil { panic("production db must not be nil") @@ -33,6 +38,11 @@ func (s *ProductionStore) WithRedisOnlineKeyCounter(counter redisOnlineKeyCounte return s } +func (s *ProductionStore) WithCapacityChecker(checker capacityChecker) *ProductionStore { + s.capacityCheck = checker + return s +} + func OpenSQL(ctx context.Context, driver, dsn string) (*sql.DB, error) { db, err := sql.Open(driver, dsn) if err != nil { @@ -895,6 +905,7 @@ func (s *ProductionStore) OpsHealth(ctx context.Context) (OpsHealth, error) { locationHealth := s.mysqlTableReadHealth(ctx, "vehicle_realtime_location", "读取实时位置表正常") tdengineHealth := s.tdengineRawFrameHealth(ctx) redisHealth, redisOnlineKeys := s.redisOnlineKeyHealth(ctx) + capacityHealth, kafkaLag := s.capacityCheckHealth(ctx) mysqlWritable := mysqlStatus == "ok" && snapshotHealth.Status == "ok" && locationHealth.Status == "ok" return OpsHealth{ LinkHealth: []LinkHealth{ @@ -902,15 +913,41 @@ func (s *ProductionStore) OpsHealth(ctx context.Context) (OpsHealth, error) { snapshotHealth, locationHealth, tdengineHealth, - {Name: "Kafka lag", Status: "warning", Detail: "平台暂未接入 Kafka consumer lag 监控"}, + capacityHealth, redisHealth, }, + KafkaLag: kafkaLag, RedisOnlineKeys: redisOnlineKeys, TDengineWritable: tdengineHealth.Status == "ok", MySQLWritable: mysqlWritable, }, nil } +func (s *ProductionStore) capacityCheckHealth(ctx context.Context) (LinkHealth, *int) { + health := LinkHealth{Name: "Capacity check", Status: "warning", Detail: "平台暂未接入 capacity-check"} + if s.capacityCheck == nil { + return health, nil + } + result, err := s.capacityCheck.CheckCapacity(ctx) + if err != nil { + health.Status = "error" + health.Detail = err.Error() + return health, nil + } + kafkaLag := int(result.KafkaLag) + detail := "active_connections=" + strconv.Itoa(int(result.ActiveConnections)) + ", kafka_lag=" + strconv.Itoa(kafkaLag) + if len(result.Findings) > 0 { + detail += ", findings=" + strings.Join(result.Findings, "; ") + } + if strings.EqualFold(result.Status, "ok") { + health.Status = "ok" + } else { + health.Status = "warning" + } + health.Detail = detail + return health, &kafkaLag +} + func (s *ProductionStore) redisOnlineKeyHealth(ctx context.Context) (LinkHealth, *int) { health := LinkHealth{Name: "Redis online keys", Status: "warning", Detail: "平台暂未接入 Redis 在线 key 读取"} if s.redisOnline == nil { diff --git a/vehicle-data-platform/apps/api/internal/platform/production_store_test.go b/vehicle-data-platform/apps/api/internal/platform/production_store_test.go index 13f96360..8889dbde 100644 --- a/vehicle-data-platform/apps/api/internal/platform/production_store_test.go +++ b/vehicle-data-platform/apps/api/internal/platform/production_store_test.go @@ -113,6 +113,15 @@ func (p fakeRedisOnlineProbe) CountOnlineKeys(context.Context) (int, error) { return p.count, p.err } +type fakeCapacityCheckProbe struct { + result CapacityCheckResult + err error +} + +func (p fakeCapacityCheckProbe) CheckCapacity(context.Context) (CapacityCheckResult, error) { + return p.result, p.err +} + func TestBuildVehicleServiceOverviewBatchSQLUsesFuzzyKeywordMatching(t *testing.T) { built := buildVehicleServiceOverviewBatchSQL(VehicleOverviewBatchQuery{ Keywords: []string{"AG183", "R0LS1426"}, @@ -195,6 +204,41 @@ func TestOpsHealthUsesRedisOnlineKeyProbe(t *testing.T) { } } +func TestOpsHealthUsesCapacityCheckProbeForKafkaLag(t *testing.T) { + db, err := sql.Open("ops_health_test", "") + if err != nil { + t.Fatalf("open ops health db: %v", err) + } + defer db.Close() + store := &ProductionStore{ + db: db, + capacityCheck: fakeCapacityCheckProbe{result: CapacityCheckResult{ + Status: "degraded", + ActiveConnections: 120000, + KafkaLag: 42, + Findings: []string{"kafka lag 42"}, + }}, + } + + health, err := store.OpsHealth(context.Background()) + if err != nil { + t.Fatalf("OpsHealth returned error: %v", err) + } + if health.KafkaLag == nil || *health.KafkaLag != 42 { + t.Fatalf("OpsHealth should expose capacity-check kafka lag, got %+v", health.KafkaLag) + } + var capacityHealth *LinkHealth + for index := range health.LinkHealth { + if health.LinkHealth[index].Name == "Capacity check" { + capacityHealth = &health.LinkHealth[index] + break + } + } + if capacityHealth == nil || capacityHealth.Status != "warning" || !strings.Contains(capacityHealth.Detail, "kafka lag 42") { + t.Fatalf("OpsHealth should expose capacity-check warning, got %+v in %+v", capacityHealth, health.LinkHealth) + } +} + func TestVehicleServiceOverviewMatchesPartialKeyword(t *testing.T) { overview := VehicleServiceOverview{ VIN: "LB9A32A24R0LS1426",