feature: 支持 upgrade_to_agent_id 升级为 AgentID 配置 (closed #1309)

apps/adapters/api/gse/base.py

@@ -91,6 +91,15 @@ def _operate_proc_multi(self, proc_operate_req: InfoDictList, **options) -> str:
         """
         raise NotImplementedError
 
+    @abc.abstractmethod
+    def _upgrade_to_agent_id(self, hosts: InfoDictList) -> InfoDict:
+        """
+        将基于Host IP的配置升级到基于Agent-ID的配置
+        :param hosts: 源 hosts 主机信息 [{"ip": "127.0.0.1", "bk_cloud_id": 0, "bk_agent_id": "xxxx"}]
+        :return: {"failed":[],"success":["0:127.0.0.1"]}
+        """
+        raise NotImplementedError
+
     @abc.abstractmethod
     def get_proc_operate_result(self, task_id: str) -> InfoDict:
         """
@@ -198,3 +207,6 @@ def operate_proc_multi(self, proc_operate_req: InfoDictList, **options) -> str:
             hosts = proc_operate_info.pop("hosts")
             preprocessed_proc_operate_req.append(self.preprocessing_proc_operate_info(hosts, proc_operate_info))
         return self._operate_proc_multi(preprocessed_proc_operate_req, **options)
+
+    def upgrade_to_agent_id(self, hosts: InfoDictList) -> InfoDict:
+        return self._upgrade_to_agent_id(hosts)

apps/adapters/api/gse/v1.py

@@ -102,3 +102,6 @@ def _operate_proc_multi(self, proc_operate_req: base.InfoDictList, **options) ->
 
     def get_proc_operate_result(self, task_id: str) -> base.InfoDict:
         return self.gse_api_obj.get_proc_operate_result({"task_id": task_id}, raw=True)
+
+    def _upgrade_to_agent_id(self, hosts: base.InfoDictList) -> base.InfoDict:
+        raise EnvironmentError("v1 gse api should not call upgrade_to_agent_id")

apps/adapters/api/gse/v2.py

@@ -113,5 +113,8 @@ def preprocessing_proc_operate_info(
     def _operate_proc_multi(self, proc_operate_req: base.InfoDictList, **options) -> str:
         return self.gse_api_obj.v2_proc_operate_proc_multi({"proc_operate_req": proc_operate_req})["task_id"]
 
+    def _upgrade_to_agent_id(self, hosts: base.InfoDictList) -> base.InfoDict:
+        return self.gse_api_obj.v2_proc_upgrade_to_agent_id({"hosts": hosts})
+
     def get_proc_operate_result(self, task_id: str) -> base.InfoDict:
         return self.gse_api_obj.v2_proc_get_proc_operate_result_v2({"task_id": task_id}, raw=True)

apps/backend/agent/manager.py

@@ -233,3 +233,11 @@ def unbind_host_agent(cls):
             component_code=components.UnBindHostAgentComponent.code, name=components.UnBindHostAgentComponent.name
         )
         return act
+
+    @classmethod
+    def upgrade_to_agent_id(cls):
+        """升级为 Agent-ID 配置"""
+        act = AgentServiceActivity(
+            component_code=components.UpgradeToAgentIDComponent.code, name=components.UpgradeToAgentIDComponent.name
+        )
+        return act

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

@@ -34,6 +34,7 @@
 from .unbind_host_agent import UnBindHostAgentService
 from .update_install_info import UpdateInstallInfoService
 from .update_process_status import UpdateProcessStatusService
+from .upgrade_to_agent_id import UpgradeToAgentIDService
 from .wait import WaitService
 
 
@@ -179,3 +180,9 @@ class AddOrUpdateHostsComponent(Component):
     name = _("新增或更新主机信息")
     code = "add_or_update_hosts"
     bound_service = AddOrUpdateHostsService
+
+
+class UpgradeToAgentIDComponent(Component):
+    name = _("升级为 Agent-ID 配置")
+    code = "upgrade_to_agent_id"
+    bound_service = UpgradeToAgentIDService

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

@@ -0,0 +1,44 @@
+# -*- coding: utf-8 -*-
+"""
+TencentBlueKing is pleased to support the open source community by making 蓝鲸智云-节点管理(BlueKing-BK-NODEMAN) available.
+Copyright (C) 2017-2022 THL A29 Limited, a Tencent company. All rights reserved.
+Licensed under the MIT License (the "License"); you may not use this file except in compliance with the License.
+You may obtain a copy of the License at https://opensource.org/licenses/MIT
+Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on
+an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the
+specific language governing permissions and limitations under the License.
+"""
+
+from typing import Dict, List, Union
+
+from django.utils.translation import ugettext_lazy as _
+
+from apps.adapters.api.gse import GseApiHelper
+
+from .base import AgentBaseService, AgentCommonData
+
+
+class UpgradeToAgentIDService(AgentBaseService):
+    def _execute(self, data, parent_data, common_data: AgentCommonData):
+        upgrade_hosts: List[Dict[str, Union[int, str]]] = []
+        cloud_ip__sub_inst_id_map: Dict[str, int] = {}
+        # 如果主机有AgentID，则调用 upgrade_to_agent_id 将基于 Host IP 的配置升级到基于 Agent-ID 的配置
+        for host_id, sub_inst_id in common_data.host_id__sub_inst_id_map.items():
+            host = common_data.host_id_obj_map[host_id]
+            upgrade_hosts.append(
+                {"ip": host.inner_ip, "bk_cloud_id": host.bk_cloud_id, "bk_agent_id": host.bk_agent_id}
+            )
+            cloud_ip__sub_inst_id_map[f"{host.bk_cloud_id}:{host.inner_ip}"] = sub_inst_id
+
+        upgrade_hosts = [
+            {"ip": host.inner_ip, "bk_cloud_id": host.bk_cloud_id, "bk_agent_id": host.bk_agent_id}
+            for host in common_data.host_id_obj_map.values()
+            if host.bk_agent_id
+        ]
+
+        if not upgrade_hosts:
+            return True
+        result = GseApiHelper.upgrade_to_agent_id(hosts=upgrade_hosts)
+        failed_cloud_ips = result.get("failed") or []
+        failed_sub_inst_ids = [cloud_ip__sub_inst_id_map[cloud_ip] for cloud_ip in failed_cloud_ips]
+        self.move_insts_to_failed(failed_sub_inst_ids, log_content=_("升级 Agent-ID 配置失败"))

apps/backend/subscription/steps/agent.py

@@ -491,6 +491,7 @@ def _generate_activities(self, agent_manager: AgentManager):
             agent_manager.choose_ap(),
             agent_manager.install(),
             agent_manager.bind_host_agent(),
+            agent_manager.upgrade_to_agent_id(),
             agent_manager.get_agent_status(expect_status=constants.ProcStateType.RUNNING),
             # 等待 JOB 完成 AgentID 同步
             agent_manager.wait(15),
@@ -525,6 +526,7 @@ def _generate_activities(self, agent_manager: AgentManager):
             agent_manager.choose_ap(),
             agent_manager.install(),
             agent_manager.bind_host_agent(),
+            agent_manager.upgrade_to_agent_id(),
             agent_manager.get_agent_status(expect_status=constants.ProcStateType.RUNNING, name=_("查询Proxy状态")),
             agent_manager.check_policy_gse_to_proxy(),
         ]

