feat(dbm-services): 导出数据在放到中转机器操作 #11134

ymakedaq · iSecloud · commit d24c4c6a4433 · 2025-06-27T19:50:25.000+08:00
diff --git a/dbm-services/mysql/db-tools/dbactuator/pkg/components/mysql/dbconsole_dump.go b/dbm-services/mysql/db-tools/dbactuator/pkg/components/mysql/dbconsole_dump.go
@@ -121,6 +121,10 @@ func (c *DbConsoleDumpComp) Init() (err error) {
 		User: user,
 		Pwd:  pwd,
 	}.Conn()
+	if err != nil {
+		logger.Error("connect to %s:%d failed,err:%s", host, port, err.Error())
+		return err
+	}
 	defer func() {
 		if conn != nil {
 			conn.Close()
diff --git a/dbm-services/mysql/db-tools/dbactuator/pkg/util/osutil/mountpoint.go b/dbm-services/mysql/db-tools/dbactuator/pkg/util/osutil/mountpoint.go
@@ -170,5 +170,8 @@ func GetMostSuitableMountPoint() (FileSystem, error) {
 			}
 		}
 	}
+	if maxAvFs == (FileSystem{}) {
+		return FileSystem{}, fmt.Errorf("no suitable data mount point found")
+	}
 	return maxAvFs, nil
 }
diff --git a/dbm-ui/backend/db_proxy/constants.py b/dbm-ui/backend/db_proxy/constants.py
@@ -38,6 +38,7 @@ class ExtensionType(str, StructuredEnum):
     DRS = EnumField("DRS", _("SQL 远程执行服务"))
     DBHA = EnumField("DBHA", _("数据库高可用服务"))
     REDIS_DTS = EnumField("REDIS_DTS", _("Redis DTS服务"))
+    CONSOLE_DUMP_CENTER = EnumField("CONSOLE_DUMP_CENTER", _("控制台转储中心"))
 
 
 class ExtensionServiceStatus(str, StructuredEnum):
diff --git a/dbm-ui/backend/flow/engine/bamboo/scene/mysql/dbconsole.py b/dbm-ui/backend/flow/engine/bamboo/scene/mysql/dbconsole.py
@@ -12,6 +12,7 @@
 from dataclasses import asdict
 from typing import Dict, Optional
 
+from django.utils.crypto import get_random_string
 from django.utils.translation import ugettext as _
 
 from backend.components.db_remote_service.client import DRSApi
@@ -20,12 +21,21 @@
 from backend.db_meta.enums.cluster_type import ClusterType
 from backend.db_meta.enums.instance_role import InstanceRole, TenDBClusterSpiderRole
 from backend.db_meta.models import Cluster, ProxyInstance, StorageInstance
-from backend.flow.consts import LONG_JOB_TIMEOUT
+from backend.db_proxy.constants import ExtensionType
+from backend.db_proxy.models import DBExtension
+from backend.flow.consts import ACCOUNT_PREFIX, LONG_JOB_TIMEOUT
 from backend.flow.engine.bamboo.scene.common.builder import Builder
 from backend.flow.engine.bamboo.scene.common.get_file_list import GetFileList
+from backend.flow.plugins.components.collections.mysql.create_user import AddUserComponent
+from backend.flow.plugins.components.collections.mysql.drop_user import DropUserComponent
 from backend.flow.plugins.components.collections.mysql.exec_actuator_script import ExecuteDBActuatorScriptComponent
 from backend.flow.plugins.components.collections.mysql.trans_flies import TransFileComponent
-from backend.flow.utils.mysql.mysql_act_dataclass import DownloadMediaKwargs, ExecActuatorKwargs
+from backend.flow.utils.mysql.mysql_act_dataclass import (
+    AddTempUserKwargs,
+    DownloadMediaKwargs,
+    DropUserKwargs,
+    ExecActuatorKwargs,
+)
 from backend.flow.utils.mysql.mysql_act_playload import MysqlActPayload
 
 logger = logging.getLogger("flow")
@@ -46,15 +56,53 @@ def __init__(self, root_id: str, data: Optional[Dict]):
         self.dbconsole_dump_file_name = self.data["dump_file_name"]
 
     def dump_flow(self):
-        p = Builder(root_id=self.root_id, data=self.data, need_random_pass_cluster_ids=[self.cluster_id])
         ro_instance_info = self.__get_read_instance(self.cluster)
         bk_cloud_id = ro_instance_info["bk_cloud_id"]
         exec_ip = ro_instance_info["ip"]
+        # 判断是否配置中转机
+        dump_center = DBExtension.get_latest_extension(
+            bk_cloud_id=bk_cloud_id, extension_type=ExtensionType.CONSOLE_DUMP_CENTER
+        )
+        if dump_center:
+            p = Builder(root_id=self.root_id, data=self.data)
+        else:
+            p = Builder(root_id=self.root_id, data=self.data, need_random_pass_cluster_ids=[self.cluster_id])
         # 此处可以根据延迟来考虑是否需要抛出错误
         if ro_instance_info["instance_role"] == InstanceRole.BACKEND_SLAVE:
             behind_master_sec = self.get_slave_delay_second(exec_ip, ro_instance_info["port"], bk_cloud_id)
             logger.info(f"slave delay sec: {behind_master_sec}")
 
