tiny-rdm/backend/services/connection_service.go

1454 lines
35 KiB
Go
Raw Normal View History

2023-06-27 15:53:29 +08:00
package services
import (
"context"
2023-10-15 02:27:23 +08:00
"crypto/tls"
"crypto/x509"
2023-06-27 15:53:29 +08:00
"errors"
"fmt"
"github.com/redis/go-redis/v9"
"golang.org/x/crypto/ssh"
2023-10-14 21:26:47 +08:00
"log"
"net"
"os"
2023-06-27 15:53:29 +08:00
"strconv"
"strings"
"sync"
2023-10-14 21:26:47 +08:00
"sync/atomic"
2023-06-27 15:53:29 +08:00
"time"
. "tinyrdm/backend/storage"
"tinyrdm/backend/types"
"tinyrdm/backend/utils/coll"
2023-06-27 15:53:29 +08:00
maputil "tinyrdm/backend/utils/map"
redis2 "tinyrdm/backend/utils/redis"
sliceutil "tinyrdm/backend/utils/slice"
strutil "tinyrdm/backend/utils/string"
2023-06-27 15:53:29 +08:00
)
2023-07-16 01:50:01 +08:00
type cmdHistoryItem struct {
Timestamp int64 `json:"timestamp"`
2023-07-16 01:50:01 +08:00
Server string `json:"server"`
Cmd string `json:"cmd"`
Cost int64 `json:"cost"`
2023-07-16 01:50:01 +08:00
}
2023-06-27 15:53:29 +08:00
type connectionService struct {
2023-07-16 01:50:01 +08:00
ctx context.Context
conns *ConnectionsStorage
connMap map[string]connectionItem
cmdHistory []cmdHistoryItem
2023-06-27 15:53:29 +08:00
}
type connectionItem struct {
2023-10-14 21:26:47 +08:00
client redis.UniversalClient
2023-06-27 15:53:29 +08:00
ctx context.Context
cancelFunc context.CancelFunc
}
type keyItem struct {
Type string `json:"t"`
}
var connection *connectionService
var onceConnection sync.Once
func Connection() *connectionService {
if connection == nil {
onceConnection.Do(func() {
connection = &connectionService{
conns: NewConnections(),
connMap: map[string]connectionItem{},
}
})
}
return connection
}
func (c *connectionService) Start(ctx context.Context) {
c.ctx = ctx
}
func (c *connectionService) Stop(ctx context.Context) {
for _, item := range c.connMap {
2023-10-14 21:26:47 +08:00
if item.client != nil {
2023-06-27 15:53:29 +08:00
item.cancelFunc()
2023-10-14 21:26:47 +08:00
item.client.Close()
2023-06-27 15:53:29 +08:00
}
}
c.connMap = map[string]connectionItem{}
}
func (c *connectionService) buildOption(config types.ConnectionConfig) (*redis.Options, error) {
var sshClient *ssh.Client
if config.SSH.Enable {
sshConfig := &ssh.ClientConfig{
User: config.SSH.Username,
Auth: []ssh.AuthMethod{ssh.Password(config.SSH.Password)},
HostKeyCallback: ssh.InsecureIgnoreHostKey(),
Timeout: time.Duration(config.ConnTimeout) * time.Second,
}
switch config.SSH.LoginType {
case "pwd":
sshConfig.Auth = []ssh.AuthMethod{ssh.Password(config.SSH.Password)}
case "pkfile":
key, err := os.ReadFile(config.SSH.PKFile)
if err != nil {
return nil, err
}
var signer ssh.Signer
if len(config.SSH.Passphrase) > 0 {
signer, err = ssh.ParsePrivateKeyWithPassphrase(key, []byte(config.SSH.Passphrase))
} else {
signer, err = ssh.ParsePrivateKey(key)
}
if err != nil {
return nil, err
}
sshConfig.Auth = []ssh.AuthMethod{ssh.PublicKeys(signer)}
default:
return nil, errors.New("invalid login type")
}
var err error
sshClient, err = ssh.Dial("tcp", fmt.Sprintf("%s:%d", config.SSH.Addr, config.SSH.Port), sshConfig)
if err != nil {
return nil, err
}
}
2023-10-15 02:27:23 +08:00
var tlsConfig *tls.Config
if config.SSL.Enable {
// setup tls config
var certs []tls.Certificate
if len(config.SSL.CertFile) > 0 && len(config.SSL.KeyFile) > 0 {
if cert, err := tls.LoadX509KeyPair(config.SSL.CertFile, config.SSL.KeyFile); err != nil {
return nil, err
} else {
certs = []tls.Certificate{cert}
}
}
var caCertPool *x509.CertPool
if len(config.SSL.CAFile) > 0 {
ca, err := os.ReadFile(config.SSL.CAFile)
if err != nil {
return nil, err
}
caCertPool = x509.NewCertPool()
caCertPool.AppendCertsFromPEM(ca)
}
if len(certs) <= 0 {
return nil, errors.New("tls config error")
}
tlsConfig = &tls.Config{
RootCAs: caCertPool,
InsecureSkipVerify: false,
Certificates: certs,
}
}
option := &redis.Options{
2023-10-14 21:26:47 +08:00
ClientName: config.Name,
Addr: fmt.Sprintf("%s:%d", config.Addr, config.Port),
Username: config.Username,
Password: config.Password,
DialTimeout: time.Duration(config.ConnTimeout) * time.Second,
ReadTimeout: time.Duration(config.ExecTimeout) * time.Second,
WriteTimeout: time.Duration(config.ExecTimeout) * time.Second,
2023-10-15 02:27:23 +08:00
TLSConfig: tlsConfig,
}
if sshClient != nil {
option.Dialer = func(ctx context.Context, network, addr string) (net.Conn, error) {
return sshClient.Dial(network, addr)
}
option.ReadTimeout = -2
option.WriteTimeout = -2
}
return option, nil
}
2023-10-14 21:26:47 +08:00
func (c *connectionService) createRedisClient(config types.ConnectionConfig) (redis.UniversalClient, error) {
option, err := c.buildOption(config)
if err != nil {
return nil, err
}
2023-10-08 01:31:54 +08:00
if config.Sentinel.Enable {
2023-10-14 21:26:47 +08:00
// get master address via sentinel node
2023-10-08 01:31:54 +08:00
sentinel := redis.NewSentinelClient(option)
defer sentinel.Close()
var addr []string
addr, err = sentinel.GetMasterAddrByName(c.ctx, config.Sentinel.Master).Result()
2023-10-08 01:31:54 +08:00
if err != nil {
return nil, err
}
if len(addr) < 2 {
return nil, errors.New("cannot get master address")
}
option.Addr = fmt.Sprintf("%s:%s", addr[0], addr[1])
option.Username = config.Sentinel.Username
option.Password = config.Sentinel.Password
}
rdb := redis.NewClient(option)
2023-10-14 21:26:47 +08:00
if config.Cluster.Enable {
// connect to cluster
var slots []redis.ClusterSlot
if slots, err = rdb.ClusterSlots(c.ctx).Result(); err == nil {
log.Println(slots)
clusterOptions := &redis.ClusterOptions{
//NewClient: nil,
//MaxRedirects: 0,
//RouteByLatency: false,
//RouteRandomly: false,
//ClusterSlots: nil,
ClientName: option.ClientName,
Dialer: option.Dialer,
OnConnect: option.OnConnect,
Protocol: option.Protocol,
Username: option.Username,
Password: option.Password,
MaxRetries: option.MaxRetries,
MinRetryBackoff: option.MinRetryBackoff,
MaxRetryBackoff: option.MaxRetryBackoff,
DialTimeout: option.DialTimeout,
ContextTimeoutEnabled: option.ContextTimeoutEnabled,
PoolFIFO: option.PoolFIFO,
PoolSize: option.PoolSize,
PoolTimeout: option.PoolTimeout,
MinIdleConns: option.MinIdleConns,
MaxIdleConns: option.MaxIdleConns,
ConnMaxIdleTime: option.ConnMaxIdleTime,
ConnMaxLifetime: option.ConnMaxLifetime,
TLSConfig: option.TLSConfig,
DisableIndentity: option.DisableIndentity,
}
if option.Dialer != nil {
clusterOptions.Dialer = option.Dialer
clusterOptions.ReadTimeout = -2
clusterOptions.WriteTimeout = -2
}
var addrs []string
for _, slot := range slots {
for _, node := range slot.Nodes {
addrs = append(addrs, node.Addr)
}
}
clusterOptions.Addrs = addrs
clusterClient := redis.NewClusterClient(clusterOptions)
return clusterClient, nil
} else {
return nil, err
}
}
return rdb, nil
}
// ListSentinelMasters list all master info by sentinel
func (c *connectionService) ListSentinelMasters(config types.ConnectionConfig) (resp types.JSResp) {
option, err := c.buildOption(config)
if err != nil {
resp.Msg = err.Error()
return
}
if option.DialTimeout > 0 {
option.DialTimeout = 10 * time.Second
}
sentinel := redis.NewSentinelClient(option)
defer sentinel.Close()
var retInfo []map[string]string
masterInfos, err := sentinel.Masters(c.ctx).Result()
if err != nil {
resp.Msg = err.Error()
return
}
for _, info := range masterInfos {
if infoMap, ok := info.(map[any]any); ok {
retInfo = append(retInfo, map[string]string{
"name": infoMap["name"].(string),
"addr": fmt.Sprintf("%s:%s", infoMap["ip"].(string), infoMap["port"].(string)),
})
}
}
resp.Data = retInfo
resp.Success = true
return
}
func (c *connectionService) TestConnection(config types.ConnectionConfig) (resp types.JSResp) {
2023-10-14 21:26:47 +08:00
client, err := c.createRedisClient(config)
if err != nil {
resp.Msg = err.Error()
return
}
2023-10-14 21:26:47 +08:00
defer client.Close()
2023-10-14 21:26:47 +08:00
if _, err = client.Ping(c.ctx).Result(); err != nil && err != redis.Nil {
2023-06-27 15:53:29 +08:00
resp.Msg = err.Error()
} else {
resp.Success = true
}
return
}
// ListConnection list all saved connection in local profile
func (c *connectionService) ListConnection() (resp types.JSResp) {
resp.Success = true
resp.Data = c.conns.GetConnections()
return
}
// GetConnection get connection profile by name
func (c *connectionService) GetConnection(name string) (resp types.JSResp) {
conn := c.conns.GetConnection(name)
resp.Success = conn != nil
resp.Data = conn
return
}
2023-06-27 15:53:29 +08:00
// SaveConnection save connection config to local profile
func (c *connectionService) SaveConnection(name string, param types.ConnectionConfig) (resp types.JSResp) {
var err error
if strings.ContainsAny(param.Name, "/") {
err = errors.New("connection name contains illegal characters")
} else {
if len(name) > 0 {
// update connection
err = c.conns.UpdateConnection(name, param)
} else {
err = c.conns.CreateConnection(param)
}
}
if err != nil {
2023-06-27 15:53:29 +08:00
resp.Msg = err.Error()
} else {
resp.Success = true
}
return
}
// DeleteConnection remove connection by name
func (c *connectionService) DeleteConnection(name string) (resp types.JSResp) {
err := c.conns.DeleteConnection(name)
if err != nil {
resp.Msg = err.Error()
return
}
resp.Success = true
return
}
// SaveSortedConnection save sorted connection after drag
func (c *connectionService) SaveSortedConnection(sortedConns types.Connections) (resp types.JSResp) {
err := c.conns.SaveSortedConnection(sortedConns)
if err != nil {
resp.Msg = err.Error()
return
}
resp.Success = true
return
}
// CreateGroup create a new group
func (c *connectionService) CreateGroup(name string) (resp types.JSResp) {
err := c.conns.CreateGroup(name)
if err != nil {
resp.Msg = err.Error()
return
}
resp.Success = true
return
}
// RenameGroup rename group
func (c *connectionService) RenameGroup(name, newName string) (resp types.JSResp) {
err := c.conns.RenameGroup(name, newName)
if err != nil {
resp.Msg = err.Error()
return
}
resp.Success = true
return
}
// DeleteGroup remove a group by name
func (c *connectionService) DeleteGroup(name string, includeConn bool) (resp types.JSResp) {
err := c.conns.DeleteGroup(name, includeConn)
if err != nil {
resp.Msg = err.Error()
return
}
resp.Success = true
return
}
2023-06-27 15:53:29 +08:00
// OpenConnection open redis server connection
func (c *connectionService) OpenConnection(name string) (resp types.JSResp) {
2023-10-14 21:26:47 +08:00
client, ctx, err := c.getRedisClient(name, 0)
2023-06-27 15:53:29 +08:00
if err != nil {
resp.Msg = err.Error()
return
}
// get connection config
selConn := c.conns.GetConnection(name)
var totaldb int
if selConn.DBFilterType == "" || selConn.DBFilterType == "none" {
// get total databases
2023-10-14 21:26:47 +08:00
if config, err := client.ConfigGet(ctx, "databases").Result(); err == nil {
if total, err := strconv.Atoi(config["databases"]); err == nil {
totaldb = total
}
}
2023-06-27 15:53:29 +08:00
}
// parse all db, response content like below
2023-06-27 15:53:29 +08:00
var dbs []types.ConnectionDB
2023-10-14 21:26:47 +08:00
var clusterKeyCount int64
cluster, isCluster := client.(*redis.ClusterClient)
if isCluster {
var keyCount atomic.Int64
err = cluster.ForEachMaster(ctx, func(ctx context.Context, cli *redis.Client) error {
if size, serr := cli.DBSize(ctx).Result(); serr != nil {
return serr
} else {
keyCount.Add(size)
}
2023-10-14 21:26:47 +08:00
return nil
})
if err != nil {
resp.Msg = "get db size error:" + err.Error()
return
}
2023-10-14 21:26:47 +08:00
clusterKeyCount = keyCount.Load()
// only one database in cluster mode
dbs = []types.ConnectionDB{
{
Name: "db0",
Index: 0,
Keys: int(clusterKeyCount),
},
}
} else {
// get database info
var res string
res, err = client.Info(ctx, "keyspace").Result()
if err != nil {
resp.Msg = "get server info fail:" + err.Error()
return
}
info := c.parseInfo(res)
if totaldb <= 0 {
// cannot retrieve the database count by "CONFIG GET databases", try to get max index from keyspace
keyspace := info["Keyspace"]
var db, maxDB int
for dbName := range keyspace {
if db, err = strconv.Atoi(strings.TrimLeft(dbName, "db")); err == nil {
if maxDB < db {
maxDB = db
}
}
}
2023-10-14 21:26:47 +08:00
totaldb = maxDB + 1
}
2023-10-14 21:26:47 +08:00
queryDB := func(idx int) types.ConnectionDB {
dbName := "db" + strconv.Itoa(idx)
dbInfoStr := info["Keyspace"][dbName]
if len(dbInfoStr) > 0 {
dbInfo := c.parseDBItemInfo(dbInfoStr)
return types.ConnectionDB{
Name: dbName,
Index: idx,
Keys: dbInfo["keys"],
Expires: dbInfo["expires"],
AvgTTL: dbInfo["avg_ttl"],
}
} else {
return types.ConnectionDB{
Name: dbName,
Index: idx,
}
}
}
2023-10-14 21:26:47 +08:00
switch selConn.DBFilterType {
case "show":
filterList := sliceutil.Unique(selConn.DBFilterList)
for _, idx := range filterList {
dbs = append(dbs, queryDB(idx))
}
case "hide":
hiddenList := coll.NewSet(selConn.DBFilterList...)
for idx := 0; idx < totaldb; idx++ {
if !hiddenList.Contains(idx) {
dbs = append(dbs, queryDB(idx))
}
}
default:
for idx := 0; idx < totaldb; idx++ {
dbs = append(dbs, queryDB(idx))
}
}
2023-06-27 15:53:29 +08:00
}
resp.Success = true
resp.Data = map[string]any{
"db": dbs,
}
return
}
// CloseConnection close redis server connection
func (c *connectionService) CloseConnection(name string) (resp types.JSResp) {
item, ok := c.connMap[name]
if ok {
delete(c.connMap, name)
2023-10-14 21:26:47 +08:00
if item.client != nil {
2023-06-27 15:53:29 +08:00
item.cancelFunc()
2023-10-14 21:26:47 +08:00
item.client.Close()
2023-06-27 15:53:29 +08:00
}
}
resp.Success = true
return
}
// get redis client from local cache or create a new open
2023-07-16 01:50:01 +08:00
// if db >= 0, will also switch to db index
2023-10-14 21:26:47 +08:00
func (c *connectionService) getRedisClient(connName string, db int) (redis.UniversalClient, context.Context, error) {
2023-06-27 15:53:29 +08:00
item, ok := c.connMap[connName]
2023-10-14 21:26:47 +08:00
var client redis.UniversalClient
2023-06-27 15:53:29 +08:00
var ctx context.Context
if ok {
2023-10-14 21:26:47 +08:00
client, ctx = item.client, item.ctx
2023-06-27 15:53:29 +08:00
} else {
selConn := c.conns.GetConnection(connName)
2023-06-27 15:53:29 +08:00
if selConn == nil {
return nil, nil, fmt.Errorf("no match connection \"%s\"", connName)
2023-06-27 15:53:29 +08:00
}
2023-10-14 21:26:47 +08:00
hook := redis2.NewHook(connName, func(cmd string, cost int64) {
2023-07-16 01:50:01 +08:00
now := time.Now()
//last := strings.LastIndex(cmd, ":")
//if last != -1 {
// cmd = cmd[:last]
//}
2023-07-16 01:50:01 +08:00
c.cmdHistory = append(c.cmdHistory, cmdHistoryItem{
Timestamp: now.UnixMilli(),
2023-07-16 01:50:01 +08:00
Server: connName,
Cmd: cmd,
Cost: cost,
2023-07-16 01:50:01 +08:00
})
2023-10-14 21:26:47 +08:00
})
var err error
client, err = c.createRedisClient(selConn.ConnectionConfig)
if err != nil {
return nil, nil, fmt.Errorf("create conenction error: %s", err.Error())
}
// add hook to each node in cluster mode
var cluster *redis.ClusterClient
if cluster, ok = client.(*redis.ClusterClient); ok {
err = cluster.ForEachShard(c.ctx, func(ctx context.Context, cli *redis.Client) error {
cli.AddHook(hook)
return nil
})
if err != nil {
return nil, nil, fmt.Errorf("get cluster nodes error: %s", err.Error())
}
} else {
client.AddHook(hook)
}
2023-06-27 15:53:29 +08:00
2023-10-14 21:26:47 +08:00
if _, err = client.Ping(c.ctx).Result(); err != nil && err != redis.Nil {
2023-06-27 15:53:29 +08:00
return nil, nil, errors.New("can not connect to redis server:" + err.Error())
}
var cancelFunc context.CancelFunc
ctx, cancelFunc = context.WithCancel(c.ctx)
c.connMap[connName] = connectionItem{
2023-10-14 21:26:47 +08:00
client: client,
2023-06-27 15:53:29 +08:00
ctx: ctx,
cancelFunc: cancelFunc,
}
}
if db >= 0 {
2023-10-14 21:26:47 +08:00
var rdb *redis.Client
if rdb, ok = client.(*redis.Client); ok && rdb != nil {
if err := rdb.Do(ctx, "select", strconv.Itoa(db)).Err(); err != nil {
return nil, nil, err
}
2023-06-27 15:53:29 +08:00
}
}
2023-10-14 21:26:47 +08:00
return client, ctx, nil
2023-06-27 15:53:29 +08:00
}
// parse command response content which use "redis info"
// # Keyspace\r\ndb0:keys=2,expires=1,avg_ttl=1877111749\r\ndb1:keys=33,expires=0,avg_ttl=0\r\ndb3:keys=17,expires=0,avg_ttl=0\r\ndb5:keys=3,expires=0,avg_ttl=0\r\n
2023-07-25 11:51:23 +08:00
func (c *connectionService) parseInfo(info string) map[string]map[string]string {
parsedInfo := map[string]map[string]string{}
2023-06-27 15:53:29 +08:00
lines := strings.Split(info, "\r\n")
if len(lines) > 0 {
2023-07-25 11:51:23 +08:00
var subInfo map[string]string
2023-06-27 15:53:29 +08:00
for _, line := range lines {
2023-07-25 11:51:23 +08:00
if strings.HasPrefix(line, "#") {
subInfo = map[string]string{}
parsedInfo[strings.TrimSpace(strings.TrimLeft(line, "#"))] = subInfo
} else {
2023-06-27 15:53:29 +08:00
items := strings.SplitN(line, ":", 2)
if len(items) < 2 {
continue
}
2023-07-25 11:51:23 +08:00
subInfo[items[0]] = items[1]
2023-06-27 15:53:29 +08:00
}
}
}
return parsedInfo
}
// parse db item value, content format like below
// keys=2,expires=1,avg_ttl=1877111749
func (c *connectionService) parseDBItemInfo(info string) map[string]int {
ret := map[string]int{}
items := strings.Split(info, ",")
for _, item := range items {
kv := strings.SplitN(item, "=", 2)
if len(kv) > 1 {
ret[kv[0]], _ = strconv.Atoi(kv[1])
}
}
return ret
}
2023-07-14 02:03:45 +08:00
// ServerInfo get server info
func (c *connectionService) ServerInfo(name string) (resp types.JSResp) {
2023-10-14 21:26:47 +08:00
client, ctx, err := c.getRedisClient(name, 0)
2023-07-14 02:03:45 +08:00
if err != nil {
resp.Msg = err.Error()
return
}
// get database info
2023-10-14 21:26:47 +08:00
res, err := client.Info(ctx).Result()
2023-07-14 02:03:45 +08:00
if err != nil {
resp.Msg = "get server info fail:" + err.Error()
return
}
resp.Success = true
resp.Data = c.parseInfo(res)
return
}
2023-06-27 15:53:29 +08:00
// OpenDatabase open select database, and list all keys
// @param path contain connection name and db name
func (c *connectionService) OpenDatabase(connName string, db int, match string, keyType string) (resp types.JSResp) {
return c.ScanKeys(connName, db, match, keyType)
}
2023-07-17 17:17:52 +08:00
// ScanKeys scan all keys
func (c *connectionService) ScanKeys(connName string, db int, match, keyType string) (resp types.JSResp) {
2023-10-14 21:26:47 +08:00
client, ctx, err := c.getRedisClient(connName, db)
2023-06-27 15:53:29 +08:00
if err != nil {
resp.Msg = err.Error()
return
}
filterType := len(keyType) > 0
2023-10-14 21:26:47 +08:00
var countPerScan int64 = 10000
// define sub scan function
scan := func(ctx context.Context, cli redis.UniversalClient, appendFunc func(k any)) error {
var iter *redis.ScanIterator
if filterType {
2023-10-14 21:26:47 +08:00
iter = cli.ScanType(ctx, 0, match, countPerScan, keyType).Iterator()
} else {
2023-10-14 21:26:47 +08:00
iter = cli.Scan(ctx, 0, match, countPerScan).Iterator()
}
2023-10-14 21:26:47 +08:00
for iter.Next(ctx) {
appendFunc(strutil.EncodeRedisKey(iter.Val()))
2023-06-27 15:53:29 +08:00
}
2023-10-14 21:26:47 +08:00
return iter.Err()
}
var keys []any
if cluster, ok := client.(*redis.ClusterClient); ok {
// cluster mode
var mutex sync.Mutex
err = cluster.ForEachMaster(ctx, func(ctx context.Context, cli *redis.Client) error {
return scan(ctx, cli, func(k any) {
mutex.Lock()
keys = append(keys, k)
mutex.Unlock()
})
})
} else {
err = scan(ctx, client, func(k any) {
keys = append(keys, k)
})
}
if err != nil {
resp.Msg = err.Error()
return
2023-06-27 15:53:29 +08:00
}
resp.Success = true
resp.Data = map[string]any{
"keys": keys,
}
return
}
// GetKeyValue get value by key
func (c *connectionService) GetKeyValue(connName string, db int, k any, viewAs string) (resp types.JSResp) {
2023-10-14 21:26:47 +08:00
client, ctx, err := c.getRedisClient(connName, db)
2023-06-27 15:53:29 +08:00
if err != nil {
resp.Msg = err.Error()
return
}
key := strutil.DecodeRedisKey(k)
2023-06-27 15:53:29 +08:00
var keyType string
var dur time.Duration
2023-10-14 21:26:47 +08:00
keyType, err = client.Type(ctx, key).Result()
2023-06-27 15:53:29 +08:00
if err != nil {
resp.Msg = err.Error()
return
}
2023-07-10 18:06:14 +08:00
if keyType == "none" {
resp.Msg = "key not exists"
return
}
2023-06-27 15:53:29 +08:00
var ttl int64
2023-10-14 21:26:47 +08:00
if dur, err = client.TTL(ctx, key).Result(); err != nil {
2023-06-27 15:53:29 +08:00
ttl = -1
} else {
if dur < 0 {
ttl = -1
} else {
ttl = int64(dur.Seconds())
}
}
var value any
var size int64
2023-06-27 15:53:29 +08:00
var cursor uint64
switch strings.ToLower(keyType) {
case "string":
var str string
2023-10-14 21:26:47 +08:00
str, err = client.Get(ctx, key).Result()
value, viewAs = strutil.ConvertTo(str, viewAs)
2023-10-14 21:26:47 +08:00
size, _ = client.StrLen(ctx, key).Result()
2023-06-27 15:53:29 +08:00
case "list":
2023-10-14 21:26:47 +08:00
value, err = client.LRange(ctx, key, 0, -1).Result()
size, _ = client.LLen(ctx, key).Result()
2023-06-27 15:53:29 +08:00
case "hash":
2023-10-14 21:26:47 +08:00
//value, err = client.HGetAll(ctx, key).Result()
2023-06-27 15:53:29 +08:00
items := map[string]string{}
for {
var loadedVal []string
2023-10-14 21:26:47 +08:00
loadedVal, cursor, err = client.HScan(ctx, key, cursor, "*", 10000).Result()
2023-06-27 15:53:29 +08:00
if err != nil {
resp.Msg = err.Error()
return
}
for i := 0; i < len(loadedVal); i += 2 {
items[loadedVal[i]] = loadedVal[i+1]
}
if cursor == 0 {
break
}
}
value = items
2023-10-14 21:26:47 +08:00
size, _ = client.HLen(ctx, key).Result()
2023-06-27 15:53:29 +08:00
case "set":
2023-10-14 21:26:47 +08:00
//value, err = client.SMembers(ctx, key).Result()
2023-06-27 15:53:29 +08:00
items := []string{}
for {
var loadedKey []string
2023-10-14 21:26:47 +08:00
loadedKey, cursor, err = client.SScan(ctx, key, cursor, "*", 10000).Result()
2023-06-27 15:53:29 +08:00
if err != nil {
resp.Msg = err.Error()
return
}
items = append(items, loadedKey...)
if cursor == 0 {
break
}
}
value = items
2023-10-14 21:26:47 +08:00
size, _ = client.SCard(ctx, key).Result()
2023-06-27 15:53:29 +08:00
case "zset":
2023-10-14 21:26:47 +08:00
//value, err = client.ZRangeWithScores(ctx, key, 0, -1).Result()
2023-06-27 15:53:29 +08:00
var items []types.ZSetItem
for {
var loadedVal []string
2023-10-14 21:26:47 +08:00
loadedVal, cursor, err = client.ZScan(ctx, key, cursor, "*", 10000).Result()
2023-06-27 15:53:29 +08:00
if err != nil {
resp.Msg = err.Error()
return
}
var score float64
for i := 0; i < len(loadedVal); i += 2 {
if score, err = strconv.ParseFloat(loadedVal[i+1], 64); err == nil {
items = append(items, types.ZSetItem{
Value: loadedVal[i],
Score: score,
})
}
}
if cursor == 0 {
break
}
}
value = items
2023-10-14 21:26:47 +08:00
size, _ = client.ZCard(ctx, key).Result()
2023-07-25 19:12:50 +08:00
case "stream":
var msgs []redis.XMessage
items := []types.StreamItem{}
2023-10-14 21:26:47 +08:00
msgs, err = client.XRevRange(ctx, key, "+", "-").Result()
2023-07-25 19:12:50 +08:00
if err != nil {
resp.Msg = err.Error()
return
}
for _, msg := range msgs {
items = append(items, types.StreamItem{
ID: msg.ID,
Value: msg.Values,
})
}
value = items
2023-10-14 21:26:47 +08:00
size, _ = client.XLen(ctx, key).Result()
2023-06-27 15:53:29 +08:00
}
if err != nil {
resp.Msg = err.Error()
return
}
resp.Success = true
resp.Data = map[string]any{
"type": keyType,
"ttl": ttl,
"value": value,
"size": size,
"viewAs": viewAs,
2023-06-27 15:53:29 +08:00
}
return
}
// SetKeyValue set value by key
// @param ttl <= 0 means keep current ttl
func (c *connectionService) SetKeyValue(connName string, db int, k any, keyType string, value any, ttl int64, viewAs string) (resp types.JSResp) {
2023-10-14 21:26:47 +08:00
client, ctx, err := c.getRedisClient(connName, db)
2023-06-27 15:53:29 +08:00
if err != nil {
resp.Msg = err.Error()
return
}
key := strutil.DecodeRedisKey(k)
2023-06-27 15:53:29 +08:00
var expiration time.Duration
if ttl < 0 {
2023-10-14 21:26:47 +08:00
if expiration, err = client.PTTL(ctx, key).Result(); err != nil {
expiration = redis.KeepTTL
}
2023-06-27 15:53:29 +08:00
} else {
expiration = time.Duration(ttl) * time.Second
}
switch strings.ToLower(keyType) {
case "string":
if str, ok := value.(string); !ok {
resp.Msg = "invalid string value"
return
} else {
var saveStr string
if saveStr, err = strutil.SaveAs(str, viewAs); err != nil {
resp.Msg = fmt.Sprintf(`save to "%s" type fail: %s`, viewAs, err.Error())
return
}
2023-10-14 21:26:47 +08:00
_, err = client.Set(ctx, key, saveStr, 0).Result()
// set expiration lonely, not "keepttl"
if err == nil && expiration > 0 {
2023-10-14 21:26:47 +08:00
client.Expire(ctx, key, expiration)
}
2023-06-27 15:53:29 +08:00
}
case "list":
if strs, ok := value.([]any); !ok {
resp.Msg = "invalid list value"
return
} else {
2023-10-14 21:26:47 +08:00
err = client.LPush(ctx, key, strs...).Err()
2023-07-25 19:12:50 +08:00
if err == nil && expiration > 0 {
2023-10-14 21:26:47 +08:00
client.Expire(ctx, key, expiration)
2023-07-25 19:12:50 +08:00
}
2023-06-27 15:53:29 +08:00
}
case "hash":
if strs, ok := value.([]any); !ok {
resp.Msg = "invalid hash value"
return
} else {
total := len(strs)
if total > 1 {
2023-10-14 21:26:47 +08:00
_, err = client.Pipelined(ctx, func(pipe redis.Pipeliner) error {
for i := 0; i < total; i += 2 {
pipe.HSet(ctx, key, strs[i], strs[i+1])
}
if expiration > 0 {
pipe.Expire(ctx, key, expiration)
}
return nil
})
}
2023-06-27 15:53:29 +08:00
}
case "set":
if strs, ok := value.([]any); !ok || len(strs) <= 0 {
resp.Msg = "invalid set value"
return
} else {
if len(strs) > 0 {
2023-10-14 21:26:47 +08:00
err = client.SAdd(ctx, key, strs...).Err()
if err == nil && expiration > 0 {
2023-10-14 21:26:47 +08:00
client.Expire(ctx, key, expiration)
2023-06-27 15:53:29 +08:00
}
}
2023-06-27 15:53:29 +08:00
}
case "zset":
if strs, ok := value.([]any); !ok || len(strs) <= 0 {
resp.Msg = "invalid zset value"
return
} else {
if len(strs) > 1 {
2023-06-27 15:53:29 +08:00
var members []redis.Z
for i := 0; i < len(strs); i += 2 {
score, _ := strconv.ParseFloat(strs[i+1].(string), 64)
2023-06-27 15:53:29 +08:00
members = append(members, redis.Z{
Score: score,
Member: strs[i],
2023-06-27 15:53:29 +08:00
})
}
2023-10-14 21:26:47 +08:00
err = client.ZAdd(ctx, key, members...).Err()
if err == nil && expiration > 0 {
2023-10-14 21:26:47 +08:00
client.Expire(ctx, key, expiration)
2023-06-27 15:53:29 +08:00
}
}
2023-06-27 15:53:29 +08:00
}
2023-07-25 19:12:50 +08:00
case "stream":
if strs, ok := value.([]any); !ok {
resp.Msg = "invalid stream value"
return
} else {
if len(strs) > 2 {
2023-10-14 21:26:47 +08:00
err = client.XAdd(ctx, &redis.XAddArgs{
2023-07-25 19:12:50 +08:00
Stream: key,
ID: strs[0].(string),
Values: strs[1:],
}).Err()
if err == nil && expiration > 0 {
2023-10-14 21:26:47 +08:00
client.Expire(ctx, key, expiration)
2023-07-25 19:12:50 +08:00
}
}
}
2023-06-27 15:53:29 +08:00
}
if err != nil {
resp.Msg = err.Error()
return
}
resp.Success = true
resp.Data = map[string]any{
"value": value,
}
return
}
// SetHashValue set hash field
func (c *connectionService) SetHashValue(connName string, db int, k any, field, newField, value string) (resp types.JSResp) {
2023-10-14 21:26:47 +08:00
client, ctx, err := c.getRedisClient(connName, db)
2023-06-27 15:53:29 +08:00
if err != nil {
resp.Msg = err.Error()
return
}
key := strutil.DecodeRedisKey(k)
2023-06-27 15:53:29 +08:00
var removedField []string
updatedField := map[string]string{}
if len(field) <= 0 {
// old filed is empty, add new field
2023-10-14 21:26:47 +08:00
_, err = client.HSet(ctx, key, newField, value).Result()
2023-06-27 15:53:29 +08:00
updatedField[newField] = value
} else if len(newField) <= 0 {
// new field is empty, delete old field
2023-10-14 21:26:47 +08:00
_, err = client.HDel(ctx, key, field, value).Result()
2023-06-27 15:53:29 +08:00
removedField = append(removedField, field)
} else if field == newField {
// replace field
2023-10-14 21:26:47 +08:00
_, err = client.HSet(ctx, key, newField, value).Result()
2023-06-27 15:53:29 +08:00
updatedField[newField] = value
} else {
// remove old field and add new field
2023-10-14 21:26:47 +08:00
if _, err = client.HDel(ctx, key, field).Result(); err != nil {
2023-06-27 15:53:29 +08:00
resp.Msg = err.Error()
return
}
2023-10-14 21:26:47 +08:00
_, err = client.HSet(ctx, key, newField, value).Result()
2023-06-27 15:53:29 +08:00
removedField = append(removedField, field)
updatedField[newField] = value
}
if err != nil {
resp.Msg = err.Error()
return
}
resp.Success = true
resp.Data = map[string]any{
"removed": removedField,
"updated": updatedField,
}
return
}
// AddHashField add or update hash field
func (c *connectionService) AddHashField(connName string, db int, k any, action int, fieldItems []any) (resp types.JSResp) {
2023-10-14 21:26:47 +08:00
client, ctx, err := c.getRedisClient(connName, db)
2023-06-27 15:53:29 +08:00
if err != nil {
resp.Msg = err.Error()
return
}
key := strutil.DecodeRedisKey(k)
2023-06-27 15:53:29 +08:00
updated := map[string]any{}
switch action {
case 1:
// ignore duplicated fields
for i := 0; i < len(fieldItems); i += 2 {
2023-10-14 21:26:47 +08:00
_, err = client.HSetNX(ctx, key, fieldItems[i].(string), fieldItems[i+1]).Result()
2023-06-27 15:53:29 +08:00
if err == nil {
updated[fieldItems[i].(string)] = fieldItems[i+1]
}
}
default:
// overwrite duplicated fields
total := len(fieldItems)
if total > 1 {
2023-10-14 21:26:47 +08:00
_, err = client.Pipelined(ctx, func(pipe redis.Pipeliner) error {
for i := 0; i < total; i += 2 {
2023-10-14 21:26:47 +08:00
client.HSet(ctx, key, fieldItems[i], fieldItems[i+1])
}
return nil
})
for i := 0; i < total; i += 2 {
updated[fieldItems[i].(string)] = fieldItems[i+1]
}
2023-06-27 15:53:29 +08:00
}
}
if err != nil {
resp.Msg = err.Error()
return
}
resp.Success = true
resp.Data = map[string]any{
"updated": updated,
}
return
}
// AddListItem add item to list or remove from it
func (c *connectionService) AddListItem(connName string, db int, k any, action int, items []any) (resp types.JSResp) {
2023-10-14 21:26:47 +08:00
client, ctx, err := c.getRedisClient(connName, db)
2023-06-27 15:53:29 +08:00
if err != nil {
resp.Msg = err.Error()
return
}
key := strutil.DecodeRedisKey(k)
2023-06-27 15:53:29 +08:00
var leftPush, rightPush []any
switch action {
case 0:
// push to head
2023-10-14 21:26:47 +08:00
_, err = client.LPush(ctx, key, items...).Result()
2023-06-27 15:53:29 +08:00
leftPush = append(leftPush, items...)
default:
// append to tail
2023-10-14 21:26:47 +08:00
_, err = client.RPush(ctx, key, items...).Result()
2023-06-27 15:53:29 +08:00
rightPush = append(rightPush, items...)
}
if err != nil {
resp.Msg = err.Error()
return
}
resp.Success = true
resp.Data = map[string]any{
"left": leftPush,
"right": rightPush,
}
return
}
// SetListItem update or remove list item by index
func (c *connectionService) SetListItem(connName string, db int, k any, index int64, value string) (resp types.JSResp) {
2023-10-14 21:26:47 +08:00
client, ctx, err := c.getRedisClient(connName, db)
2023-06-27 15:53:29 +08:00
if err != nil {
resp.Msg = err.Error()
return
}
key := strutil.DecodeRedisKey(k)
2023-06-27 15:53:29 +08:00
var removed []int64
updated := map[int64]string{}
if len(value) <= 0 {
// remove from list
2023-10-14 21:26:47 +08:00
err = client.LSet(ctx, key, index, "---VALUE_REMOVED_BY_TINY_RDM---").Err()
2023-06-27 15:53:29 +08:00
if err != nil {
resp.Msg = err.Error()
return
}
2023-10-14 21:26:47 +08:00
err = client.LRem(ctx, key, 1, "---VALUE_REMOVED_BY_TINY_RDM---").Err()
2023-06-27 15:53:29 +08:00
if err != nil {
resp.Msg = err.Error()
return
}
removed = append(removed, index)
} else {
// replace index value
2023-10-14 21:26:47 +08:00
err = client.LSet(ctx, key, index, value).Err()
2023-06-27 15:53:29 +08:00
if err != nil {
resp.Msg = err.Error()
return
}
updated[index] = value
}
resp.Success = true
resp.Data = map[string]any{
"removed": removed,
"updated": updated,
}
return
}
// SetSetItem add members to set or remove from set
func (c *connectionService) SetSetItem(connName string, db int, k any, remove bool, members []any) (resp types.JSResp) {
2023-10-14 21:26:47 +08:00
client, ctx, err := c.getRedisClient(connName, db)
2023-06-27 15:53:29 +08:00
if err != nil {
resp.Msg = err.Error()
return
}
key := strutil.DecodeRedisKey(k)
2023-06-27 15:53:29 +08:00
if remove {
2023-10-14 21:26:47 +08:00
_, err = client.SRem(ctx, key, members...).Result()
2023-06-27 15:53:29 +08:00
} else {
2023-10-14 21:26:47 +08:00
_, err = client.SAdd(ctx, key, members...).Result()
2023-06-27 15:53:29 +08:00
}
if err != nil {
resp.Msg = err.Error()
return
}
resp.Success = true
return
}
// UpdateSetItem replace member of set
func (c *connectionService) UpdateSetItem(connName string, db int, k any, value, newValue string) (resp types.JSResp) {
2023-10-14 21:26:47 +08:00
client, ctx, err := c.getRedisClient(connName, db)
2023-06-27 15:53:29 +08:00
if err != nil {
resp.Msg = err.Error()
return
}
key := strutil.DecodeRedisKey(k)
2023-10-14 21:26:47 +08:00
_, _ = client.SRem(ctx, key, value).Result()
_, err = client.SAdd(ctx, key, newValue).Result()
2023-06-27 15:53:29 +08:00
if err != nil {
resp.Msg = err.Error()
return
}
resp.Success = true
return
}
// UpdateZSetValue update value of sorted set member
func (c *connectionService) UpdateZSetValue(connName string, db int, k any, value, newValue string, score float64) (resp types.JSResp) {
2023-10-14 21:26:47 +08:00
client, ctx, err := c.getRedisClient(connName, db)
2023-06-27 15:53:29 +08:00
if err != nil {
resp.Msg = err.Error()
return
}
key := strutil.DecodeRedisKey(k)
2023-06-27 15:53:29 +08:00
updated := map[string]any{}
var removed []string
if len(newValue) <= 0 {
// blank new value, delete value
2023-10-14 21:26:47 +08:00
_, err = client.ZRem(ctx, key, value).Result()
2023-06-27 15:53:29 +08:00
if err == nil {
removed = append(removed, value)
}
} else if newValue == value {
// update score only
2023-10-14 21:26:47 +08:00
_, err = client.ZAdd(ctx, key, redis.Z{
2023-06-27 15:53:29 +08:00
Score: score,
Member: value,
}).Result()
} else {
// remove old value and add new one
2023-10-14 21:26:47 +08:00
_, err = client.ZRem(ctx, key, value).Result()
2023-06-27 15:53:29 +08:00
if err == nil {
removed = append(removed, value)
}
2023-10-14 21:26:47 +08:00
_, err = client.ZAdd(ctx, key, redis.Z{
2023-06-27 15:53:29 +08:00
Score: score,
Member: newValue,
}).Result()
if err == nil {
updated[newValue] = score
}
}
if err != nil {
resp.Msg = err.Error()
return
}
resp.Success = true
resp.Data = map[string]any{
"updated": updated,
"removed": removed,
}
return
}
// AddZSetValue add item to sorted set
func (c *connectionService) AddZSetValue(connName string, db int, k any, action int, valueScore map[string]float64) (resp types.JSResp) {
2023-10-14 21:26:47 +08:00
client, ctx, err := c.getRedisClient(connName, db)
2023-06-27 15:53:29 +08:00
if err != nil {
resp.Msg = err.Error()
return
}
key := strutil.DecodeRedisKey(k)
2023-06-27 15:53:29 +08:00
members := maputil.ToSlice(valueScore, func(k string) redis.Z {
return redis.Z{
Score: valueScore[k],
Member: k,
}
})
switch action {
case 1:
// ignore duplicated fields
2023-10-14 21:26:47 +08:00
_, err = client.ZAddNX(ctx, key, members...).Result()
2023-06-27 15:53:29 +08:00
default:
// overwrite duplicated fields
2023-10-14 21:26:47 +08:00
_, err = client.ZAdd(ctx, key, members...).Result()
2023-06-27 15:53:29 +08:00
}
if err != nil {
resp.Msg = err.Error()
return
}
resp.Success = true
return
}
2023-07-25 19:12:50 +08:00
// AddStreamValue add stream field
func (c *connectionService) AddStreamValue(connName string, db int, k any, ID string, fieldItems []any) (resp types.JSResp) {
2023-10-14 21:26:47 +08:00
client, ctx, err := c.getRedisClient(connName, db)
2023-07-25 19:12:50 +08:00
if err != nil {
resp.Msg = err.Error()
return
}
key := strutil.DecodeRedisKey(k)
2023-10-14 21:26:47 +08:00
_, err = client.XAdd(ctx, &redis.XAddArgs{
2023-07-25 19:12:50 +08:00
Stream: key,
ID: ID,
Values: fieldItems,
}).Result()
if err != nil {
resp.Msg = err.Error()
return
}
resp.Success = true
return
}
// RemoveStreamValues remove stream values by id
func (c *connectionService) RemoveStreamValues(connName string, db int, k any, IDs []string) (resp types.JSResp) {
2023-10-14 21:26:47 +08:00
client, ctx, err := c.getRedisClient(connName, db)
2023-07-25 19:12:50 +08:00
if err != nil {
resp.Msg = err.Error()
return
}
key := strutil.DecodeRedisKey(k)
2023-10-14 21:26:47 +08:00
_, err = client.XDel(ctx, key, IDs...).Result()
2023-07-25 19:12:50 +08:00
resp.Success = true
return
}
2023-06-27 15:53:29 +08:00
// SetKeyTTL set ttl of key
func (c *connectionService) SetKeyTTL(connName string, db int, k any, ttl int64) (resp types.JSResp) {
2023-10-14 21:26:47 +08:00
client, ctx, err := c.getRedisClient(connName, db)
2023-06-27 15:53:29 +08:00
if err != nil {
resp.Msg = err.Error()
return
}
key := strutil.DecodeRedisKey(k)
2023-06-27 15:53:29 +08:00
var expiration time.Duration
if ttl < 0 {
2023-10-14 21:26:47 +08:00
if err = client.Persist(ctx, key).Err(); err != nil {
2023-06-27 15:53:29 +08:00
resp.Msg = err.Error()
return
}
} else {
expiration = time.Duration(ttl) * time.Second
2023-10-14 21:26:47 +08:00
if err = client.Expire(ctx, key, expiration).Err(); err != nil {
2023-06-27 15:53:29 +08:00
resp.Msg = err.Error()
return
}
}
resp.Success = true
return
}
// DeleteKey remove redis key
func (c *connectionService) DeleteKey(connName string, db int, k any) (resp types.JSResp) {
2023-10-14 21:26:47 +08:00
client, ctx, err := c.getRedisClient(connName, db)
2023-06-27 15:53:29 +08:00
if err != nil {
resp.Msg = err.Error()
return
}
key := strutil.DecodeRedisKey(k)
var deletedKeys []string
if strings.HasSuffix(key, "*") {
// delete by prefix
2023-10-14 21:26:47 +08:00
var mutex sync.Mutex
del := func(ctx context.Context, cli redis.UniversalClient) error {
iter := cli.Scan(ctx, 0, key, 10000).Iterator()
for iter.Next(ctx) {
subKey := iter.Val()
if err = cli.Unlink(ctx, subKey).Err(); err != nil {
log.Println("unlink error", err.Error())
return err
} else {
2023-10-14 21:26:47 +08:00
mutex.Lock()
deletedKeys = append(deletedKeys, subKey)
mutex.Unlock()
}
}
2023-10-14 21:26:47 +08:00
return nil
}
2023-10-14 21:26:47 +08:00
if cluster, ok := client.(*redis.ClusterClient); ok {
// cluster mode
err = cluster.ForEachMaster(ctx, func(ctx context.Context, cli *redis.Client) error {
return del(ctx, cli)
})
} else {
err = del(ctx, client)
}
if err != nil {
resp.Msg = err.Error()
return
}
} else {
// delete key only
2023-10-14 21:26:47 +08:00
_, err = client.Del(ctx, key).Result()
if err != nil {
resp.Msg = err.Error()
return
}
deletedKeys = append(deletedKeys, key)
2023-06-27 15:53:29 +08:00
}
resp.Success = true
resp.Data = map[string]any{
"deleted": deletedKeys,
2023-06-27 15:53:29 +08:00
}
return
}
// RenameKey rename key
func (c *connectionService) RenameKey(connName string, db int, key, newKey string) (resp types.JSResp) {
2023-10-14 21:26:47 +08:00
client, ctx, err := c.getRedisClient(connName, db)
2023-06-27 15:53:29 +08:00
if err != nil {
resp.Msg = err.Error()
return
}
2023-10-14 21:26:47 +08:00
_, err = client.RenameNX(ctx, key, newKey).Result()
2023-06-27 15:53:29 +08:00
if err != nil {
resp.Msg = err.Error()
return
}
resp.Success = true
return
}
2023-08-28 15:21:08 +08:00
// GetCmdHistory get redis command history
2023-07-16 01:50:01 +08:00
func (c *connectionService) GetCmdHistory(pageNo, pageSize int) (resp types.JSResp) {
resp.Success = true
if pageSize <= 0 || pageNo <= 0 {
// return all history
resp.Data = map[string]any{
"list": c.cmdHistory,
"pageNo": 1,
"pageSize": -1,
}
} else {
total := len(c.cmdHistory)
startIndex := total / pageSize * (pageNo - 1)
endIndex := min(startIndex+pageSize, total)
2023-07-16 01:50:01 +08:00
resp.Data = map[string]any{
"list": c.cmdHistory[startIndex:endIndex],
"pageNo": pageNo,
"pageSize": pageSize,
}
}
return
}
2023-08-28 15:21:08 +08:00
// CleanCmdHistory clean redis command history
func (c *connectionService) CleanCmdHistory() (resp types.JSResp) {
c.cmdHistory = []cmdHistoryItem{}
resp.Success = true
return
}
2023-06-27 15:53:29 +08:00
// update or insert key info to database
//func (c *connectionService) updateDBKey(connName string, db int, keys []string, separator string) {
// dbStruct := map[string]any{}
// for _, key := range keys {
// keyPart := strings.Split(key, separator)
// prefixLen := len(keyPart)-1
// if prefixLen > 0 {
// for i := 0; i < prefixLen; i++ {
// if dbStruct[keyPart[i]]
// keyPart[i]
// }
// }
// log.Println("key", key)
// }
//}