feat: restoreFlow流程代码结构优化 #11975

Author: zfrendo

dbm-ui/backend/flow/engine/bamboo/scene/mysql/common/recover_slave_instance.py renamed to dbm-ui/backend/flow/engine/bamboo/scene/mysql/common/mysql_resotre_data_remote_sub_flow.py

@@ -9,8 +9,8 @@
 specific language governing permissions and limitations under the License.
 """
 import copy
-import logging
-import os.path
+import logging.config
+import os
 from dataclasses import asdict
 from datetime import datetime
 
@@ -36,12 +36,218 @@
     ExecActuatorKwargs,
 )
 from backend.flow.utils.mysql.mysql_act_playload import MysqlActPayload
+from backend.flow.utils.spider.tendb_cluster_info import get_remotedb_info
 
 logger = logging.getLogger("flow")
 
 
+def remote_instance_migrate_sub_flow(root_id: str, ticket_data: dict, cluster_info: dict):
+    """
+    定义 tendbHa tendbCluster remote节点 主从迁移数据恢复 流程。实例级别迁移,使用远程备份文件 (只做流程,元数据请在主流程控制)
+    @param root_id:  flow流程的root_id
+    @param ticket_data: 关联单据 ticket对象
+    @param cluster_info:  关联的cluster对象
+    """
+
+    sub_pipeline = SubBuilder(root_id=root_id, data=ticket_data)
+    #  已经安装好的2个ip，需要导入同步数据
+    # 下发dbactor》通过master/slave 获取备份的文件》判断备份文件》恢复数据》change master
+    cluster = {
+        "cluster_id": cluster_info["cluster_id"],
+        "master_ip": cluster_info["master_ip"],
+        "slave_ip": cluster_info["slave_ip"],
+        "master_port": cluster_info["master_port"],
+        "new_master_ip": cluster_info["new_master_ip"],
+        "new_slave_ip": cluster_info["new_slave_ip"],
+        "new_slave_port": cluster_info["new_slave_port"],
+        "new_master_port": cluster_info["new_master_port"],
+        "bk_cloud_id": cluster_info["bk_cloud_id"],
+        "file_target_path": cluster_info["file_target_path"],
+        "change_master_force": cluster_info["change_master_force"],
+        "backupinfo": cluster_info["backupinfo"],
+        "charset": cluster_info["charset"],
+    }
+    exec_act_kwargs = ExecActuatorKwargs(
+        bk_cloud_id=int(cluster["bk_cloud_id"]), cluster_type=ClusterType.TenDBCluster, cluster=cluster
+    )
+    exec_act_kwargs.get_mysql_payload_func = MysqlActPayload.mysql_mkdir_dir.__name__
+    exec_act_kwargs.exec_ip = [cluster["new_slave_ip"], cluster["new_master_ip"]]
+    sub_pipeline.add_act(
+        act_name=_("创建目录 {}".format(cluster["file_target_path"])),
+        act_component_code=ExecuteDBActuatorScriptComponent.code,
+        kwargs=asdict(exec_act_kwargs),
+    )
+
+    sub_pipeline.add_act(
+        act_name=_("下发db-actor到节点"),
+        act_component_code=TransFileComponent.code,
+        kwargs=asdict(
+            DownloadMediaKwargs(
+                bk_cloud_id=int(cluster["bk_cloud_id"]),
+                exec_ip=[cluster["master_ip"], cluster["new_slave_ip"], cluster["new_master_ip"]],
+                file_list=GetFileList(db_type=DBType.MySQL).get_db_actuator_package(),
+            )
+        ),
+    )
+
+    backup_info = cluster["backupinfo"]
+    #  主从并发下载备份介质 下载为异步下载，定时调起接口扫描下载结果
+    task_ids = [i["task_id"] for i in backup_info["file_list_details"]]
+    download_sub_pipeline_list = []
+    download_kwargs = DownloadBackupFileKwargs(
+        bk_cloud_id=cluster["bk_cloud_id"],
+        task_ids=task_ids,
+        dest_ip=cluster["new_master_ip"],
+        dest_dir=cluster["file_target_path"],
+        reason="spider remote node sync data",
+    )
+    download_sub_pipeline_list.append(
+        {
+            "act_name": _("下载全库备份介质到 {}".format(cluster["new_master_ip"])),
+            "act_component_code": MySQLDownloadBackupfileComponent.code,
+            "kwargs": asdict(download_kwargs),
+        }
+    )
+    download_kwargs.dest_ip = cluster["new_slave_ip"]
+    download_sub_pipeline_list.append(
+        {
+            "act_name": _("下载全库备份介质到 {}".format(cluster["new_slave_ip"])),
+            "act_component_code": MySQLDownloadBackupfileComponent.code,
+            "kwargs": asdict(download_kwargs),
+        }
+    )
+    sub_pipeline.add_parallel_acts(download_sub_pipeline_list)
+
+    # 阶段4 恢复数据remote主从节点的数据
+    restore_list = []
+    cluster["restore_ip"] = cluster["new_master_ip"]
+    cluster["restore_port"] = cluster["new_master_port"]
+    cluster["source_ip"] = cluster["master_ip"]
+    cluster["source_port"] = cluster["master_port"]
+    cluster["change_master"] = False
+    exec_act_kwargs.exec_ip = cluster["new_master_ip"]
+    exec_act_kwargs.job_timeout = MYSQL_DATA_RESTORE_TIME
+    exec_act_kwargs.cluster = copy.deepcopy(cluster)
+    exec_act_kwargs.get_mysql_payload_func = MysqlActPayload.tendb_restore_remotedb_payload.__name__
+    restore_list.append(
+        {
+            "act_name": _("恢复新主节点数据 {}:{}".format(exec_act_kwargs.exec_ip, cluster["restore_port"])),
+            "act_component_code": ExecuteDBActuatorScriptComponent.code,
+            "kwargs": asdict(exec_act_kwargs),
+            "write_payload_var": "change_master_info",
+        }
+    )
+
+    cluster["restore_ip"] = cluster["new_slave_ip"]
+    cluster["restore_port"] = cluster["new_slave_port"]
+    cluster["source_ip"] = cluster["master_ip"]
+    cluster["source_port"] = cluster["master_port"]
+    cluster["change_master"] = False
+    exec_act_kwargs.cluster = copy.deepcopy(cluster)
+    exec_act_kwargs.exec_ip = cluster["new_slave_ip"]
+    exec_act_kwargs.get_mysql_payload_func = MysqlActPayload.tendb_restore_remotedb_payload.__name__
+    restore_list.append(
+        {
+            "act_name": _("恢复新从节点数据 {}:{}".format(exec_act_kwargs.exec_ip, cluster["restore_port"])),
+            "act_component_code": ExecuteDBActuatorScriptComponent.code,
+            "kwargs": asdict(exec_act_kwargs),
+        }
+    )
+    sub_pipeline.add_parallel_acts(acts_list=restore_list)
+
+    # 阶段5 change master: 新从库指向新主库
+    cluster["target_ip"] = cluster["new_master_ip"]
+    cluster["target_port"] = cluster["new_master_port"]
+    cluster["repl_ip"] = cluster["new_slave_ip"]
+    exec_act_kwargs.cluster = copy.deepcopy(cluster)
+    exec_act_kwargs.exec_ip = cluster["new_master_ip"]
+    exec_act_kwargs.job_timeout = MYSQL_USUAL_JOB_TIME
+    exec_act_kwargs.get_mysql_payload_func = MysqlActPayload.tendb_grant_remotedb_repl_user.__name__
+    sub_pipeline.add_act(
+        act_name=_("新增repl帐户{}".format(exec_act_kwargs.exec_ip)),
+        act_component_code=ExecuteDBActuatorScriptComponent.code,
+        kwargs=asdict(exec_act_kwargs),
+        write_payload_var="show_master_status_info",
+    )
+
+    cluster["repl_ip"] = cluster["new_slave_ip"]
+    cluster["repl_port"] = cluster["new_slave_port"]
+    cluster["target_ip"] = cluster["new_master_ip"]
+    cluster["target_port"] = cluster["new_master_port"]
+    cluster["change_master_type"] = MysqlChangeMasterType.MASTERSTATUS.value
+    exec_act_kwargs.cluster = copy.deepcopy(cluster)
+    exec_act_kwargs.exec_ip = cluster["new_slave_ip"]
+    exec_act_kwargs.get_mysql_payload_func = MysqlActPayload.tendb_remotedb_change_master.__name__
+    sub_pipeline.add_act(
+        act_name=_("建立主从关系:新从库指向新主库 {} {}:".format(exec_act_kwargs.exec_ip, cluster["repl_port"])),
+        act_component_code=ExecuteDBActuatorScriptComponent.code,
+        kwargs=asdict(exec_act_kwargs),
+    )
+
+    # 阶段6 change master: 新主库指向旧主库
+    cluster["target_ip"] = cluster["master_ip"]
+    cluster["target_port"] = cluster["master_port"]
+    cluster["repl_ip"] = cluster["new_master_ip"]
+    exec_act_kwargs.cluster = copy.deepcopy(cluster)
+    exec_act_kwargs.exec_ip = cluster["master_ip"]
+    exec_act_kwargs.get_mysql_payload_func = MysqlActPayload.tendb_grant_remotedb_repl_user.__name__
+    sub_pipeline.add_act(
+        act_name=_("新增repl帐户{}".format(exec_act_kwargs.exec_ip)),
+        act_component_code=ExecuteDBActuatorScriptComponent.code,
+        kwargs=asdict(exec_act_kwargs),
+    )
+
+    cluster["repl_ip"] = cluster["new_master_ip"]
+    cluster["repl_port"] = cluster["new_master_port"]
+    cluster["target_ip"] = cluster["master_ip"]
+    cluster["target_port"] = cluster["master_port"]
+    cluster["change_master_type"] = MysqlChangeMasterType.BACKUPFILE.value
+    exec_act_kwargs.cluster = copy.deepcopy(cluster)
+    exec_act_kwargs.exec_ip = cluster["new_master_ip"]
+    exec_act_kwargs.get_mysql_payload_func = MysqlActPayload.tendb_remotedb_change_master.__name__
+    sub_pipeline.add_act(
+        act_name=_("建立主从关系:新主库指向旧主库 {}:{}".format(exec_act_kwargs.exec_ip, cluster["repl_port"])),
+        act_component_code=ExecuteDBActuatorScriptComponent.code,
+        kwargs=asdict(exec_act_kwargs),
+    )
+    return sub_pipeline.build_sub_process(sub_name=_("RemoteDB主从节点成对迁移子流程{}".format(exec_act_kwargs.exec_ip)))
+
+
+def remote_node_uninstall_sub_flow(root_id: str, ticket_data: dict, ip: str):
+    """
+    卸载remotedb 指定ip节点下的所有实例
+    @param root_id: flow流程的root_id
+    @param ticket_data: 单据 data 对象
+    @param ip: 指定卸载的ip
+    """
+    sub_pipeline = SubBuilder(root_id=root_id, data=ticket_data)
+    cluster = {"uninstall_ip": ip, "bk_cloud_id": ticket_data["bk_cloud_id"]}
+    instances = get_remotedb_info(cluster["uninstall_ip"], cluster["bk_cloud_id"])
+    sub_pipeline_list = []
+    for instance in instances:
+        cluster["backend_port"] = instance["port"]
+        sub_pipeline_list.append(
+            {
+                "act_name": _("卸载MySQL实例:{}:{}".format(cluster["uninstall_ip"], cluster["backend_port"])),
+                "act_component_code": ExecuteDBActuatorScriptComponent.code,
+                "kwargs": asdict(
+                    ExecActuatorKwargs(
+                        exec_ip=cluster["uninstall_ip"],
+                        bk_cloud_id=cluster["bk_cloud_id"],
+                        cluster=cluster,
+                        get_mysql_payload_func=MysqlActPayload.get_uninstall_mysql_payload.__name__,
+                    )
+                ),
+            }
+        )
+
+    sub_pipeline.add_parallel_acts(acts_list=sub_pipeline_list)
+    return sub_pipeline.build_sub_process(sub_name=_("Remote node {} 卸载整机实例".format(cluster["uninstall_ip"])))
+
+
 def slave_recover_sub_flow(root_id: str, ticket_data: dict, cluster_info: dict):
     """
