fix: Redis集群查询+metrics修复 — 单节点连接替代ClusterClient
- crypto.go: decryptPassword 自动处理 enc: 前缀,避免调用方遗漏 - redis.go: getRedisClient 统一使用 redis.NewClient 单节点连接 ClusterClient 拓扑发现(CLUSTER SLOTS)在Docker内网IP不可达时超时 redisClientWrapper 全部方法改为使用 Client (不再走 ClusterClient) ScanAllKeys 简化为单节点 SCAN - connection_tester.go: getRedisMetrics 统一单节点连接采集metrics 删除 isCluster 分支的 ClusterClient 代码,避免5秒超时
This commit is contained in:
1 parent
e9a700b9d9
commit
9c5cb04cfa
3 files changed
+55
-173
No files matched your search
@@ -200,48 +200,29 @@ func getRedisMetrics(instance *model.Instance) map[string]interface{} {
|
||||
}
|
||||
|
||||
var info string
|
||||
if isCluster {
|
||||
rdb := redis.NewClusterClient(&redis.ClusterOptions{
|
||||
Addrs: []string{addr},
|
||||
Password: password,
|
||||
DialTimeout: 5 * time.Second,
|
||||
ReadTimeout: 5 * time.Second,
|
||||
})
|
||||
defer rdb.Close()
|
||||
// For metrics collection, always use single-node connection even for clusters.
|
||||
// ClusterClient does topology discovery which can timeout if some nodes are unreachable.
|
||||
rdb := redis.NewClient(&redis.Options{
|
||||
Addr: addr,
|
||||
Password: password,
|
||||
DB: 0,
|
||||
DialTimeout: 5 * time.Second,
|
||||
ReadTimeout: 5 * time.Second,
|
||||
WriteTimeout: 3 * time.Second,
|
||||
})
|
||||
defer rdb.Close()
|
||||
|
||||
_, err := rdb.Ping(ctx).Result()
|
||||
if err != nil {
|
||||
metrics["status"] = "offline"
|
||||
return metrics
|
||||
}
|
||||
metrics["status"] = "online"
|
||||
info, _ = rdb.Info(ctx, "server", "clients", "stats").Result()
|
||||
// Latency from PING
|
||||
start := time.Now()
|
||||
rdb.Ping(ctx)
|
||||
metrics["latency"] = time.Since(start).Milliseconds()
|
||||
} else {
|
||||
rdb := redis.NewClient(&redis.Options{
|
||||
Addr: addr,
|
||||
Password: password,
|
||||
DB: 0,
|
||||
DialTimeout: 5 * time.Second,
|
||||
ReadTimeout: 5 * time.Second,
|
||||
})
|
||||
defer rdb.Close()
|
||||
|
||||
_, err := rdb.Ping(ctx).Result()
|
||||
if err != nil {
|
||||
metrics["status"] = "offline"
|
||||
return metrics
|
||||
}
|
||||
metrics["status"] = "online"
|
||||
info, _ = rdb.Info(ctx, "server", "clients", "stats").Result()
|
||||
// Latency from PING
|
||||
start := time.Now()
|
||||
rdb.Ping(ctx)
|
||||
metrics["latency"] = time.Since(start).Milliseconds()
|
||||
_, err := rdb.Ping(ctx).Result()
|
||||
if err != nil {
|
||||
metrics["status"] = "offline"
|
||||
return metrics
|
||||
}
|
||||
metrics["status"] = "online"
|
||||
info, _ = rdb.Info(ctx, "server", "clients", "stats").Result()
|
||||
// Latency from PING
|
||||
start := time.Now()
|
||||
rdb.Ping(ctx)
|
||||
metrics["latency"] = time.Since(start).Milliseconds()
|
||||
|
||||
if info == "" {
|
||||
return metrics
|
||||
|
||||
@@ -6,6 +6,7 @@ import (
|
||||
"crypto/rand"
|
||||
"encoding/base64"
|
||||
"os"
|
||||
"strings"
|
||||
)
|
||||
|
||||
// getSecretKey returns the AES key derived from SECRET_KEY env var.
|
||||
@@ -26,8 +27,12 @@ func getSecretKey() []byte {
|
||||
|
||||
// decryptPassword decrypts a base64-encoded AES-256-GCM ciphertext.
|
||||
// Mirrors Python's crypto.decrypt_password().
|
||||
// If the value is not valid encrypted data (e.g., plaintext), returns it as-is.
|
||||
// Handles "enc:" prefix automatically. If not valid encrypted data, returns as-is.
|
||||
func decryptPassword(encrypted string) (string, error) {
|
||||
// Strip "enc:" prefix if present
|
||||
if strings.HasPrefix(encrypted, "enc:") {
|
||||
encrypted = strings.TrimPrefix(encrypted, "enc:")
|
||||
}
|
||||
raw, err := base64.StdEncoding.DecodeString(encrypted)
|
||||
if err != nil {
|
||||
return encrypted, nil // not valid base64 — plaintext
|
||||
|
||||
@@ -21,221 +21,127 @@ type redisClientWrapper struct {
|
||||
}
|
||||
|
||||
func (w *redisClientWrapper) Close() {
|
||||
if w.isCluster {
|
||||
w.ClusterClient.Close()
|
||||
} else {
|
||||
// Always close the single-node Client (cluster uses single-node now)
|
||||
if w.Client != nil {
|
||||
w.Client.Close()
|
||||
}
|
||||
if w.ClusterClient != nil {
|
||||
w.ClusterClient.Close()
|
||||
}
|
||||
}
|
||||
|
||||
// Ping, Info, DBSize, Scan, Type, TTL, Get, LRange, SMembers, ZRangeWithScores, HGetAll,
|
||||
// SlowLogGet, ClientList, Do, Set, RPush, SAdd, HSet, Expire, Del — unified interface via wrapper methods
|
||||
// NOTE: All methods use Client (single-node). isCluster flag is only for ScanAllKeys behavior.
|
||||
|
||||
func (w *redisClientWrapper) Ping(ctx context.Context) *redis.StatusCmd {
|
||||
if w.isCluster {
|
||||
return w.ClusterClient.Ping(ctx)
|
||||
}
|
||||
return w.Client.Ping(ctx)
|
||||
}
|
||||
|
||||
func (w *redisClientWrapper) Info(ctx context.Context, sections ...string) *redis.StringCmd {
|
||||
if w.isCluster {
|
||||
return w.ClusterClient.Info(ctx, sections...)
|
||||
}
|
||||
return w.Client.Info(ctx, sections...)
|
||||
}
|
||||
|
||||
func (w *redisClientWrapper) DBSize(ctx context.Context) *redis.IntCmd {
|
||||
if w.isCluster {
|
||||
return w.ClusterClient.DBSize(ctx)
|
||||
}
|
||||
return w.Client.DBSize(ctx)
|
||||
}
|
||||
|
||||
func (w *redisClientWrapper) Scan(ctx context.Context, cursor uint64, match string, count int64) *redis.ScanCmd {
|
||||
if w.isCluster {
|
||||
return w.ClusterClient.Scan(ctx, cursor, match, count)
|
||||
}
|
||||
return w.Client.Scan(ctx, cursor, match, count)
|
||||
}
|
||||
|
||||
// ScanAllKeys scans keys across ALL master nodes in cluster mode, or single node in standalone.
|
||||
// Returns merged deduplicated keys.
|
||||
// ScanAllKeys scans keys on the connected node.
|
||||
// For clusters, this only scans keys on the single connected node (not all masters).
|
||||
func (w *redisClientWrapper) ScanAllKeys(ctx context.Context, match string, count int) ([]string, error) {
|
||||
if !w.isCluster {
|
||||
// Standalone: iterate SCAN until cursor returns to 0
|
||||
var allKeys []string
|
||||
cursor := uint64(0)
|
||||
for {
|
||||
keys, nextCursor, err := w.Client.Scan(ctx, cursor, match, int64(count)).Result()
|
||||
if err != nil {
|
||||
return allKeys, err
|
||||
}
|
||||
allKeys = append(allKeys, keys...)
|
||||
cursor = nextCursor
|
||||
if cursor == 0 {
|
||||
break
|
||||
}
|
||||
}
|
||||
return allKeys, nil
|
||||
}
|
||||
|
||||
// Cluster: iterate SCAN on every master node
|
||||
var allKeys []string
|
||||
seen := make(map[string]bool)
|
||||
err := w.ClusterClient.ForEachMaster(ctx, func(ctx context.Context, node *redis.Client) error {
|
||||
cursor := uint64(0)
|
||||
for {
|
||||
keys, nextCursor, err := node.Scan(ctx, cursor, match, int64(count)).Result()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
for _, k := range keys {
|
||||
if !seen[k] {
|
||||
seen[k] = true
|
||||
allKeys = append(allKeys, k)
|
||||
}
|
||||
}
|
||||
cursor = nextCursor
|
||||
if cursor == 0 {
|
||||
break
|
||||
}
|
||||
cursor := uint64(0)
|
||||
for {
|
||||
keys, nextCursor, err := w.Client.Scan(ctx, cursor, match, int64(count)).Result()
|
||||
if err != nil {
|
||||
return allKeys, err
|
||||
}
|
||||
return nil
|
||||
})
|
||||
return allKeys, err
|
||||
allKeys = append(allKeys, keys...)
|
||||
cursor = nextCursor
|
||||
if cursor == 0 {
|
||||
break
|
||||
}
|
||||
}
|
||||
return allKeys, nil
|
||||
}
|
||||
|
||||
func (w *redisClientWrapper) Type(ctx context.Context, key string) *redis.StatusCmd {
|
||||
if w.isCluster {
|
||||
return w.ClusterClient.Type(ctx, key)
|
||||
}
|
||||
return w.Client.Type(ctx, key)
|
||||
}
|
||||
|
||||
func (w *redisClientWrapper) TTL(ctx context.Context, key string) *redis.DurationCmd {
|
||||
if w.isCluster {
|
||||
return w.ClusterClient.TTL(ctx, key)
|
||||
}
|
||||
return w.Client.TTL(ctx, key)
|
||||
}
|
||||
|
||||
func (w *redisClientWrapper) Get(ctx context.Context, key string) *redis.StringCmd {
|
||||
if w.isCluster {
|
||||
return w.ClusterClient.Get(ctx, key)
|
||||
}
|
||||
return w.Client.Get(ctx, key)
|
||||
}
|
||||
|
||||
func (w *redisClientWrapper) LRange(ctx context.Context, key string, start, stop int64) *redis.StringSliceCmd {
|
||||
if w.isCluster {
|
||||
return w.ClusterClient.LRange(ctx, key, start, stop)
|
||||
}
|
||||
return w.Client.LRange(ctx, key, start, stop)
|
||||
}
|
||||
|
||||
func (w *redisClientWrapper) SMembers(ctx context.Context, key string) *redis.StringSliceCmd {
|
||||
if w.isCluster {
|
||||
return w.ClusterClient.SMembers(ctx, key)
|
||||
}
|
||||
return w.Client.SMembers(ctx, key)
|
||||
}
|
||||
|
||||
func (w *redisClientWrapper) ZRangeWithScores(ctx context.Context, key string, start, stop int64) *redis.ZSliceCmd {
|
||||
if w.isCluster {
|
||||
return w.ClusterClient.ZRangeWithScores(ctx, key, start, stop)
|
||||
}
|
||||
return w.Client.ZRangeWithScores(ctx, key, start, stop)
|
||||
}
|
||||
|
||||
func (w *redisClientWrapper) HGetAll(ctx context.Context, key string) *redis.MapStringStringCmd {
|
||||
if w.isCluster {
|
||||
return w.ClusterClient.HGetAll(ctx, key)
|
||||
}
|
||||
return w.Client.HGetAll(ctx, key)
|
||||
}
|
||||
|
||||
func (w *redisClientWrapper) SlowLogGet(ctx context.Context, count int64) *redis.SlowLogCmd {
|
||||
if w.isCluster {
|
||||
return w.ClusterClient.SlowLogGet(ctx, count)
|
||||
}
|
||||
return w.Client.SlowLogGet(ctx, count)
|
||||
}
|
||||
|
||||
func (w *redisClientWrapper) ClientList(ctx context.Context) *redis.StringCmd {
|
||||
if w.isCluster {
|
||||
return w.ClusterClient.ClientList(ctx)
|
||||
}
|
||||
return w.Client.ClientList(ctx)
|
||||
}
|
||||
|
||||
func (w *redisClientWrapper) Do(ctx context.Context, args ...interface{}) *redis.Cmd {
|
||||
if w.isCluster {
|
||||
return w.ClusterClient.Do(ctx, args...)
|
||||
}
|
||||
return w.Client.Do(ctx, args...)
|
||||
}
|
||||
|
||||
func (w *redisClientWrapper) Set(ctx context.Context, key string, value interface{}, expiration time.Duration) *redis.StatusCmd {
|
||||
if w.isCluster {
|
||||
return w.ClusterClient.Set(ctx, key, value, expiration)
|
||||
}
|
||||
return w.Client.Set(ctx, key, value, expiration)
|
||||
}
|
||||
|
||||
func (w *redisClientWrapper) RPush(ctx context.Context, key string, values ...interface{}) *redis.IntCmd {
|
||||
if w.isCluster {
|
||||
return w.ClusterClient.RPush(ctx, key, values...)
|
||||
}
|
||||
return w.Client.RPush(ctx, key, values...)
|
||||
}
|
||||
|
||||
func (w *redisClientWrapper) SAdd(ctx context.Context, key string, members ...interface{}) *redis.IntCmd {
|
||||
if w.isCluster {
|
||||
return w.ClusterClient.SAdd(ctx, key, members...)
|
||||
}
|
||||
return w.Client.SAdd(ctx, key, members...)
|
||||
}
|
||||
|
||||
func (w *redisClientWrapper) HSet(ctx context.Context, key string, values ...interface{}) *redis.IntCmd {
|
||||
if w.isCluster {
|
||||
return w.ClusterClient.HSet(ctx, key, values...)
|
||||
}
|
||||
return w.Client.HSet(ctx, key, values...)
|
||||
}
|
||||
|
||||
func (w *redisClientWrapper) Expire(ctx context.Context, key string, expiration time.Duration) *redis.BoolCmd {
|
||||
if w.isCluster {
|
||||
return w.ClusterClient.Expire(ctx, key, expiration)
|
||||
}
|
||||
return w.Client.Expire(ctx, key, expiration)
|
||||
}
|
||||
|
||||
func (w *redisClientWrapper) Del(ctx context.Context, keys ...string) *redis.IntCmd {
|
||||
if w.isCluster {
|
||||
return w.ClusterClient.Del(ctx, keys...)
|
||||
}
|
||||
return w.Client.Del(ctx, keys...)
|
||||
}
|
||||
|
||||
func (w *redisClientWrapper) LLen(ctx context.Context, key string) *redis.IntCmd {
|
||||
if w.isCluster {
|
||||
return w.ClusterClient.LLen(ctx, key)
|
||||
}
|
||||
return w.Client.LLen(ctx, key)
|
||||
}
|
||||
|
||||
func (w *redisClientWrapper) SCard(ctx context.Context, key string) *redis.IntCmd {
|
||||
if w.isCluster {
|
||||
return w.ClusterClient.SCard(ctx, key)
|
||||
}
|
||||
return w.Client.SCard(ctx, key)
|
||||
}
|
||||
|
||||
func (w *redisClientWrapper) SScan(ctx context.Context, key string, cursor uint64, match string, count int64) *redis.ScanCmd {
|
||||
if w.isCluster {
|
||||
return w.ClusterClient.SScan(ctx, key, cursor, match, count)
|
||||
}
|
||||
return w.Client.SScan(ctx, key, cursor, match, count)
|
||||
}
|
||||
|
||||
@@ -250,23 +156,10 @@ func getRedisClient(instance *model.Instance, db int) (*redisClientWrapper, erro
|
||||
}
|
||||
addr := fmt.Sprintf("%s:%d", instance.Host, instance.Port)
|
||||
|
||||
// Check cluster mode
|
||||
isCluster := instance.ClusterMode != nil && *instance.ClusterMode
|
||||
|
||||
if isCluster {
|
||||
clusterClient := redis.NewClusterClient(&redis.ClusterOptions{
|
||||
Addrs: []string{addr},
|
||||
Password: password,
|
||||
})
|
||||
ctx := context.Background()
|
||||
if err := clusterClient.Ping(ctx).Err(); err != nil {
|
||||
clusterClient.Close()
|
||||
return nil, fmt.Errorf("Redis Cluster connection failed: %v", err)
|
||||
}
|
||||
return &redisClientWrapper{ClusterClient: clusterClient, isCluster: true}, nil
|
||||
}
|
||||
|
||||
// Standalone mode
|
||||
// Always use single-node connection for query execution.
|
||||
// ClusterClient does topology discovery (CLUSTER SLOTS) which fails when
|
||||
// internal cluster IPs are unreachable from the client (e.g., Docker networks).
|
||||
// Single-node mode works for all read-only commands on cluster nodes.
|
||||
client := redis.NewClient(&redis.Options{
|
||||
Addr: addr,
|
||||
Password: password,
|
||||
@@ -277,7 +170,10 @@ func getRedisClient(instance *model.Instance, db int) (*redisClientWrapper, erro
|
||||
client.Close()
|
||||
return nil, fmt.Errorf("Redis connection failed: %v", err)
|
||||
}
|
||||
return &redisClientWrapper{Client: client, isCluster: false}, nil
|
||||
|
||||
// Check cluster mode flag for wrapper behavior (ScanAllKeys needs to know)
|
||||
isCluster := instance.ClusterMode != nil && *instance.ClusterMode
|
||||
return &redisClientWrapper{Client: client, isCluster: isCluster}, nil
|
||||
}
|
||||
|
||||
// GET /api/v1/redis/:id/info
|
||||
|
||||
Reference in new issue
Block a user