feat(platform): report redis online key health

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

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"

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

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

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