+        flow_context = ro_instance_info
+        # 定义切换流程中用的账号密码，密码是随机生成16位字符串，并利用公钥进行加密
+        ran_str = get_random_string(length=16)
+        random_account = f"{ACCOUNT_PREFIX}{get_random_string(length=8)}"
+        if dump_center:
+            exec_ip = dump_center.details["ip"]
+            add_temp_user_kwargs = AddTempUserKwargs(
+                bk_cloud_id=bk_cloud_id,
+                hosts=[exec_ip],
+                user=random_account,
+                psw=ran_str,
+                address="{}{}{}".format(ro_instance_info["ip"], IP_PORT_DIVIDER, ro_instance_info["port"]),
+                dbname="%",
+                dml_ddl_priv="SELECT,SHOW VIEW,TRIGGER, EVENT",
+                global_priv="",
+            )
+            drop_user_kwargs = DropUserKwargs(
+                bk_cloud_id=bk_cloud_id,
+                host=exec_ip,
+                user=random_account,
+                address="{}{}{}".format(ro_instance_info["ip"], IP_PORT_DIVIDER, ro_instance_info["port"]),
+            )
+            p.add_act(
+                act_name=_("创建临时用户"),
+                act_component_code=AddUserComponent.code,
+                kwargs=asdict(add_temp_user_kwargs),
+            )
+            flow_context["dump_center"] = True
+            flow_context["random_account"] = random_account
+            flow_context["random_password"] = ran_str
+        # 下发db-actuator介质
         p.add_act(
             act_name=_("下发db-actuator介质"),
             act_component_code=TransFileComponent.code,
@@ -75,11 +123,13 @@ def dump_flow(self):
                     job_timeout=LONG_JOB_TIMEOUT,
                     bk_cloud_id=bk_cloud_id,
                     exec_ip=exec_ip,
-                    cluster=ro_instance_info,
+                    cluster=flow_context,
                     get_mysql_payload_func=MysqlActPayload.get_dbconsole_schema_payload.__name__,
                 )
             ),
         )
+        if dump_center:
+            p.add_act(act_name=_("删除临时用户"), act_component_code=DropUserComponent.code, kwargs=asdict(drop_user_kwargs))
         # 运行pipeine
         p.run_pipeline(is_drop_random_user=True)
 
