feat: Redis cluster support + version auto-detect + metrics API
This commit is contained in:
1 parent
8384006ae9
commit
3af65f2350
10 files changed
+485
-66
No files matched your search
@@ -67,9 +67,12 @@ func Login(c *gin.Context) {
|
||||
c.JSON(http.StatusUnauthorized, gin.H{"detail": "Invalid username or password"})
|
||||
return
|
||||
}
|
||||
// LDAP 认证成功,自动创建本地用户
|
||||
// LDAP 认证成功,自动创建或恢复本地用户
|
||||
isActive := true
|
||||
defaultRole := "developer"
|
||||
if ldapInfo.DisplayName == "" {
|
||||
ldapInfo.DisplayName = ldapInfo.Username
|
||||
}
|
||||
newUser := model.User{
|
||||
Username: ldapInfo.Username,
|
||||
DisplayName: ldapInfo.DisplayName,
|
||||
@@ -79,20 +82,25 @@ func Login(c *gin.Context) {
|
||||
Role: defaultRole,
|
||||
IsActive: &isActive,
|
||||
}
|
||||
if newUser.DisplayName == "" {
|
||||
newUser.DisplayName = newUser.Username
|
||||
// FirstOrCreate: 不存在则创建,已存在则获取(避免 uniqueIndex 冲突)
|
||||
result := model.DB.Where("username = ?", ldapInfo.Username).FirstOrCreate(&newUser)
|
||||
if result.Error != nil {
|
||||
c.JSON(http.StatusInternalServerError, gin.H{"detail": "Failed to create LDAP user: " + result.Error.Error()})
|
||||
return
|
||||
}
|
||||
if createErr := model.DB.Create(&newUser).Error; createErr != nil {
|
||||
// 可能并发创建冲突,再查一次
|
||||
model.DB.Where("username = ?", ldapInfo.Username).First(&user)
|
||||
if user.ID == 0 {
|
||||
c.JSON(http.StatusInternalServerError, gin.H{"detail": "Failed to create LDAP user"})
|
||||
return
|
||||
// 更新已有用户的 display_name / email(LDAP 信息可能变更)
|
||||
if newUser.ID != 0 {
|
||||
updates := map[string]interface{}{
|
||||
"display_name": ldapInfo.DisplayName,
|
||||
"auth_source": "ldap",
|
||||
"is_active": true,
|
||||
}
|
||||
} else {
|
||||
user = newUser
|
||||
if ldapInfo.Email != nil {
|
||||
updates["email"] = *ldapInfo.Email
|
||||
}
|
||||
model.DB.Model(&newUser).Updates(updates)
|
||||
}
|
||||
// 记录 auth_source
|
||||
user = newUser
|
||||
} else {
|
||||
c.JSON(http.StatusInternalServerError, gin.H{"detail": "Database error"})
|
||||
return
|
||||
|
||||
@@ -13,6 +13,14 @@ import (
|
||||
"sql-platform/internal/model"
|
||||
)
|
||||
|
||||
// ConnectionResult holds test connection results
|
||||
type ConnectionResult struct {
|
||||
Success bool `json:"success"`
|
||||
Message string `json:"message"`
|
||||
Version string `json:"version,omitempty"`
|
||||
ClusterMode bool `json:"cluster_mode"`
|
||||
}
|
||||
|
||||
// testDBConnection tests a real database connection based on db_type.
|
||||
func testDBConnection(instance *model.Instance) (bool, string) {
|
||||
password, err := decryptPassword(instance.Password)
|
||||
@@ -33,12 +41,54 @@ func testDBConnection(instance *model.Instance) (bool, string) {
|
||||
}
|
||||
return testPostgreSQL(ctx, instance.Host, instance.Port, instance.Username, password, dbname)
|
||||
case "redis":
|
||||
return testRedis(ctx, instance.Host, instance.Port, password)
|
||||
ok, msg, _ := testRedis(ctx, instance.Host, instance.Port, password)
|
||||
return ok, msg
|
||||
default:
|
||||
return false, fmt.Sprintf("Unsupported database type: %s", instance.DBType)
|
||||
}
|
||||
}
|
||||
|
||||
// testDBConnectionFull returns detailed result including version
|
||||
func testDBConnectionFull(instance *model.Instance) *ConnectionResult {
|
||||
password, err := decryptPassword(instance.Password)
|
||||
if err != nil {
|
||||
return &ConnectionResult{Success: false, Message: fmt.Sprintf("Password decryption failed: %v", err)}
|
||||
}
|
||||
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
|
||||
defer cancel()
|
||||
|
||||
switch instance.DBType {
|
||||
case "mysql":
|
||||
ok, msg := testMySQL(ctx, instance.Host, instance.Port, instance.Username, password)
|
||||
return &ConnectionResult{Success: ok, Message: msg, Version: extractVersion(msg, "version: ")}
|
||||
case "postgresql", "postgres", "pg":
|
||||
dbname := ""
|
||||
if instance.Database != nil {
|
||||
dbname = *instance.Database
|
||||
}
|
||||
ok, msg := testPostgreSQL(ctx, instance.Host, instance.Port, instance.Username, password, dbname)
|
||||
return &ConnectionResult{Success: ok, Message: msg, Version: extractVersion(msg, "version: ")}
|
||||
case "redis":
|
||||
ok, msg, isCluster := testRedis(ctx, instance.Host, instance.Port, password)
|
||||
return &ConnectionResult{Success: ok, Message: msg, Version: extractVersion(msg, "version: "), ClusterMode: isCluster}
|
||||
default:
|
||||
return &ConnectionResult{Success: false, Message: fmt.Sprintf("Unsupported database type: %s", instance.DBType)}
|
||||
}
|
||||
}
|
||||
|
||||
func extractVersion(msg, prefix string) string {
|
||||
idx := strings.Index(msg, prefix)
|
||||
if idx < 0 {
|
||||
return ""
|
||||
}
|
||||
ver := msg[idx+len(prefix):]
|
||||
if end := strings.Index(ver, ")"); end > 0 {
|
||||
ver = ver[:end]
|
||||
}
|
||||
return strings.TrimSpace(ver)
|
||||
}
|
||||
|
||||
func testMySQL(ctx context.Context, host string, port int, user, pass string) (bool, string) {
|
||||
dsn := fmt.Sprintf("%s:%s@tcp(%s:%d)/?timeout=10s&parseTime=true", user, pass, host, port)
|
||||
db, err := sql.Open("mysql", dsn)
|
||||
@@ -60,7 +110,7 @@ func testPostgreSQL(ctx context.Context, host string, port int, user, pass, dbna
|
||||
if dbname == "" {
|
||||
dbname = "postgres"
|
||||
}
|
||||
dsn := fmt.Sprintf("postgres://%s:%s@%s:%d/%s?sslmode=disable&connect_timeout=10",
|
||||
dsn := fmt.Sprintf("postgres://%s:***@%s:%d/%s?sslmode=disable&connect_timeout=10",
|
||||
user, pass, host, port, dbname)
|
||||
db, err := sql.Open("pgx", dsn)
|
||||
if err != nil {
|
||||
@@ -77,7 +127,7 @@ func testPostgreSQL(ctx context.Context, host string, port int, user, pass, dbna
|
||||
return true, fmt.Sprintf("PostgreSQL connection successful (version: %s)", ver)
|
||||
}
|
||||
|
||||
func testRedis(ctx context.Context, host string, port int, pass string) (bool, string) {
|
||||
func testRedis(ctx context.Context, host string, port int, pass string) (bool, string, bool) {
|
||||
addr := fmt.Sprintf("%s:%d", host, port)
|
||||
rdb := redis.NewClient(&redis.Options{
|
||||
Addr: addr,
|
||||
@@ -90,26 +140,145 @@ func testRedis(ctx context.Context, host string, port int, pass string) (bool, s
|
||||
|
||||
pong, err := rdb.Ping(ctx).Result()
|
||||
if err != nil {
|
||||
return false, fmt.Sprintf("Connection failed: %v", err)
|
||||
return false, fmt.Sprintf("Connection failed: %v", err), false
|
||||
}
|
||||
|
||||
var version string
|
||||
if info, err := rdb.Info(ctx, "server").Result(); err == nil {
|
||||
var isCluster bool
|
||||
if info, err := rdb.Info(ctx, "server", "cluster").Result(); err == nil {
|
||||
for _, line := range strings.Split(info, "\n") {
|
||||
line = strings.TrimRight(line, "\r")
|
||||
if strings.HasPrefix(line, "redis_version:") {
|
||||
version = strings.TrimPrefix(line, "redis_version:")
|
||||
break
|
||||
}
|
||||
if strings.HasPrefix(line, "cluster_enabled:") {
|
||||
val := strings.TrimPrefix(line, "cluster_enabled:")
|
||||
isCluster = strings.TrimSpace(val) == "1"
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
if version != "" {
|
||||
return true, fmt.Sprintf("Redis connection successful (PING=%s, version: %s)", pong, version)
|
||||
mode := "standalone"
|
||||
if isCluster {
|
||||
mode = "cluster"
|
||||
}
|
||||
return true, fmt.Sprintf("Redis connection successful (PING=%s)", pong)
|
||||
|
||||
if version != "" {
|
||||
return true, fmt.Sprintf("Redis connection successful (PING=%s, version: %s, mode: %s)", pong, version, mode), isCluster
|
||||
}
|
||||
return true, fmt.Sprintf("Redis connection successful (PING=%s, mode: %s)", pong, mode), isCluster
|
||||
}
|
||||
|
||||
// getRedisMetrics fetches live Redis metrics for instance overview
|
||||
func getRedisMetrics(instance *model.Instance) map[string]interface{} {
|
||||
password := instance.Password
|
||||
if strings.HasPrefix(password, "enc:") {
|
||||
decrypted, err := decryptPassword(strings.TrimPrefix(password, "enc:"))
|
||||
if err == nil {
|
||||
password = decrypted
|
||||
}
|
||||
}
|
||||
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
|
||||
defer cancel()
|
||||
|
||||
addr := fmt.Sprintf("%s:%d", instance.Host, instance.Port)
|
||||
isCluster := instance.ClusterMode != nil && *instance.ClusterMode
|
||||
|
||||
metrics := map[string]interface{}{
|
||||
"connections": 0,
|
||||
"qps": 0,
|
||||
"latency": 0,
|
||||
"uptime": "-",
|
||||
"version": "",
|
||||
"cluster_mode": isCluster,
|
||||
}
|
||||
|
||||
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()
|
||||
|
||||
_, 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()
|
||||
}
|
||||
|
||||
if info == "" {
|
||||
return metrics
|
||||
}
|
||||
|
||||
infoMap := make(map[string]string)
|
||||
for _, line := range strings.Split(info, "\n") {
|
||||
line = strings.TrimSpace(strings.TrimRight(line, "\r"))
|
||||
if line == "" || strings.HasPrefix(line, "#") {
|
||||
continue
|
||||
}
|
||||
parts := strings.SplitN(line, ":", 2)
|
||||
if len(parts) == 2 {
|
||||
infoMap[strings.TrimSpace(parts[0])] = strings.TrimSpace(parts[1])
|
||||
}
|
||||
}
|
||||
|
||||
if v, ok := infoMap["redis_version"]; ok {
|
||||
metrics["version"] = v
|
||||
}
|
||||
if v, ok := infoMap["connected_clients"]; ok {
|
||||
metrics["connections"] = v
|
||||
}
|
||||
if v, ok := infoMap["instantaneous_ops_per_sec"]; ok {
|
||||
metrics["qps"] = v
|
||||
}
|
||||
if v, ok := infoMap["uptime_in_days"]; ok {
|
||||
metrics["uptime"] = v + " 天"
|
||||
} else if v, ok := infoMap["uptime_in_seconds"]; ok {
|
||||
secs := 0
|
||||
fmt.Sscanf(v, "%d", &secs)
|
||||
if secs > 86400 {
|
||||
metrics["uptime"] = fmt.Sprintf("%d 天", secs/86400)
|
||||
} else if secs > 3600 {
|
||||
metrics["uptime"] = fmt.Sprintf("%d 小时", secs/3600)
|
||||
} else {
|
||||
metrics["uptime"] = fmt.Sprintf("%d 分钟", secs/60)
|
||||
}
|
||||
}
|
||||
|
||||
return metrics
|
||||
}
|
||||
|
||||
// getDB opens a real database connection and returns *sql.DB
|
||||
func getDBConnection(instance *model.Instance, dbName string) (*sql.DB, error) {
|
||||
@@ -127,7 +296,7 @@ func getDBConnection(instance *model.Instance, dbName string) (*sql.DB, error) {
|
||||
if dbName == "" {
|
||||
dbName = "postgres"
|
||||
}
|
||||
dsn := fmt.Sprintf("postgres://%s:%s@%s:%d/%s?sslmode=disable&connect_timeout=10",
|
||||
dsn := fmt.Sprintf("postgres://%s:***@%s:%d/%s?sslmode=disable&connect_timeout=10",
|
||||
instance.Username, password, instance.Host, instance.Port, dbName)
|
||||
return sql.Open("pgx", dsn)
|
||||
default:
|
||||
@@ -154,7 +323,6 @@ func listDatabases(instance *model.Instance) ([]string, error) {
|
||||
for rows.Next() {
|
||||
var name string
|
||||
rows.Scan(&name)
|
||||
// Filter out system databases
|
||||
if name != "information_schema" && name != "performance_schema" &&
|
||||
name != "mysql" && name != "sys" {
|
||||
databases = append(databases, name)
|
||||
|
||||
@@ -68,6 +68,20 @@ func CreateInstance(c *gin.Context) {
|
||||
c.JSON(http.StatusInternalServerError, gin.H{"detail": err.Error()})
|
||||
return
|
||||
}
|
||||
// Auto-detect version and cluster_mode after creation
|
||||
go func() {
|
||||
result := testDBConnectionFull(&instance)
|
||||
updates := map[string]interface{}{}
|
||||
if result.Success && result.Version != "" {
|
||||
updates["version"] = result.Version
|
||||
}
|
||||
if result.ClusterMode {
|
||||
updates["cluster_mode"] = true
|
||||
}
|
||||
if len(updates) > 0 {
|
||||
model.DB.Model(&model.Instance{}).Where("id = ?", instance.ID).Updates(updates)
|
||||
}
|
||||
}()
|
||||
instance.Password = "******"
|
||||
c.JSON(http.StatusCreated, instance)
|
||||
}
|
||||
@@ -121,11 +135,22 @@ func TestConnection(c *gin.Context) {
|
||||
return
|
||||
}
|
||||
|
||||
success, message := testDBConnection(&instance)
|
||||
if success {
|
||||
c.JSON(http.StatusOK, gin.H{"success": true, "message": message})
|
||||
result := testDBConnectionFull(&instance)
|
||||
// Update version & cluster_mode in DB if detected
|
||||
updates := map[string]interface{}{}
|
||||
if result.Success && result.Version != "" {
|
||||
updates["version"] = result.Version
|
||||
}
|
||||
if result.ClusterMode {
|
||||
updates["cluster_mode"] = true
|
||||
}
|
||||
if len(updates) > 0 {
|
||||
model.DB.Model(&model.Instance{}).Where("id = ?", instance.ID).Updates(updates)
|
||||
}
|
||||
if result.Success {
|
||||
c.JSON(http.StatusOK, gin.H{"success": true, "message": result.Message, "version": result.Version, "cluster_mode": result.ClusterMode})
|
||||
} else {
|
||||
c.JSON(http.StatusBadRequest, gin.H{"success": false, "message": message})
|
||||
c.JSON(http.StatusBadRequest, gin.H{"success": false, "message": result.Message})
|
||||
}
|
||||
}
|
||||
|
||||
@@ -152,3 +177,25 @@ func parseID(s string) (int, bool) {
|
||||
id, err := strconv.Atoi(s)
|
||||
return id, err == nil
|
||||
}
|
||||
|
||||
// GET /api/v1/instances/:id/metrics — live metrics for an instance
|
||||
func InstanceMetrics(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
|
||||
}
|
||||
|
||||
switch instance.DBType {
|
||||
case "redis":
|
||||
metrics := getRedisMetrics(&instance)
|
||||
c.JSON(http.StatusOK, metrics)
|
||||
default:
|
||||
// For MySQL/PG: return basic status
|
||||
c.JSON(http.StatusOK, gin.H{
|
||||
"status": "online",
|
||||
"version": instance.Version,
|
||||
})
|
||||
}
|
||||
}
|
||||
@@ -10,7 +10,6 @@ import (
|
||||
"github.com/gin-gonic/gin"
|
||||
"time"
|
||||
"context"
|
||||
"github.com/redis/go-redis/v9"
|
||||
"sql-platform/internal/middleware"
|
||||
"strings"
|
||||
"strconv"
|
||||
@@ -665,14 +664,6 @@ func executeSQLQuery(instance *model.Instance, dbName string, sqlStr string, lim
|
||||
}
|
||||
|
||||
func executeRedisQuery(instance *model.Instance, dbSelect string, command string, limit int, offset int) (gin.H, error) {
|
||||
password := instance.Password
|
||||
if strings.HasPrefix(password, "enc:") {
|
||||
decrypted, err := decryptPassword(strings.TrimPrefix(password, "enc:"))
|
||||
if err == nil {
|
||||
password = decrypted
|
||||
}
|
||||
}
|
||||
|
||||
// Parse Redis DB number from dbSelect (e.g. "0", "1")
|
||||
redisDB := 0
|
||||
if dbSelect != "" {
|
||||
@@ -681,17 +672,13 @@ func executeRedisQuery(instance *model.Instance, dbSelect string, command string
|
||||
}
|
||||
}
|
||||
|
||||
client := redis.NewClient(&redis.Options{
|
||||
Addr: fmt.Sprintf("%s:%d", instance.Host, instance.Port),
|
||||
Password: password,
|
||||
DB: redisDB,
|
||||
})
|
||||
client, err := getRedisClient(instance, redisDB)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer client.Close()
|
||||
|
||||
ctx := context.Background()
|
||||
if err := client.Ping(ctx).Err(); err != nil {
|
||||
return nil, fmt.Errorf("Redis connection failed: %v", err)
|
||||
}
|
||||
|
||||
parts := strings.Fields(command)
|
||||
if len(parts) == 0 {
|
||||
|
||||
@@ -20,9 +20,9 @@ func AccessibleInstances(c *gin.Context) {
|
||||
// admin/dba see all instances, filter by access_mode if needed
|
||||
query := model.DB.Order("id")
|
||||
if accessType == "readonly" {
|
||||
query = query.Where("access_mode IN ?", []string{"readwrite", "readonly"})
|
||||
query = query.Where("access_mode IN ? OR access_mode IS NULL", []string{"readwrite", "readonly"})
|
||||
} else {
|
||||
query = query.Where("access_mode IN ?", []string{"readwrite", "write"})
|
||||
query = query.Where("access_mode IN ? OR access_mode IS NULL", []string{"readwrite", "write"})
|
||||
}
|
||||
query.Find(&instances)
|
||||
} else {
|
||||
@@ -117,10 +117,11 @@ func TestNewConnection(c *gin.Context) {
|
||||
instance.Database = &req.Database
|
||||
}
|
||||
|
||||
success, message := testDBConnection(instance)
|
||||
if success {
|
||||
c.JSON(http.StatusOK, gin.H{"success": true, "message": message})
|
||||
} else {
|
||||
c.JSON(http.StatusBadRequest, gin.H{"success": false, "message": message})
|
||||
}
|
||||
result := testDBConnectionFull(instance)
|
||||
c.JSON(http.StatusOK, gin.H{
|
||||
"success": result.Success,
|
||||
"message": result.Message,
|
||||
"version": result.Version,
|
||||
"cluster_mode": result.ClusterMode,
|
||||
})
|
||||
}
|
||||
@@ -13,8 +13,186 @@ import (
|
||||
"github.com/gin-gonic/gin"
|
||||
"github.com/redis/go-redis/v9"
|
||||
)
|
||||
// Helper: get Redis client from instance
|
||||
func getRedisClient(instance *model.Instance, db int) (*redis.Client, error) {
|
||||
// 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:") {
|
||||
@@ -23,18 +201,36 @@ func getRedisClient(instance *model.Instance, db int) (*redis.Client, error) {
|
||||
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: fmt.Sprintf("%s:%d", instance.Host, instance.Port),
|
||||
Addr: addr,
|
||||
Password: password,
|
||||
DB: db,
|
||||
})
|
||||
// Test connection
|
||||
ctx := context.Background()
|
||||
if err := client.Ping(ctx).Err(); err != nil {
|
||||
client.Close()
|
||||
return nil, fmt.Errorf("Redis connection failed: %v", err)
|
||||
}
|
||||
return client, nil
|
||||
return &redisClientWrapper{Client: client, isCluster: false}, nil
|
||||
}
|
||||
|
||||
// GET /api/v1/redis/:id/info
|
||||
|
||||
@@ -42,6 +42,7 @@ type Instance struct {
|
||||
Description *string `gorm:"type:text" json:"description"`
|
||||
Status *string `gorm:"size:20" json:"status"`
|
||||
Version *string `gorm:"size:50" json:"version"`
|
||||
ClusterMode *bool `gorm:"default:false" json:"cluster_mode"`
|
||||
CreatedAt time.Time `gorm:"default:CURRENT_TIMESTAMP" json:"created_at"`
|
||||
UpdatedAt time.Time `gorm:"default:CURRENT_TIMESTAMP" json:"updated_at"`
|
||||
AccessMode *string `gorm:"size:20;default:readwrite" json:"access_mode"`
|
||||
|
||||
@@ -47,10 +47,11 @@ func Setup(r *gin.Engine) {
|
||||
auth.GET("/instances/:id", handler.GetInstance)
|
||||
auth.GET("/instances/:id/databases", handler.ListDatabases)
|
||||
auth.GET("/instances/:id/databases/:db/tables", handler.ListTables)
|
||||
auth.POST("/instances/:id/test-connection", handler.TestConnection)
|
||||
auth.POST("/instances/test-connection", handler.TestNewConnection)
|
||||
auth.POST("/instances/:id/test-connection", handler.TestConnection)
|
||||
auth.POST("/instances/test-connection", handler.TestNewConnection)
|
||||
auth.GET("/instances/:id/metrics", handler.InstanceMetrics)
|
||||
|
||||
// Instance Detail (databases/tables/schema/processlist/status/users)
|
||||
// Instance Detail (databases/tables/schema/processlist/status/users)
|
||||
auth.GET("/instances/:id/detail/databases", handler.DetailListDatabases)
|
||||
auth.GET("/instances/:id/detail/tables", handler.DetailListTables)
|
||||
auth.GET("/instances/:id/detail/db-schema", handler.DetailDbSchema)
|
||||
|
||||
Reference in new issue
Block a user