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

apps/backend/subscription/errors.py

@@ -169,3 +169,19 @@ class SubscriptionIncludeGrayBizError(AppBaseException):
     ERROR_CODE = 19
     MESSAGE = _("订阅任务包含Gse2.0灰度业务，任务将暂缓执行无需重复点击")
     MESSAGE_TPL = _("订阅任务包含Gse2.0灰度业务，任务将暂缓执行无需重复点击")
+
+
+class MultiBizSetError(AppBaseException):
+    """订阅范围不支持多业务集"""
+
+    ERROR_CODE = 20
+    MESSAGE = _("订阅范围不支持多业务集")
+    MESSAGE_TPL = _("{订阅范围不支持多业务集}")
+
+
+class BizSetNotExsitError(AppBaseException):
+    """业务集不存在"""
+
+    ERROR_CODE = 21
+    MESSAGE = _("业务集不存在")
+    MESSAGE_TPL = _("{业务集[id: {biz_set_id}] 不存在}")

apps/backend/subscription/serializers.py

@@ -19,7 +19,11 @@
 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,
+    biz_set_scope_params_valid,
+)
 from apps.utils import basic
 
 
@@ -28,16 +32,15 @@ class GatewaySerializer(serializers.Serializer):
     bk_app_code = serializers.CharField()
 
 
-class ScopeSerializer(SubScopeInstSelectorSerializer):
-    bk_biz_id = serializers.IntegerField(required=False, default=None)
+class ScopeSerializer(SubScopeInstSelectorSerializer, ScopeTypeSerializer):
     # TODO: 是否取消掉这个范围内的scope
     bk_biz_scope = serializers.ListField(required=False)
     object_type = serializers.ChoiceField(choices=models.Subscription.OBJECT_TYPE_CHOICES, label="对象类型")
     node_type = serializers.ChoiceField(choices=models.Subscription.NODE_TYPE_CHOICES, label="节点类别")
     need_register = serializers.BooleanField(required=False, default=False, label="是否需要注册到CMDB")
-    nodes = serializers.ListField(child=serializers.DictField())
 
     def validate(self, attrs):
