fix(redis): load modules修复配置更新bug和libB2RedisModule加载时需要关闭aof问题 #6643

dbm-services/redis/db-tools/dbactuator/pkg/atomjobs/atomproxy/predixy_add_modules_cmds.go

@@ -10,10 +10,10 @@ import (
 
 	"github.com/go-playground/validator/v10"
 
-	"dbm-services/mongodb/db-tools/dbmon/util"
 	"dbm-services/redis/db-tools/dbactuator/models/myredis"
 	"dbm-services/redis/db-tools/dbactuator/pkg/consts"
 	"dbm-services/redis/db-tools/dbactuator/pkg/jobruntime"
+	"dbm-services/redis/db-tools/dbactuator/pkg/util"
 )
 
 // PredixyAddModulesCmdsParams 参数
@@ -68,7 +68,7 @@ func (job *PredixyAddModulesCmds) Init(m *jobruntime.JobGenericRuntime) error {
 
 // Name 原子任务名
 func (job *PredixyAddModulesCmds) Name() string {
-	return "predixy_add_modules_cmds"
+	return "redis_predixy_add_modules_cmds"
 }
 
 // Run 执行
@@ -98,11 +98,19 @@ func (job *PredixyAddModulesCmds) Run() (err error) {
 			job.params.LoadModules)
 		return nil
 	}
+	// 备份配置文件
+	bakConfFile := job.configFile + "_old_" + time.Now().Format(consts.FilenameTimeLayout)
+	cpCmd := fmt.Sprintf(`cp %s %s`, job.configFile, bakConfFile)
+	job.runtime.Logger.Info(cpCmd)
+	_, err = util.RunBashCmd(cpCmd, "", nil, 10*time.Second)
+	if err != nil {
+		return err
+	}
 	// 删除配置文件中 CustomCommand 配置中的内容
 	// sed命令意思是删除 CustomCommand 到 ###### 所有行
 	sedCmd := fmt.Sprintf(`sed -i '/CustomCommand/,/######/d' %s`, job.configFile)
 	job.runtime.Logger.Info(sedCmd)
-	_, err = util.RunBashCmd(sedCmd, job.configFile, nil, 10*time.Second)
+	_, err = util.RunBashCmd(sedCmd, "", nil, 10*time.Second)
 	if err != nil {
 		return err
 	}
