package handler // Redis 键扫描(不透明游标分页),设计对标 opsxdb: // - 游标 = {pattern, db, type, nodes, index, cursor} 的 base64 编码,服务端校验搜索条件一致 // - 每页上限 redisKeyPageCap,单页最多 redisScanBatchMax 次 SCAN,整批返回 // - 集群模式逐 master 分片 SCAN 后归并(ClusterClient.Scan 只命中单节点会漏键) // - 无通配符输入:先 TYPE 精确探测(exact 置顶),再按 *关键词* 包含匹配扫描 import ( "context" "encoding/base64" "encoding/json" "fmt" "net/http" "sort" "strconv" "strings" "time" "sql-platform/internal/handler/resp" "sql-platform/internal/model" "github.com/gin-gonic/gin" "github.com/redis/go-redis/v9" ) const ( redisKeyPageCap = 200 // 每页最多返回的 key 数 redisScanBatchMax = 20 // 单页最多 SCAN 调用次数(凑满一页即提前返回) redisScanCountHint = 100 // SCAN COUNT 提示值 ) // RedisScanKeyInfo 单个键摘要(TTL 毫秒,-1 永久;与既有前端契约一致) type RedisScanKeyInfo struct { Key string `json:"key"` Type string `json:"type"` TTL int64 `json:"ttl"` } // keyScanPosition 不透明游标内容:记录搜索条件与扫描位置 type keyScanPosition struct { Pattern string `json:"pattern"` DB int `json:"db"` Type string `json:"type,omitempty"` Nodes []string `json:"nodes"` // 集群:各 master 地址(排序);单机:单地址 Index int `json:"index"` // 当前扫描的节点下标 Cursor uint64 `json:"cursor"` } func encodeScanPosition(p keyScanPosition) string { b, _ := json.Marshal(p) return base64.RawURLEncoding.EncodeToString(b) } func decodeScanPosition(s string) (keyScanPosition, error) { var p keyScanPosition b, err := base64.RawURLEncoding.DecodeString(strings.TrimSpace(s)) if err != nil { return p, fmt.Errorf("无效的游标") } if err := json.Unmarshal(b, &p); err != nil { return p, fmt.Errorf("无效的游标") } return p, nil } // redisBrowsePattern 模式规则:无通配符自动包 *关键词*(包含匹配);有则按 glob 原样 func redisBrowsePattern(raw string) string { raw = strings.TrimSpace(raw) if raw == "" { return "*" } if strings.ContainsAny(raw, "*?[") { return raw } return "*" + raw + "*" } func ttlMilliseconds(d time.Duration) int64 { if d > 0 { return d.Milliseconds() } return -1 } // redisMovedHint 单机模式收到 MOVED 时提示切换部署模式(opsxdb 同款) func redisErrHint(err error) error { if err == nil { return nil } msg := err.Error() if strings.Contains(msg, "MOVED") { return fmt.Errorf("%w;该键已被集群分片管理,请把实例的部署模式改为集群并填写种子节点", err) } return err } // scanStandaloneNode 用独立连接扫描单个节点(集群分片 SCAN 必须直连 master) func scanStandaloneNode(ctx context.Context, w *redisClientWrapper, addr string, cursor uint64, match string, callBudget, softCap int) (keys []string, nextCursor uint64, calls int, err error) { var client *redis.Client if !w.IsCluster() || len(w.MasterAddrs) == 1 && addr == w.MasterAddrs[0] && w.Client != nil { client = w.Client // 单机直接复用主连接 } else { client = redis.NewClient(&redis.Options{ Addr: addr, Username: w.Username, Password: w.Password, DB: 0, // 集群恒 DB0 DialTimeout: 5 * time.Second, ReadTimeout: 5 * time.Second, WriteTimeout: 5 * time.Second, }) defer client.Close() } for calls < callBudget && len(keys) < softCap { var batch []string batch, nextCursor, err = client.Scan(ctx, cursor, match, redisScanCountHint).Result() if err != nil { return nil, 0, calls, err } keys = append(keys, batch...) calls++ cursor = nextCursor if nextCursor == 0 { break // 该节点扫描完毕 } } return keys, cursor, calls, nil } // pipelineTypeTTL 批量取 TYPE/TTL,替换逐 key 串行 N+1 func pipelineTypeTTL(ctx context.Context, w *redisClientWrapper, keys []string) (types map[string]string, ttls map[string]int64) { types = make(map[string]string, len(keys)) ttls = make(map[string]int64, len(keys)) if len(keys) == 0 { return } var pipe redis.Pipeliner if w.IsCluster() && w.ClusterClient != nil { pipe = w.ClusterClient.Pipeline() } else { pipe = w.Client.Pipeline() } typeCmds := make([]*redis.StatusCmd, len(keys)) ttlCmds := make([]*redis.DurationCmd, len(keys)) for i, k := range keys { typeCmds[i] = pipe.Type(ctx, k) ttlCmds[i] = pipe.TTL(ctx, k) } _, _ = pipe.Exec(ctx) for i, k := range keys { t, err := typeCmds[i].Result() if err != nil { t = "none" } types[k] = t ttls[k] = ttlMilliseconds(ttlCmds[i].Val()) } return } // GET /api/v1/redis/:id/keys — 游标分页扫描 func RedisScanKeysPaged(c *gin.Context) { id := c.Param("id") db := 0 if v := c.Query("db"); v != "" { db, _ = strconv.Atoi(v) } rawPattern := strings.TrimSpace(c.Query("pattern")) reqType := strings.TrimSpace(c.Query("type")) cursorParam := strings.TrimSpace(c.Query("cursor")) var instance model.Instance if err := model.DB.First(&instance, id).Error; err != nil { resp.NotFound(c, "Instance not found") return } // 集群只允许 DB0 if instance.ClusterMode != nil && *instance.ClusterMode { db = 0 } client, err := getRedisClient(&instance, db) if err != nil { resp.InternalErr(c, err) return } defer client.Close() ctx, cancel := context.WithTimeout(context.Background(), 8*time.Second) defer cancel() pattern := redisBrowsePattern(rawPattern) // 精确命中探测:无通配符且非全量时,先 TYPE 直查(集群走槽位路由,不受 NAT 影响) var exact *RedisScanKeyInfo rawHasWildcard := strings.ContainsAny(rawPattern, "*?[") if rawPattern != "" && rawPattern != "*" && !rawHasWildcard { if keyType, err := client.Type(ctx, rawPattern).Result(); err == nil && keyType != "none" { ttl, _ := client.TTL(ctx, rawPattern).Result() exact = &RedisScanKeyInfo{Key: rawPattern, Type: keyType, TTL: ttlMilliseconds(ttl)} } } // 还原或初始化游标 var pos keyScanPosition if cursorParam != "" { pos, err = decodeScanPosition(cursorParam) if err != nil { resp.BadRequest(c, err.Error()) return } if pos.Pattern != pattern || pos.DB != db || pos.Type != reqType || !equalStrings(pos.Nodes, client.MasterAddrs) { resp.BadRequest(c, "搜索条件已变化,请重新搜索") return } } else { pos = keyScanPosition{Pattern: pattern, DB: db, Type: reqType, Nodes: client.MasterAddrs, Index: 0, Cursor: 0} } // 逐节点增量扫描,凑满一页即返回 nameSet := map[string]bool{} budget := redisScanBatchMax for pos.Index < len(pos.Nodes) && budget > 0 && len(nameSet) < redisKeyPageCap { batch, nextCursor, calls, err := scanStandaloneNode(ctx, client, pos.Nodes[pos.Index], pos.Cursor, pattern, budget, redisKeyPageCap-len(nameSet)) if err != nil { resp.BadRequest(c, redisErrHint(err).Error()) return } budget -= calls for _, k := range batch { nameSet[k] = true } pos.Cursor = nextCursor if nextCursor == 0 { pos.Index++ // 该节点扫完,移到下一分片 pos.Cursor = 0 } } names := make([]string, 0, len(nameSet)) for k := range nameSet { names = append(names, k) } sort.Strings(names) types, ttls := pipelineTypeTTL(ctx, client, names) keys := make([]RedisScanKeyInfo, 0, len(names)) for _, k := range names { t := types[k] if reqType != "" && t != reqType { continue } if t == "none" { continue // 扫描与 TYPE 之间已过期 } keys = append(keys, RedisScanKeyInfo{Key: k, Type: t, TTL: ttls[k]}) } // 所有节点扫完前都可继续翻页(SCAN 次数用尽同样可续) nextCursor := "" if pos.Index < len(pos.Nodes) { nextCursor = encodeScanPosition(pos) } c.JSON(http.StatusOK, gin.H{ "keys": keys, "next_cursor": nextCursor, "exact": exact, "total": len(keys), "cluster": client.IsCluster(), }) } // GET /api/v1/redis/:id/dbs — 库列表(集群恒 [0]) func RedisDbs(c *gin.Context) { id := c.Param("id") var instance model.Instance if err := model.DB.First(&instance, id).Error; err != nil { resp.NotFound(c, "Instance not found") return } isCluster := instance.ClusterMode != nil && *instance.ClusterMode if !isCluster { client, err := getRedisClient(&instance, 0) if err != nil { resp.InternalErr(c, err) return } defer client.Close() isCluster = client.IsCluster() } dbs := []int{0} if !isCluster { for i := 1; i <= 15; i++ { dbs = append(dbs, i) } } c.JSON(http.StatusOK, gin.H{"dbs": dbs, "cluster": isCluster}) } // POST /api/v1/redis/:id/keys/:key/field — Hash 行级编辑(HSET/HDEL,不整键重写) func RedisUpdateKeyField(c *gin.Context) { id := c.Param("id") key := c.Param("key") db := 0 if v := c.Query("db"); v != "" { db, _ = strconv.Atoi(v) } var input struct { Op string `json:"op" binding:"required"` // set | del Field string `json:"field" binding:"required"` Value string `json:"value"` } if err := c.ShouldBindJSON(&input); err != nil { resp.BadRequest(c, err.Error()) return } if input.Op != "set" && input.Op != "del" { resp.BadRequest(c, "op 仅支持 set / del") return } var instance model.Instance if err := model.DB.First(&instance, id).Error; err != nil { resp.NotFound(c, "Instance not found") return } client, err := getRedisClient(&instance, db) if err != nil { resp.InternalErr(c, err) return } defer client.Close() ctx := context.Background() if keyType, err := client.Type(ctx, key).Result(); err != nil || keyType != "hash" { resp.BadRequest(c, "仅支持 Hash 类型键的行级编辑") return } if input.Op == "del" { if err := client.HDel(ctx, key, input.Field).Err(); err != nil { resp.InternalErr(c, err) return } } else { if err := client.HSet(ctx, key, input.Field, input.Value).Err(); err != nil { resp.InternalErr(c, err) return } } resp.Message(c, "Field updated") } func equalStrings(a, b []string) bool { if len(a) != len(b) { return false } for i := range a { if a[i] != b[i] { return false } } return true }