feat(platform): report capacity check health

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")
 		}

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
+}

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 {

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",