@@ -114,7 +122,7 @@ cat >>%s<<EOF
 EOF
 	`, job.configFile, customCmdData)
 	job.runtime.Logger.Info(catCmd)
-	_, err = util.RunBashCmd(catCmd, job.configFile, nil, 10*time.Second)
+	_, err = util.RunBashCmd(catCmd, "", nil, 10*time.Second)
 	if err != nil {
 		return err
 	}
@@ -135,12 +143,12 @@ EOF
 // (对于用户自定义的module,很多时候我们也不知道对应command,此时不用在predixy的customCommands中增加这些module的命令)
 // moduleCmdInFile: true 表示 在配置文件中包含该module的命令.
 func (job *PredixyAddModulesCmds) IsModuleCmdInConfFile(confFile, module string) (knownModule, moduleCmdInFile bool) {
-	if module != consts.ModuleRedisBloom && module != consts.ModuleRedisJson && module != consts.ModuleRedisCell {
-		// 目前只认识这三个module的命令
+	if !consts.IsKnownModule(module) {
 		return false, false
 	}
 	knownModule = true
-	grepCmd := fmt.Sprintf(`grep -i %q %s`, module, confFile)
+	firstCmd := consts.GetFirstCommandByModule(module)
+	grepCmd := fmt.Sprintf(`grep -i %q %s`, firstCmd, confFile)
 	ret, _ := util.RunBashCmd(grepCmd, "", nil, 10*time.Second)
 	if ret != "" {
 		moduleCmdInFile = true

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

@@ -493,6 +493,10 @@ func (job *RedisInstall) getRedisConfTemplate() error {
 	// 加载module
 	soFile := ""
 	for _, moduleItem := range job.params.LoadModulesDetail {
+		if strings.Contains(moduleItem.SoFile, "libB2RedisModule") {
+			// libB2RedisModule 相关module需要关闭 aof,否则会报错
+			sb.WriteString("appendonly no\n")
+		}
 		soFile = filepath.Join(consts.RedisModulePath, moduleItem.SoFile)
 		sb.WriteString("loadmodule " + soFile + "\n")
 	}

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

@@ -5,6 +5,7 @@ import (
 	"errors"
 	"fmt"
 	"path/filepath"
+	"strings"
 	"time"
 
 	"github.com/go-playground/validator/v10"
@@ -88,16 +89,23 @@ func (job *RedisLoadModules) Run() (err error) {
 	if err != nil {
 		return err
 	}
+	// 加载b2 module时,需要先关闭aof
+	err = job.disableAofWhenB2Module()
+	if err != nil {
+		return err
+	}
+
 	soFile := ""
 	for _, cli := range job.AddrMapCli {
 		for _, moduleItem := range job.params.LoadModulesDetail {
 			soFile = filepath.Join(consts.RedisModulePath, moduleItem.SoFile)
 			err = cli.ModuleLoad(soFile)
-			if err != nil {
+			if err != nil && !strings.Contains(err.Error(), "loading the extension") {
 				job.runtime.Logger.Error(fmt.Sprintf("redis:%s load module(%s) fail,err:%v",
 					cli.Addr, moduleItem.SoFile, err))
 				return
 			}
+			job.runtime.Logger.Info(fmt.Sprintf("redis:%s load module(%s) success", cli.Addr, moduleItem.SoFile))
 		}
 	}
 	for addr, cli := range job.AddrMapCli {
@@ -121,6 +129,26 @@ func (job *RedisLoadModules) Run() (err error) {
 	return nil
 }
 
+// disableAofWhenB2Module 如果需要加载 libB2RedisModule,则先关闭aof
+func (job *RedisLoadModules) disableAofWhenB2Module() (err error) {
+	if len(job.params.LoadModulesDetail) == 0 {
+		return nil
+	}
+	for _, moduleItem := range job.params.LoadModulesDetail {
+		if strings.Contains(moduleItem.SoFile, "libB2RedisModule") {
+			for _, cli := range job.AddrMapCli {
+				// 关闭aof
+				_, err = cli.ConfigSet("appendonly", "no")
+				if err != nil {
+					return err
+				}
+			}
+			return nil
+		}
+	}
+	return nil
+}
+
 // isModulesSoFileExists 判断module so文件是否存在
 func (job *RedisLoadModules) isModulesSoFileExists() (err error) {
 	if len(job.params.LoadModulesDetail) == 0 {

dbm-services/redis/db-tools/dbactuator/pkg/consts/consts.go

@@ -1,6 +1,8 @@
 // Package consts 常量
 package consts
 
+import "strings"
+
 const (
 	// TendisTypePredixyRedisCluster predixy + RedisCluster架构
 	TendisTypePredixyRedisCluster = "PredixyRedisCluster"
@@ -241,6 +243,25 @@ const (
 	ModuleRedisJson  = "redisjson"
 )
 
+// IsKnownModule 是否是认识的module
+func IsKnownModule(module string) bool {
+	return module == ModuleRedisCell ||
+		module == ModuleRedisJson ||
+		module == ModuleRedisBloom
+}
+
+// ConfItemWithMultiLines  redis存在多行的配置项
+// 如:
+// rename-command config  xxxx
+// rename-command flushdb  yyyy
+func ConfItemWithMultiLines(confItem string) bool {
+	confItem = strings.ToLower(confItem)
+	return confItem == "rename-command" ||
+		confItem == "save" ||
+		confItem == "loadmodule" ||
+		confItem == "client-output-buffer-limit"
+}
+
 // IsClusterDbType 存储端是否是cluster类型
 func IsClusterDbType(dbType string) bool {
 	if dbType == TendisTypePredixyRedisCluster ||

dbm-services/redis/db-tools/dbactuator/pkg/consts/redis_modules.go

@@ -109,3 +109,20 @@ CustomCommand {
 	`)
 	return builder.String()
 }
+
+// GetFirstCommandByModule 获取module的第一个命令
+func GetFirstCommandByModule(module string) string {
+	if !IsKnownModule(module) {
+		return ""
+	}
+	if module == ModuleRedisBloom {
+		return RedisBloomCmdItems[0].Command
+	}
+	if module == ModuleRedisJson {
+		return RedisJsonCmdItems[0].Command
+	}
+	if module == ModuleRedisCell {
+		return RedisCellCmdItems[0].Command
+	}
+	return ""
+}

dbm-services/redis/db-tools/dbactuator/pkg/util/redisutil.go

