optimization: ip 选择器动态分组先过滤失效分组再进行查询

gcloud/utils/cmdb.py

@@ -332,6 +332,16 @@ def get_notify_receivers(client, biz_cc_id, supplier_account, receiver_group, mo
     return result
 
 
+def get_dynamic_group_list(username, bk_biz_id, bk_supplier_account):
+    """获取业务下的所有动态分组列表"""
+    client = get_client_by_user(username)
+    kwargs = {"bk_biz_id": bk_biz_id, "bk_supplier_account": bk_supplier_account}
+    result = batch_request(client.cc.search_dynamic_group, kwargs, limit=200, check_iam_auth_fail=True)
+
+    dynamic_groups = [{"id": dynamic_group["id"]} for dynamic_group in result if dynamic_group["bk_obj_id"] == "host"]
+    return dynamic_groups
+
+
 def get_dynamic_group_host_list(username, bk_biz_id, bk_supplier_account, dynamic_group_id):
     """获取动态分组中对应主机列表"""
     client = get_client_by_user(username)

pipeline_plugins/cmdb_ip_picker/utils.py

@@ -20,6 +20,7 @@
 from gcloud.conf import settings
 from gcloud.exceptions import ApiRequestError
 from gcloud.utils import cmdb
+from gcloud.utils.cmdb import get_dynamic_group_list
 from gcloud.utils.handlers import handle_api_error
 from gcloud.utils.ip import extract_ip_from_ip_str, format_sundry_ip
 
@@ -283,6 +284,13 @@ def group_picker_handler(self, inputted_group):
         :params inputted_group: 动态分组信息列表, list
         """
         dynamic_group_ids = [dynamic_group["id"] for dynamic_group in inputted_group]
+        try:
+            existing_dynamic_groups = get_dynamic_group_list(self.username, self.bk_biz_id, self.bk_supplier_account)
+            existing_dynamic_group_ids = set([dynamic_group["id"] for dynamic_group in existing_dynamic_groups])
+            dynamic_group_ids = set(dynamic_group_ids) & existing_dynamic_group_ids
+        except Exception as e:
+            # 如果获取动态分组失败，则不做过滤
+            logger.info(f"[group_picker_handler]: get_dynamic_group_list error {e}")
         dynamic_groups_host = {}
         for dynamic_group_id in dynamic_group_ids:
             success, result = cmdb.get_dynamic_group_host_list(