feat(dbm-services): 升级单据忽略子版本的校验 #14467

Author: ymakedaq

dbm-ui/backend/flow/engine/bamboo/scene/mysql/mysql_proxy_upgrade.py

@@ -86,6 +86,29 @@ def upgrade_mysql_proxy_flow(self):
             for proxy_instance in proxies:
                 proxy_ips.append(proxy_instance.machine.ip)
             proxy_ips = list(set(proxy_ips))
+
+            # 获取目标版本
+            target_version = get_sub_version_by_pkg_name(proxy_pkg.name)
+
+            # 筛选需要升级的 proxy IP（跳过所有 proxy 版本都等于目标版本的机器）
+            upgrade_proxy_ips = []
+            for proxy_ip in proxy_ips:
+                ip_proxies = ProxyInstance.objects.filter(cluster__in=clusters, machine__ip=proxy_ip)
+                all_version_match = all(p.version == target_version for p in ip_proxies)
+                if all_version_match:
+                    logger.info(_("机器 {} 上所有 proxy 版本已是目标版本 {}, 跳过升级").format(proxy_ip, target_version))
+                else:
+                    upgrade_proxy_ips.append(proxy_ip)
+
+            # 如果没有需要升级的机器，跳过整个集群
+            if not upgrade_proxy_ips:
+                logger.info(
+                    _("集群 {} 所有 proxy 已是目标版本 {}, 跳过升级").format(
+                        ",".join([c.immute_domain for c in clusters]), target_version
+                    )
+                )
+                continue
+
             # 切换前做预检测
             check_db_connect_sub_flow_list = []
             for cluster_id in cluster_ids:
@@ -114,27 +137,33 @@ def upgrade_mysql_proxy_flow(self):
                 kwargs=asdict(
                     DownloadMediaKwargs(
                         bk_cloud_id=bk_cloud_id,
-                        exec_ip=proxy_ips,
+                        exec_ip=upgrade_proxy_ips,
                         file_list=GetFileList(db_type=DBType.MySQL).mysql_proxy_upgrade_package(pkg_id=pkg_id),
                     )
                 ),
             )
-            for index, proxy_ip in enumerate(proxy_ips):
+            for index, proxy_ip in enumerate(upgrade_proxy_ips):
                 sub_pipeline.add_sub_pipeline(
                     sub_flow=self.upgrade_mysql_proxy_subflow(
                         bk_cloud_id=bk_cloud_id,
                         ip=proxy_ip,
                         pkg_id=pkg_id,
-                        proxy_version=get_sub_version_by_pkg_name(proxy_pkg.name),
+                        proxy_version=target_version,
                         cluster_ids=cluster_ids,
                         force_upgrade=True,
                     )
                 )
                 # 最后一个节点无需再确认
-                if index < len(proxy_ips) - 1:
+                if index < len(upgrade_proxy_ips) - 1:
                     sub_pipeline.add_act(act_name=_("人工确认"), act_component_code=PauseComponent.code, kwargs={})
 
             sub_pipelines.append(sub_pipeline.build_sub_process(sub_name=sub_process_name))
+
+        # 如果所有集群都跳过了升级，则不执行流水线
+        if not sub_pipelines:
+            logger.info(_("所有集群的 proxy 都已是目标版本, 无需升级"))
+            return
+
         proxy_upgrade_pipeline.add_parallel_sub_pipeline(sub_flow_list=sub_pipelines)
         proxy_upgrade_pipeline.run_pipeline()
         return

dbm-ui/backend/flow/engine/bamboo/scene/mysql/validate/mysql_proxy_upgrade_validator.py

@@ -112,23 +112,29 @@ def run_check_for_info(self, info: dict, index: int) -> list:
         # 检查proxy版本升级合法性
         try:
             new_proxy_version_num = proxy_version_parse(package.name)
-            proxies = ProxyInstance.objects.filter(cluster__in=clusters)
+            proxies = ProxyInstance.objects.filter(cluster__in=clusters).prefetch_related("cluster")
 
             if len(proxies) <= 0:
                 error_msg = _("第{}行根据cluster_ids {}无法找到对应的proxy实例").format(index + 1, cluster_ids)
                 error_msgs.append(error_msg)
                 return error_msgs
 
             for proxy_instance in proxies:
+                cluster = proxy_instance.cluster.first()
+                if not cluster:
+                    error_msg = _("第{}行proxy实例 {} 未关联任何集群").format(index + 1, proxy_instance.ip_port)
+                    error_msgs.append(error_msg)
+                    continue
+
                 current_version = proxy_version_parse(proxy_instance.version)
-                if current_version >= new_proxy_version_num:
+                if current_version > new_proxy_version_num:
                     logger.error(
-                        _("集群 {} 的proxy实例 {} 当前版本 {} 大于等于升级版本 {}").format(
-                            proxy_instance.cluster.id, proxy_instance.ip_port, current_version, new_proxy_version_num
+                        _("集群 {} 的proxy实例 {} 当前版本 {} 大于升级版本 {}").format(
+                            cluster.id, proxy_instance.ip_port, current_version, new_proxy_version_num
                         )
                     )
                     error_msg = _("第{}行集群 {} 的proxy实例 {} 待升级版本大于等于当前版本，请确认升级的版本").format(
-                        index + 1, proxy_instance.cluster.id, proxy_instance.ip_port
+                        index + 1, cluster.id, proxy_instance.ip_port
                     )
                     error_msgs.append(error_msg)
 

dbm-ui/backend/flow/engine/bamboo/scene/spider/upgrade/upgrade_spider_node.py

@@ -41,13 +41,7 @@
     add_spider_upgrade_check_act,
     build_spider_upgrade_subflow,
 )
-from .upgrade_utils import (
-    check_cross_major_version_upgrade,
-    check_spider_node_count_compatibility,
-    check_spider_upgrade_version_compatibility,
-    get_spider_master_instances,
-    get_spider_upgrade_instances,
-)
+from .upgrade_utils import check_cross_major_version_upgrade, get_spider_master_instances, get_spider_upgrade_instances
 
 logger = logging.getLogger("flow")
 
@@ -119,7 +113,7 @@ def run(self):
         执行spider升级流程的主入口方法
 
         执行流程：
-        1. 执行前置检查(__pre_check)：验证升级版本和节点数量
+        1. 前置校验已由 TenDBClusterSpiderUpgradeValidator 完成（包括版本兼容性和节点数量检查）
         2. 根据upgrade_local参数选择升级模式：
            - True: 执行本地升级(local_upgrade)
            - False: 执行迁移升级(migrate_upgrade)
@@ -128,9 +122,6 @@ def run(self):
         - 本地升级：在现有机器上直接升级spider版本，适用于版本兼容性好的场景
         - 迁移升级：通过新增机器替换旧机器的方式进行升级，适用于需要保证服务连续性的场景
         """
-        # 执行前置检查：验证升级版本和节点数量
-        self.__pre_check()
-
         # 根据升级模式选择执行路径
         if self.upgrade_local:
             # 本地升级：在现有机器上直接升级版本
@@ -139,17 +130,6 @@ def run(self):
             # 迁移升级：通过新增机器替换旧机器进行升级
             self.migrate_upgrade()
 
-    # spider_ins.tendbclusterspiderext.spider_role
-    def __pre_check(self):
-        """
-        检查升级版本和源版本
-        """
-        # 检查版本兼容性
-        check_spider_upgrade_version_compatibility(self.data)
-
-        # 检查节点数量兼容性
-        check_spider_node_count_compatibility(self.data)
-
     def local_upgrade(self):
         """
         spider 本地升级场景

dbm-ui/backend/flow/engine/bamboo/scene/spider/upgrade/upgrade_utils.py

@@ -492,9 +492,9 @@ def check_spider_upgrade_version_compatibility(data: Dict) -> None:
         for spider_ins in spiders:
             current_version = tspider_version_parse(spider_ins.version)
             current_versions.add(spider_ins.version)
-            if current_version >= new_spider_version_num:
+            if current_version > new_spider_version_num:
                 logger.error(_("待升级版本 {} 需要大于当前版本 {}").format(new_spider_version_num, current_version))
-                raise DBMetaException(message=_("待升级版本大于等于新版本，请确认升级的版本"))
+                raise DBMetaException(message=_("待升级版本大于新版本，请确认升级的版本"))
 
 
 def check_spider_node_count_compatibility(data: Dict) -> None:

