feat(backend): redis key scan service - cursor pagination + cluster + CLI fix
Modeled on opsxdb's redis scan design:
- opaque base64 scan cursor {pattern,db,type,nodes,index,cursor}; server
validates search conditions match, rejects stale cursors (400)
- soft page cap 200 keys (whole scan batches returned, no key loss),
max 20 SCAN calls per page; next_cursor until all shards exhausted
- pattern rule: no glob chars -> *kw* contains match; non-glob input
also probed with TYPE for exact hit (returned as exact, slot-routed
so cluster/NAT safe)
- pipeline batch TYPE/TTL replaces per-key serial N+1; type filter now
effective server-side
- cluster: per-master direct SCAN merged across shards (MasterAddrs on
wrapper), DB0-only, MOVED errors hint at switching deploy mode;
SlowLog/ClientList aggregate across masters (were cluster-broken)
- GET /redis/:id/dbs (0-15, cluster -> [0])
- POST /redis/:id/keys/:key/field: hash row-level HSET/HDEL (no whole-
key rewrite)
- RedisExecuteCommand fixed (was always-403): accepts {db, command}
full-string, quote-aware tokenizer, whitelist on first word, expanded
read-only set
- RedisUpdateKey preserves previous TTL unless explicitly set/cleared
(was silently dropping TTL); string type skips the DEL
Verified live against local redis: 200/page no overlap across pages,
cursor 400 on condition change, exact hit, type filter, hash field
edit round-trip, CLI DBSIZE/HGET ok, SET rejected
This commit is contained in:
1 parent
6945c53879
commit
6604fdf48e
3 files changed
+521
-129
No files matched your search
+147
-128
@@ -20,8 +20,14 @@ type redisClientWrapper struct {
|
||||
Client *redis.Client
|
||||
ClusterClient *redis.ClusterClient
|
||||
isCluster bool
|
||||
// MasterAddrs 集群各 master 地址(排序);单机为单地址。扫描分片与游标校验依赖
|
||||
MasterAddrs []string
|
||||
Username string
|
||||
Password string
|
||||
}
|
||||
|
||||
func (w *redisClientWrapper) IsCluster() bool { return w.isCluster }
|
||||
|
||||
func (w *redisClientWrapper) Close() {
|
||||
if w.Client != nil {
|
||||
w.Client.Close()
|
||||
@@ -142,12 +148,41 @@ func (w *redisClientWrapper) HGetAll(ctx context.Context, key string) *redis.Map
|
||||
return w.Client.HGetAll(ctx, key)
|
||||
}
|
||||
|
||||
func (w *redisClientWrapper) SlowLogGet(ctx context.Context, count int64) *redis.SlowLogCmd {
|
||||
return w.Client.SlowLogGet(ctx, count)
|
||||
// SlowLogGet 集群模式逐 master 聚合(单机连接打不到全部分片)
|
||||
func (w *redisClientWrapper) SlowLogGet(ctx context.Context, count int64) ([]redis.SlowLog, error) {
|
||||
if w.isCluster && w.ClusterClient != nil {
|
||||
var all []redis.SlowLog
|
||||
err := w.ClusterClient.ForEachMaster(ctx, func(ctx context.Context, shard *redis.Client) error {
|
||||
logs, err := shard.SlowLogGet(ctx, count).Result()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
all = append(all, logs...)
|
||||
return nil
|
||||
})
|
||||
return all, err
|
||||
}
|
||||
return w.Client.SlowLogGet(ctx, count).Result()
|
||||
}
|
||||
|
||||
func (w *redisClientWrapper) ClientList(ctx context.Context) *redis.StringCmd {
|
||||
return w.Client.ClientList(ctx)
|
||||
// ClientList 集群模式逐 master 聚合
|
||||
func (w *redisClientWrapper) ClientList(ctx context.Context) (string, error) {
|
||||
if w.isCluster && w.ClusterClient != nil {
|
||||
var sb strings.Builder
|
||||
err := w.ClusterClient.ForEachMaster(ctx, func(ctx context.Context, shard *redis.Client) error {
|
||||
list, err := shard.ClientList(ctx).Result()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
sb.WriteString(list)
|
||||
if !strings.HasSuffix(list, "\n") {
|
||||
sb.WriteString("\n")
|
||||
}
|
||||
return nil
|
||||
})
|
||||
return sb.String(), err
|
||||
}
|
||||
return w.Client.ClientList(ctx).Result()
|
||||
}
|
||||
|
||||
func (w *redisClientWrapper) Do(ctx context.Context, args ...interface{}) *redis.Cmd {
|
||||
@@ -192,6 +227,20 @@ func (w *redisClientWrapper) ZAdd(ctx context.Context, key string, members ...re
|
||||
return w.Client.ZAdd(ctx, key, members...)
|
||||
}
|
||||
|
||||
func (w *redisClientWrapper) Persist(ctx context.Context, key string) *redis.BoolCmd {
|
||||
if w.isCluster && w.ClusterClient != nil {
|
||||
return w.ClusterClient.Persist(ctx, key)
|
||||
}
|
||||
return w.Client.Persist(ctx, key)
|
||||
}
|
||||
|
||||
func (w *redisClientWrapper) HDel(ctx context.Context, key string, fields ...string) *redis.IntCmd {
|
||||
if w.isCluster && w.ClusterClient != nil {
|
||||
return w.ClusterClient.HDel(ctx, key, fields...)
|
||||
}
|
||||
return w.Client.HDel(ctx, key, fields...)
|
||||
}
|
||||
|
||||
func (w *redisClientWrapper) Expire(ctx context.Context, key string, expiration time.Duration) *redis.BoolCmd {
|
||||
if w.isCluster && w.ClusterClient != nil {
|
||||
return w.ClusterClient.Expire(ctx, key, expiration)
|
||||
@@ -358,7 +407,13 @@ func getRedisClient(instance *model.Instance, db int) (*redisClientWrapper, erro
|
||||
}
|
||||
}
|
||||
|
||||
wrapper := &redisClientWrapper{Client: client, isCluster: isCluster}
|
||||
wrapper := &redisClientWrapper{
|
||||
Client: client,
|
||||
isCluster: isCluster,
|
||||
MasterAddrs: []string{addr},
|
||||
Username: redisUsername,
|
||||
Password: password,
|
||||
}
|
||||
if !isCluster {
|
||||
return wrapper, nil
|
||||
}
|
||||
@@ -377,6 +432,7 @@ func getRedisClient(instance *model.Instance, db int) (*redisClientWrapper, erro
|
||||
return nil, fmt.Errorf("Redis cluster connection failed: %v", err)
|
||||
}
|
||||
wrapper.ClusterClient = clusterClient
|
||||
wrapper.MasterAddrs = addrs
|
||||
return wrapper, nil
|
||||
}
|
||||
|
||||
@@ -481,100 +537,6 @@ func RedisDbsize(c *gin.Context) {
|
||||
c.JSON(http.StatusOK, gin.H{"db_size": dbSize})
|
||||
}
|
||||
|
||||
// GET /api/v1/redis/:id/keys
|
||||
func RedisScanKeys(c *gin.Context) {
|
||||
id := c.Param("id")
|
||||
dbParam := c.Query("db")
|
||||
type KeyInfo struct {
|
||||
Key string `json:"key"`
|
||||
Type string `json:"type"`
|
||||
TTL int64 `json:"ttl"`
|
||||
}
|
||||
|
||||
rawPattern := strings.TrimSpace(c.Query("pattern"))
|
||||
if rawPattern == "" {
|
||||
c.JSON(http.StatusOK, gin.H{
|
||||
"keys": []KeyInfo{},
|
||||
"cursor": 0,
|
||||
"total": 0,
|
||||
})
|
||||
return
|
||||
}
|
||||
pattern := rawPattern
|
||||
hasWildcard := strings.ContainsAny(pattern, "*?[")
|
||||
count, _ := strconv.Atoi(c.DefaultQuery("count", "100"))
|
||||
|
||||
db := 0
|
||||
if dbParam != "" {
|
||||
db, _ = strconv.Atoi(dbParam)
|
||||
}
|
||||
|
||||
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()
|
||||
|
||||
// Redis Cluster 精确 key 查询应优先走槽位路由直查。全量 SCAN 需要访问各 master,
|
||||
// 在 NAT/代理暴露单入口时容易受节点通告地址影响。
|
||||
if !hasWildcard && pattern != "*" {
|
||||
keyType, err := client.Type(ctx, pattern).Result()
|
||||
if err == nil && keyType != "none" {
|
||||
ttl, _ := client.TTL(ctx, pattern).Result()
|
||||
ttlMs := int64(-1)
|
||||
if ttl > 0 {
|
||||
ttlMs = ttl.Milliseconds()
|
||||
}
|
||||
c.JSON(http.StatusOK, gin.H{
|
||||
"keys": []KeyInfo{{Key: pattern, Type: keyType, TTL: ttlMs}},
|
||||
"cursor": 0,
|
||||
"total": 1,
|
||||
})
|
||||
return
|
||||
}
|
||||
}
|
||||
|
||||
// Auto-append * if pattern has no wildcard (user-friendly fuzzy search)
|
||||
if !hasWildcard {
|
||||
pattern = pattern + "*"
|
||||
}
|
||||
|
||||
allKeys, err := client.ScanAllKeys(ctx, pattern, count)
|
||||
if err != nil {
|
||||
resp.InternalErr(c, err)
|
||||
return
|
||||
}
|
||||
|
||||
var keyInfos []KeyInfo
|
||||
for _, key := range allKeys {
|
||||
keyType, _ := client.Type(ctx, key).Result()
|
||||
ttl, _ := client.TTL(ctx, key).Result()
|
||||
ttlMs := int64(-1)
|
||||
if ttl > 0 {
|
||||
ttlMs = ttl.Milliseconds()
|
||||
}
|
||||
keyInfos = append(keyInfos, KeyInfo{Key: key, Type: keyType, TTL: ttlMs})
|
||||
}
|
||||
if keyInfos == nil {
|
||||
keyInfos = []KeyInfo{}
|
||||
}
|
||||
|
||||
c.JSON(http.StatusOK, gin.H{
|
||||
"keys": keyInfos,
|
||||
"cursor": 0,
|
||||
"total": len(keyInfos),
|
||||
})
|
||||
}
|
||||
|
||||
// GET /api/v1/redis/:id/keys/:key
|
||||
func RedisGetKeyDetail(c *gin.Context) {
|
||||
id := c.Param("id")
|
||||
@@ -649,7 +611,7 @@ func RedisSlowlog(c *gin.Context) {
|
||||
defer client.Close()
|
||||
|
||||
ctx := context.Background()
|
||||
logs, err := client.SlowLogGet(ctx, int64(count)).Result()
|
||||
logs, err := client.SlowLogGet(ctx, int64(count))
|
||||
if err != nil {
|
||||
resp.InternalErr(c, err)
|
||||
return
|
||||
@@ -688,7 +650,7 @@ func RedisClients(c *gin.Context) {
|
||||
defer client.Close()
|
||||
|
||||
ctx := context.Background()
|
||||
clientList, err := client.ClientList(ctx).Result()
|
||||
clientList, err := client.ClientList(ctx)
|
||||
if err != nil {
|
||||
resp.InternalErr(c, err)
|
||||
return
|
||||
@@ -716,23 +678,22 @@ func RedisClients(c *gin.Context) {
|
||||
c.JSON(http.StatusOK, clients)
|
||||
}
|
||||
|
||||
// POST /api/v1/redis/:id/command
|
||||
// POST /api/v1/redis/:id/command — 只读命令行(引号感知拆参,白名单按首词判定)
|
||||
func RedisExecuteCommand(c *gin.Context) {
|
||||
id := c.Param("id")
|
||||
dbParam := c.Query("db")
|
||||
db := 0
|
||||
if dbParam != "" {
|
||||
db, _ = strconv.Atoi(dbParam)
|
||||
}
|
||||
|
||||
var input struct {
|
||||
DB int `json:"db"`
|
||||
Command string `json:"command" binding:"required"`
|
||||
Args []string `json:"args"`
|
||||
Args []string `json:"args"` // 兼容旧调用:显式参数
|
||||
}
|
||||
if err := c.ShouldBindJSON(&input); err != nil {
|
||||
resp.BadRequest(c, err.Error())
|
||||
return
|
||||
}
|
||||
if v := c.Query("db"); v != "" && input.DB == 0 {
|
||||
input.DB, _ = strconv.Atoi(v)
|
||||
}
|
||||
|
||||
var instance model.Instance
|
||||
if err := model.DB.First(&instance, id).Error; err != nil {
|
||||
@@ -740,7 +701,7 @@ func RedisExecuteCommand(c *gin.Context) {
|
||||
return
|
||||
}
|
||||
|
||||
client, err := getRedisClient(&instance, db)
|
||||
client, err := getRedisClient(&instance, input.DB)
|
||||
if err != nil {
|
||||
resp.InternalErr(c, err)
|
||||
return
|
||||
@@ -748,38 +709,84 @@ func RedisExecuteCommand(c *gin.Context) {
|
||||
defer client.Close()
|
||||
|
||||
ctx := context.Background()
|
||||
cmd := strings.ToUpper(input.Command)
|
||||
|
||||
// Safe read-only commands
|
||||
readOnly := map[string]bool{
|
||||
"GET": true, "MGET": true, "HGET": true, "HGETALL": true,
|
||||
"LRANGE": true, "SMEMBERS": true, "ZRANGE": true,
|
||||
"TYPE": true, "TTL": true, "EXISTS": true,
|
||||
"INFO": true, "DBSIZE": true, "KEYS": true, "SCAN": true,
|
||||
"STRLEN": true, "LLEN": true, "SCARD": true, "ZCARD": true,
|
||||
"HLEN": true, "GETRANGE": true,
|
||||
// 命令解析:优先按整串引号感知拆分;整串为空时兼容旧的显式 args 调用
|
||||
var verb string
|
||||
var args []string
|
||||
if _, parts := splitRedisCommand(input.Command); len(parts) > 0 {
|
||||
verb = strings.ToUpper(parts[0])
|
||||
args = parts[1:]
|
||||
} else if len(input.Args) > 0 {
|
||||
verb = strings.ToUpper(strings.TrimSpace(input.Args[0]))
|
||||
args = input.Args[1:]
|
||||
}
|
||||
if verb == "" {
|
||||
resp.BadRequest(c, "命令不能为空")
|
||||
return
|
||||
}
|
||||
|
||||
if !readOnly[cmd] {
|
||||
resp.Forbidden(c, "Only read-only commands allowed via query interface")
|
||||
readOnly := map[string]bool{
|
||||
"GET": true, "MGET": true, "HGET": true, "HGETALL": true, "HKEYS": true, "HVALS": true, "HSTRLEN": true,
|
||||
"LRANGE": true, "LINDEX": true, "LLEN": true, "SMEMBERS": true, "SISMEMBER": true, "SCARD": true,
|
||||
"ZRANGE": true, "ZREVRANGE": true, "ZSCORE": true, "ZCARD": true,
|
||||
"TYPE": true, "TTL": true, "PTTL": true, "EXISTS": true, "STRLEN": true,
|
||||
"INFO": true, "DBSIZE": true, "SCAN": true, "HSCAN": true, "SSCAN": true, "ZSCAN": true,
|
||||
"GETRANGE": true, "OBJECT": true, "RANDOMKEY": true, "PING": true, "ECHO": true,
|
||||
"XRANGE": true, "XLEN": true, "MEMORY": true, "BITCOUNT": true, "GETBIT": true,
|
||||
}
|
||||
if !readOnly[verb] {
|
||||
resp.Forbidden(c, "仅允许只读命令,数据变更请走工单流程")
|
||||
return
|
||||
}
|
||||
|
||||
// Execute command
|
||||
args := input.Args
|
||||
cmdArgs := make([]interface{}, 0, len(args)+1)
|
||||
cmdArgs = append(cmdArgs, input.Command)
|
||||
cmdArgs = append(cmdArgs, verb)
|
||||
for _, a := range args {
|
||||
cmdArgs = append(cmdArgs, a)
|
||||
}
|
||||
result, err := client.Do(ctx, cmdArgs...).Result()
|
||||
if err != nil {
|
||||
c.JSON(http.StatusOK, gin.H{"result": nil, "error": err.Error()})
|
||||
c.JSON(http.StatusOK, gin.H{"result": nil, "error": redisErrHint(err).Error()})
|
||||
return
|
||||
}
|
||||
c.JSON(http.StatusOK, gin.H{"result": result})
|
||||
}
|
||||
|
||||
// splitRedisCommand 按空白拆分命令串(双引号/单引号感知,支持引号内空格)
|
||||
func splitRedisCommand(cmdline string) (verb string, args []string) {
|
||||
var parts []string
|
||||
var cur strings.Builder
|
||||
inQuote := byte(0)
|
||||
flush := func() {
|
||||
if cur.Len() > 0 {
|
||||
parts = append(parts, cur.String())
|
||||
cur.Reset()
|
||||
}
|
||||
}
|
||||
for i := 0; i < len(cmdline); i++ {
|
||||
ch := cmdline[i]
|
||||
switch {
|
||||
case inQuote != 0:
|
||||
if ch == inQuote {
|
||||
inQuote = 0
|
||||
} else {
|
||||
cur.WriteByte(ch)
|
||||
}
|
||||
case ch == '"' || ch == '\'':
|
||||
inQuote = ch
|
||||
case ch == ' ' || ch == '\t':
|
||||
flush()
|
||||
default:
|
||||
cur.WriteByte(ch)
|
||||
}
|
||||
}
|
||||
flush()
|
||||
if len(parts) == 0 {
|
||||
return "", nil
|
||||
}
|
||||
return strings.ToUpper(parts[0]), parts
|
||||
}
|
||||
|
||||
func stringsToInterfaces(ss []string) []interface{} {
|
||||
result := make([]interface{}, len(ss))
|
||||
for i, s := range ss {
|
||||
@@ -869,16 +876,28 @@ func RedisUpdateKey(c *gin.Context) {
|
||||
defer client.Close()
|
||||
|
||||
ctx := context.Background()
|
||||
if err := client.Del(ctx, key).Err(); err != nil {
|
||||
resp.InternalErr(c, err)
|
||||
return
|
||||
|
||||
// 记录原 TTL:未显式指定新 TTL 时保留原值(旧行为会静默丢 TTL)
|
||||
prevTTL, _ := client.TTL(ctx, key).Result()
|
||||
|
||||
// string 覆盖写无需 DEL;容器类型可能收缩,需先清空再重写
|
||||
if input.Type != "string" {
|
||||
if err := client.Del(ctx, key).Err(); err != nil {
|
||||
resp.InternalErr(c, err)
|
||||
return
|
||||
}
|
||||
}
|
||||
if err := writeRedisKeyValue(ctx, client, key, input.Type, input.Value); err != nil {
|
||||
resp.BadRequest(c, err.Error())
|
||||
return
|
||||
}
|
||||
if input.TTL > 0 {
|
||||
switch {
|
||||
case input.TTL > 0:
|
||||
client.Expire(ctx, key, time.Duration(input.TTL)*time.Second)
|
||||
case input.TTL == -1:
|
||||
client.Persist(ctx, key)
|
||||
case prevTTL > 0:
|
||||
client.Expire(ctx, key, prevTTL)
|
||||
}
|
||||
|
||||
c.JSON(http.StatusOK, gin.H{"message": "Key updated"})
|
||||
|
||||
@@ -0,0 +1,371 @@
|
||||
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
|
||||
}
|
||||
Reference in new issue
Block a user