+    定义 tendbHa tendbCluster 从节点恢复流程，从远程备份下载备份文件.
     tendb remote slave 节点 恢复。(只做流程,元数据请在主流程控制)
     @param root_id:  flow流程的root_id
     @param ticket_data: 关联单据 ticket对象
@@ -175,6 +381,8 @@ def slave_recover_sub_flow(root_id: str, ticket_data: dict, cluster_info: dict):
 
 def priv_recover_sub_flow(root_id: str, ticket_data: dict, cluster_info: dict, ips: list):
     """
+    定义 tendbHa 从集群备份的主从节点恢复权限。这里主要用于恢复从节点权限,补充主从权限差异时slave重建直接从
+    主节点克隆权限可能有权限不全的问题。
     tendb privilege recover 指定实例权限恢复。
     @param root_id:  flow流程的root_id
     @param ticket_data: 关联单据 ticket对象
@@ -195,6 +403,7 @@ def priv_recover_sub_flow(root_id: str, ticket_data: dict, cluster_info: dict, i
 
     priv_file_task_ids = [file["task_id"] for file in backup_info["file_list"].values()]
     priv_files = [os.path.basename(file["file_name"]) for file in backup_info["file_list"].values()]
+    logger.info(priv_files)
     storages = cluster_model.storageinstance_set.filter(machine__ip__in=ips)
     priv_sub_pipeline_list = []
     for storage in storages:
@@ -246,7 +455,7 @@ def priv_recover_sub_flow(root_id: str, ticket_data: dict, cluster_info: dict, i
 
     if len(priv_sub_pipeline_list) > 0:
         sub_pipeline.add_parallel_sub_pipeline(sub_flow_list=priv_sub_pipeline_list)
-        return sub_pipeline.build_sub_process(sub_name=_("{}恢复权限".format(cluster_model.id)))
+        return sub_pipeline.build_sub_process(sub_name=_("集群{}恢复权限".format(cluster_model.id)))
     else:
         return None
     #  如果流程未空，如何？