@@ -195,27 +195,30 @@ func ClearUsrLocalPredixy(clearTarget bool) (err error) {
 // SaveKvToConfigFile 保存key/value到配置文件
 func SaveKvToConfigFile(confFile, item, value string) (err error) {
 	var ret string
-	grepCmd := fmt.Sprintf("grep -iP '^%s' %s|awk '{print $2}'", item, confFile)
+	grepCmd := fmt.Sprintf("grep -iP '^%s' %s|tail -1|awk '{print $2}'", item, confFile)
 	mylog.Logger.Info(grepCmd)
 	ret, _ = RunBashCmd(grepCmd, "", nil, 10*time.Second)
+	ret = strings.TrimPrefix(ret, "\"")
+	ret = strings.TrimSuffix(ret, "\"")
 	if ret == value {
+		// 如果存在,且值正确
 		return nil
 	}
-	if ret != "" {
-		// 如果存在,但值不对
-		// 先删除
-		sedCmd := fmt.Sprintf("sed -i -e '/^%s/d' %s", item, confFile)
-		mylog.Logger.Info(sedCmd)
-		_, err = RunBashCmd(sedCmd, "", nil, 10*time.Second)
-		// 再添加
+	if ret == "" || consts.ConfItemWithMultiLines(item) {
+		// 如果配置文件中不存在该配置项 或者 该配置项是多行配置项,则直接添加
 		echoCmd := fmt.Sprintf("echo '%s %s' >> %s", item, value, confFile)
 		mylog.Logger.Info(echoCmd)
 		_, err = RunBashCmd(echoCmd, "", nil, 10*time.Second)
-	} else {
-		// 直接添加
-		echoCmd := fmt.Sprintf("echo '%s %s' >> %s", item, value, confFile)
-		mylog.Logger.Info(echoCmd)
-		_, err = RunBashCmd(echoCmd, "", nil, 10*time.Second)
-	}
+		return err
+	}
+	// 如果存在,但值不对
+	// 先删除
+	sedCmd := fmt.Sprintf("sed -i -e '/^%s/d' %s", item, confFile)
+	mylog.Logger.Info(sedCmd)
+	_, err = RunBashCmd(sedCmd, "", nil, 10*time.Second)
+	// 再添加
+	echoCmd := fmt.Sprintf("echo '%s %s' >> %s", item, value, confFile)
+	mylog.Logger.Info(echoCmd)
+	_, err = RunBashCmd(echoCmd, "", nil, 10*time.Second)
 	return err
 }

dbm-services/redis/redis-dts/pkg/dtsJob/base.go

@@ -191,9 +191,9 @@ func (job *DtsJobBase) BgDtsTaskRunnerWithoutLimit(taskType, dbType string) {
 
 // BgOldRunningSyncTaskWatcher 目的:
 // 很多时候 redis-sync 已经拉起,状态为runnig(taskrow.status==1 taskrow.taskType="makeSync")
-// 而此时我们需要暂停 dbm-services/redis/redis-dts 升级 dbm-services/redis/redis-dts的介质
-// 再次拉起后, 以前(taskrow.status==1 taskrow.taskType="makeSync")的task其相关状态依然需要我们不断watch
-// 注意: 该函数只在 dbm-services/redis/redis-dts 被拉起时执行,启动goroutine监听属于当前dts_server的属于running状态的tasks
+// 而此时我们需要关闭 redis_dts_server进程，进行 redis_dts_server 的介质升级
+// 再次拉起redis_dts_server进程后, 以前处于增量同步的(taskrow.status==1 taskrow.taskType="makeSync")的task其相关状态依然需要我们不断watch
+// 注意: 该函数只在 redis_dts_server 被拉起时执行一次,启动goroutine监听处于 增量同步的(taskrow.status==1 taskrow.taskType="makeSync")的task
 // 对于后续新增的 (taskrow.status==1 taskrow.taskType="makeSync")的task,不归该函数处理
 func (job *DtsJobBase) BgOldRunningSyncTaskWatcher(taskType, dbType string, status int) {
 	limit := 100000
@@ -243,7 +243,7 @@ func (job *DtsJobBase) IsMyselfInBlacklist() bool {
 }
 
 // CheckSrcSlaveServerConcurrency 检查源slave机器是否还能新增迁移task
-// 如源slave机器上有20个redis,同时启动迁移是危险的,需做并发控制
+// 如源slave机器上有20个redis,同时启动迁移所有redis是危险的,需做并发控制
 func (job *DtsJobBase) CheckSrcSlaveServerConcurrency(taskRow *tendisdb.TbTendisDTSTask, taskTypes []string) (ok bool,
 	err error) {
 

dbm-ui/backend/db_services/redis/redis_modules/migrations/0003_alter_tbredismodulesupport_so_file.py

@@ -0,0 +1,18 @@
+# Generated by Django 3.2.25 on 2024-09-04 15:59
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ("redis_modules", "0002_auto_20240827_1912"),
+    ]
+
+    operations = [
+        migrations.AlterField(
+            model_name="tbredismodulesupport",
+            name="so_file",
+            field=models.CharField(default="", max_length=64, verbose_name="so文件名"),
+        ),
+    ]

dbm-ui/backend/db_services/redis/redis_modules/models/redis_module_support.py

@@ -12,7 +12,7 @@
 class TbRedisModuleSupport(models.Model):
     major_version = models.CharField(_("主版本号"), max_length=32, default="")
     module_name = models.CharField(_("module名"), max_length=32, default="")
-    so_file = models.CharField(_("so文件名"), max_length=32, default="")
+    so_file = models.CharField(_("so文件名"), max_length=64, default="")
 
     class Meta:
         verbose_name = _("Redis module支持")

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

@@ -2312,6 +2312,9 @@ def redis_cluster_add_modules_update_dbconfig(self, cluster_map: dict) -> dict:
         dst_conf_upsert_items = [
             {"conf_name": "loadmodule", "conf_value": ",".join(load_modules), "op_type": "update"}
         ]
+        # 如果 load_modules中包含 jlsy-b2,则 appendonly 为no
+        if "jlsy-b2" in load_modules:
+            dst_conf_upsert_items.append({"conf_name": "appendonly", "conf_value": "no", "op_type": "update"})
 
         upsert_param = {
             "conf_file_info": {