+        biz_set_scope_params_valid(attrs=attrs)
         for node in attrs["nodes"]:
             basic.ipv6_formatter(data=node, ipv6_field_names=["ip"])
             basic.ipv6_formatter(
@@ -129,6 +132,7 @@ class UpdateScopeSerializer(SubScopeInstSelectorSerializer):
         node_type = serializers.ChoiceField(choices=models.Subscription.NODE_TYPE_CHOICES)
         nodes = serializers.ListField()
         bk_biz_id = serializers.IntegerField(required=False, default=None)
+        scope_id = serializers.IntegerField(required=False, default=None)
 
     class UpdateStepSerializer(serializers.Serializer):
         id = serializers.CharField()

apps/backend/subscription/tasks.py

@@ -590,6 +590,8 @@ def run_subscription_task_and_create_instance(
     else:
         scope["object_type"] = subscription.object_type
         scope["bk_biz_id"] = subscription.bk_biz_id
+        scope["scope_type"] = subscription.scope_type
+        scope["scope_id"] = subscription.scope_id
 
     # 获取订阅范围内全部实例
     instances = tools.get_instances_by_scope(scope, source="run_subscription_task_and_create_instance")
@@ -697,7 +699,8 @@ def run_subscription_task_and_create_instance(
         )
         deleted_instance_info = tools.get_instances_by_scope(
             {
-                "bk_biz_id": subscription.bk_biz_id,
+                "scope_id": subscription.scope["scope_id"],
+                "scope_type": subscription.scope_type,
                 "object_type": subscription.object_type,
                 "node_type": models.Subscription.NodeType.INSTANCE,
                 "nodes": deleted_id_not_in_scope,

apps/backend/subscription/tools.py

@@ -20,7 +20,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
@@ -31,6 +31,7 @@
 from apps.backend.subscription.constants import SUBSCRIPTION_SCOPE_CACHE_TIME
 from apps.backend.subscription.errors import (
     ConfigRenderFailed,
+    MultiBizSetError,
     MultipleObjectError,
     PipelineTreeParseError,
 )
@@ -91,7 +92,7 @@ def parse_group_id(group_id: str) -> Dict:
         "id": 1,
     }
     """
-    source_type, subscription_id, object_type, _id = group_id.split("_")
+    __, subscription_id, object_type, _id = group_id.split("_")
     return {
         "subscription_id": subscription_id,
         "object_type": object_type,
@@ -322,7 +323,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")
@@ -373,7 +374,9 @@ def fetch_biz_info(bk_biz_ids: typing.List[int]) -> typing.Dict[int, typing.Dict
     return {bk_biz_id: biz_info_map.get(str(bk_biz_id)) or {} for bk_biz_id in bk_biz_ids}
 
 
-def get_host_detail_by_template(bk_obj_id, template_info_list: list, bk_biz_id: int = None):
+def get_host_detail_by_template(
+    bk_obj_id, template_info_list: typing.List[typing.Dict[str, Any]], bk_biz_id: int = None
+):
     """
     根据集群模板ID/服务模板ID获得主机的详细信息
     :param bk_obj_id: 模板类型
@@ -678,22 +681,37 @@ 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("bk_biz_id") is not None:
-            return get_instances_by_scope_func(scope, **kwargs)
-        # 兼容只传bk_host_id的情况
-        if (
-            scope["object_type"] == models.Subscription.ObjectType.HOST
-            and scope["node_type"] == models.Subscription.NodeType.INSTANCE
-        ):
-            if None in [node.get("bk_biz_id") for node in scope["nodes"]]:
+        scope_type: str = scope.get("scope_type")
+        scope_id: int = scope.get("scope_id")
+        # 如果是业务集 直接进入转换逻辑
+        # 如果是 实例主机类型，需要处理 BIZ 类型
+        if scope_type == models.Subscription.ScopeType.BIZ_SET:
+            covert_biz_set_scope_to_scope(biz_set_scope=scope)
+        else:
+            # 非业务集类型的订阅自动填充为业务类型
+            scope["scope_type"] = models.Subscription.ScopeType.BIZ
+            if scope_id:
                 return get_instances_by_scope_func(scope, **kwargs)
+            else:
+                bk_biz_id: int = scope.get("bk_biz_id")
+                if bk_biz_id:
+                    scope["scope_id"] = scope.get("bk_biz_id")
+                    return get_instances_by_scope_func(scope, **kwargs)
+            if (
+                scope["object_type"] == models.Subscription.ObjectType.HOST
+                and scope["node_type"] == models.Subscription.NodeType.INSTANCE
+            ):
+                if None in [node.get("bk_biz_id") for node in scope["nodes"]]:
+                    # 兼容只传 bk_host_id 的情况
+                    return get_instances_by_scope_func(scope, **kwargs)
 
         instance_id_info_map = {}
-        nodes = sorted(scope["nodes"], key=lambda node: node["bk_biz_id"])
-        params_list = [
+        nodes: List[Dict[str, Union[str, int]]] = sorted(scope["nodes"], key=lambda node: node["bk_biz_id"])
+        params_list: List[Dict[str, Dict[str, Any]]] = [
             {
                 "scope": {
-                    "bk_biz_id": bk_biz_id,
+                    "scope_type": models.Subscription.ScopeType.BIZ,
+                    "scope_id": bk_biz_id,
                     "object_type": scope["object_type"],
                     "node_type": scope["node_type"],
                     "nodes": list(nodes),
@@ -730,6 +748,56 @@ def get_scope_labels_func(
     }
 
 
+def covert_biz_set_scope_to_scope(biz_set_scope: Dict[str, Union[int, str, List[Dict[str, Union[str, int]]]]]):
+    """
+     对于整个业务集范围的类型转换为多业务范围
+     对于其他类型只过滤掉所有不在当前业务集中的 node
+
+    {
+    "nodes": [
+        {
+            "bk_obj_id": "biz_set",
+            "bk_inst_id": 10
+        }
+    }
+    转换为
+    {
+     "nodes": [
+        {
+            "bk_obj_id": "biz",
+            "bk_inst_id": 2,
+            "bk_biz_id": 2
+        },
+        {
+            "bk_obj_id": "biz",
+            "bk_inst_id": 3,
+            "bk_biz_id": 3
+        }
+    ]
+    }
+
+    """
+    from apps.node_man.handlers.cmdb import CmdbHandler
+
+    nodes: List[Dict[str, Union[int, str]]] = biz_set_scope["nodes"]
+    biz_set_id: int = biz_set_scope["scope_id"]
+
+    bk_biz_ids: List[int] = CmdbHandler.list_biz_ids_in_biz_set(biz_set_id=biz_set_id)
+    bk_obj_id_set: Set[str] = {node.get("bk_obj_id") for node in nodes}
+
+    if "biz_set" in list(bk_obj_id_set):
+        if len(nodes) > 1:
+            raise MultiBizSetError
+        else:
+            # 业务集范围转换为多业务
+            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_ids
+            ]
+    else:
+        # 不包括在业务集中的业务和未指定具体业务的 node 将被剔除
+        biz_set_scope["nodes"] = [node for node in nodes if node.get("bk_biz_id") in bk_biz_ids]
+
+
 @support_multi_biz
 @SetupObserve(histogram=metrics.app_task_get_instances_by_scope_duration_seconds, get_labels_func=get_scope_labels_func)
 @FuncCacheDecorator(cache_time=SUBSCRIPTION_SCOPE_CACHE_TIME)
@@ -771,7 +839,8 @@ def get_instances_by_scope(scope: Dict[str, Union[Dict, int, Any]]) -> Dict[str,
         return {}
 
     instances = []
-    bk_biz_id = scope["bk_biz_id"]
+    # 业务类型的订阅范围，认为 scope_id 就是 bk_biz_id
+    bk_biz_id = scope["scope_id"]
     if bk_biz_id:
         module_to_topo = get_module_to_topo_dict(bk_biz_id)
     else:

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,
@@ -209,6 +213,8 @@ def update_subscription(self, request):
             # 策略部署新增
             subscription.plugin_name = params.get("plugin_name")
             subscription.bk_biz_scope = params.get("bk_biz_scope")
+            # 业务集
+            subscription.scope_id = params.get("scope_id")
             subscription.save()
 
             step_ids: Set[str] = set()

apps/backend/tests/components/collections/plugin/test_check_agent_status.py

@@ -16,6 +16,7 @@
     BK_HOST_ID,
     PluginTestObjFactory,
 )
+from apps.node_man import models
 from pipeline.component_framework.test import (
     ComponentTestCase,
     ComponentTestMixin,
@@ -27,6 +28,9 @@
 class CheckAgentStatusTest(TestCase, ComponentTestMixin):
     def setUp(self):
         self.ids = PluginTestObjFactory.init_db()
+        models.ProcessStatus.objects.filter(bk_host_id=self.ids["bk_host_id"]).update(
+            name=models.ProcessStatus.GSE_AGENT_PROCESS_NAME
+        )
         self.COMMON_INPUTS = PluginTestObjFactory.inputs(
             attr_values={
                 "description": "description",