feature: Agent 2.0 Proxy 配置结构变更、升级、重载适配 (closed #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/base.py

@@ -135,7 +135,7 @@ def get_agent_pkg_name(
         package_type = ("client", "proxy")[host.node_type == constants.NodeType.PROXY]
         agent_step_adapter = common_data.agent_step_adapter
         if not agent_step_adapter.is_legacy:
-            setup_info = agent_step_adapter.setup_info
+            setup_info = agent_step_adapter.get_host_setup_info()
             return f"{setup_info.name}-{setup_info.version}.tgz"
 
         # GSE1.0 的升级包是独立的，添加了 _upgrade 后缀
@@ -262,7 +262,7 @@ def get_host_id__installation_tool_map(
             host for host in hosts_need_gen_commands if host.bk_host_id in host_id__install_channel_map
         ]
         host_id__installation_tool_map = batch_gen_commands(
-            base_agent_setup_info=common_data.agent_step_adapter.setup_info,
+            base_agent_setup_info=common_data.agent_step_adapter.get_host_setup_info(),
             hosts=hosts_need_gen_commands,
             pipeline_id=self.id,
             is_uninstall=is_uninstall,

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

@@ -9,7 +9,7 @@
 specific language governing permissions and limitations under the License.
 """
 
-from typing import Any, Dict, List
+from typing import Any, Dict, List, Union
 
 from apps.node_man import models
 from apps.utils.files import PathHandler
@@ -18,18 +18,50 @@
 
 
 class RenderAndPushGseConfigService(AgentPushConfigService):
+    def host_id_agent_version_map(self, data, common_data: AgentCommonData) -> Dict[int, str]:
+        # 重载2.0配置实时获取Agent Version信息
+        is_need_request_agent_version: bool = data.get_one_of_inputs("is_need_request_agent_version")
+        if not is_need_request_agent_version or common_data.agent_step_adapter.is_legacy:
+            return
+        # 实时获取Agent Version信息
+        # 构造 gse 请求参数
+        hosts: List[Dict[str, Union[int, str]]] = []
+        agent_host_id_map: Dict[int, str] = {}
+        for host_id in common_data.bk_host_ids:
+            host_obj: models.Host = common_data.host_id_obj_map[host_id]
+            host: Dict[str, Union[int, str]] = {
+                "ip": host_obj.inner_ip or host_obj.inner_ipv6,
+                "bk_cloud_id": host_obj.bk_cloud_id,
+                "bk_agent_id": host_obj.bk_agent_id,
+            }
+            hosts.append(host)
+            agent_host_id_map[common_data.gse_api_helper.get_agent_id(host)] = host_id
+        agent_id__agent_state_info_map: Dict[str, Dict] = common_data.gse_api_helper.list_agent_state(hosts)
+
+        host_id_agent_version_map: Dict[int, str] = {}
+        for agent_id, agent_state_info in agent_id__agent_state_info_map.items():
+            host_id_agent_version_map[agent_host_id_map[agent_id]] = agent_state_info["version"]
+
+        return host_id_agent_version_map
+
     def get_config_info_list(self, data, common_data: AgentCommonData, host: models.Host) -> 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)
 
+        host_id_agent_version_map: Dict[int, str] = self.host_id_agent_version_map(data=data, common_data=common_data)
+
         config_file_list: List[Dict[str, Any]] = []
         for file_name in file_name_list:
             config_file_list.append(
                 {
                     "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,
+                        host_id_agent_version_map=host_id_agent_version_map,
                     ),
                 }
             )

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

@@ -57,13 +57,15 @@ class AgentStepAdapter:
     log_prefix: str = field(init=False)
     # 配置处理模块缓存
     _config_handler_cache: typing.Dict[str, GseConfigHandler] = field(init=False)
+    _agent_setup_info: typing.Dict[str, base.AgentSetupInfo] = field(init=False)
 
     def __post_init__(self):
         self.is_legacy = self.gse_version == GseVersion.V1.value
         self.log_prefix: str = (
             f"[{self.__class__.__name__}({self.subscription_step.step_id})] | {self.subscription_step} |"
         )
         self._config_handler_cache: typing.Dict[str, GseConfigHandler] = {}
+        self._agent_setup_info: typing.Dict[str, base.AgentSetupInfo] = {}
 
     def get_config_handler(self, agent_name: str, target_version: str) -> GseConfigHandler:
 
@@ -102,8 +104,11 @@ def _get_config(
         ap: models.AccessPoint,
         proxies: typing.List[models.Host],
         install_channel: typing.Tuple[typing.Optional[models.Host], typing.Dict[str, typing.List]],
+        host_id_agent_version_map: typing.Optional[typing.Dict[int, str]] = None,
     ) -> str:
-        agent_setup_info: base.AgentSetupInfo = self.setup_info
+        agent_setup_info: base.AgentSetupInfo = self.get_host_setup_info(
+            agent_version=(host_id_agent_version_map or {}).get(host.bk_host_id)
+        )
         config_handler: GseConfigHandler = self.get_config_handler(agent_setup_info.name, agent_setup_info.version)
         config_tmpl_obj: base.AgentConfigTemplate = config_handler.get_matching_config_tmpl(
             os_type=host.os_type,
@@ -134,6 +139,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,
+        host_id_agent_version_map: typing.Optional[typing.Dict[int, str]] = None,
     ) -> str:
         """
         获取配置
@@ -145,19 +151,25 @@ def get_config(
         :param install_channel: 安装通道
         :return:
         """
+        if host_id_agent_version_map is None:
+            host_id_agent_version_map: typing.Dict = {}
 
         ap = ap or host.ap
         proxies = proxies if proxies is not None else fetch_proxies(host, ap)
         install_channel = install_channel or host.install_channel(ap_id=ap.id)
 
         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,
+            host_id_agent_version_map=host_id_agent_version_map,
         )
 
-    @property
-    @cache.class_member_cache()
-    def setup_info(self) -> base.AgentSetupInfo:
+    def get_host_setup_info(self, agent_version: typing.Optional[str] = None) -> base.AgentSetupInfo:
         """
         获取 Agent 设置信息
         TODO 后续如需支持多版本，该方法改造为 `get_host_setup_info`，根据维度进行缓存，参考 _config_handler_cache
@@ -171,15 +183,22 @@ def setup_info(self) -> base.AgentSetupInfo:
         else:
             target_version: str = get_target_helper(TargetType.AGENT.value).get_target_version(
                 target_id=AGENT_NAME_TARGET_ID_MAP[agent_name],
-                target_version=self.config.get("version"),
+                target_version=agent_version or self.config.get("version"),
             )
 
-        return base.AgentSetupInfo(
+        cache_key: str = f"agent_name:{agent_name}:version:{target_version}"
+        agent_setup_info: typing.Optional[base.AgentSetupInfo] = self._agent_setup_info.get(cache_key)
+        if agent_setup_info:
+            return agent_setup_info
+
+        agent_setup_info: base.AgentSetupInfo = base.AgentSetupInfo(
             is_legacy=self.is_legacy,
             agent_tools_relative_dir=("agent_tools/agent2", "")[self.is_legacy],
-            name=self.config.get("name"),
+            name=agent_name,
             version=target_version,
         )
+        self._agent_setup_info[cache_key] = agent_setup_info
+        return agent_setup_info
 
     @staticmethod
     def validated_data(data, serializer) -> OrderedDict:

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]],
+    host_id_agent_version_map: Optional[Dict[int, str]] = None,
 ):
     """
     生成 GSE 相关配置

apps/backend/subscription/views.py

@@ -690,7 +690,7 @@ def fetch_commands(self, request):
         host_ap: models.AccessPoint = ap_id_obj_map[host_ap_id]
 
         base_agent_setup_info_dict: Dict[str, Any] = asdict(
-            AgentStepAdapter(subscription_step=sub_step_obj, gse_version=host_ap.gse_version).setup_info
+            AgentStepAdapter(subscription_step=sub_step_obj, gse_version=host_ap.gse_version).get_host_setup_info()
         )
         agent_setup_extra_info_dict = sub_inst.instance_info["host"].get("agent_setup_extra_info") or {}
         installation_tool = gen_commands(

apps/backend/tests/components/collections/agent_new/test_install.py

@@ -405,7 +405,7 @@ def test_batch_solution(self):
             gse_version=GseVersion.V2.value,
         )
         installation_tool = gen_commands(
-            agent_step_adapter.setup_info,
+            agent_step_adapter.get_host_setup_info(),
             host,
             mock_data_utils.JOB_TASK_PIPELINE_ID,
             is_uninstall=False,
@@ -878,7 +878,7 @@ def test_shell_solution(self):
             gse_version=GseVersion.V2.value,
         )
         installation_tool = gen_commands(
-            agent_step_adapter.setup_info,
+            agent_step_adapter.get_host_setup_info(),
             host,
             mock_data_utils.JOB_TASK_PIPELINE_ID,
             is_uninstall=False,

apps/backend/tests/subscription/agent_adapter/test_adapter.py

@@ -97,8 +97,8 @@ def test_get_config(self):
             self.get_config(config_name)
             self.assertEqual(
                 self.agent_step_adapter.get_config_handler(
-                    agent_name=self.agent_step_adapter.setup_info.name,
-                    target_version=self.agent_step_adapter.setup_info.version,
+                    agent_name=self.agent_step_adapter.get_host_setup_info().name,
+                    target_version=self.agent_step_adapter.get_host_setup_info().version,
                 )
                 .get_matching_config_tmpl(self.host.os_type, self.host.cpu_arch, config_name)
                 .agent_name_from,
@@ -108,8 +108,8 @@ def test_get_config(self):
             self.get_config(config_name)
             self.assertEqual(
                 self.agent_step_adapter.get_config_handler(
-                    agent_name=self.agent_step_adapter.setup_info.name,
-                    target_version=self.agent_step_adapter.setup_info.version,
+                    agent_name=self.agent_step_adapter.get_host_setup_info().name,
+                    target_version=self.agent_step_adapter.get_host_setup_info().version,
                 )
                 .get_matching_config_tmpl(self.host.os_type, self.host.cpu_arch, config_name)
                 .agent_name_from,
@@ -129,15 +129,15 @@ def clear_agent_data(cls):
 
     def test_get_env(self):
         agent_env = self.agent_step_adapter.get_config_handler(
-            agent_name=self.agent_step_adapter.setup_info.name,
-            target_version=self.agent_step_adapter.setup_info.version,
+            agent_name=self.agent_step_adapter.get_host_setup_info().name,
+            target_version=self.agent_step_adapter.get_host_setup_info().version,
         ).get_matching_template_env(self.host.os_type, self.host.cpu_arch, constants.GsePackageCode.AGENT.value)
 
         self.assertEqual(agent_env["BK_GSE_HOME_DIR"], "/usr/local/gse/agent")
 
         proxy_env = self.agent_step_adapter.get_config_handler(
-            agent_name=self.agent_step_adapter.setup_info.name,
-            target_version=self.agent_step_adapter.setup_info.version,
+            agent_name=self.agent_step_adapter.get_host_setup_info().name,
+            target_version=self.agent_step_adapter.get_host_setup_info().version,
         ).get_matching_template_env(self.host.os_type, self.host.cpu_arch, constants.GsePackageCode.PROXY.value)
 
         self.assertEqual(proxy_env["BK_GSE_HOME_DIR"], "/usr/local/gse/proxy")

apps/backend/tests/view/test_get_gse_config.py

@@ -123,7 +123,7 @@ def setUpTestData(cls):
             file_name=cls.agent_step_adapter.get_main_config_filename(), sub_inst_id=cls.sub_inst_record_obj.id
         )
         cls.agent_step_adapter = AgentStepAdapter(subscription_step=cls.sub_step_obj, gse_version=cls.GSE_VERSION)
-        target_version = cls.agent_step_adapter.setup_info.version
+        target_version = cls.agent_step_adapter.get_host_setup_info().version
         models.GseConfigEnv.objects.create(
             agent_name="gse_agent",
             version=target_version,