fix: 服务节点条件下查询服务实例详情因查询数量过多导致esb内存增加 (closed #2193)

apps/backend/constants.py

@@ -142,3 +142,21 @@ def _get_member__alias_map(cls) -> Dict[Enum, str]:
 class ProxyConfigFile(enum.EnhanceEnum):
     V1 = ["agent.conf", "btsvr.conf", "transit.conf", "opts.conf", "plugin_info.json", "data.conf", "dataflow.conf"]
     V2 = ["gse_agent.conf", "gse_data_proxy.conf", "gse_file_proxy.conf"]
+
+
+class FilterFieldName(enum.EnhanceEnum):
+    BK_HOST_LIST = "bk_host_list"
+    SERVICE_INSTANCE_IDS = "service_instance_ids"
+
+    @classmethod
+    def _get_member__alias_map(cls) -> Dict[Enum, str]:
+        return {cls.BK_HOST_LIST: _("主机id列表"), cls.SERVICE_INSTANCE_IDS: _("服务实例id列表")}
+
+    @property
+    def needs_batch_request(self) -> bool:
+        if self == FilterFieldName.BK_HOST_LIST:
+            return True
+        elif self == FilterFieldName.SERVICE_INSTANCE_IDS:
+            return False
+
+        return True

apps/backend/plugin/views.py

@@ -38,6 +38,7 @@
 
 from apps.backend import constants as backend_const
 from apps.backend import exceptions
+from apps.backend.constants import FilterFieldName
 from apps.backend.plugin import serializers, tasks, tools
 from apps.backend.plugin.handler import PluginHandler
 from apps.backend.subscription.errors import (
@@ -46,7 +47,7 @@
 )
 from apps.backend.subscription.handler import SubscriptionHandler
 from apps.backend.subscription.tasks import run_subscription_task_and_create_instance
-from apps.backend.subscription.tools import get_service_instance_by_ids
+from apps.backend.subscription.tools import get_service_instances
 from apps.core.files import core_files_constants
 from apps.core.files.storage import get_storage
 from apps.exceptions import AppBaseException, ValidationError
@@ -574,8 +575,11 @@ def start_debug(self, request):
         else:
             bk_biz_id: int = instance_info["bk_biz_id"]
             service_instance_id: Optional[int] = instance_info["id"]
-            service_instance_result: List[Dict[str, Any]] = get_service_instance_by_ids(
-                bk_biz_id=instance_info["bk_biz_id"], ids=[service_instance_id]
+            service_instance_result: List[Dict[str, Any]] = get_service_instances(
+                bk_biz_id=instance_info["bk_biz_id"],
+                filter_id_list=[service_instance_id],
+                filter_field_name=FilterFieldName.SERVICE_INSTANCE_IDS,
+                ignore_exception=False,
             )
             try:
                 bk_host_id: int = service_instance_result[0]["bk_host_id"]

apps/backend/subscription/tools.py

@@ -27,7 +27,7 @@
 from django.utils import timezone
 
 from apps.backend.components.collections import core
-from apps.backend.constants import InstNodeType
+from apps.backend.constants import FilterFieldName, InstNodeType
 from apps.backend.subscription import task_tools
 from apps.backend.subscription.commons import get_host_by_inst, list_biz_hosts
 from apps.backend.subscription.constants import SUBSCRIPTION_SCOPE_CACHE_TIME
@@ -48,6 +48,7 @@
 from apps.utils import concurrent
 from apps.utils.basic import chunk_lists, distinct_dict_list, order_dict
 from apps.utils.batch_request import batch_request, request_multi_thread
+from apps.utils.concurrent import batch_call
 from apps.utils.time_handler import strftime_local
 from common.api import CCApi
 
@@ -278,31 +279,46 @@ def find_host_biz_relations(bk_host_ids: List[int]) -> List[Dict]:
 
 
 @controller.ConcurrentController(
-    data_list_name="bk_host_list",
+    data_list_name="filter_id_list",
     batch_call_func=concurrent.batch_call,
     get_config_dict_func=core.get_config_dict,
     get_config_dict_kwargs={"config_name": core.ServiceCCConfigName.CMDB_QUERY.value},
 )
-def get_service_instances(bk_biz_id: int, bk_host_list: List[int]) -> List[Dict]:
+def get_service_instances(
+    bk_biz_id: int, filter_id_list: List[int], filter_field_name: FilterFieldName, ignore_exception: bool = True
+) -> List[Dict]:
     """
     分批查询业务主机进程
     :param bk_biz_id: 业务ID
-    :param bk_host_list: 实例主机id列表
+    :param filter_id_list: 筛选字段列表
+    :param filter_field_name: 筛选字段名字
+    :param ignore_exception: 是否忽略错误
     :return: 主机进程
     """
+    if not filter_id_list:
+        return []
+
+    params = {
+        "bk_biz_id": int(bk_biz_id),
+        "with_name": True,
+        filter_field_name.value: filter_id_list,
+    }
     try:
-        params = {
-            "bk_biz_id": int(bk_biz_id),
-            "with_name": True,
-            "bk_host_list": bk_host_list,
-            "page": {
+        if filter_field_name.needs_batch_request:
+            result = batch_request(CCApi.list_service_instance_detail, params)
+        else:
+            params["page"] = {
                 "start": 0,
-                "limit": len(bk_host_list),
-            },
-        }
-        result = CCApi.list_service_instance_detail(params)["info"]
+                "limit": len(filter_id_list),
+            }
+            result = CCApi.list_service_instance_detail(params)["info"]
     except Exception:
-        logger.exception(f"Failed to list_service_instance_detail with biz_id={bk_biz_id}, bk_host_list={bk_host_list}")
+        logger.exception(
+            f"Failed to list_service_instance_detail with biz_id={bk_biz_id}, "
+            f"{filter_field_name.value}={filter_id_list}"
+        )
+        if not ignore_exception:
+            raise
         service_instances = []
     else:
         service_instances = result
@@ -322,7 +338,12 @@ def get_process_by_biz_id(bk_biz_id: int, bk_host_list: List[int]) -> Dict:
         }
     }
     """
-    service_instances = get_service_instances(bk_biz_id=bk_biz_id, bk_host_list=bk_host_list)
+    service_instances = get_service_instances(
+        bk_biz_id=bk_biz_id,
+        filter_id_list=bk_host_list,
+        filter_field_name=FilterFieldName.BK_HOST_LIST,
+        ignore_exception=True,
+    )
 
     host_processes = defaultdict(dict)
 
@@ -354,9 +375,27 @@ 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)
-    params = {"bk_biz_id": int(bk_biz_id), "with_name": True}
+    if not module_ids:
+        return []
 
-    service_instances = batch_request(client_v2.cc.list_service_instance_detail, params, sort="id")
+    if len(module_ids) <= constants.QUERY_MODULE_ID_THRESHOLD:
+        params = [
+            {
+                "func": CCApi.list_service_instance_detail,
+                "params": {
+                    "bk_biz_id": int(bk_biz_id),
+                    "with_name": True,
+                    "bk_module_id": bk_module_id,
+                },
+                "sort": "id",
+            }
+            for bk_module_id in module_ids
+        ]
+
+        service_instances = batch_call(batch_request, params, extend_result=True)
+    else:
+        params = {"bk_biz_id": int(bk_biz_id), "with_name": True}
+        service_instances = batch_request(CCApi.list_service_instance_detail, params, sort="id")
 
     service_instances = [
         service_instance for service_instance in service_instances if service_instance["bk_module_id"] in module_ids
@@ -365,23 +404,6 @@ def get_service_instance_by_inst(bk_biz_id, inst_list, module_to_topo):
     return service_instances
 
 
-def get_service_instance_by_ids(bk_biz_id, ids):
-    """
-    根据服务实例id获取服务实例详情
-    :param bk_biz_id: int 业务id
-    :param ids: list 服务实例id
-    :return:
-    """
-    params = {
-        "bk_biz_id": int(bk_biz_id),
-        "with_name": True,
-        "service_instance_ids": ids,
-    }
-
-    result = batch_request(client_v2.cc.list_service_instance_detail, params)
-    return result
-
-
 @FuncCacheDecorator(cache_time=1 * constants.TimeUnit.MINUTE)
 def fetch_biz_info_map(fields: typing.Optional[typing.List[str]] = None) -> typing.Dict[str, typing.Dict]:
     """