dbm-ui/backend/flow/engine/bamboo/scene/spider/validate/spider_upgrade_validate.py

@@ -10,9 +10,15 @@
 from django.utils.translation import gettext as _
 
 from backend.db_meta.enums import TenDBClusterSpiderRole
-from backend.db_meta.models import Cluster, Machine
-from backend.flow.engine.bamboo.scene.spider.validate.exception import SpiderRoleFailedException
+from backend.db_meta.models import Cluster, Machine, ProxyInstance
+from backend.db_package.models import Package
+from backend.flow.consts import MediumEnum
+from backend.flow.engine.bamboo.scene.spider.validate.exception import (
+    SpiderRoleFailedException,
+    UpgradeVersionFailedException,
+)
 from backend.flow.engine.validate.mysql_base_validate import MysqlBaseValidator
+from backend.flow.utils.mysql.mysql_version_parse import spider_cross_major_version, tspider_version_parse
 
 
 class TenDBClusterSpiderUpgradeValidator(MysqlBaseValidator):
@@ -119,27 +125,167 @@ def pre_check_spider_spec_consistency(self):
 
         return error_msgs
 
+    def pre_check_spider_upgrade_version_compatibility(self):
+        """
+        检查spider升级版本兼容性
+
+        校验逻辑：
+        1. 遍历所有待升级的集群信息
+        2. 获取目标版本包信息和当前集群所有spider实例版本
+        3. 确保目标版本大于所有当前版本
+
+        返回：
+        - list: 错误信息列表，如果没有错误则返回空列表
+        """
+        error_msgs = []
+
+        for info in self.data["infos"]:
+            pkg_id = info["pkg_id"]
+            cluster_id = info["cluster_id"]
+            spider_pkg = Package.objects.get(id=pkg_id, pkg_type=MediumEnum.Spider)
+            new_spider_version_num = tspider_version_parse(spider_pkg.name)
+            cluster = Cluster.objects.get(id=cluster_id)
+            spiders = ProxyInstance.objects.filter(cluster=cluster)
+
+            for spider_ins in spiders:
+                current_version = tspider_version_parse(spider_ins.version)
+                if current_version > new_spider_version_num:
+                    error_msg = _("集群 {} 待升级版本 {} 需要大于当前版本 {}").format(
+                        cluster.immute_domain, new_spider_version_num, current_version
+                    )
+                    error_msgs.append(error_msg)
+                    break  # 一个集群发现一个不兼容版本就跳出
+
+        return error_msgs
+
+    def pre_check_spider_node_count_compatibility(self):
+        """
+        检查spider节点数量兼容性（仅迁移升级场景）
+
+        校验逻辑：
+        1. 仅在非本地升级（迁移升级）场景下进行检查
+        2. 检查新传入的spider master IP数量与现有节点数量是否一致
+        3. 检查新传入的spider slave IP数量与现有节点数量是否一致
+
+        返回：
+        - list: 错误信息列表，如果没有错误则返回空列表
+        """
+        error_msgs = []
+
+        # 仅在迁移升级场景下进行检查
+        if self.data.get("upgrade_local", False):
+            return error_msgs
+
+        for info in self.data["infos"]:
+            cluster_id = info["cluster_id"]
+            cluster = Cluster.objects.get(id=cluster_id)
+
+            spider_master_ip_list = info.get("spider_master_ip_list", [])
+            spider_slave_ip_list = info.get("spider_slave_ip_list", [])
+
+            master_spiders_count = cluster.proxyinstance_set.filter(
+                tendbclusterspiderext__spider_role=TenDBClusterSpiderRole.SPIDER_MASTER
+            ).count()
+            if master_spiders_count != len(spider_master_ip_list):
+                error_msg = _("集群 {} 待升级spiderMaster节点数({})与传入ip节点数({})不一致,请确认").format(
+                    cluster.immute_domain, master_spiders_count, len(spider_master_ip_list)
+                )
+                error_msgs.append(error_msg)
+
+            slave_spiders_count = cluster.proxyinstance_set.filter(
+                tendbclusterspiderext__spider_role=TenDBClusterSpiderRole.SPIDER_SLAVE
+            ).count()
+            if slave_spiders_count > 0 and len(spider_slave_ip_list) != slave_spiders_count:
+                error_msg = _("集群 {} 待升级spiderSlave节点数({})与传入ip节点数({})不一致,请确认").format(
+                    cluster.immute_domain, slave_spiders_count, len(spider_slave_ip_list)
+                )
+                error_msgs.append(error_msg)
+
+        return error_msgs
+
+    def pre_check_local_upgrade_cross_major_version(self):
+        """
+        检查本地升级时spider实例之间是否存在跨主版本
+
+        校验逻辑：
+        1. 仅在本地升级场景（upgrade_local=True）下进行检查
+        2. 遍历所有待升级的集群信息
+        3. 获取每个集群的所有spider实例版本
+        4. 比较最大版本和最小版本是否跨主版本
+        5. 如果发现跨主版本，记录错误信息
+
+        返回：
+        - list: 错误信息列表，如果没有错误则返回空列表
+        """
+        error_msgs = []
+
+        # 仅在本地升级场景下进行检查
+        if not self.data.get("upgrade_local", False):
+            return error_msgs
+
+        for info in self.data["infos"]:
+            cluster_id = info["cluster_id"]
+            cluster = Cluster.objects.get(id=cluster_id)
+            spiders = ProxyInstance.objects.filter(cluster=cluster)
+
+            if not spiders.exists():
+                continue
+
+            # 收集所有spider实例的版本号
+            version_nums = [tspider_version_parse(spider.version) for spider in spiders]
+
+            # 比较最大版本和最小版本是否跨主版本
+            max_version = max(version_nums)
+            min_version = min(version_nums)
+
+            if spider_cross_major_version(max_version, min_version):
+                all_versions = set(spider.version for spider in spiders)
+                error_msg = _("集群 {} 本地升级不允许spider实例之间存在跨主版本，当前存在版本: {}").format(
+                    cluster.immute_domain, ", ".join(all_versions)
+                )
+                error_msgs.append(error_msg)
+
+        return error_msgs
+
     def __call__(self):
         """
         发起校验，实例函数化
 
         执行流程：
-        1. 调用pre_check_spider_master_spec_consistency方法检查规格一致性
-        2. 如果发现错误，抛出SpiderRoleFailedException异常
-        3. 如果没有错误，返回None表示校验通过
+        1. 调用pre_check_spider_spec_consistency方法检查规格一致性
+        2. 调用pre_check_spider_upgrade_version_compatibility方法检查版本兼容性
+        3. 调用pre_check_spider_node_count_compatibility方法检查节点数量兼容性
+        4. 如果发现错误，抛出对应的异常
+        5. 如果没有错误，返回None表示校验通过
 
         异常处理：
         - 当发现规格不一致时，抛出SpiderRoleFailedException
-        - 异常消息包含所有发现的规格不一致问题
+        - 当发现版本不兼容时，抛出UpgradeVersionFailedException
+        - 异常消息包含所有发现的问题
 
         返回：
         - None: 校验通过
-        - 异常: 校验失败时抛出SpiderRoleFailedException
+        - 异常: 校验失败时抛出对应异常
         """
         # 检查spider机器规格一致性（包括spider_master和spider_slave）
         error_msgs = self.pre_check_spider_spec_consistency()
         if error_msgs:
             # 将所有错误信息合并为一个字符串，并抛出异常
             raise SpiderRoleFailedException("\n".join(error_msgs))
 
+        # 检查spider升级版本兼容性
+        error_msgs = self.pre_check_spider_upgrade_version_compatibility()
+        if error_msgs:
+            raise UpgradeVersionFailedException("\n".join(error_msgs))
+
+        # 检查spider节点数量兼容性（仅迁移升级场景）
+        error_msgs = self.pre_check_spider_node_count_compatibility()
+        if error_msgs:
+            raise UpgradeVersionFailedException("\n".join(error_msgs))
+
+        # 检查本地升级时是否跨主版本
+        error_msgs = self.pre_check_local_upgrade_cross_major_version()
+        if error_msgs:
+            raise UpgradeVersionFailedException("\n".join(error_msgs))
+
         return None