fix(backend): 调整proxy升级的连接检查 TencentBlueKing#7890

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

@@ -664,6 +664,7 @@ def check_sub_flow(
     check_client_conn_inst: list = None,
     is_verify_checksum: bool = False,
     verify_checksum_tuples: list = None,
+    is_proxy: bool = False,
 ):
     """
     设计预检测的公共子流程，主要服务于切换类的流程，做前置检查，方便管控
@@ -689,8 +690,7 @@ def check_sub_flow(
                 "act_component_code": CheckClientConnComponent.code,
                 "kwargs": asdict(
                     CheckClientConnKwargs(
-                        bk_cloud_id=cluster.bk_cloud_id,
-                        check_instances=check_client_conn_inst,
+                        bk_cloud_id=cluster.bk_cloud_id, check_instances=check_client_conn_inst, is_proxy=is_proxy
                     )
                 ),
             }

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

@@ -22,6 +22,7 @@
 from backend.db_package.models import Package
 from backend.flow.engine.bamboo.scene.common.builder import Builder, SubBuilder
 from backend.flow.engine.bamboo.scene.common.get_file_list import GetFileList
+from backend.flow.engine.bamboo.scene.mysql.common.common_sub_flow import check_sub_flow
 from backend.flow.plugins.components.collections.common.pause import PauseComponent
 from backend.flow.plugins.components.collections.mysql.exec_actuator_script import ExecuteDBActuatorScriptComponent
 from backend.flow.plugins.components.collections.mysql.mysql_db_meta import MySQLDBMetaComponent
@@ -58,10 +59,12 @@ def __init__(self, root_id: str, data: Optional[Dict]):
         self.data = data
         self.uid = data["uid"]
         self.upgrade_cluster_list = data["infos"]
+        self.force_upgrade = data.get("force", True)
 
     def upgrade_mysql_proxy_flow(self):
         proxy_upgrade_pipeline = Builder(root_id=self.root_id, data=self.data)
         sub_pipelines = []
+        # 统一下发文件
         # 声明子流程
         for upgrade_info in self.upgrade_cluster_list:
             cluster_ids = upgrade_info["cluster_ids"]
@@ -77,7 +80,8 @@ def upgrade_mysql_proxy_flow(self):
             sub_flow_context = copy.deepcopy(self.data)
             sub_pipeline = SubBuilder(root_id=self.root_id, data=copy.deepcopy(sub_flow_context))
             ports_map = defaultdict(list)
-
+            proxy_ips = []
+            # 元数据版本检查
             for proxy_instance in proxies:
                 current_version = proxy_version_parse(proxy_instance.version)
                 if current_version >= new_proxy_version_num:
@@ -88,6 +92,38 @@ def upgrade_mysql_proxy_flow(self):
                     )
                     raise DBMetaException(message=_("待升级版本大于等于新版本，请确认升级的版本"))
                 ports_map[proxy_instance.machine.ip].append(proxy_instance.port)
+                proxy_ips.append(proxy_instance.machine.ip)
+            # 切换前做预检测
+            check_db_connect_sub_flow_lis = []
+            for cluster_id in cluster_ids:
+                cluster_obj = Cluster.objects.get(id=cluster_id)
+                ps = ProxyInstance.objects.filter(cluster=cluster_obj)
+                proxy_ins = []
+                for p in ps:
+                    proxy_ins.append(f"{p.machine.ip}:{p.port}")
+                sub_flow = check_sub_flow(
+                    uid=self.uid,
+                    root_id=self.root_id,
+                    cluster=cluster_obj,
+                    is_check_client_conn=self.force_upgrade,
+                    is_proxy=True,
+                    check_client_conn_inst=proxy_ins,
+                )
+            check_db_connect_sub_flow_lis.append(sub_flow)
+            if len(check_db_connect_sub_flow_lis) > 0:
+                sub_pipeline.add_parallel_sub_pipeline(sub_flow_list=check_db_connect_sub_flow_lis)
+            # 提前下发文件
+            sub_pipeline.add_act(
+                act_name=_("下发升级的安装包"),
+                act_component_code=TransFileComponent.code,
+                kwargs=asdict(
+                    DownloadMediaKwargs(
+                        bk_cloud_id=bk_cloud_id,
+                        exec_ip=proxy_ips,
+                        file_list=GetFileList(db_type=DBType.MySQL).mysql_proxy_upgrade_package(pkg_id=pkg_id),
+                    )
+                ),
+            )
             for index, (proxy_ip, ports) in enumerate(ports_map.items()):
                 sub_pipeline.add_sub_pipeline(
                     sub_flow=self.upgrade_mysql_proxy_subflow(
@@ -96,6 +132,7 @@ def upgrade_mysql_proxy_flow(self):
                         pkg_id=pkg_id,
                         proxy_version=get_sub_version_by_pkg_name(proxy_pkg.name),
                         proxy_ports=ports,
+                        force_upgrade=False,
                     )
                 )
                 # 最后一个节点无需再确认
@@ -114,25 +151,13 @@ def upgrade_mysql_proxy_subflow(
         pkg_id: int,
         proxy_version: str,
         proxy_ports: list = None,
+        force_upgrade: bool = True,
     ):
         """
         定义upgrade mysql proxy 的flow
         """
         sub_pipeline = SubBuilder(root_id=self.root_id, data=self.data)
-
-        sub_pipeline.add_act(
-            act_name=_("下发升级的安装包"),
-            act_component_code=TransFileComponent.code,
-            kwargs=asdict(
-                DownloadMediaKwargs(
-                    bk_cloud_id=bk_cloud_id,
-                    exec_ip=ip,
-                    file_list=GetFileList(db_type=DBType.MySQL).mysql_proxy_upgrade_package(pkg_id=pkg_id),
-                )
-            ),
-        )
-
-        cluster = {"proxy_ports": proxy_ports, "pkg_id": pkg_id}
+        cluster = {"proxy_ports": proxy_ports, "pkg_id": pkg_id, "force_upgrade": force_upgrade}
         exec_act_kwargs = ExecActuatorKwargs(cluster=cluster, bk_cloud_id=bk_cloud_id)
         exec_act_kwargs.exec_ip = ip
         exec_act_kwargs.get_mysql_payload_func = ProxyActPayload.get_proxy_upgrade_payload.__name__

dbm-ui/backend/flow/utils/mysql/proxy_act_payload.py

@@ -88,7 +88,7 @@ def get_proxy_upgrade_payload(self, **kwargs) -> dict:
                 "extend": {
                     "host": kwargs["ip"],
                     "ports": self.cluster["proxy_ports"],
-                    "force": self.ticket_data["force"],
+                    "force": self.cluster["force_upgrade"],
                     "pkg": proxy_pkg.name,
                     "pkg_md5": proxy_pkg.md5,
                 },