feature: 订阅支持业务集 (closed #1724)

apps/backend/subscription/commons.py

@@ -86,8 +86,6 @@ def get_host_by_inst(bk_biz_id, inst_list):
             topo_cond = {"bk_obj_id": inst["bk_obj_id"], "bk_inst_id": inst["bk_inst_id"]}
             hosts.extend(list_biz_hosts(bk_biz_id, topo_cond, "find_host_by_topo"))
 
-        # TODO: 增加 bk_obj_id: biz_set
-
     if bk_biz_ids:
         # 业务查询
         for bk_biz_id in bk_biz_ids:

apps/backend/subscription/errors.py

@@ -169,3 +169,11 @@ class SubscriptionIncludeGrayBizError(AppBaseException):
     ERROR_CODE = 19
     MESSAGE = _("订阅任务包含Gse2.0灰度业务，任务将暂缓执行无需重复点击")
     MESSAGE_TPL = _("订阅任务包含Gse2.0灰度业务，任务将暂缓执行无需重复点击")
+
+
+class BizSetNullError(AppBaseException):
+    """订阅范围中的业务集包含的业务为空"""
+
+    ERROR_CODE = 20
+    MESSAGE = _("订阅范围中的业务集包含的业务为空")
+    MESSAGE_TPL = _("{订阅范围中的业务集包含的业务为空}")

apps/backend/subscription/serializers.py

@@ -19,7 +19,10 @@
 from apps.node_man import constants, models, tools
 from apps.node_man.models import ProcessStatus
 from apps.node_man.serializers import policy
-from apps.node_man.serializers.base import SubScopeInstSelectorSerializer
+from apps.node_man.serializers.base import (
+    ScopeTypeSerializer,
+    SubScopeInstSelectorSerializer,
+)
 from apps.utils import basic
 
 
@@ -127,7 +130,7 @@ class GetSubscriptionSerializer(GatewaySerializer):
 
 
 class UpdateSubscriptionSerializer(GatewaySerializer):
-    class UpdateScopeSerializer(SubScopeInstSelectorSerializer):
+    class UpdateScopeSerializer(SubScopeInstSelectorSerializer, ScopeTypeSerializer):
         node_type = serializers.ChoiceField(choices=models.Subscription.NODE_TYPE_CHOICES)
         nodes = serializers.ListField()
         bk_biz_id = serializers.IntegerField(required=False, default=None)
@@ -160,7 +163,7 @@ class SwitchSubscriptionSerializer(GatewaySerializer):
 
 
 class RunSubscriptionSerializer(GatewaySerializer):
-    class RunScopeSerializer(SubScopeInstSelectorSerializer):
+    class RunScopeSerializer(SubScopeInstSelectorSerializer, ScopeTypeSerializer):
         node_type = serializers.ChoiceField(choices=models.Subscription.NODE_TYPE_CHOICES, label="节点类型")
         nodes = serializers.ListField(child=serializers.DictField(), label="拓扑节点列表")
 

apps/backend/subscription/tools.py

@@ -18,7 +18,7 @@
 from concurrent.futures import ThreadPoolExecutor, as_completed
 from functools import wraps
 from itertools import groupby
-from typing import Any, Dict, List, Union
+from typing import Any, Dict, List, Set, Union
 
 from django.conf import settings
 from django.db.models import Q
@@ -28,6 +28,7 @@
 from apps.backend.subscription.commons import get_host_by_inst, list_biz_hosts
 from apps.backend.subscription.constants import SUBSCRIPTION_SCOPE_CACHE_TIME
 from apps.backend.subscription.errors import (
+    BizSetNullError,
     ConfigRenderFailed,
     MultipleObjectError,
     PipelineTreeParseError,
@@ -317,7 +318,7 @@ def get_modules_by_inst_list(inst_list, module_to_topo):
 
 
 def get_service_instance_by_inst(bk_biz_id, inst_list, module_to_topo):
-    module_ids, no_module_inst_list = get_modules_by_inst_list(inst_list, module_to_topo)
+    module_ids, __ = get_modules_by_inst_list(inst_list, module_to_topo)
     params = {"bk_biz_id": int(bk_biz_id), "with_name": True}
 
     service_instances = batch_request(client_v2.cc.list_service_instance_detail, params, sort="id")
@@ -665,8 +666,8 @@ def support_multi_biz(get_instances_by_scope_func):
     @wraps(get_instances_by_scope_func)
     def wrapper(scope: Dict[str, Union[Dict, Any]], *args, **kwargs) -> Dict[str, Dict[str, Union[Dict, Any]]]:
         if scope.get("scope_type") == models.Subscription.ScopeType.BIZ_SET:
-            covert_biz_set_scope_to_scope(biz_set_scope=scope)
-        if scope.get("bk_biz_id") is not None:
+            covert_biz_set_scope_to_scope(biz_set_scope=scope, biz_set_id=scope.get("scope_id"))
+        elif scope.get("bk_biz_id") is not None:
             return get_instances_by_scope_func(scope, **kwargs)
         # 兼容只传bk_host_id的情况
         if (
@@ -699,7 +700,7 @@ def wrapper(scope: Dict[str, Union[Dict, Any]], *args, **kwargs) -> Dict[str, Di
     return wrapper
 
 
-def covert_biz_set_scope_to_scope(biz_set_scope: Dict[str, Union[Dict, int, str]]):
+def covert_biz_set_scope_to_scope(biz_set_scope: Dict[str, Union[Dict, int, str]], biz_set_id: int = None):
     """
      对于整个业务集范围的类型转换为多业务范围
      对于其他类型只过滤掉所有不在当前业务集中的 node
@@ -728,31 +729,27 @@ def covert_biz_set_scope_to_scope(biz_set_scope: Dict[str, Union[Dict, int, str]
     }
 
     """
-    fields: List[str] = ["bk_biz_id", "bk_biz_name"]
-    nodes_bizs = list(set([node["bk_biz_id"] for node in biz_set_scope["nodes"]]))
-    business_set_query_rule = {"field": "bk_biz_id", "operator": "in", "value": nodes_bizs}
-    biz_set_info = client_v2.cc.list_business_in_business_set(
-        {
-            "fields": fields,
-            "bk_biz_set_id": biz_set_scope["scope_id"],
-            "filter": {"condition": "AND", "rules": business_set_query_rule},
-            "page": {"start": 0, "limit": 500, "enable_count": False, "sort": "bk_biz_id"},
-        }
-    )
+    from apps.node_man.handlers.cmdb import CmdbHandler
 
-    bk_biz_list: List[int] = []
-    for biz in biz_set_info.get("info") or []:
-        bk_biz_list.append(int(biz["bk_biz_id"]))
+    bk_obj_id_set: Set[str] = check_instances_object_type(biz_set_scope["nodes"])
+    # 如果 node 层级上面指定了业务集， 则只对改业务集进行下发, 否则多业务集都转换为业务 ID
+    if biz_set_id is not None:
+        biz_set_ids: List[int] = [biz_set_id]
+    else:
+        biz_set_ids: List[int] = list(set([node["bk_inst_id"] for node in biz_set_scope["nodes"]]))
 
-    # 过滤 nodes 中的混合部分，如果是混合的就报错
-    bk_obj_id_set = check_instances_object_type(biz_set_scope["nodes"])
-    if bk_obj_id_set[0] == "biz_set":
+    bk_biz_ids: List[int] = CmdbHandler.list_biz_ids_in_biz_set(biz_set_ids=biz_set_ids)
+    if not bk_biz_ids:
+        raise BizSetNullError
 
+    if list(bk_obj_id_set)[0] == "biz_set":
+        # 业务集转换为多业务
         biz_set_scope["nodes"] = [
-            {"bk_obj_id": "biz", "bk_inst_id": bk_biz_id, "bk_biz_id": bk_biz_id} for bk_biz_id in bk_biz_list
+            {"bk_obj_id": "biz", "bk_inst_id": bk_biz_id, "bk_biz_id": bk_biz_id} for bk_biz_id in bk_biz_ids
         ]
     else:
-        biz_set_scope["nodes"] = [node for node in biz_set_scope["nodes"] if node["bk_biz_id"] in bk_biz_list]
+        # 不包括在业务集中的业务和未指定具体业务的 node 将被剔除
+        biz_set_scope["nodes"] = [node for node in biz_set_scope["nodes"] if node.get("bk_biz_id") in bk_biz_ids]
 
 
 @support_multi_biz

apps/backend/subscription/views.py

@@ -83,6 +83,8 @@ def create_subscription(self, request):
                 bk_biz_id=scope["bk_biz_id"],
                 object_type=scope["object_type"],
                 node_type=scope["node_type"],
+                scope_type=scope["scope_type"],
+                scope_id=scope["scope_id"],
                 nodes=scope["nodes"],
                 instance_selector=scope.get("instance_selector"),
                 target_hosts=params.get("target_hosts"),
@@ -160,6 +162,8 @@ def info(self, request):
                     "bk_biz_id": subscription.bk_biz_id,
                     "object_type": subscription.object_type,
                     "node_type": subscription.node_type,
+                    "scope_type": subscription.scope_type,
+                    "scope_id": subscription.scope_id,
                     "nodes": subscription.nodes,
                 },
                 "pid": subscription.pid,

apps/node_man/handlers/cmdb.py

@@ -35,7 +35,7 @@
     cache_all_biz_topo_delay_task,
     get_and_cache_format_biz_topo,
 )
-from apps.utils import APIModel
+from apps.utils import APIModel, basic
 from apps.utils.batch_request import batch_request, request_multi_thread
 from apps.utils.local import get_request_username
 from common.log import logger
@@ -582,3 +582,32 @@ def find_host_service_template(bk_host_ids: List[int]) -> List[Dict]:
     @staticmethod
     def get_biz_service_template(bk_biz_id: int) -> List[Dict]:
         return batch_request(client_v2.cc.list_service_template, {"bk_biz_id": bk_biz_id})
+
+    @staticmethod
+    def list_biz_ids_in_biz_set(biz_set_ids: List[int]):
+        group_bk_biz_ids: List[int] = []
+        for biz_set_id in biz_set_ids:
+            # 将所有的业务集包括的业务 ID 查询出来后聚合去重复
+            biz_count: int = int(
+                client_v2.cc.list_business_in_business_set(
+                    {"bk_biz_set_id": biz_set_id, "page": {"enable_count": True}}
+                )["count"]
+            )
+            prams_list = [
+                {
+                    "fields": ["bk_biz_id"],
+                    "bk_biz_set_id": biz_set_id,
+                    "page": {
+                        "start": index * constants.QUERY_CMDB_LIMIT,
+                        "limit": count,
+                        "enable_count": False,
+                        "sort": "bk_biz_id",
+                    },
+                }
+                for index, count in enumerate(basic.number_slice(biz_count, constants.QUERY_CMDB_LIMIT))
+            ]
+            set__biz_info = request_multi_thread(
+                client_v2.cc.list_business_in_business_set, prams_list, get_data=lambda x: x["info"]
+            )
+            group_bk_biz_ids.extend([x["bk_biz_id"] for x in set__biz_info])
+        return list(set(group_bk_biz_ids))

apps/node_man/migrations/0076_auto_20230926_1745.py

@@ -0,0 +1,29 @@
+# Generated by Django 3.2.4 on 2023-09-26 09:45
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ("node_man", "0075_pluginconfigtemplate_variables"),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name="subscription",
+            name="scope_id",
+            field=models.IntegerField(db_index=True, null=True, verbose_name="订阅集合类型ID"),
+        ),
+        migrations.AddField(
+            model_name="subscription",
+            name="scope_type",
+            field=models.CharField(
+                choices=[("HOST", "主机"), ("SERVICE", "服务")],
+                db_index=True,
+                max_length=20,
+                null=True,
+                verbose_name="订阅集合类型",
+            ),
+        ),
+    ]

