From 2cac206d023c806c259086f55e87e5bc2854ac69 Mon Sep 17 00:00:00 2001 From: zfrendo <842557664@qq.com> Date: Thu, 5 Dec 2024 10:02:23 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E5=AE=9A=E7=82=B9=E5=9B=9E=E6=A1=A3?= =?UTF-8?q?=E7=89=A9=E7=90=86=E5=A4=87=E4=BB=BDchangeMasterTEST3=20#8338?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../scene/mysql/mysql_rollback_data_flow.py | 59 +++++++++++-------- 1 file changed, 33 insertions(+), 26 deletions(-) diff --git a/dbm-ui/backend/flow/engine/bamboo/scene/mysql/mysql_rollback_data_flow.py b/dbm-ui/backend/flow/engine/bamboo/scene/mysql/mysql_rollback_data_flow.py index 2aa97da612..14362c17d6 100644 --- a/dbm-ui/backend/flow/engine/bamboo/scene/mysql/mysql_rollback_data_flow.py +++ b/dbm-ui/backend/flow/engine/bamboo/scene/mysql/mysql_rollback_data_flow.py @@ -186,7 +186,8 @@ def rollback_data_flow(self): backupinfo = get_local_backup(inst_list, cluster_class, mycluster["rollback_time"]) if backupinfo is None: logger.error("cluster {} backup info not exists".format(cluster_class.id)) - raise TendbGetBackupInfoFailedException(message=_("获取集群 {} 的备份信息失败".format(cluster_class.id))) + raise TendbGetBackupInfoFailedException( + message=_("获取集群 {} 的备份信息失败".format(cluster_class.id))) mycluster["backupinfo"] = copy.deepcopy(backupinfo) sub_pipeline.add_sub_pipeline( @@ -278,6 +279,34 @@ def rollback_to_cluster_flow(self): db_module_id=self.data["db_module_id"], cluster_type=self.data["cluster_type"], ) + # 为了回档的备份信息各个节点统一，获取备份的信息提前 + backupinfo = copy.deepcopy(self.data["backupinfo"]) + if self.data["rollback_type"] == RollbackType.LOCAL_AND_TIME: + # 可从主库获取,但需要排除掉被回档的机器 + inst_list = ["{}{}{}".format(master.machine.ip, IP_PORT_DIVIDER, master.port)] + stand_by_slaves = cluster_class.storageinstance_set.filter( + instance_inner_role=InstanceInnerRole.SLAVE.value, + is_stand_by=True, + status=InstanceStatus.RUNNING.value, + ) + if len(stand_by_slaves) > 0: + inst_list.append( + "{}{}{}".format(stand_by_slaves[0].machine.ip, IP_PORT_DIVIDER, stand_by_slaves[0].port) + ) + backupinfo = get_local_backup(inst_list, cluster_class, self.data["rollback_time"]) + if backupinfo is None: + logger.error("cluster {} backup info not exists".format(cluster_class.id)) + raise TendbGetBackupInfoFailedException( + message=_("获取集群 {} 的备份信息失败".format(cluster_class.id))) + + elif self.data["rollback_type"] == RollbackType.REMOTE_AND_TIME.value: + rollback_handler = FixPointRollbackHandler(cluster_class.id) + backupinfo = rollback_handler.query_latest_backup_log(str2datetime(self.data["rollback_time"])) + if backupinfo is None: + logger.error("cluster {} backup info not exists".format(["cluster_id"])) + raise TendbGetBackupInfoFailedException( + message=_("获取集群 {} 的备份信息失败".format(cluster_class.id)) + ) sub_pipeline = SubBuilder(root_id=self.root_id, data=copy.deepcopy(self.data)) rollback_class = Cluster.objects.get(id=self.data["rollback_cluster_id"]) @@ -285,13 +314,13 @@ def rollback_to_cluster_flow(self): rollback_pipeline_list = [] for rollback_storage in storages: if not check_storage_database( - rollback_class.bk_cloud_id, rollback_storage.machine.ip, rollback_storage.port + rollback_class.bk_cloud_id, rollback_storage.machine.ip, rollback_storage.port ): logger.error("cluster {} check database fail".format(rollback_class.id)) raise NormalSpiderFlowException( - message=_("回档集群 {} 空闲检查不通过，请确认回档集群是否存在非系统数据库".format(rollback_class.id)) + message=_( + "回档集群 {} 空闲检查不通过，请确认回档集群是否存在非系统数据库".format(rollback_class.id)) ) - backupinfo = copy.deepcopy(self.data["backupinfo"]) mycluster = { "name": cluster_class.name, "cluster_id": cluster_class.id, @@ -369,20 +398,6 @@ def rollback_to_cluster_flow(self): ) # 本地备份+时间 if self.data["rollback_type"] == RollbackType.LOCAL_AND_TIME: - inst_list = ["{}{}{}".format(master.machine.ip, IP_PORT_DIVIDER, master.port)] - stand_by_slaves = cluster_class.storageinstance_set.filter( - instance_inner_role=InstanceInnerRole.SLAVE.value, - is_stand_by=True, - status=InstanceStatus.RUNNING.value, - ).exclude(machine__ip__in=[rollback_storage.machine.ip]) - if len(stand_by_slaves) > 0: - inst_list.append( - "{}{}{}".format(stand_by_slaves[0].machine.ip, IP_PORT_DIVIDER, stand_by_slaves[0].port) - ) - backupinfo = get_local_backup(inst_list, cluster_class, mycluster["rollback_time"]) - if backupinfo is None: - logger.error("cluster {} backup info not exists".format(cluster_class.id)) - raise TendbGetBackupInfoFailedException(message=_("获取集群 {} 的备份信息失败".format(cluster_class.id))) mycluster["backupinfo"] = copy.deepcopy(backupinfo) rollback_pipeline.add_sub_pipeline( sub_flow=rollback_local_and_time( @@ -392,16 +407,8 @@ def rollback_to_cluster_flow(self): cluster_model=cluster_class, ) ) - # 远程备份+时间 elif self.data["rollback_type"] == RollbackType.REMOTE_AND_TIME.value: - rollback_handler = FixPointRollbackHandler(cluster_class.id) - backupinfo = rollback_handler.query_latest_backup_log(str2datetime(mycluster["rollback_time"])) - if backupinfo is None: - logger.error("cluster {} backup info not exists".format(mycluster["cluster_id"])) - raise TendbGetBackupInfoFailedException( - message=_("获取集群 {} 的备份信息失败".format(mycluster["cluster_id"])) - ) mycluster["backupinfo"] = copy.deepcopy(backupinfo) rollback_pipeline.add_sub_pipeline( sub_flow=rollback_remote_and_time(