fix(mongodb): mongodb迁移flow增加节点 TencentBlueKing#8013

dbm-ui/backend/flow/consts.py

@@ -293,6 +293,7 @@ class MediumEnum(str, StructuredEnum):
     MongoDB = EnumField("mongodb", _("mongodb"))
     MongoToolKit = EnumField("mongo-toolkit", _("Mongo 工具集"))
     MongoMonitor = EnumField("mongo-monitor", _("Mongo 监控"))
+    DBTools = EnumField("dbtools", _("工具包"))
     Doris = EnumField("doris", _("doris"))
     Vm = EnumField("vm", _("vm"))
 
@@ -451,7 +452,6 @@ class RedisActuatorActionEnum(str, StructuredEnum):
     Install = EnumField("install", _("install"))
     REPLICA_BATCH = EnumField("replica_batch", _("replica_batch"))
     Replicaof = EnumField("replicaof", _("replicaof"))
-    LocalReDoDR = EnumField("local_redo_dr", _("local_redo_dr"))
     CLUSTER_MEET = EnumField("clustermeet_slotsassign", _("clustermeet_slotsassign"))
     KEYS_PATTERN = EnumField("keyspattern", _("keyspattern"))
     KEYS_DELETE_REGEX = EnumField("keysdelete_regex", _("keysdelete_regex"))

dbm-ui/backend/flow/engine/bamboo/scene/mongodb/mongodb_install.py

@@ -107,7 +107,7 @@ def install_dbmon(self, data: dict, pipeline: Builder):
         ip_list = self.get_kwargs.payload["hosts"]
         exec_ips = [host["ip"] for host in ip_list]
         bk_cloud_id = ip_list[0]["bk_cloud_id"]