apps/node_man/models.py

@@ -1792,7 +1792,7 @@ class CategoryType(object):
 
     name = models.CharField(_("任务名称"), max_length=64, null=True, blank=True)
     bk_biz_id = models.IntegerField(_("业务ID"), db_index=True, null=True)
-    scope_type = models.CharField(_("订阅集合类型"), db_index=True, null=True, choices=SCOPE_TYPE_CHOICES)
+    scope_type = models.CharField(_("订阅集合类型"), db_index=True, null=True, choices=OBJECT_TYPE_CHOICES, max_length=20)
     scope_id = models.IntegerField(_("订阅集合类型ID"), db_index=True, null=True)
     object_type = models.CharField(_("对象类型"), max_length=20, choices=OBJECT_TYPE_CHOICES, db_index=True)
     node_type = models.CharField(_("节点类型"), max_length=20, choices=NODE_TYPE_CHOICES, db_index=True)

apps/node_man/serializers/base.py

@@ -17,11 +17,12 @@
 
 # 放在后台会导致循坏导入
 class SubScopeInstSelectorSerializer(serializers.Serializer):
-    instance_selector = serializers.ListField(
-        child=serializers.DictField(),
-        required=False,
-        label="实例筛选器"
-    )
+    instance_selector = serializers.ListField(child=serializers.DictField(), required=False, label="实例筛选器")
+
+
+class ScopeTypeSerializer(serializers.Serializer):
+    scope_type = serializers.CharField(required=False, default=None)
+    scope_id = serializers.IntegerField(required=False, default=None)
 
 
 # 安装插件配置

