feat(redis): redis 版本原地升级

dbm-services/redis/db-tools/dbactuator/example/redis_cluster_failover.example.md

@@ -0,0 +1,24 @@
+### redis cluster failover
+redis cluster failover
+`./dbactuator_redis  --uid={{uid}} --root_id={{root_id}} --node_id={{node_id}} --version_id={{version_id}} --atom-job-list="redis_cluster_failover" --data_dir=/path/to/data  --backup_dir=/path/to/backup --payload='{{payload_base64}}'`
+
+`--data_dir`、`--backup_dir` 可以留空.  
+
+
+原始payload
+```json
+{
+    "redis_password":"xxxx",
+    "redis_master_slave_pairs":[
+        {
+            "master": {"ip":"a.a.a.a","port":"30000"},
+            "slave": {"ip":"b.b.b.b","port":"30000"}
+        },
+        {
+            "master": {"ip":"a.a.a.a","port":"30001"},
+            "slave": {"ip":"b.b.b.b","port":"30001"}
+        }
+    ],
+    "force":false
+}
+```

dbm-services/redis/db-tools/dbactuator/example/redis_version_update.example.md

@@ -0,0 +1,21 @@
+### redis version update
+redis版本更新  
+`./dbactuator_redis  --uid={{uid}} --root_id={{root_id}} --node_id={{node_id}} --version_id={{version_id}} --atom-job-list="redis_version_update" --data_dir=/path/to/data  --backup_dir=/path/to/backup --payload='{{payload_base64}}'`
+
+`--data_dir`、`--backup_dir` 可以留空.  
+
+前置工作:  
+- 将`redis-6.2.7.tar.gz`下载到`/data/install`目录下;  
+
+
+原始payload
+```json
+{
+    "pkg":"redis-6.2.7.tar.gz",
+    "pkg_md5":"1fc9e5c3a044ce523844a6f2717e5ac3",
+    "ip":"127.0.0.1",
+    "ports":[30000,30001,30002], //端口不连续
+    "password":"xxx",
+    "role":"redis_master" // or redis_slave
+}
+```

dbm-services/redis/db-tools/dbactuator/models/myredis/client.go

@@ -1751,6 +1751,32 @@ func (db *RedisClient) ClusterReset() (err error) {
 	return nil
 }
 
+// ClusterFailOver 执行cluster failover,opt: force or takeover
+func (db *RedisClient) ClusterFailOver(opt string) (err error) {
+	var ret string
+	if opt == "" {
+		ret, err = db.InstanceClient.ClusterFailover(context.Background()).Result()
+	} else if strings.ToLower(opt) == "force" || strings.ToLower(opt) == "takeover" {
+		cmds := []string{"cluster", "failover", opt}
+		tmpRet, err := db.DoCommand(cmds, 0)
+		if err != nil {
+			return err
+		}
+		ret = tmpRet.(string)
+	}
+	if err != nil {
+		err = fmt.Errorf("ClusterFailOver fail,err:%v,addr:%s", err, db.Addr)
+		mylog.Logger.Error(err.Error())
+		return
+	}
+	if strings.ToLower(ret) != "ok" {
+		err = fmt.Errorf("ClusterFailOver fail,ret:%s,addr:%s", ret, db.Addr)
+		mylog.Logger.Error(err.Error())
+		return
+	}
+	return nil
+}
+
 // GetClusterNodesStr 获取tendisplus集群cluster nodes命令结果,并返回字符串
 func (db *RedisClient) GetClusterNodesStr() (ret string, err error) {
 	ret, err = db.InstanceClient.ClusterNodes(context.TODO()).Result()

dbm-services/redis/db-tools/dbactuator/models/myredis/myredis.go

@@ -106,3 +106,44 @@ func LocalRedisConnectTest(ip string, ports []int, password string) (err error)
 	}
 	return
 }
+
+// CheckMultiRedisConnected 检查多个redis是否可连接
+func CheckMultiRedisConnected(addrs []string, password string) (err error) {
+	l01 := make([]*connTestItem, 0, len(addrs))
+	for _, addr := range addrs {
+		ip, port, err := util.AddrToIpPort(addr)
+		if err != nil {
+			return err
+		}
+		l01 = append(l01, &connTestItem{
+			IP:       ip,
+			Port:     port,
+			Password: password,
+		})
+	}
+	// 并发测试
+	wg := sync.WaitGroup{}
+	for _, item := range l01 {
+		test01 := item
+		wg.Add(1)
+		go func(test01 *connTestItem) {
+			defer wg.Done()
+			cli01, err := NewRedisClientWithTimeout(test01.addr(), test01.Password, 0,
+				consts.TendisTypeRedisInstance, 10*time.Second)
+			if err != nil {
+				test01.Err = err
+				return
+			}
+			cli01.Close()
+		}(test01)
+	}
+	wg.Wait()
+
+	for _, item := range l01 {
+		test01 := item
+		if test01.Err != nil {
+			return test01.Err
+		}
+	}
+	return
+}

dbm-services/redis/db-tools/dbactuator/models/myredis/tendisplus_infoRepl.go

@@ -12,15 +12,22 @@ import (
 	"dbm-services/redis/db-tools/dbactuator/pkg/consts"
 )
 
