feat(redis): Job超时时间改成3天;可以执行执行账户 #7104

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

@@ -46,7 +46,7 @@ func (r *RedisLocalDoDR) Init(runtime *jobruntime.JobGenericRuntime) error {
 	// 获取安装参数
 	r.DataDir = consts.GetRedisDataDir()
 	r.runtime = runtime
-	r.runtime.Logger.Info("start to redis local do dr .")
+	r.runtime.Logger.Info("start to redis local do dr . with dir :%s", r.DataDir)
 
 	// 加载参数
 	if err := json.Unmarshal([]byte(r.runtime.PayloadDecoded), &r.params); err != nil {
@@ -87,10 +87,12 @@ func (r *RedisLocalDoDR) Run() error {
 		return fmt.Errorf("X-%s-X", r.params.ClusterType)
 	}
 
+	r.runtime.Logger.Warn("first stop dbmon.")
 	if rst, err := util.RunBashCmd("/home/mysql/bk-dbmon/stop.sh", "", nil, 10*time.Second); err != nil {
 		r.runtime.Logger.Warn("try stop dbmon failed %s:+%+v", rst, err)
 	}
 	defer func() {
+		r.runtime.Logger.Warn("finally start dbmon.")
 		if rst, err := util.RunBashCmd("/home/mysql/bk-dbmon/start.sh", "", nil, 10*time.Second); err != nil {
 			r.runtime.Logger.Error("try start dbmon failed %s:+%+v", rst, err)
 		}
@@ -116,19 +118,19 @@ func (r *RedisLocalDoDR) Run() error {
 		if err := r.backupFiles(addr, instance); err != nil {
 			return err
 		}
-		// start , slaveof ,wait,rewirte
+		// start , slaveof ,==,rewirte
 		if err := r.startAndWaitLinkUp(addr, password, idx, instance); err != nil {
 			return err
 		}
-		r.runtime.Logger.Info("done local redo dr %d:%s", idx, addr)
+		r.runtime.Logger.Info("done local redo dr %d:%s ^_^ \n", idx, addr)
 	}
 	return nil
 }
 
 // start-redis.sh
 // slaveof master xxx xx
 // config rewite
-// wating for link up .
+// == for link up .
 func (r *RedisLocalDoDR) startAndWaitLinkUp(addr, pass string, idx int, instance ReplicaItem) error {
 	if rst, err := util.RunBashCmd(fmt.Sprintf("/usr/local/redis/bin/start-redis.sh %d",
 		instance.SlavePort), "", nil, 10*time.Second); err != nil || rst != "" {
@@ -191,11 +193,11 @@ func (r *RedisLocalDoDR) WaitMasterLinkUp(rConn *myredis.RedisClient, addr strin
 }
 
 func (r *RedisLocalDoDR) backupFiles(addr string, instance ReplicaItem) error {
-	bashPath := filepath.Join(r.DataDir, strconv.Itoa(instance.SlavePort))
+	bashPath := filepath.Join(r.DataDir, "redis", strconv.Itoa(instance.SlavePort))
 
 	rdbFile, aofFile := filepath.Join(bashPath, "data", "dump.rdb"), filepath.Join(bashPath, "data", "appendonly.aof")
-	bkRdb := filepath.Join("/data/dbbak", fmt.Sprintf("backup.%s.%d.dump.rdb", r.runtime.UID, r.startTime))
-	bkAof := filepath.Join("/data/dbbak", fmt.Sprintf("backup.%s.%d.appendonly.aof", r.runtime.UID, r.startTime))
+	bkRdb := filepath.Join("/data/dbbak", fmt.Sprintf("backup.%s.%d.%d.dump.rdb", r.runtime.UID, r.startTime, instance.SlavePort))
+	bkAof := filepath.Join("/data/dbbak", fmt.Sprintf("backup.%s.%d.%d.appendonly.aof", r.runtime.UID, r.startTime, instance.SlavePort))
 	if err := r.tryBackupData(aofFile, bkAof, addr); err != nil {
 		return err
 	}
@@ -221,7 +223,8 @@ func (r *RedisLocalDoDR) tryCommentSlaveOf(cnf, addr string) error {
 		return nil
 	}
 
-	if rst, err := util.RunBashCmd(fmt.Sprintf("sed -i 's/slaveof /#slaveof /g' %s", cnf),
+	// replicaof slaveof
+	if rst, err := util.RunBashCmd(fmt.Sprintf("sed -i 's/slaveof /#slaveof /g; s/replicaof /#replicaof /g' %s", cnf),
 		"", nil, 10*time.Second); err != nil || rst != "" {
 		r.runtime.Logger.Error("backup file %s ,failed:%s:%+v", cnf, rst, err)
 		return fmt.Errorf("failed by mv %s:%+v", rst, err)
@@ -239,7 +242,7 @@ func (r *RedisLocalDoDR) tryBackupData(src, dst, addr string) error {
 		r.runtime.Logger.Error("backup file %s ,failed:%s:%+v", src, rst, err)
 		return fmt.Errorf("failed by mv %s:%+v", rst, err)
 	}
-	r.runtime.Logger.Info("%s rdb[%s] and aof[%s] backuped succ ^_^", addr, src, dst)
+	r.runtime.Logger.Info("backup %s [%s] 2-> [%s] succ ^_^", addr, src, dst)
 	return nil
 }
 

dbm-ui/backend/flow/engine/bamboo/scene/redis/atom_jobs/redis_cluster_master_rep.py

@@ -228,6 +228,7 @@ def TwemproxyClusterMasterReplaceJob(
     ]:
         sub_pipelines, resync_args = [], deepcopy(act_kwargs)
         resync_args.cluster = {
+            "run_as_system_user": "mysql",
             "bk_biz_id": int(act_kwargs.cluster["bk_biz_id"]),
             "cluster_id": int(act_kwargs.cluster["cluster_id"]),
             "cluster_type": act_kwargs.cluster["cluster_type"],

dbm-ui/backend/flow/plugins/components/collections/redis/exec_actuator_script.py

@@ -8,6 +8,7 @@
 an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the
 specific language governing permissions and limitations under the License.
 """
+import copy
 import json
 import logging
 import re
@@ -118,11 +119,13 @@ def _execute(self, data, parent_data) -> bool:
             "target_server": {"ip_list": target_ip_info},
             "timeout": 86400,
         }
+        # 外边可以指定执行账户，不指定则采用默认账户Root
+        common_kwargs = copy.deepcopy(redis_fast_execute_script_common_kwargs)
+        if kwargs["cluster"].get("run_as_system_user"):
+            common_kwargs["account_alias"] = kwargs["cluster"]["run_as_system_user"]
 
-        self.log_info(
-            "[{}] ready start task with body {} {}".format(node_name, redis_fast_execute_script_common_kwargs, body)
-        )
-        resp = JobApi.fast_execute_script({**redis_fast_execute_script_common_kwargs, **body}, raw=True)
+        self.log_info("[{}] ready start task with body {} {}".format(node_name, common_kwargs, body))
+        resp = JobApi.fast_execute_script({**common_kwargs, **body}, raw=True)
 
         # 传入调用结果，并单调监听任务状态
         data.outputs.ext_result = resp

dbm-ui/backend/flow/utils/redis/redis_script_template.py

@@ -11,7 +11,7 @@
 # fast_execute_script接口固定参数
 # 这里独立出来，遇到过全局变量被其他db修改，导致用户错乱的问题
 redis_fast_execute_script_common_kwargs = {
-    "timeout": 3600,
+    "timeout": 10800,
     "account_alias": "root",
     "is_param_sensitive": 0,
 }