feat: 订阅下发支持动态分组 (closed TencentBlueKing#2507)

# Reviewed, transaction id: 27776

apps/backend/subscription/tools.py

@@ -459,6 +459,9 @@ def get_host_detail_by_template(bk_obj_id, template_info_list: list, bk_biz_id:
     if not template_info_list:
         return []
 
+    # 补充实例所属模块ID
+    host_biz_relations = []
+
     fields = constants.CC_HOST_FIELDS
 
     if bk_obj_id == models.Subscription.NodeType.SERVICE_TEMPLATE:
@@ -468,6 +471,14 @@ def get_host_detail_by_template(bk_obj_id, template_info_list: list, bk_biz_id:
         host_info_result = batch_request(
             call_func, dict(bk_service_template_ids=template_ids, bk_biz_id=bk_biz_id, fields=fields)
         )
+    elif bk_obj_id == models.Subscription.NodeType.DYNAMIC_GROUP:
+        # 集群模板
+        call_func = client_v2.cc.find_host_by_set_template
+        template_ids = [info["bk_inst_id"] for info in template_info_list]
+        bk_set_ids = [info["bk_set_id"] for info in template_info_list]
+        host_info_result = batch_request(
+            call_func, dict(bk_set_template_ids=template_ids, bk_set_ids=bk_set_ids, bk_biz_id=bk_biz_id, fields=fields)
+        )
     else:
         # 集群模板
         call_func = client_v2.cc.find_host_by_set_template
@@ -486,6 +497,17 @@ def get_host_detail_by_template(bk_obj_id, template_info_list: list, bk_biz_id:
         host["bk_biz_name"] = host["bk_biz_name"] = biz_info[bk_biz_id].get("bk_biz_name")
         host["bk_cloud_name"] = cloud_id_name_map.get(str(host["bk_cloud_id"]))
 
+    bk_host_id_chunks = chunk_lists([instance["host"]["bk_host_id"] for instance in host_info_result], 500)
+    with ThreadPoolExecutor(max_workers=settings.CONCURRENT_NUMBER) as ex:
+        tasks = [
+            ex.submit(client_v2.cc.find_host_biz_relations, dict(bk_host_id=chunk, bk_biz_id=bk_biz_id))
+            for chunk in bk_host_id_chunks
+        ]
+        for future in as_completed(tasks):
+            host_biz_relations.extend(future.result())
+
+    host_info_result = add_host_module_info(host_biz_relations, host_info_result)
+
     return host_info_result
 
 
@@ -718,6 +740,14 @@ def set_template_scope_nodes(scope):
             # 兼容 service_template_id 不存在的场景
             if "service_template_id" in node and node["service_template_id"] in template_ids
         ]
+    elif scope["node_type"] == models.Subscription.NodeType.DYNAMIC_GROUP:
+        # 转化服务模板为node
+        scope["nodes"] = [
+            {"bk_inst_id": node["bk_module_id"], "bk_obj_id": "module"}
+            for node in modules_info
+            # 兼容 bk_set_id 不存在的场景
+            if "bk_set_id" in node and node["bk_set_id"] in template_ids
+        ]
     else:
         # 转化集群模板为node
         scope["nodes"] = [
@@ -823,6 +853,28 @@ def get_scope_labels_func(
     }
 
 
+def execute_dynamic_groups(nodes: List[dict], bk_biz_id: int, bk_obj_id: str, fields: List[str]):
+    params = [
+        {
+            "func": CCApi.execute_dynamic_group,
+            "params": {
+                "fields": fields,
+                "bk_biz_id": bk_biz_id,
+                "id": node["bk_group_id"],
+                "no_request": True,
+            },
+            "sort": "id",
+            "limit": constants.LIST_SERVICE_INSTANCE_DETAIL_LIMIT,
+        }
+        for node in nodes
+        if node["bk_obj_id"] == bk_obj_id
+    ]
+
+    return batch_call(
+        batch_request, params, extend_result=True, interval=constants.LIST_SERVICE_INSTANCE_DETAIL_INTERVAL
+    )
+
+
 def get_instances_by_scope_with_checker(
     scope: Dict[str, Union[Dict, int, Any]], steps: List[models.SubscriptionStep], *args, **kwargs
 ) -> Dict[str, Dict[str, Union[Dict, Any]]]:
@@ -888,7 +940,7 @@ def get_instances_by_scope(scope: Dict[str, Union[Dict, int, Any]]) -> Dict[str,
     else:
         module_to_topo = {}
 
-    nodes = scope["nodes"]
+    nodes: List[dict] = scope["nodes"]
     if not nodes:
         # 兼容节点为空的情况
         return {}
@@ -935,27 +987,15 @@ def get_instances_by_scope(scope: Dict[str, Union[Dict, int, Any]]) -> Dict[str,
         # 校验是否都选择了同一种模板
         bk_obj_id_set = check_instances_object_type(nodes)
         if scope["object_type"] == models.Subscription.ObjectType.HOST:
-            # 补充实例所属模块ID
-            host_biz_relations = []
             instances.extend(
                 [
                     {"host": inst}
                     for inst in get_host_detail_by_template(list(bk_obj_id_set)[0], nodes, bk_biz_id=bk_biz_id)
                 ]
             )
-            bk_host_id_chunks = chunk_lists([instance["host"]["bk_host_id"] for instance in instances], 500)
-            with ThreadPoolExecutor(max_workers=settings.CONCURRENT_NUMBER) as ex:
-                tasks = [
-                    ex.submit(client_v2.cc.find_host_biz_relations, dict(bk_host_id=chunk, bk_biz_id=bk_biz_id))
-                    for chunk in bk_host_id_chunks
-                ]
-                for future in as_completed(tasks):
-                    host_biz_relations.extend(future.result())
 
             # 转化模板为节点
             nodes = set_template_scope_nodes(scope)
-            instances = add_host_module_info(host_biz_relations, instances)
-
         else:
             # 补充服务实例中的信息
             # 转化模板为节点，**注意不可在get_service_instance_by_inst之后才转换**
@@ -964,6 +1004,119 @@ def get_instances_by_scope(scope: Dict[str, Union[Dict, int, Any]]) -> Dict[str,
                 [{"service": inst} for inst in get_service_instance_by_inst(bk_biz_id, nodes, module_to_topo)]
             )
 
+    # 按照动态分组查询
+    elif scope["node_type"] == models.Subscription.NodeType.DYNAMIC_GROUP:
+        # 获取动态分组主机信息
+        host_infos = execute_dynamic_groups(
+            nodes=nodes,
+            bk_biz_id=bk_biz_id,
+            bk_obj_id=constants.CmdbGroupObjId.HOST.value,
+            fields=["bk_host_id"],
+        )
+
+        # 获取动态分组集群信息
+        set_infos = execute_dynamic_groups(
+            nodes=nodes,
+            bk_biz_id=bk_biz_id,
+            bk_obj_id=constants.CmdbGroupObjId.SET.value,
+            fields=["bk_set_id", "set_template_id"],
+        )
+
+        if scope["object_type"] == models.Subscription.ObjectType.HOST:
+            # 根据主机信息填充主机实例
+            if host_infos:
+                instances.extend(
+                    [
+                        {"host": inst, "source": "host_infos"}
+                        for inst in get_host_detail(
+                            host_info_list=[
+                                {
+                                    "bk_biz_id": bk_biz_id,
+                                    "bk_host_id": host_info["bk_host_id"],
+                                }
+                                for host_info in host_infos
+                            ],
+                            bk_biz_id=bk_biz_id,
+                            source="get_instances_by_scope",
+                        )
+                    ]
+                )
+
+            # 根据集群信息填充主机实例
+            if set_infos:
+                instances.extend(
+                    [
+                        {"host": inst, "source": "set_infos"}
+                        for inst in get_host_detail_by_template(
+                            bk_obj_id=models.Subscription.NodeType.DYNAMIC_GROUP,
+                            template_info_list=[
+                                {
+                                    "bk_set_id": set_info["bk_set_id"],
+                                    "bk_inst_id": set_info["set_template_id"],
+                                }
+                                for set_info in set_infos
+                            ],
+                            bk_biz_id=bk_biz_id,
+                        )
+                    ]
+                )
+
+                # 转化模板为节点
+                nodes = set_template_scope_nodes(
+                    scope={
+                        "bk_biz_id": bk_biz_id,
+                        "node_type": models.Subscription.NodeType.DYNAMIC_GROUP,
+                        "nodes": [
+                            {
+                                "bk_inst_id": set_info["bk_set_id"],
+                            }
+                            for set_info in set_infos
+                        ],
+                    }
+                )
+
+            # 去重主机id去重
+            instances = list({instance["host"]["bk_host_id"]: instance for instance in instances}.values())
+
+        else:
+            # 根据主机信息填充服务实例
+            if host_infos:
+                instances.extend(
+                    [
+                        {"service": inst, "source": "host_infos"}
+                        for inst in get_service_instances(
+                            bk_biz_id=bk_biz_id,
+                            filter_id_list=[host_info["bk_host_id"] for host_info in host_infos],
+                            filter_field_name=FilterFieldName.BK_HOST_LIST,
+                            ignore_exception=False,
+                        )
+                    ]
+                )
+
+            # 根据集群西南西填充服务实例
+            if set_infos:
+                nodes = set_template_scope_nodes(
+                    scope={
+                        "bk_biz_id": bk_biz_id,
+                        "node_type": models.Subscription.NodeType.DYNAMIC_GROUP,
+                        "nodes": [
+                            {
+                                "bk_inst_id": set_info["bk_set_id"],
+                            }
+                            for set_info in set_infos
+                        ],
+                    }
+                )
+                instances.extend(
+                    [
+                        {"service": inst, "source": "set_infos"}
+                        for inst in get_service_instance_by_inst(bk_biz_id, nodes, module_to_topo)
+                    ]
+                )
+
+            # 根据服务实例id去重
+            instances = list({instance["service"]["id"]: instance for instance in instances}.values())
+
     if not need_register:
         # 补充必要的主机或实例相关信息
 
@@ -1010,7 +1163,7 @@ def get_instances_by_scope(scope: Dict[str, Union[Dict, int, Any]]) -> Dict[str,
     return instances_dict
 
 
-def add_host_info_to_instances(bk_biz_id: int, scope: Dict, instances: Dict):
+def add_host_info_to_instances(bk_biz_id: int, scope: Dict, instances: List):
     """
     补全实例的主机信息
     :param bk_biz_id: 业务ID
@@ -1087,7 +1240,7 @@ def add_scope_info_to_instances(nodes: List, scope: Dict, instances: List[Dict],
     :return:
     """
     for instance in instances:
-        if scope["node_type"] == models.Subscription.NodeType.INSTANCE:
+        if scope["node_type"] == models.Subscription.NodeType.INSTANCE or instance.pop("source", "") == "host_infos":
             _add_scope_info_to_inst_instances(scope, instance)
         else:
             _add_scope_info_to_topo_instances(scope, instance, nodes, module_to_topo)

apps/node_man/constants.py

@@ -1089,6 +1089,17 @@ def _get_member__alias_map(cls) -> Dict[Enum, str]:
         return {cls.V4: _("IPv4"), cls.V6: _("IPv6")}
 
 
+class CmdbGroupObjId(EnhanceEnum):
+    """IP 版本"""
+
+    HOST = "host"
+    SET = "set"
+
+    @classmethod
+    def _get_member__alias_map(cls) -> Dict[Enum, str]:
+        return {cls.HOST: _("主机"), cls.SET: _("集群")}
+
+
 class PolicyRollBackType:
     SUPPRESSED = "SUPPRESSED"
     LOSE_CONTROL = "LOSE_CONTROL"

apps/node_man/models.py

@@ -1824,12 +1824,14 @@ class NodeType(object):
         INSTANCE = "INSTANCE"
         SERVICE_TEMPLATE = "SERVICE_TEMPLATE"
         SET_TEMPLATE = "SET_TEMPLATE"
+        DYNAMIC_GROUP = "DYNAMIC_GROUP"
 
     NODE_TYPE_CHOICES = (
         (NodeType.TOPO, _("动态实例（拓扑）")),
         (NodeType.INSTANCE, _("静态实例")),
         (NodeType.SERVICE_TEMPLATE, _("服务模板")),
         (NodeType.SET_TEMPLATE, _("集群模板")),
+        (NodeType.DYNAMIC_GROUP, _("动态分组")),
     )
 
     class CategoryType(object):

common/api/domains.py

@@ -33,6 +33,7 @@ def gen_api_root(api_gw_env_key: str, suffix: str) -> str:
 
 # 蓝鲸平台模块域名
 CC_APIGATEWAY_ROOT_V2 = gen_api_root("BKAPP_BK_CC_APIGATEWAY", "cc")
+CC_APIGATEWAY_ROOT_V3 = gen_api_root("BKAPP_BK_CC_APIGATEWAY_V3", "cc")
 GSE_APIGATEWAY_ROOT = gen_api_root("BKAPP_BK_GSE_LEGACY_APIGATEWAY", "gse")
 GSE_APIGATEWAY_ROOT_V2 = gen_api_root("BKAPP_BK_GSE_APIGATEWAY", "gse")
 ESB_APIGATEWAY_ROOT_V2 = gen_api_root("BKAPP_BK_ESB_APIGATEWAY", "esb")

common/api/modules/cc.py

@@ -12,7 +12,7 @@
 from django.utils.translation import ugettext_lazy as _
 
 from ..base import BaseApi, DataAPI
-from ..domains import CC_APIGATEWAY_ROOT_V2
+from ..domains import CC_APIGATEWAY_ROOT_V2, CC_APIGATEWAY_ROOT_V3
 from .utils import add_esb_info_before_request
 
 
@@ -256,3 +256,39 @@ def __init__(self):
             before_request=add_esb_info_before_request,
             api_name="list_service_instance_detail",
         )
+        self.get_dynamic_group = DataAPI(
+            method="GET",
+            url=CC_APIGATEWAY_ROOT_V3 + "dynamicgroup/{bk_biz_id}/{id}",
+            module=self.MODULE,
+            simple_module=self.SIMPLE_MODULE,
+            description="查询指定动态分组",
+            before_request=add_esb_info_before_request,
+            api_name="get_dynamic_group",
+        )
+        self.execute_dynamic_group = DataAPI(
+            method="POST",
+            url=CC_APIGATEWAY_ROOT_V3 + "dynamicgroup/data/{bk_biz_id}/{id}",
+            module=self.MODULE,
+            simple_module=self.SIMPLE_MODULE,
+            description="执行动态分组",
+            before_request=add_esb_info_before_request,
+            api_name="search_set_v2",
+        )
+        self.post_findmany_hosts_search_with_biz = DataAPI(
+            method="POST",
+            url=CC_APIGATEWAY_ROOT_V3 + "findmany/hosts/search/with_biz",
+            module=self.MODULE,
+            simple_module=self.SIMPLE_MODULE,
+            description="根据动态分组主机筛选条件获取主机",
+            before_request=add_esb_info_before_request,
+            api_name="post_findmany_hosts_search_with_biz",
+        )
+        self.search_set_v2 = DataAPI(
+            method="POST",
+            url=CC_APIGATEWAY_ROOT_V3 + "set/search/{bk_supplier_account}/{bk_biz_id}",
+            module=self.MODULE,
+            simple_module=self.SIMPLE_MODULE,
+            description="查询集群",
+            before_request=add_esb_info_before_request,
+            api_name="search_set_v2",
+        )