@@ -847,7 +869,15 @@ def get_instances_by_scope(scope: Dict[str, Union[Dict, int, Any]]) -> Dict[str,
         else:
             service_instance_ids = [int(node["id"]) for node in nodes]
             instances.extend(
-                [{"service": inst} for inst in get_service_instance_by_ids(bk_biz_id, service_instance_ids)]
+                [
+                    {"service": inst}
+                    for inst in get_service_instances(
+                        bk_biz_id=bk_biz_id,
+                        filter_id_list=service_instance_ids,
+                        filter_field_name=FilterFieldName.SERVICE_INSTANCE_IDS,
+                        ignore_exception=False,
+                    )
+                ]
             )
 
     # 按照模板查询

apps/backend/tests/api/test_debug.py

@@ -42,9 +42,7 @@ def setUp(self):
             }
         ]
 
-        mock.patch(
-            "apps.backend.plugin.views.get_service_instance_by_ids", return_value=service_instance_result
-        ).start()
+        mock.patch("apps.backend.plugin.views.get_service_instances", return_value=service_instance_result).start()
 
     def test_host_info_debug(self):
         base_host_info_data: typing.Dict[str, typing.Union[str, int]] = {

apps/node_man/constants.py

@@ -557,6 +557,7 @@ def _get_member__alias_map(cls) -> Dict[Enum, str]:
 QUERY_CMDB_MODULE_LIMIT = 500
 QUERY_CLOUD_LIMIT = 200
 QUERY_HOST_SERVICE_TEMPLATE_LIMIT = 200
+QUERY_MODULE_ID_THRESHOLD = 15
 VERSION_PATTERN = re.compile(r"[vV]?(\d+\.){1,5}\d+(-rc\d)?$")
 # 语义化版本正则，参考：https://semver.org/#is-there-a-suggested-regular-expression-regex-to-check-a-semver-string
 SEMANTIC_VERSION_PATTERN = re.compile(