feature: Agent 2.0 Proxy 配置结构变更、升级、重载适配 (closed TencentBlueKing#2033)

apps/backend/agent/manager.py

@@ -191,9 +191,12 @@ def start_nginx(cls):
         return act
 
     @classmethod
-    def render_and_push_gse_config(cls, name=components.RenderAndPushGseConfigComponent.name):
+    def render_and_push_gse_config(
+        cls, is_need_request_agent_version: bool = False, name=components.RenderAndPushGseConfigComponent.name
+    ):
         """渲染并下载 agent 配置"""
         act = AgentServiceActivity(component_code=components.RenderAndPushGseConfigComponent.code, name=name)
+        act.component.inputs.is_need_request_agent_version = Var(type=Var.PLAIN, value=is_need_request_agent_version)
         return act
 
     @classmethod

apps/backend/components/collections/agent_new/push_environ_files.py

@@ -22,7 +22,11 @@
 
 class PushEnvironFilesService(AgentPushConfigService):
     def get_config_info_list(
-        self, data, common_data: AgentCommonData, host: models.Host
+        self,
+        data,
+        common_data: AgentCommonData,
+        host: models.Host,
+        host_id__agent_state_info: typing.Dict[int, typing.Dict[str, typing.Any]],
     ) -> typing.List[typing.Dict[str, typing.Any]]:
 
         if host.os_type == constants.OsType.WINDOWS:

apps/backend/components/collections/agent_new/render_and_push_gse_config.py

@@ -18,7 +18,13 @@
 
 
 class RenderAndPushGseConfigService(AgentPushConfigService):
-    def get_config_info_list(self, data, common_data: AgentCommonData, host: models.Host) -> List[Dict[str, Any]]:
+    def get_config_info_list(
+        self,
+        data,
+        common_data: AgentCommonData,
+        host: models.Host,
+        host_id__agent_state_info: Dict[int, Dict[str, Any]],
+    ) -> List[Dict[str, Any]]:
         file_name_list: List[str] = common_data.agent_step_adapter.get_config_filename_by_node_type(host.node_type)
         general_node_type = self.get_general_node_type(host.node_type)
         host_ap: models.AccessPoint = self.get_host_ap(common_data=common_data, host=host)
@@ -29,7 +35,11 @@ def get_config_info_list(self, data, common_data: AgentCommonData, host: models.
                 {
                     "file_name": file_name,
                     "content": common_data.agent_step_adapter.get_config(
-                        host=host, filename=file_name, node_type=general_node_type, ap=host_ap
+                        host=host,
+                        filename=file_name,
+                        node_type=general_node_type,
+                        ap=host_ap,
+                        agent_version=host_id__agent_state_info.get(host.bk_host_id, {}).get("version"),
                     ),
                 }
             )

apps/backend/components/collections/job.py

@@ -27,6 +27,9 @@
 from apps.core.files.storage import get_storage
 from apps.exceptions import AppBaseException
 from apps.node_man import constants, models
+from apps.node_man.periodic_tasks.sync_agent_status_task import (
+    update_or_create_host_agent_status,
+)
 from apps.utils.batch_request import request_multi_thread
 from common.api import JobApi
 from pipeline.component_framework.component import Component
@@ -592,13 +595,28 @@ def cal_job_unique_key(self, config_info_list: List[Dict[str, Any]], file_target
 
     def _execute(self, data, parent_data, common_data: CommonData):
         timeout = data.get_one_of_inputs("timeout")
+        # 重载配置实时获取Agent版本
+        is_need_request_agent_version: bool = data.get_one_of_inputs("is_need_request_agent_version")
+        if not is_need_request_agent_version:
+            host_id__agent_state_info: Dict = {}
+        else:
+            host_id__agent_state_info: Dict[int, Dict[str, Any]] = update_or_create_host_agent_status(
+                task_id="[reload_agent_fill_agent_state_info_to_hosts]",
+                host_queryset=models.Host.objects.filter(bk_host_id__in=common_data.bk_host_ids),
+            )
+
         # 批量请求作业平台的参数
         multi_job_params_map: Dict[str, Dict[str, Any]] = {}
         for sub_inst in common_data.subscription_instances:
             bk_host_id = sub_inst.instance_info["host"]["bk_host_id"]
             host_obj = common_data.host_id_obj_map[bk_host_id]
 
-            config_info_list = self.get_config_info_list(data=data, common_data=common_data, host=host_obj)
+            config_info_list = self.get_config_info_list(
+                data=data,
+                common_data=common_data,
+                host=host_obj,
+                host_id__agent_state_info=host_id__agent_state_info,
+            )
             file_target_path = self.get_file_target_path(data=data, common_data=common_data, host=host_obj)
 
             job_unique_key = self.cal_job_unique_key(config_info_list, file_target_path)
@@ -638,7 +656,13 @@ def _execute(self, data, parent_data, common_data: CommonData):
 
         self.run_job_or_finish_schedule(multi_job_params_map)
 
-    def get_config_info_list(self, data, common_data: CommonData, host: models.Host) -> List[Dict[str, Any]]:
+    def get_config_info_list(
+        self,
+        data,
+        common_data: CommonData,
+        host: models.Host,
+        host_id__agent_state_info: Dict[int, Dict[str, Any]],
+    ) -> List[Dict[str, Any]]:
         """
         获取主机所需的配置文件信息列表
         :param data:

apps/backend/subscription/steps/agent.py

@@ -316,6 +316,7 @@ class UpgradeAgent(ReinstallAgent):
     def _generate_activities(self, agent_manager: AgentManager):
         activities = [
             agent_manager.push_upgrade_package(),
+            agent_manager.render_and_push_gse_config(),
             agent_manager.run_upgrade_command(),
             agent_manager.wait(30),
             agent_manager.get_agent_status(expect_status=constants.ProcStateType.RUNNING),
@@ -457,6 +458,7 @@ class UpgradeProxy(ReinstallProxy):
     def _generate_activities(self, agent_manager: AgentManager):
         activities = [
             agent_manager.push_upgrade_package(),
+            agent_manager.render_and_push_gse_config(),
             agent_manager.run_upgrade_command(),
             agent_manager.wait(30),
             agent_manager.get_agent_status(expect_status=constants.ProcStateType.RUNNING),
@@ -533,7 +535,7 @@ def _generate_activities(self, agent_manager: AgentManager):
         activities = [
             agent_manager.check_agent_status(),
             agent_manager.update_install_info(),
-            agent_manager.render_and_push_gse_config(),
+            agent_manager.render_and_push_gse_config(is_need_request_agent_version=True),
             agent_manager.reload_agent(skip_polling_result=True),
             agent_manager.wait(5),
             agent_manager.get_agent_status(expect_status=constants.ProcStateType.RUNNING),

apps/backend/subscription/steps/agent_adapter/adapter.py

@@ -102,9 +102,12 @@ def _get_config(
         ap: models.AccessPoint,
         proxies: typing.List[models.Host],
         install_channel: typing.Tuple[typing.Optional[models.Host], typing.Dict[str, typing.List]],
+        agent_version: typing.Optional[typing.Dict[int, str]] = None,
     ) -> str:
         agent_setup_info: base.AgentSetupInfo = self.setup_info
-        config_handler: GseConfigHandler = self.get_config_handler(agent_setup_info.name, agent_setup_info.version)
+        # 目标版本优先使用传入版本，传入版本必不会是标签所以可直接使用
+        target_version: str = agent_version or agent_setup_info.version
+        config_handler: GseConfigHandler = self.get_config_handler(agent_setup_info.name, target_version)
         config_tmpl_obj: base.AgentConfigTemplate = config_handler.get_matching_config_tmpl(
             os_type=host.os_type,
             cpu_arch=host.cpu_arch,
@@ -134,6 +137,7 @@ def get_config(
         ap: typing.Optional[models.AccessPoint] = None,
         proxies: typing.Optional[typing.List[models.Host]] = None,
         install_channel: typing.Tuple[typing.Optional[models.Host], typing.Dict[str, typing.List]] = None,
+        agent_version: typing.Optional[str] = None,
     ) -> str:
         """
         获取配置
@@ -152,11 +156,17 @@ def get_config(
 
         func: typing.Callable[..., str] = (self._get_config, legacy.generate_gse_config)[self.is_legacy]
         return func(
-            host=host, filename=filename, node_type=node_type, ap=ap, proxies=proxies, install_channel=install_channel
+            host=host,
+            filename=filename,
+            node_type=node_type,
+            ap=ap,
+            proxies=proxies,
+            install_channel=install_channel,
+            agent_version=agent_version,
         )
 
     @property
-    @cache.class_member_cache()
+    @cache.class_member_cache
     def setup_info(self) -> base.AgentSetupInfo:
         """
         获取 Agent 设置信息

apps/backend/subscription/steps/agent_adapter/config_context/context_dataclass.py

@@ -194,6 +194,12 @@ class DataAgentConfigContext(GseConfigContext, TlsBaseConfigContext):
     tcp_server_thread_num: int = 32
     tcp_server_max_message_size: int = 10485760
 
+    # 兼容Agent 2.0 >= 2.1.5 gse_data_proxy 配置文件结构变更
+    bind_port: int = 0
+    bind_ip: str = "::"
+
+    thread_num: int = 32
+
 
 @dataclass
 class DataProxyConfigContext(GseConfigContext, TlsBaseConfigContext):

apps/backend/subscription/steps/agent_adapter/config_context/context_helper.py

@@ -153,6 +153,8 @@ def __post_init__(self):
             context_dataclass.DataMetricConfigContext(exporter_bind_port=self.ap.port_config["data_prometheus_port"]),
             context_dataclass.DataAgentConfigContext(
                 tcp_bind_port=self.ap.port_config["data_port"],
+                # 兼容Agent 2.0 >= 2.1.5 gse_data_proxy 配置文件结构变更
+                bind_port=self.ap.port_config["data_port"],
                 tls_ca_file=proxy_tls_ca_file,
                 tls_cert_file=proxy_tls_cert_file,
                 tls_key_file=proxy_tls_key_file,

apps/backend/subscription/steps/agent_adapter/legacy.py

@@ -458,6 +458,7 @@ def generate_gse_config(
     ap: models.AccessPoint,
     proxies: List[models.Host],
     install_channel: Tuple[Optional[models.Host], Dict[str, List]],
+    agent_version: Optional[Dict[int, str]] = None,
 ):
     """
     生成 GSE 相关配置