diff --git a/vehicle-data-platform/apps/api/internal/app/server.go b/vehicle-data-platform/apps/api/internal/app/server.go index 752c45e1..cbc0e7c3 100644 --- a/vehicle-data-platform/apps/api/internal/app/server.go +++ b/vehicle-data-platform/apps/api/internal/app/server.go @@ -33,7 +33,12 @@ func NewServer(cfg config.Config) http.Handler { log.Printf("production tdengine store enabled") } } - store = platform.NewProductionStore(db, tdengine, cfg.TDengineDatabase) + productionStore := platform.NewProductionStore(db, tdengine, cfg.TDengineDatabase) + if cfg.RedisAddr != "" { + productionStore.WithRedisOnlineKeyCounter(platform.NewRedisOnlineKeyCounter(cfg.RedisAddr, cfg.RedisUsername, cfg.RedisPassword, cfg.RedisDB)) + log.Printf("production redis health probe enabled") + } + store = productionStore log.Printf("production mysql store enabled") } } 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 93231a1c..8a4d797c 100644 --- a/vehicle-data-platform/apps/api/internal/platform/production_store.go +++ b/vehicle-data-platform/apps/api/internal/platform/production_store.go @@ -11,9 +11,14 @@ import ( ) type ProductionStore struct { - db *sql.DB - tdengine *sql.DB - tdDatabase string + db *sql.DB + tdengine *sql.DB + tdDatabase string + redisOnline redisOnlineKeyCounter +} + +type redisOnlineKeyCounter interface { + CountOnlineKeys(context.Context) (int, error) } func NewProductionStore(db *sql.DB, tdengine *sql.DB, tdengineDatabase string) *ProductionStore { @@ -23,6 +28,11 @@ func NewProductionStore(db *sql.DB, tdengine *sql.DB, tdengineDatabase string) * return &ProductionStore{db: db, tdengine: tdengine, tdDatabase: tdengineDatabase} } +func (s *ProductionStore) WithRedisOnlineKeyCounter(counter redisOnlineKeyCounter) *ProductionStore { + s.redisOnline = counter + return s +} + func OpenSQL(ctx context.Context, driver, dsn string) (*sql.DB, error) { db, err := sql.Open(driver, dsn) if err != nil { @@ -884,6 +894,7 @@ func (s *ProductionStore) OpsHealth(ctx context.Context) (OpsHealth, error) { snapshotHealth := s.mysqlTableReadHealth(ctx, "vehicle_realtime_snapshot", "读取实时快照表正常") locationHealth := s.mysqlTableReadHealth(ctx, "vehicle_realtime_location", "读取实时位置表正常") tdengineHealth := s.tdengineRawFrameHealth(ctx) + redisHealth, redisOnlineKeys := s.redisOnlineKeyHealth(ctx) mysqlWritable := mysqlStatus == "ok" && snapshotHealth.Status == "ok" && locationHealth.Status == "ok" return OpsHealth{ LinkHealth: []LinkHealth{ @@ -892,13 +903,30 @@ func (s *ProductionStore) OpsHealth(ctx context.Context) (OpsHealth, error) { locationHealth, tdengineHealth, {Name: "Kafka lag", Status: "warning", Detail: "平台暂未接入 Kafka consumer lag 监控"}, - {Name: "Redis online keys", Status: "warning", Detail: "平台暂未接入 Redis 在线 key 读取"}, + redisHealth, }, + RedisOnlineKeys: redisOnlineKeys, TDengineWritable: tdengineHealth.Status == "ok", MySQLWritable: mysqlWritable, }, nil } +func (s *ProductionStore) redisOnlineKeyHealth(ctx context.Context) (LinkHealth, *int) { + health := LinkHealth{Name: "Redis online keys", Status: "warning", Detail: "平台暂未接入 Redis 在线 key 读取"} + if s.redisOnline == nil { + return health, nil + } + count, err := s.redisOnline.CountOnlineKeys(ctx) + if err != nil { + health.Status = "error" + health.Detail = err.Error() + return health, nil + } + health.Status = "ok" + health.Detail = "Redis 在线 key 读取正常：" + strconv.Itoa(count) + return health, &count +} + 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" 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 85d4eb2c..13f96360 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 @@ -12,6 +12,7 @@ import ( func init() { sql.Register("tdengine_missing_table_test", missingTableDriver{}) + sql.Register("ops_health_test", opsHealthDriver{}) } type missingTableDriver struct{} @@ -52,6 +53,66 @@ func (emptyRows) Next(_ []driver.Value) error { return io.EOF } +type opsHealthDriver struct{} + +func (opsHealthDriver) Open(_ string) (driver.Conn, error) { + return opsHealthConn{}, nil +} + +type opsHealthConn struct{} + +func (opsHealthConn) Prepare(_ string) (driver.Stmt, error) { + return nil, errors.New("not implemented") +} + +func (opsHealthConn) Close() error { + return nil +} + +func (opsHealthConn) Begin() (driver.Tx, error) { + return nil, errors.New("not implemented") +} + +func (opsHealthConn) Ping(_ context.Context) error { + return nil +} + +func (opsHealthConn) QueryContext(_ context.Context, _ string, _ []driver.NamedValue) (driver.Rows, error) { + return &singleValueRows{columns: []string{"ok"}, values: []driver.Value{1}}, nil +} + +type singleValueRows struct { + columns []string + values []driver.Value + read bool +} + +func (r singleValueRows) Columns() []string { + return r.columns +} + +func (r singleValueRows) Close() error { + return nil +} + +func (r *singleValueRows) Next(dest []driver.Value) error { + if r.read { + return io.EOF + } + r.read = true + copy(dest, r.values) + return nil +} + +type fakeRedisOnlineProbe struct { + count int + err error +} + +func (p fakeRedisOnlineProbe) CountOnlineKeys(context.Context) (int, error) { + return p.count, p.err +} + func TestBuildVehicleServiceOverviewBatchSQLUsesFuzzyKeywordMatching(t *testing.T) { built := buildVehicleServiceOverviewBatchSQL(VehicleOverviewBatchQuery{ Keywords: []string{"AG183", "R0LS1426"}, @@ -107,6 +168,33 @@ func TestRawFramesReturnsEmptyPageWhenTDengineTableIsMissing(t *testing.T) { } } +func TestOpsHealthUsesRedisOnlineKeyProbe(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, redisOnline: fakeRedisOnlineProbe{count: 73}} + + health, err := store.OpsHealth(context.Background()) + if err != nil { + t.Fatalf("OpsHealth returned error: %v", err) + } + if health.RedisOnlineKeys == nil || *health.RedisOnlineKeys != 73 { + t.Fatalf("OpsHealth should expose Redis online key count, got %+v", health.RedisOnlineKeys) + } + var redisHealth *LinkHealth + for index := range health.LinkHealth { + if health.LinkHealth[index].Name == "Redis online keys" { + redisHealth = &health.LinkHealth[index] + break + } + } + if redisHealth == nil || redisHealth.Status != "ok" || !strings.Contains(redisHealth.Detail, "73") { + t.Fatalf("OpsHealth should expose Redis online key probe status, got %+v in %+v", redisHealth, health.LinkHealth) + } +} + func TestVehicleServiceOverviewMatchesPartialKeyword(t *testing.T) { overview := VehicleServiceOverview{ VIN: "LB9A32A24R0LS1426", diff --git a/vehicle-data-platform/apps/api/internal/platform/redis_online.go b/vehicle-data-platform/apps/api/internal/platform/redis_online.go new file mode 100644 index 00000000..50678435 --- /dev/null +++ b/vehicle-data-platform/apps/api/internal/platform/redis_online.go @@ -0,0 +1,165 @@ +package platform + +import ( + "bufio" + "context" + "errors" + "fmt" + "io" + "net" + "strconv" + "strings" + "time" +) + +type RedisOnlineKeyCounter struct { + addr string + username string + password string + db int +} + +func NewRedisOnlineKeyCounter(addr, username, password string, db int) *RedisOnlineKeyCounter { + return &RedisOnlineKeyCounter{addr: strings.TrimSpace(addr), username: username, password: password, db: db} +} + +func (c *RedisOnlineKeyCounter) CountOnlineKeys(ctx context.Context) (int, error) { + if c.addr == "" { + return 0, errors.New("redis addr is empty") + } + dialer := net.Dialer{Timeout: 2 * time.Second} + conn, err := dialer.DialContext(ctx, "tcp", c.addr) + if err != nil { + return 0, err + } + defer conn.Close() + if deadline, ok := ctx.Deadline(); ok { + _ = conn.SetDeadline(deadline) + } else { + _ = conn.SetDeadline(time.Now().Add(3 * time.Second)) + } + reader := bufio.NewReader(conn) + if c.password != "" { + args := []string{"AUTH"} + if c.username != "" { + args = append(args, c.username) + } + args = append(args, c.password) + if _, err := redisCommand(conn, reader, args...); err != nil { + return 0, err + } + } + if _, err := redisCommand(conn, reader, "PING"); err != nil { + return 0, err + } + if c.db >= 0 { + if _, err := redisCommand(conn, reader, "SELECT", strconv.Itoa(c.db)); err != nil { + return 0, err + } + } + total := 0 + cursor := "0" + for { + value, err := redisCommand(conn, reader, "SCAN", cursor, "MATCH", "vehicle:online:*", "COUNT", "1000") + if err != nil { + return 0, err + } + reply, ok := value.([]any) + if !ok || len(reply) != 2 { + return 0, fmt.Errorf("unexpected redis scan reply: %#v", value) + } + nextCursor, ok := reply[0].(string) + if !ok { + return 0, fmt.Errorf("unexpected redis scan cursor: %#v", reply[0]) + } + keys, ok := reply[1].([]any) + if !ok { + return 0, fmt.Errorf("unexpected redis scan keys: %#v", reply[1]) + } + total += len(keys) + if nextCursor == "0" { + return total, nil + } + cursor = nextCursor + } +} + +func redisCommand(conn net.Conn, reader *bufio.Reader, args ...string) (any, error) { + if _, err := fmt.Fprintf(conn, "*%d\r\n", len(args)); err != nil { + return nil, err + } + for _, arg := range args { + if _, err := fmt.Fprintf(conn, "$%d\r\n%s\r\n", len(arg), arg); err != nil { + return nil, err + } + } + return readRedisReply(reader) +} + +func readRedisReply(reader *bufio.Reader) (any, error) { + prefix, err := reader.ReadByte() + if err != nil { + return nil, err + } + switch prefix { + case '+': + return readRedisLine(reader) + case '-': + line, err := readRedisLine(reader) + if err != nil { + return nil, err + } + return nil, errors.New(line) + case ':': + line, err := readRedisLine(reader) + if err != nil { + return nil, err + } + return strconv.Atoi(line) + case '$': + line, err := readRedisLine(reader) + if err != nil { + return nil, err + } + size, err := strconv.Atoi(line) + if err != nil { + return nil, err + } + if size < 0 { + return "", nil + } + buf := make([]byte, size+2) + if _, err := io.ReadFull(reader, buf); err != nil { + return nil, err + } + return string(buf[:size]), nil + case '*': + line, err := readRedisLine(reader) + if err != nil { + return nil, err + } + count, err := strconv.Atoi(line) + if err != nil { + return nil, err + } + values := make([]any, 0, count) + for i := 0; i < count; i++ { + value, err := readRedisReply(reader) + if err != nil { + return nil, err + } + values = append(values, value) + } + return values, nil + default: + return nil, fmt.Errorf("unexpected redis reply prefix %q", prefix) + } +} + +func readRedisLine(reader *bufio.Reader) (string, error) { + line, err := reader.ReadString('\n') + if err != nil { + return "", err + } + return strings.TrimSuffix(strings.TrimSuffix(line, "\n"), "\r"), nil +} diff --git a/vehicle-data-platform/apps/api/internal/platform/redis_online_test.go b/vehicle-data-platform/apps/api/internal/platform/redis_online_test.go new file mode 100644 index 00000000..db89954f --- /dev/null +++ b/vehicle-data-platform/apps/api/internal/platform/redis_online_test.go @@ -0,0 +1,94 @@ +package platform + +import ( + "bufio" + "context" + "fmt" + "net" + "strings" + "testing" +) + +func TestRedisOnlineKeyCounterScansVehicleOnlineKeys(t *testing.T) { + listener, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatalf("listen fake redis: %v", err) + } + defer listener.Close() + done := make(chan struct{}) + go func() { + defer close(done) + conn, err := listener.Accept() + if err != nil { + return + } + defer conn.Close() + reader := bufio.NewReader(conn) + for { + command, err := readFakeRedisCommand(reader) + if err != nil { + return + } + switch strings.ToUpper(command[0]) { + case "PING": + _, _ = conn.Write([]byte("+PONG\r\n")) + case "SELECT": + _, _ = conn.Write([]byte("+OK\r\n")) + case "SCAN": + if command[1] == "0" { + _, _ = conn.Write(fakeRedisScanReply("7", []string{"vehicle:online:GB32960:VIN001", "vehicle:online:JT808:VIN002"})) + } else { + _, _ = conn.Write(fakeRedisScanReply("0", []string{"vehicle:online:YUTONG_MQTT:VIN003"})) + } + default: + t.Errorf("unexpected fake redis command: %#v", command) + return + } + } + }() + + counter := NewRedisOnlineKeyCounter(listener.Addr().String(), "", "", 50) + count, err := counter.CountOnlineKeys(context.Background()) + if err != nil { + t.Fatalf("CountOnlineKeys returned error: %v", err) + } + if count != 3 { + t.Fatalf("expected 3 online keys, got %d", count) + } + _ = listener.Close() + <-done +} + +func fakeRedisScanReply(cursor string, keys []string) []byte { + var builder strings.Builder + builder.WriteString("*2\r\n") + builder.WriteString(fmt.Sprintf("$%d\r\n%s\r\n", len(cursor), cursor)) + builder.WriteString(fmt.Sprintf("*%d\r\n", len(keys))) + for _, key := range keys { + builder.WriteString(fmt.Sprintf("$%d\r\n%s\r\n", len(key), key)) + } + return []byte(builder.String()) +} + +func readFakeRedisCommand(reader *bufio.Reader) ([]string, error) { + line, err := reader.ReadString('\n') + if err != nil { + return nil, err + } + var count int + if _, err := fmt.Sscanf(strings.TrimSpace(line), "*%d", &count); err != nil { + return nil, err + } + parts := make([]string, 0, count) + for i := 0; i < count; i++ { + if _, err := reader.ReadString('\n'); err != nil { + return nil, err + } + value, err := reader.ReadString('\n') + if err != nil { + return nil, err + } + parts = append(parts, strings.TrimSuffix(strings.TrimSuffix(value, "\n"), "\r")) + } + return parts, nil +}