feature: JOB API 适配 (closed #785)

apps/backend/components/collections/agent_new/check_policy_gse_to_proxy.py

@@ -8,7 +8,7 @@
 an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the
 specific language governing permissions and limitations under the License.
 """
-from typing import Any, Dict, List
+from typing import Any, Dict, List, Union
 
 from django.utils.translation import ugettext_lazy as _
 
@@ -50,17 +50,26 @@ class CheckPolicyGseToProxyService(AgentExecuteScriptService):
     def script_name(self):
         return "is_target_reachable"
 
-    def get_target_servers(self, data, common_data: CommonData, host: models.Host) -> List[Dict[str, Any]]:
+    def get_target_servers(
+        self, data, common_data: CommonData, host: models.Host
+    ) -> Dict[str, Union[List[Dict[str, Any]], List[int]]]:
         # 取接入点
         ap = common_data.ap_id_obj_map[host.ap_id]
-        return [
-            {
-                "bk_cloud_id": constants.DEFAULT_CLOUD,
-                "ip": bt_server["inner_ip"],
-                "bk_host_id": bt_server.get("bk_host_id", ""),
-            }
-            for bt_server in ap.btfileserver
-        ]
+        bt_file_server_queryset = models.Host.objects.filter(
+            inner_ip__in=[bt_server["inner_ip"] for bt_server in ap.btfileserver], bk_cloud_id=constants.DEFAULT_CLOUD
+        ).values_list("bk_host_id", flat=True)
+        bt_file_server_ids: List[int] = [bk_host_id for bk_host_id in bt_file_server_queryset]
+
+        return {
+            "ip_list": [
+                {
+                    "bk_cloud_id": constants.DEFAULT_CLOUD,
+                    "ip": bt_server["inner_ip"],
+                }
+                for bt_server in ap.btfileserver
+            ],
+            "host_id_list": bt_file_server_ids,
+        }
 
     def get_script_content(self, data, common_data: AgentCommonData, host: models.Host) -> str:
         port_config = common_data.host_id__ap_map[host.bk_host_id].port_config

apps/backend/components/collections/agent_new/install.py

@@ -388,7 +388,7 @@ def execute_job_commands(self, sub_inst_id, installation_tool: InstallationTools
         bk_biz_id = settings.BLUEKING_BIZ_ID
         target_server: Dict[str, List[Union[int, Dict[str, Union[int, str]]]]] = (
             {"host_id_list": [jump_server.bk_host_id]}
-            if settings.ENABLE_DBCP
+            if settings.BKAPP_ENABLE_DHCP
             else {
                 "ip_list": [
                     {

apps/backend/components/collections/job.py

@@ -14,7 +14,8 @@
 import hashlib
 import json
 import logging
-from typing import Any, Callable, Dict, List, Optional, Set
+from collections import defaultdict
+from typing import Any, Callable, Dict, List, Optional, Set, Union
 
 import six
 from django.conf import settings
@@ -206,20 +207,33 @@ def handler_job_result(self, job_sub_map: models.JobSubscriptionInstanceMap) ->
         # 构造主机作业状态映射表
         cloud_ip_status_map: Dict[str, Dict] = {}
         for ip_result in ip_results["step_instance_list"][0].get("step_ip_result_list") or []:
-            cloud_ip_status_map[f'{ip_result["bk_cloud_id"]}-{ip_result["ip"]}-{ip_result["bk_host_id"]}'] = ip_result
+            if settings.BKAPP_ENABLE_DHCP:
+                cloud_ip_status_map[
+                    f'{ip_result["bk_cloud_id"]}-{ip_result["ip"]}-{ip_result["bk_host_id"]}'
+                ] = ip_result
+            else:
+                cloud_ip_status_map[f'{ip_result["bk_cloud_id"]}-{ip_result["ip"]}'] = ip_result
 
         succeed_sub_inst_ids: List[int] = []
         subscription_instances = models.SubscriptionInstanceRecord.objects.filter(
             id__in=job_sub_map.subscription_instance_ids
         )
 
         bk_host_ids = [sub_inst.instance_info["host"]["bk_host_id"] for sub_inst in subscription_instances]
-        host_id__cloud_ip_map = {
-            host_info["bk_host_id"]: f"{host_info['bk_cloud_id']}-{host_info['inner_ip']}-{host_info['bk_host_id']}"
-            for host_info in models.Host.objects.filter(bk_host_id__in=bk_host_ids).values(
-                "bk_host_id", "inner_ip", "bk_cloud_id"
-            )
-        }
+        if settings.BKAPP_ENABLE_DHCP:
+            host_id__cloud_ip_map = {
+                host_info["bk_host_id"]: f"{host_info['bk_cloud_id']}-{host_info['inner_ip']}-{host_info['bk_host_id']}"
+                for host_info in models.Host.objects.filter(bk_host_id__in=bk_host_ids).values(
+                    "bk_host_id", "inner_ip", "bk_cloud_id"
+                )
+            }
+        else:
+            host_id__cloud_ip_map = {
+                host_info["bk_host_id"]: f"{host_info['bk_cloud_id']}-{host_info['inner_ip']}"
+                for host_info in models.Host.objects.filter(bk_host_id__in=bk_host_ids).values(
+                    "bk_host_id", "inner_ip", "bk_cloud_id"
+                )
+            }
 
         for sub_inst in subscription_instances:
             ip = sub_inst.instance_info["host"]["bk_host_innerip"]
@@ -378,7 +392,7 @@ def _execute(self, data, parent_data, common_data: CommonData):
 
         timeout = data.get_one_of_inputs("timeout")
         # 批量请求作业平台的参数
-        multi_job_params_map: Dict[str, Dict[str, Any]] = {}
+        multi_job_params_map: Dict[str, Dict[str, Any]] = defaultdict(lambda: defaultdict(list))
         for sub_inst in common_data.subscription_instances:
             bk_host_id = sub_inst.instance_info["host"]["bk_host_id"]
             host_obj = common_data.host_id_obj_map[bk_host_id]
@@ -404,7 +418,7 @@ def _execute(self, data, parent_data, common_data: CommonData):
                     "subscription_instance_id": [sub_inst.id],
                     "subscription_id": common_data.subscription.id,
                     "job_params": {
-                        "target_server": {"ip_list": target_servers, "host_id_list": host_obj.bk_host_id},
+                        "target_server": target_servers,
                         "script_content": script_content,
                         "script_param": script_param,
                         "timeout": timeout,
@@ -434,15 +448,17 @@ def get_script_param(self, data, common_data: CommonData, host: models.Host) ->
         """
         return data.get_one_of_inputs("script_param", default="")
 
-    def get_target_servers(self, data, common_data: CommonData, host: models.Host) -> List[Dict[str, Any]]:
+    def get_target_servers(
+        self, data, common_data: CommonData, host: models.Host
+    ) -> Dict[str, Union[List[Dict[str, Union[int, str]]], List[int]]]:
         """
         获取目标服务器
         :param data:
         :param common_data:
         :param host: 主机类型
         :return: 目标服务器
         """
-        return [{"bk_cloud_id": host.bk_cloud_id, "ip": host.inner_ip, "bk_host_id": host.bk_host_id}]
+        return {"ip_list": [{"bk_cloud_id": host.bk_cloud_id, "ip": host.inner_ip}], "host_id_list": [host.bk_host_id]}
 
 
 class JobTransferFileService(JobV3BaseService, metaclass=abc.ABCMeta):

apps/backend/components/collections/plugin.py

@@ -1357,7 +1357,6 @@ def _schedule(self, data, parent_data, callback_data=None):
             "bk_scope_id": settings.BLUEKING_BIZ_ID,
             "bk_username": settings.BACKEND_JOB_OPERATOR,
             "step_instance_id": step_instance_id,
-            "bk_host_id": bk_host_id,
         }
         host_interaction_params: Dict[str, Union[str, int]] = (
             {"bk_host_id": bk_host_id} if settings.BKAPP_ENABLE_DHCP else {"ip": ip, "bk_cloud_id": bk_cloud_id}

apps/backend/tests/api/test_job.py

@@ -153,7 +153,6 @@ class TestJob(TestCase):
                             {
                                 "total_time": 0.001,
                                 "ip": "127.0.0.1",
-                                "host_id": 1,
                                 "start_time": "2019-08-12 19:04:07 +0800",
                                 "log_content": "\u30102019-08-12 19:04:07.757\u3011 FileName\uff1a/data/dev_"
                                 "pipeline_unit_test  FileSize\uff1a9.0 Byte  State\uff1adownload"
@@ -213,7 +212,6 @@ def test_fast_execute_script(self):
         assert len(task_result["success"]) == 1
         assert set(task_result["success"][0].keys()) == {
             "ip",
-            "host_id",
             "log_content",
             "bk_cloud_id",
             "error_code",
@@ -238,7 +236,6 @@ def test_fast_push_file(self):
         assert len(task_result["success"]) == 1
         assert set(task_result["success"][0].keys()) == {
             "ip",
-            "host_id",
             "log_content",
             "bk_cloud_id",
             "error_code",

apps/backend/tests/components/collections/agent_new/test_check_policy_gse_to_proxy.py

@@ -11,7 +11,6 @@
 import base64
 from typing import Dict
 
-from django.test import override_settings
 from django.utils.translation import ugettext_lazy as _
 
 from apps.backend.components.collections.agent_new import (
@@ -32,7 +31,6 @@ def get_default_case_name(cls) -> str:
     def component_cls(self):
         return components.CheckPolicyGseToProxyComponent
 
-    @override_settings(BKAPP_ENABLE_DHCP=False)
     def tearDown(self) -> None:
         record = self.job_api_mock_client.call_recorder.record
         host_key__script_content_map: Dict[str, str] = {}

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

@@ -432,6 +432,7 @@ def get_job_instance_status(cls, *args, **kwargs):
                     "step_ip_result_list": [
                         {
                             "ip": TEST_IP,
+                            "bk_host_id": BK_HOST_ID,
                             "bk_cloud_id": const.DEFAULT_CLOUD,
                             "status": const.BkJobIpStatus.SUCCEEDED,
                             "error_code": 0,