apps/utils/basic.py

@@ -10,6 +10,7 @@
 """
 import ipaddress
 import json
+import math
 from collections import Counter, namedtuple
 from copy import deepcopy
 from functools import partial
@@ -140,6 +141,18 @@ def list_slice(lst: List[Any], limit: int) -> List[List[Any]]:
     return slice_list
 
 
+def number_slice(number: int, limit: int) -> List[int]:
+    num_chunks = math.ceil(number / limit)
+    chunks = []
+    for i in range(num_chunks):
+        if i == num_chunks - 1:
+            # 如果是最后一个分片，加入余数
+            chunks.append(number % limit)
+        else:
+            chunks.append(limit)
+    return chunks
+
+
 def to_int_or_default(val: Any, default: Any = None) -> Union[int, Any, None]:
     try:
         return int(val)

blueking/component/apis/cc.py

@@ -779,3 +779,9 @@ def __init__(self, client):
             path="/api/c/compapi{bk_api_ver}/cc/list_service_template/",
             description="查询业务服务模板列表",
         )
+        self.list_business_in_business_set = ComponentAPI(
+            client=self.client,
+            method="POST",
+            path="/api/c/compapi{bk_api_ver}/cc/list_business_in_business_set/",
+            description="查询业务集中的业务列表",
+        )