-// InfoReplSlave Tendisplus master中执行info replication结果中slave状态
+// InfoReplSlave master中执行info replication结果中slave状态
 // 如: slave0:ip=luketest03-redis-rdsplus4-1.luketest03-svc.dmc,port=30000,state=online,offset=930327677,lag=0
 type InfoReplSlave struct {
-	Name   string `json:"name"`
-	IP     string `json:"ip"`
-	Port   int    `json:"port"`
-	State  string `json:"state"`
-	Offset int64  `json:"offset"`
-	Lag    int64  `json:"lag"`
+	Name      string `json:"name"`
+	IP        string `json:"ip"`
+	Port      int    `json:"port"`
+	State     string `json:"state"`
+	Offset    int64  `json:"offset"`
+	Seq       int64  `json:"seq"`
+	Lag       int64  `json:"lag"`
+	BinlogLag int64  `json:"binlog_lag"`
+}
+
+// Addr addr字符串
+func (slave *InfoReplSlave) Addr() string {
+	return slave.IP + ":" + strconv.Itoa(slave.Port)
 }
 
 func (slave *InfoReplSlave) decode(line string) error {
@@ -40,10 +47,14 @@ func (slave *InfoReplSlave) decode(line string) error {
 			slave.Port, _ = strconv.Atoi(list02[1])
 		} else if list02[0] == "state" {
 			slave.State = list02[1]
+		} else if list02[0] == "seq" {
+			slave.Seq, _ = strconv.ParseInt(list02[1], 10, 64)
 		} else if list02[0] == "offset" {
 			slave.Offset, _ = strconv.ParseInt(list02[1], 10, 64)
 		} else if list02[0] == "lag" {
 			slave.Lag, _ = strconv.ParseInt(list02[1], 10, 64)
+		} else if list02[0] == "binlog_lag" {
+			slave.BinlogLag, _ = strconv.ParseInt(list02[1], 10, 64)
 		}
 	}
 	return nil
@@ -411,3 +422,46 @@ func (db *RedisClient) TendisSSDInfoSlaves() (ret TendisSSDInfoSlavesData, err e
 	}
 	return
 }
+
+// GetInfoReplSlaves 获取info replication中slave列表
+func (db *RedisClient) GetInfoReplSlaves() (slaveList []*InfoReplSlave, slaveMap map[string]*InfoReplSlave, err error) {
+	var replRet string
+	if db.DbType == consts.TendisTypeRedisCluster {
+		replRet, err = db.ClusterClient.Info(context.TODO(), "replication").Result()
+	} else {
+		replRet, err = db.InstanceClient.Info(context.TODO(), "replication").Result()
+	}
+	if err != nil {
+		err = fmt.Errorf("info replication fail,err:%v,aadr:%s", err, db.Addr)
+		mylog.Logger.Error(err.Error())
+		return
+	}
+	infoList := strings.Split(replRet, "\n")
+	slaveReg := regexp.MustCompile(`^slave\d+$`)
+	slaveMap = make(map[string]*InfoReplSlave)
+	for _, infoItem := range infoList {
+		infoItem = strings.TrimSpace(infoItem)
+		if strings.HasPrefix(infoItem, "#") {
+			continue
+		}
+		if len(infoItem) == 0 {
+			continue
+		}
+		list01 := strings.SplitN(infoItem, ":", 2)
+		if len(list01) < 2 {
+			continue
+		}
+		list01[0] = strings.TrimSpace(list01[0])
+		list01[1] = strings.TrimSpace(list01[1])
+		if slaveReg.MatchString(list01[0]) == true {
+			slave01 := &InfoReplSlave{}
+			err = slave01.decode(infoItem)
+			if err != nil {
+				return
+			}
+			slaveList = append(slaveList, slave01)
+			slaveMap[slave01.Addr()] = slave01
+		}
+	}
+	return
+}

dbm-services/redis/db-tools/dbactuator/pkg/atomjobs/atomredis/redis_backup.go

@@ -419,6 +419,7 @@ func (task *BackupTask) newConnect() {
 	if task.Err != nil {
 		return
 	}
+	mylog.Logger.Info("redis(%s) role:%s,dataDir:%s,dbType:%s", task.Addr(), task.Role, task.DataDir, task.DbType)
 	// 获取数据量大小
 	if task.DbType == consts.TendisTypeRedisInstance {
 		task.DataSize, task.Err = task.Cli.RedisInstanceDataSize()
@@ -439,11 +440,13 @@ func (task *BackupTask) PrecheckDisk() {
 	bakDiskUsg, err := util.GetLocalDirDiskUsg(task.BackupDir)
 	task.Err = err
 	if task.Err != nil {
+		mylog.Logger.Error("redis:%s backupDir:%s GetLocalDirDiskUsg fail,err:%v", task.Addr(), task.BackupDir, err)
 		return
 	}
 	dataDiskUsg, err := util.GetLocalDirDiskUsg(task.DataDir)
 	task.Err = err
 	if task.Err != nil {
+		mylog.Logger.Error("redis:%s dataDir:%s GetLocalDirDiskUsg fail,err:%v", task.Addr(), task.DataDir, err)
 		return
 	}
 	// 磁盘空间使用已有85%,则报错