diff --git a/dbm-ui/backend/flow/engine/bamboo/scene/mysql/mysql_master_fail_over.py b/dbm-ui/backend/flow/engine/bamboo/scene/mysql/mysql_master_fail_over.py
@@ -183,7 +183,7 @@ def master_fail_over_flow(self):
                         cluster_sw_kwargs.get_mysql_payload_func = MysqlActPayload.get_change_master_payload.__name__
                         acts_list.append(
                             {
-                                "act_name": _("salve节点同步新master数据"),
+                                "act_name": _("slave节点同步新master数据"),
                                 "act_component_code": ExecuteDBActuatorScriptComponent.code,
                                 "kwargs": asdict(cluster_sw_kwargs),
                             }
diff --git a/dbm-ui/backend/flow/engine/bamboo/scene/mysql/mysql_master_slave_switch.py b/dbm-ui/backend/flow/engine/bamboo/scene/mysql/mysql_master_slave_switch.py
@@ -347,7 +347,7 @@ def master_slave_switch_flow(self):
                     cluster_sw_kwargs.get_mysql_payload_func = MysqlActPayload.get_change_master_payload.__name__
                     acts_list.append(
                         {
-                            "act_name": _("salve节点同步新master数据"),
+                            "act_name": _("slave节点同步新master数据"),
                             "act_component_code": ExecuteDBActuatorScriptComponent.code,
                             "kwargs": asdict(cluster_sw_kwargs),
                         }
diff --git a/dbm-ui/backend/flow/engine/bamboo/scene/mysql/mysql_restore_slave_remote_flow.py b/dbm-ui/backend/flow/engine/bamboo/scene/mysql/mysql_restore_slave_remote_flow.py
@@ -690,7 +690,7 @@ def restore_local_slave_flow(self):
                 kwargs=asdict(InstanceUserCloneKwargs(clone_data=clone_data)),
             )
 
-            # 这里区分是standby还是普通salve添加域名
+            # 这里区分是standby还是普通slave添加域名
             if target_slave.is_stand_by:
                 domain_map = get_standby_dns(cluster_model.id)
                 domain_add_list = []
diff --git a/dbm-ui/backend/flow/plugins/components/collections/mysql/create_user.py b/dbm-ui/backend/flow/plugins/components/collections/mysql/create_user.py
@@ -57,3 +57,41 @@ class CreateUserComponent(Component):
     name = __name__
     code = "mysql_create_user"
     bound_service = CreateUserService
+
+
+class AddUserService(BaseService):
+    """
+    通用add user组件
+    """
+
+    def _execute(self, data, parent_data) -> bool:
+        kwargs = data.get_one_of_inputs("kwargs")
+        global_data = data.get_one_of_inputs("global_data")
+
+        try:
+            DBPrivManagerApi.add_priv_without_account_rule(
+                params={
+                    "bk_cloud_id": kwargs["bk_cloud_id"],
+                    "bk_biz_id": global_data["bk_biz_id"],
+                    "operator": global_data["created_by"],
+                    "user": kwargs["user"],
+                    "psw": kwargs["psw"],
+                    "hosts": kwargs["hosts"],
+                    "dbname": kwargs["dbname"],
+                    "dml_ddl_priv": kwargs["dml_ddl_priv"],
+                    "global_priv": kwargs["global_priv"],
+                    "address": kwargs["address"],
+                }
+            )
+            self.log_info(_("在「{}」创建临时用户「{}@{}」成功").format(kwargs["address"], kwargs["user"], kwargs["hosts"]))
+        except Exception as e:  # pylint: disable=broad-except
+            self.log_error(_("创建用户接口异常，相关信息: {}").format(e))
+            return False
+
+        return True
+
+
+class AddUserComponent(Component):
+    name = __name__
+    code = "mysql_add_user"
+    bound_service = AddUserService
diff --git a/dbm-ui/backend/flow/utils/mysql/mysql_act_playload.py b/dbm-ui/backend/flow/utils/mysql/mysql_act_playload.py
@@ -2474,14 +2474,16 @@ def get_dbconsole_schema_payload(self, **kwargs):
         db_cloud_token = AsymmetricHandler.encrypt(
             name=AsymmetricCipherConfigType.PROXYPASS.value, content=f"{bk_cloud_id}_dbactuator_token"
         )
-
+        if self.cluster.get("dump_center"):
+            self.account["admin_user"] = self.cluster["random_account"]
+            self.account["admin_pwd"] = self.cluster["random_password"]
         return {
             "db_type": DBActuatorTypeEnum.MySQL.value,
             "action": DBActuatorActionEnum.MysqlDumpData.value,
             "payload": {
                 "general": {"runtime_account": self.account},
                 "extend": {
-                    "host": kwargs["ip"],
+                    "host": self.cluster["ip"],
                     "port": self.cluster["port"],
                     "charset": self.ticket_data["charset"],
                     "dump_detail": {

Original file line number	Diff line number	Diff line change
`@@ -170,5 +170,8 @@ func GetMostSuitableMountPoint() (FileSystem, error) {`
`170`	`170`	`}`
`171`	`171`	`}`
`172`	`172`	`}`
	`173`	`+ if maxAvFs == (FileSystem{}) {`
	`174`	`+ return FileSystem{}, fmt.Errorf("no suitable data mount point found")`
	`175`	`+ }`
`173`	`176`	`return maxAvFs, nil`
`174`	`177`	`}`
Original file line number	Diff line number	Diff line change
`@@ -183,7 +183,7 @@ def master_fail_over_flow(self):`
`183`	`183`	`cluster_sw_kwargs.get_mysql_payload_func = MysqlActPayload.get_change_master_payload.__name__`
`184`	`184`	`acts_list.append(`
`185`	`185`	`{`
`186`		`- "act_name": _("salve节点同步新master数据"),`
	`186`	`+ "act_name": _("slave节点同步新master数据"),`
`187`	`187`	`"act_component_code": ExecuteDBActuatorScriptComponent.code,`
`188`	`188`	`"kwargs": asdict(cluster_sw_kwargs),`
`189`	`189`	`}`
Original file line number	Diff line number	Diff line change
`@@ -347,7 +347,7 @@ def master_slave_switch_flow(self):`
`347`	`347`	`cluster_sw_kwargs.get_mysql_payload_func = MysqlActPayload.get_change_master_payload.__name__`
`348`	`348`	`acts_list.append(`
`349`	`349`	`{`
`350`		`- "act_name": _("salve节点同步新master数据"),`
	`350`	`+ "act_name": _("slave节点同步新master数据"),`
`351`	`351`	`"act_component_code": ExecuteDBActuatorScriptComponent.code,`
`352`	`352`	`"kwargs": asdict(cluster_sw_kwargs),`
`353`	`353`	`}`
Original file line number	Diff line number	Diff line change
`@@ -690,7 +690,7 @@ def restore_local_slave_flow(self):`
`690`	`690`	`kwargs=asdict(InstanceUserCloneKwargs(clone_data=clone_data)),`
`691`	`691`	`)`
`692`	`692`
`693`		`- # 这里区分是standby还是普通salve添加域名`
	`693`	`+ # 这里区分是standby还是普通slave添加域名`
`694`	`694`	`if target_slave.is_stand_by:`
`695`	`695`	`domain_map = get_standby_dns(cluster_model.id)`
`696`	`696`	`domain_add_list = []`