-        add_install_dbmon(self, data, pipeline, exec_ips, bk_cloud_id, allow_empty_instance=True)
+        add_install_dbmon(self.root_id, data, pipeline, exec_ips, bk_cloud_id, allow_empty_instance=True)
 
     def multi_replicaset_install_flow(self):
         """

dbm-ui/backend/flow/engine/bamboo/scene/mongodb/mongodb_install_dbmon.py

@@ -56,7 +56,7 @@ def get_pkg_info():
     }
 
 
-def add_install_dbmon(flow, flow_data, pipeline, iplist, bk_cloud_id, allow_empty_instance=False):
+def add_install_dbmon(root_id, flow_data, pipeline, iplist, bk_cloud_id, allow_empty_instance=False):
     """
     allow_empty_instance 上架流程中，允许ip没有实例. allow_empty_instance = True
     """
@@ -87,7 +87,7 @@ def add_install_dbmon(flow, flow_data, pipeline, iplist, bk_cloud_id, allow_empt
     for ip in iplist:
         nodes = instances_by_ip[ip]
         sub_pl, sub_bk_host_list = InstallDBMonSubTask.process_server(
-            root_id=flow.root_id,
+            root_id=root_id,
             flow_data=flow_data,
             ip=ip,
             bk_cloud_id=bk_cloud_id,

dbm-ui/backend/flow/engine/bamboo/scene/mongodb/sub_task/migrate_meta.py

@@ -15,10 +15,16 @@
 from django.utils.translation import ugettext as _
 
 from backend.db_meta.enums.cluster_type import ClusterType
+from backend.flow.consts import DEPENDENCIES_PLUGINS
 from backend.flow.engine.bamboo.scene.common.builder import SubBuilder
+from backend.flow.engine.bamboo.scene.mongodb.mongodb_install_dbmon import add_install_dbmon
+from backend.flow.plugins.components.collections.common.install_nodeman_plugin import (
+    InstallNodemanPluginServiceComponent,
+)
 from backend.flow.plugins.components.collections.mongodb.add_relationship_to_meta import (
     ExecAddRelationshipOperationComponent,
 )
+from backend.flow.plugins.components.collections.mongodb.exec_actuator_job import ExecuteDBActuatorJobComponent
 from backend.flow.plugins.components.collections.mongodb.migrate_meta import MongoDBMigrateMetaComponent
 from backend.flow.utils.mongodb.mongodb_migrate_dataclass import MigrateActKwargs
 
@@ -104,6 +110,37 @@ def cluster_migrate(
         name = "replicaset"
         cluster_name = sub_get_kwargs.source_cluster_info["replsetname"]
 
+    # 安装蓝鲸插件
+    acts_list = []
+    for plugin_name in DEPENDENCIES_PLUGINS:
+        acts_list.append(
+            {
+                "act_name": _("安装[{}]插件".format(plugin_name)),
+                "act_component_code": InstallNodemanPluginServiceComponent.code,
+                "kwargs": sub_get_kwargs.get_install_plugin_info(plugin_name=plugin_name),
+            }
+        )
+    sub_pipeline.add_parallel_acts(acts_list=acts_list)
+
+    # os初始化
+    kwargs = sub_get_kwargs.get_os_init_info()
+    sub_pipeline.add_act(
+        act_name=_("MongoDB-机器初始化"), act_component_code=ExecuteDBActuatorJobComponent.code, kwargs=kwargs
+    )
+
+    # 关闭老的dbmon
+    kwargs = sub_get_kwargs.get_stop_old_dbmon_info()
+    sub_pipeline.add_act(
+        act_name=_("MongoDB-关闭老dbmon"), act_component_code=ExecuteDBActuatorJobComponent.code, kwargs=kwargs
+    )
+
+    # 安装新的dbmon
+    ip_list = sub_get_kwargs.hosts
+    exec_ips = [host["ip"] for host in ip_list]
+    add_install_dbmon(
+        root_id, sub_get_kwargs.payload, sub_pipeline, exec_ips, sub_get_kwargs.bk_cloud_id, allow_empty_instance=True
+    )
+
     return sub_pipeline.build_sub_process(
         sub_name=_("MongoDB--{}-meta迁移-{}-{}".format(name, sub_get_kwargs.payload["app"], cluster_name))
     )

dbm-ui/backend/flow/plugins/components/collections/mongodb/exec_actuator_job.py

@@ -27,10 +27,11 @@
 from backend.flow.plugins.components.collections.common.base_service import BkJobService
 from backend.flow.utils.mongodb.mongodb_dataclass import ActKwargs
 from backend.flow.utils.mongodb.mongodb_script_template import (
+    make_script_common_kwargs,
     mongodb_actuator_template,
     mongodb_create_actuator_dir_template,
-    mongodb_fast_execute_script_common_kwargs,
     mongodb_os_init_actuator_template,
+    mongodb_stop_old_dbmon_script,
 )
 
 logger = logging.getLogger("json")
@@ -170,11 +171,15 @@ def _execute(self, data, parent_data) -> bool:
 
         # 脚本内容
         jinja_env = Environment()
+        exec_account = "root"
         template = jinja_env.from_string(mongodb_actuator_template)
         if kwargs.get("init_flag", False):
             template = jinja_env.from_string(mongodb_os_init_actuator_template)
         elif kwargs.get("create_dir", False):
             template = jinja_env.from_string(mongodb_create_actuator_dir_template)
+        elif kwargs.get("stop_old_dbmon", False):
+            template = jinja_env.from_string(mongodb_stop_old_dbmon_script)
+            exec_account = "mysql"
         body = {
             "bk_biz_id": env.JOB_BLUEKING_BIZ_ID,
             "task_name": f"DBM_{node_name}_{node_id}",
@@ -184,9 +189,11 @@ def _execute(self, data, parent_data) -> bool:
         }
 
         self.log_info(
-            "[{}] ready start task with body {} {}".format(node_name, mongodb_fast_execute_script_common_kwargs, body)
+            "[{}] ready start task with body {} {}".format(
+                node_name, make_script_common_kwargs(exec_account=exec_account), body
+            )
         )
-        resp = JobApi.fast_execute_script({**mongodb_fast_execute_script_common_kwargs, **body}, raw=True)
+        resp = JobApi.fast_execute_script({**make_script_common_kwargs(exec_account=exec_account), **body}, raw=True)
 
         # 传入调用结果，并单调监听任务状态
         data.outputs.ext_result = resp

dbm-ui/backend/flow/utils/mongodb/mongodb_migrate_dataclass.py

@@ -10,15 +10,25 @@
 """
 from dataclasses import dataclass
 
-from backend.components.dbconfig.constants import OpType
+from backend.components import DBConfigApi
+from backend.components.dbconfig.constants import FormatType, LevelName, OpType
 from backend.configuration.constants import DBType
 from backend.db_meta.enums.cluster_type import ClusterType
 from backend.db_meta.enums.instance_role import InstanceRole
 from backend.db_meta.enums.spec import SpecClusterType, SpecMachineType
 from backend.db_meta.models import Machine
 from backend.db_meta.models.spec import Spec
-from backend.flow.consts import DEFAULT_DB_MODULE_ID, ConfigTypeEnum, MongoDBManagerUser
+from backend.flow.consts import (
+    DEFAULT_DB_MODULE_ID,
+    ConfigFileEnum,
+    ConfigTypeEnum,
+    MongoDBActuatorActionEnum,
+    MongoDBManagerUser,
+    NameSpaceEnum,
+)
 from backend.flow.utils.mongodb.migrate_meta import MongoDBMigrateMeta
+from backend.flow.utils.mongodb.mongodb_dataclass import CommonContext
+from backend.flow.utils.mongodb.mongodb_password import MongoDBPassword
 
 
 @dataclass()
@@ -81,6 +91,10 @@ def __init__(self):
             MongoDBManagerUser.MonitorUser.value,
             MongoDBManagerUser.AppMonitorUser.value,
         ]
+        # 集群的主机
+        self.hosts: list = None
+        # 集群bk_cloud_id
+        self.bk_cloud_id: int = None
 
     def skip_machine(self):
         """副本集机器复用"""
