Files
opsxsql/backend/internal/handler/redis.go
T

706 lines
18 KiB
Go

package handler
import (
"context"
"fmt"
"net/http"
"strconv"
"strings"
"time"
"sql-platform/internal/model"
"github.com/gin-gonic/gin"
"github.com/redis/go-redis/v9"
)
// Helper: get Redis client from instance (auto-detect standalone vs cluster)
type redisClientWrapper struct {
Client *redis.Client
ClusterClient *redis.ClusterClient
isCluster bool
}
func (w *redisClientWrapper) Close() {
if w.isCluster {
w.ClusterClient.Close()
} else {
w.Client.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
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)
}
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)
}
func getRedisClient(instance *model.Instance, db int) (*redisClientWrapper, error) {
password := instance.Password
// Decrypt if encrypted
if strings.HasPrefix(password, "enc:") {
decrypted, err := decryptPassword(strings.TrimPrefix(password, "enc:"))
if err == nil {
password = decrypted
}
}
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
client := redis.NewClient(&redis.Options{
Addr: addr,
Password: password,
DB: db,
})
ctx := context.Background()
if err := client.Ping(ctx).Err(); err != nil {
client.Close()
return nil, fmt.Errorf("Redis connection failed: %v", err)
}
return &redisClientWrapper{Client: client, isCluster: false}, nil
}
// GET /api/v1/redis/:id/info
func RedisInfo(c *gin.Context) {
id := c.Param("id")
var instance model.Instance
if err := model.DB.First(&instance, id).Error; err != nil {
c.JSON(http.StatusNotFound, gin.H{"detail": "Instance not found"})
return
}
client, err := getRedisClient(&instance, 0)
if err != nil {
c.JSON(http.StatusInternalServerError, gin.H{"detail": err.Error()})
return
}
defer client.Close()
ctx := context.Background()
info, err := client.Info(ctx, "server", "memory", "clients", "stats", "keyspace").Result()
if err != nil {
c.JSON(http.StatusInternalServerError, gin.H{"detail": err.Error()})
return
}
// Parse info into structured map
result := make(map[string]interface{})
sections := strings.Split(info, "\n# ")
for _, section := range sections {
lines := strings.Split(section, "\n")
for _, line := range lines {
line = strings.TrimSpace(line)
if line == "" || strings.HasPrefix(line, "#") {
continue
}
parts := strings.SplitN(line, ":", 2)
if len(parts) == 2 {
result[strings.TrimSpace(parts[0])] = strings.TrimSpace(parts[1])
}
}
}
// Add db_size
dbSize, _ := client.DBSize(ctx).Result()
result["db_size"] = dbSize
c.JSON(http.StatusOK, result)
}
// GET /api/v1/redis/:id/dbsize
func RedisDbsize(c *gin.Context) {
id := c.Param("id")
dbParam := c.Query("db")
db := 0
if dbParam != "" {
db, _ = strconv.Atoi(dbParam)
}
var instance model.Instance
if err := model.DB.First(&instance, id).Error; err != nil {
c.JSON(http.StatusNotFound, gin.H{"detail": "Instance not found"})
return
}
client, err := getRedisClient(&instance, db)
if err != nil {
c.JSON(http.StatusInternalServerError, gin.H{"detail": err.Error()})
return
}
defer client.Close()
ctx := context.Background()
dbSize, err := client.DBSize(ctx).Result()
if err != nil {
c.JSON(http.StatusInternalServerError, gin.H{"detail": err.Error()})
return
}
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")
pattern := c.DefaultQuery("pattern", "*")
// Auto-append * if pattern has no wildcard (user-friendly fuzzy search)
if !strings.Contains(pattern, "*") && !strings.Contains(pattern, "?") && !strings.Contains(pattern, "[") {
pattern = pattern + "*"
}
count, _ := strconv.Atoi(c.DefaultQuery("count", "100"))
cursorParam := c.DefaultQuery("cursor", "0")
db := 0
if dbParam != "" {
db, _ = strconv.Atoi(dbParam)
}
var instance model.Instance
if err := model.DB.First(&instance, id).Error; err != nil {
c.JSON(http.StatusNotFound, gin.H{"detail": "Instance not found"})
return
}
client, err := getRedisClient(&instance, db)
if err != nil {
c.JSON(http.StatusInternalServerError, gin.H{"detail": err.Error()})
return
}
defer client.Close()
ctx := context.Background()
cursor, _ := strconv.ParseUint(cursorParam, 10, 64)
keys, nextCursor, err := client.Scan(ctx, cursor, pattern, int64(count)).Result()
if err != nil {
c.JSON(http.StatusInternalServerError, gin.H{"detail": err.Error()})
return
}
// Get types for each key
type KeyInfo struct {
Key string `json:"key"`
Type string `json:"type"`
TTL int64 `json:"ttl"`
}
var keyInfos []KeyInfo
for _, key := range keys {
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": nextCursor,
"total": len(keyInfos),
})
}
// GET /api/v1/redis/:id/keys/:key
func RedisGetKeyDetail(c *gin.Context) {
id := c.Param("id")
key := c.Param("key")
dbParam := c.Query("db")
db := 0
if dbParam != "" {
db, _ = strconv.Atoi(dbParam)
}
var instance model.Instance
if err := model.DB.First(&instance, id).Error; err != nil {
c.JSON(http.StatusNotFound, gin.H{"detail": "Instance not found"})
return
}
client, err := getRedisClient(&instance, db)
if err != nil {
c.JSON(http.StatusInternalServerError, gin.H{"detail": err.Error()})
return
}
defer client.Close()
ctx := context.Background()
keyType, _ := client.Type(ctx, key).Result()
ttl, _ := client.TTL(ctx, key).Result()
ttlMs := int64(-1)
if ttl > 0 {
ttlMs = ttl.Milliseconds()
}
var value interface{}
switch keyType {
case "string":
value, _ = client.Get(ctx, key).Result()
case "list":
value, _ = client.LRange(ctx, key, 0, -1).Result()
case "set":
value, _ = client.SMembers(ctx, key).Result()
case "zset":
value, _ = client.ZRangeWithScores(ctx, key, 0, -1).Result()
case "hash":
value, _ = client.HGetAll(ctx, key).Result()
default:
value = nil
}
c.JSON(http.StatusOK, gin.H{
"key": key,
"type": keyType,
"value": value,
"ttl": ttlMs,
})
}
// GET /api/v1/redis/:id/slowlog
func RedisSlowlog(c *gin.Context) {
id := c.Param("id")
count, _ := strconv.Atoi(c.DefaultQuery("count", "10"))
var instance model.Instance
if err := model.DB.First(&instance, id).Error; err != nil {
c.JSON(http.StatusNotFound, gin.H{"detail": "Instance not found"})
return
}
client, err := getRedisClient(&instance, 0)
if err != nil {
c.JSON(http.StatusInternalServerError, gin.H{"detail": err.Error()})
return
}
defer client.Close()
ctx := context.Background()
logs, err := client.SlowLogGet(ctx, int64(count)).Result()
if err != nil {
c.JSON(http.StatusInternalServerError, gin.H{"detail": err.Error()})
return
}
var result []map[string]interface{}
for _, log := range logs {
result = append(result, map[string]interface{}{
"id": log.ID,
"time": log.Time.Unix(),
"duration": log.Duration.Microseconds(),
"command": fmt.Sprintf("%v", log.Args),
"client": log.ClientAddr,
})
}
if result == nil {
result = []map[string]interface{}{}
}
c.JSON(http.StatusOK, result)
}
// GET /api/v1/redis/:id/clients
func RedisClients(c *gin.Context) {
id := c.Param("id")
var instance model.Instance
if err := model.DB.First(&instance, id).Error; err != nil {
c.JSON(http.StatusNotFound, gin.H{"detail": "Instance not found"})
return
}
client, err := getRedisClient(&instance, 0)
if err != nil {
c.JSON(http.StatusInternalServerError, gin.H{"detail": err.Error()})
return
}
defer client.Close()
ctx := context.Background()
clientList, err := client.ClientList(ctx).Result()
if err != nil {
c.JSON(http.StatusInternalServerError, gin.H{"detail": err.Error()})
return
}
// Parse client list
var clients []map[string]interface{}
for _, line := range strings.Split(clientList, "\n") {
line = strings.TrimSpace(line)
if line == "" {
continue
}
clientInfo := make(map[string]interface{})
for _, field := range strings.Split(line, " ") {
parts := strings.SplitN(field, "=", 2)
if len(parts) == 2 {
clientInfo[parts[0]] = parts[1]
}
}
clients = append(clients, clientInfo)
}
if clients == nil {
clients = []map[string]interface{}{}
}
c.JSON(http.StatusOK, clients)
}
// 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 {
Command string `json:"command" binding:"required"`
Args []string `json:"args"`
}
if err := c.ShouldBindJSON(&input); err != nil {
c.JSON(http.StatusBadRequest, gin.H{"detail": err.Error()})
return
}
var instance model.Instance
if err := model.DB.First(&instance, id).Error; err != nil {
c.JSON(http.StatusNotFound, gin.H{"detail": "Instance not found"})
return
}
client, err := getRedisClient(&instance, db)
if err != nil {
c.JSON(http.StatusInternalServerError, gin.H{"detail": err.Error()})
return
}
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,
}
if !readOnly[cmd] {
c.JSON(http.StatusForbidden, gin.H{"detail": "Only read-only commands allowed via query interface"})
return
}
// Execute command
args := input.Args
cmdArgs := make([]interface{}, 0, len(args)+1)
cmdArgs = append(cmdArgs, input.Command)
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()})
return
}
c.JSON(http.StatusOK, gin.H{"result": result})
}
func stringsToInterfaces(ss []string) []interface{} {
result := make([]interface{}, len(ss))
for i, s := range ss {
result[i] = s
}
return result
}
// POST /api/v1/redis/:id/keys
func RedisCreateKey(c *gin.Context) {
id := c.Param("id")
dbParam := c.Query("db")
db := 0
if dbParam != "" {
db, _ = strconv.Atoi(dbParam)
}
var input struct {
Key string `json:"key" binding:"required"`
Type string `json:"type" binding:"required"`
Value interface{} `json:"value"`
TTL int `json:"ttl"`
}
if err := c.ShouldBindJSON(&input); err != nil {
c.JSON(http.StatusBadRequest, gin.H{"detail": err.Error()})
return
}
var instance model.Instance
if err := model.DB.First(&instance, id).Error; err != nil {
c.JSON(http.StatusNotFound, gin.H{"detail": "Instance not found"})
return
}
client, err := getRedisClient(&instance, db)
if err != nil {
c.JSON(http.StatusInternalServerError, gin.H{"detail": err.Error()})
return
}
defer client.Close()
ctx := context.Background()
switch input.Type {
case "string":
client.Set(ctx, input.Key, input.Value, 0)
case "list":
if vals, ok := input.Value.([]interface{}); ok {
for _, v := range vals {
client.RPush(ctx, input.Key, fmt.Sprintf("%v", v))
}
}
case "set":
if vals, ok := input.Value.([]interface{}); ok {
for _, v := range vals {
client.SAdd(ctx, input.Key, fmt.Sprintf("%v", v))
}
}
case "hash":
if vals, ok := input.Value.(map[string]interface{}); ok {
client.HSet(ctx, input.Key, vals)
}
}
if input.TTL > 0 {
client.Expire(ctx, input.Key, time.Duration(input.TTL)*time.Second)
}
c.JSON(http.StatusOK, gin.H{"message": "Key created"})
}
// DELETE /api/v1/redis/:id/keys/:key
func RedisDeleteKey(c *gin.Context) {
id := c.Param("id")
key := c.Param("key")
dbParam := c.Query("db")
db := 0
if dbParam != "" {
db, _ = strconv.Atoi(dbParam)
}
var instance model.Instance
if err := model.DB.First(&instance, id).Error; err != nil {
c.JSON(http.StatusNotFound, gin.H{"detail": "Instance not found"})
return
}
client, err := getRedisClient(&instance, db)
if err != nil {
c.JSON(http.StatusInternalServerError, gin.H{"detail": err.Error()})
return
}
defer client.Close()
ctx := context.Background()
deleted, err := client.Del(ctx, key).Result()
if err != nil {
c.JSON(http.StatusInternalServerError, gin.H{"detail": err.Error()})
return
}
c.JSON(http.StatusOK, gin.H{"message": "Key deleted", "deleted": deleted})
}
// POST /api/v1/workflows/:id/clone
func CloneWorkflow(c *gin.Context) {
id, _ := strconv.Atoi(c.Param("id"))
var wf model.WorkflowDefinition
if err := model.DB.First(&wf, id).Error; err != nil {
c.JSON(http.StatusNotFound, gin.H{"detail": "Workflow not found"})
return
}
// clone
newWf := model.WorkflowDefinition{
Name: wf.Name + " (副本)",
Description: wf.Description,
Category: wf.Category,
IsActive: wf.IsActive,
FlowConfig: wf.FlowConfig,
}
model.DB.Create(&newWf)
c.JSON(http.StatusCreated, newWf)
}