apps/backend/subscription/tasks.py

@@ -25,8 +25,7 @@
 from apps.backend.subscription import tools
 from apps.backend.subscription.constants import TASK_HOST_LIMIT
 from apps.backend.subscription.errors import SubscriptionInstanceEmpty
-from apps.backend.subscription.steps import StepFactory
-from apps.backend.subscription.steps.agent import InstallAgent, InstallProxy
+from apps.backend.subscription.steps import StepFactory, agent
 from apps.node_man import constants, models
 from apps.node_man import tools as node_man_tools
 from apps.node_man.handlers.cmdb import CmdbHandler
@@ -276,8 +275,12 @@ def create_task(
             continue
 
         # 新装AGENT或PROXY会保存安装信息，需要清理
-        # TODO IPv6 待修改
-        need_clean = step_action.get("agent") in [InstallAgent.ACTION_NAME, InstallProxy.ACTION_NAME]
+        need_clean = step_action.get("agent") in [
+            agent.InstallAgent.ACTION_NAME,
+            agent.InstallAgent2.ACTION_NAME,
+            agent.InstallProxy.ACTION_NAME,
+            agent.InstallProxy2.ACTION_NAME,
+        ]
         instance_info = instances[instance_id]
         host_info = instance_info["host"]
         record = models.SubscriptionInstanceRecord(

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

@@ -0,0 +1,193 @@
+# -*- coding: utf-8 -*-
+"""
+TencentBlueKing is pleased to support the open source community by making 蓝鲸智云-节点管理(BlueKing-BK-NODEMAN) available.
+Copyright (C) 2017-2022 THL A29 Limited, a Tencent company. All rights reserved.
+Licensed under the MIT License (the "License"); you may not use this file except in compliance with the License.
+You may obtain a copy of the License at https://opensource.org/licenses/MIT
+Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on
+an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the
+specific language governing permissions and limitations under the License.
+"""
+from typing import Callable, Dict, List
+
+import mock
+
+from apps.adapters.api import gse
+from apps.backend.components.collections.agent_new.components import (
+    UpgradeToAgentIDComponent,
+)
+from apps.mock_data import api_mkd
+from apps.mock_data import utils as mock_data_utils
+from apps.node_man import models
+from env.constants import GseVersion
+from pipeline.component_framework.test import ComponentTestCase, ExecuteAssertion
+
+from . import utils
+
+
+class UpgradeToAgentIDTestCaseMixin:
+    GSE_API_MOCK_PATHS: List[str] = ["apps.backend.components.collections.agent_new.upgrade_to_agent_id.GseApiHelper"]
+    GSE_VERSION = GseVersion.V2.value
+    upgrade_to_agent_id_func: Callable[[Dict], Dict] = lambda: {"success": [], "failed": []}
+
+    @classmethod
+    def setup_obj_factory(cls):
+        """设置 obj_factory"""
+        cls.obj_factory.init_host_num = 5
+
+    @classmethod
+    def get_default_case_name(cls) -> str:
+        return UpgradeToAgentIDComponent.name
+
+    @classmethod
+    def adjust_test_data_in_db(cls):
+        """
+        调整DB中的测试数据
+        :return:
+        """
+        for host_obj in cls.obj_factory.host_objs:
+            host_obj.bk_agent_id = f"{host_obj.bk_cloud_id}:{host_obj.inner_ip}"
+        models.Host.objects.bulk_update(cls.obj_factory.host_objs, fields=["bk_agent_id"])
+
+    @classmethod
+    def setUpTestData(cls):
+        super().setUpTestData()
+        cls.adjust_test_data_in_db()
+
+    def fetch_succeeded_sub_inst_ids(self) -> List[int]:
+        return self.common_inputs["subscription_instance_ids"]
+
+    def component_cls(self):
+        return UpgradeToAgentIDComponent
+
+    def init_mock_clients(self):
+        self.gse_api_mock_client = api_mkd.gse.utils.GseApiMockClient(
+            v2_proc_upgrade_to_agent_id_return=mock_data_utils.MockReturn(
+                return_type=mock_data_utils.MockReturnType.SIDE_EFFECT.value, return_obj=self.upgrade_to_agent_id_func
+            )
+        )
+
+    def setUp(self) -> None:
+        self.init_mock_clients()
+        for gse_api_mock_path in self.GSE_API_MOCK_PATHS:
+            mock.patch(
+                gse_api_mock_path,
+                gse.get_gse_api_helper(self.GSE_VERSION)(
+                    version=self.GSE_VERSION, gse_api_obj=self.gse_api_mock_client
+                ),
+            ).start()
+        super().setUp()
+
+
+class UpgradeToAgentIDSuccessTestCase(UpgradeToAgentIDTestCaseMixin, utils.AgentServiceBaseTestCase):
+    @staticmethod
+    def upgrade_to_agent_id_func(params):
+        return {
+            "success": [f'{host_info["bk_cloud_id"]}:{host_info["ip"]}' for host_info in params["hosts"]],
+            "failed": [],
+        }
+
+    @classmethod
+    def get_default_case_name(cls) -> str:
+        return "升级Agent-ID全部成功"
+
+    def cases(self):
+        return [
+            ComponentTestCase(
+                name=self.get_default_case_name(),
+                inputs=self.common_inputs,
+                parent_data={},
+                execute_assertion=ExecuteAssertion(
+                    success=bool(self.fetch_succeeded_sub_inst_ids()),
+                    outputs={"succeeded_subscription_instance_ids": self.fetch_succeeded_sub_inst_ids()},
+                ),
+                schedule_assertion=None,
+                execute_call_assertion=None,
+            )
+        ]
+
+
+class UpgradeToAgentIDFailedTestCase(UpgradeToAgentIDTestCaseMixin, utils.AgentServiceBaseTestCase):
+    @staticmethod
+    def upgrade_to_agent_id_func(params):
+        return {
+            "success": [],
+            "failed": [f'{host_info["bk_cloud_id"]}:{host_info["ip"]}' for host_info in params["hosts"]],
+        }
+
+    @classmethod
+    def get_default_case_name(cls) -> str:
+        return "升级Agent-ID全部失败"
+
+    def cases(self):
+        return [
+            ComponentTestCase(
+                name=self.get_default_case_name(),
+                inputs=self.common_inputs,
+                parent_data={},
+                execute_assertion=ExecuteAssertion(
+                    success=False,
+                    outputs={"succeeded_subscription_instance_ids": []},
+                ),
+                schedule_assertion=None,
+                execute_call_assertion=None,
+            )
+        ]
+
+
+class UpgradeToAgentIDHalfSuccessTestCase(UpgradeToAgentIDTestCaseMixin, utils.AgentServiceBaseTestCase):
+    @staticmethod
+    def upgrade_to_agent_id_func(params):
+        hosts = sorted(params["hosts"], key=lambda host: host["ip"])
+        # mock 前一半的主机为失败，后一半为成功
+        half_index = int(len(hosts) / 2)
+        return {
+            "success": [f'{host_info["bk_cloud_id"]}:{host_info["ip"]}' for host_info in hosts[:half_index]],
+            "failed": [f'{host_info["bk_cloud_id"]}:{host_info["ip"]}' for host_info in hosts[half_index:]],
+        }
+
+    @classmethod
+    def get_default_case_name(cls) -> str:
+        return "升级Agent-ID部分成功部分失败"
+
+    def fetch_succeeded_sub_inst_ids(self) -> List[int]:
+        # 最终成功的订阅实例只有前一半
+        return self.common_inputs["subscription_instance_ids"][: int(self.obj_factory.init_host_num / 2)]
+
+    def cases(self):
+        return [
+            ComponentTestCase(
+                name=self.get_default_case_name(),
+                inputs=self.common_inputs,
+                parent_data={},
+                execute_assertion=ExecuteAssertion(
+                    success=bool(self.fetch_succeeded_sub_inst_ids()),
+                    outputs={"succeeded_subscription_instance_ids": self.fetch_succeeded_sub_inst_ids()},
+                ),
+                schedule_assertion=None,
+                execute_call_assertion=None,
+            )
+        ]
+
+
+class UpgradeToAgentIDWithV1GseTestCase(UpgradeToAgentIDTestCaseMixin, utils.AgentServiceBaseTestCase):
+    GSE_VERSION = GseVersion.V1.value
+
+    @classmethod
+    def get_default_case_name(cls) -> str:
+        return "V1 GSE API 无 upgrade_to_agent_id，执行失败"
+
+    def cases(self):
+        return [
+            ComponentTestCase(
+                name=self.get_default_case_name(),
+                inputs=self.common_inputs,
+                parent_data={},
+                execute_assertion=ExecuteAssertion(
+                    success=False,
+                    outputs={},
+                ),
+                schedule_assertion=None,
+                execute_call_assertion=None,
+            )
+        ]

apps/backend/tests/utils/__init__.py

@@ -0,0 +1,10 @@
+# -*- coding: utf-8 -*-
+"""
+TencentBlueKing is pleased to support the open source community by making 蓝鲸智云-节点管理(BlueKing-BK-NODEMAN) available.
+Copyright (C) 2017-2022 THL A29 Limited, a Tencent company. All rights reserved.
+Licensed under the MIT License (the "License"); you may not use this file except in compliance with the License.
+You may obtain a copy of the License at https://opensource.org/licenses/MIT
+Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on
+an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the
+specific language governing permissions and limitations under the License.
+"""