@@ -346,6 +360,7 @@ def get_save_password_info(self) -> dict:
                 {"ip": node.get("ip"), "port": node.get("port"), "bk_cloud_id": node.get("bk_cloud_id")}
                 for node in self.source_cluster_info.get("storages")
             ]
+            self.hosts = nodes
             info["password_infos"].append(
                 {
                     "nodes": nodes,
@@ -370,12 +385,23 @@ def get_save_password_info(self) -> dict:
                 {"nodes": config_nodes, "password": self.source_cluster_info.get("configsvr").get("password")}
             )
             # shard
+            storages_hosts_set = set()
             for shard in self.source_cluster_info.get("storages"):
                 nodes = [
                     {"ip": node.get("ip"), "port": node.get("port"), "bk_cloud_id": node.get("bk_cloud_id")}
                     for node in shard.get("nodes")
                 ]
+                for node in nodes:
+                    storages_hosts_set.add(node["ip"])
                 info["password_infos"].append({"nodes": nodes, "password": shard.get("password")})
+            storages_hosts_list = [
+                {
+                    "ip": host.get("ip"),
+                    "bk_cloud_id": self.source_cluster_info.get("storages")[0]["nodes"][0]["bk_cloud_id"],
+                }
+                for host in storages_hosts_set
+            ]
+            self.hosts = proxie_nodes + config_nodes + storages_hosts_list
         return info
 
     def get_change_dns_app_info(self) -> dict:
@@ -439,3 +465,70 @@ def get_clb_info(self) -> dict:
             "created_by": self.source_cluster_info.get("mongodb_dbas").split(",")[0],
             "meta_func_name": MongoDBMigrateMeta.add_clb_domain.__name__,
         }
+
+    def get_os_init_info(self) -> dict:
+        """进行host初始化"""
+
+        # 获取os配置
+        os_conf = DBConfigApi.query_conf_item(
+            params={
+                "bk_biz_id": self.bk_biz_id,
+                "level_name": LevelName.APP,
+                "level_value": self.bk_biz_id,
+                "conf_file": ConfigFileEnum.OsConf.value,
+                "conf_type": ConfigTypeEnum.Config.value,
+                "namespace": NameSpaceEnum.MongoDBCommon.value,
+                "format": FormatType.MAP.value,
+            }
+        )["content"]
+        user = os_conf["user"]
+        file_path = os_conf["file_path"]
+        self.bk_cloud_id = self.hosts[0]["bk_cloud_id"]
+        # 获取os user密码
+        password = MongoDBPassword().get_password_from_db(
+            ip="0.0.0.0", port=0, bk_cloud_id=self.bk_cloud_id, username=user
+        )["password"]
+        return {
+            "init_flag": True,
+            "set_trans_data_dataclass": CommonContext.__name__,
+            "get_trans_data_ip_var": None,
+            "bk_cloud_id": self.bk_cloud_id,
+            "exec_ip": self.hosts,
+            "file_target_path": file_path + "/install",
+            "db_act_template": {
+                "action": MongoDBActuatorActionEnum.OsInit,
+                "file_path": file_path,
+                "user": user,
+                "group": os_conf["group"],
+                "data_dir": os_conf["install_dir"],
+                "backup_dir": os_conf["backup_dir"],
+                "payload": {
+                    "user": user,
+                    "password": password,
+                },
+            },
+        }
+
+    def get_stop_old_dbmon_info(self) -> dict:
+        """关闭集群老的dbmon"""
+
+        return {
+            "stop_old_dbmon": True,
+            "set_trans_data_dataclass": CommonContext.__name__,
+            "get_trans_data_ip_var": None,
+            "bk_cloud_id": self.bk_cloud_id,
+            "exec_ip": self.hosts,
+            "db_act_template": {
+                "payload": {},
+            },
+        }
+
+    def get_install_plugin_info(self, plugin_name: str) -> dict:
+        """安装蓝鲸插件"""
+
+        ips = [host["ip"] for host in self.hosts]
+        return {
+            "plugin_name": plugin_name,
+            "ips": ips,
+            "bk_cloud_id": self.bk_cloud_id,
+        }

dbm-ui/backend/flow/utils/mongodb/mongodb_script_template.py

@@ -155,3 +155,17 @@ def prepare_recover_dir_script(dest_dir: str) -> str:
 chown -R {} {}
 echo return code $?"""
     return script.format(dest_dir, "mysql", dest_dir)
+
+
+# 关闭老的dbmon
+mongodb_stop_old_dbmon_script = """
+source /home/mysql/.bash_profile
+/home/mysql/dbmon/stop.sh
+/home/mysql/filebeat-deploy/remove
+/home/mysql/filebeat-deploy/stop_watcher
+/home/mysql/filebeat-deploy/stop_filebeat
+num=`pgrep prome_exporter_manager |wc -l`
+if [[ $num > 0 ]]; then
+killall -9 prome_exporter_manager
+fi
+"""