fix: 优化资源池导入的flow，执行window机器 TencentBlueKing#7130

dbm-ui/backend/db_services/dbresource/serializers.py

@@ -25,6 +25,7 @@
     RESOURCE_UPDATE_PARAMS,
     SPEC_DATA,
 )
+from backend.db_services.ipchooser.constants import BkOsTypeCode
 from backend.db_services.ipchooser.serializers.base import QueryHostsBaseSer
 from backend.ticket.builders.common.field import DBTimezoneField
 from backend.ticket.constants import TicketStatus
@@ -35,6 +36,7 @@ class HostInfoSerializer(serializers.Serializer):
         ip = serializers.CharField()
         host_id = serializers.IntegerField()
         bk_cloud_id = serializers.IntegerField()
+        os_type = serializers.CharField(required=False, default=BkOsTypeCode.LINUX)
 
     for_bizs = serializers.ListSerializer(help_text=_("专属业务的ID列表"), child=serializers.IntegerField())
     resource_types = serializers.ListField(

dbm-ui/backend/db_services/dbresource/views/resource.py

@@ -51,7 +51,7 @@
     SpecCountResourceResponseSerializer,
     SpecCountResourceSerializer,
 )
-from backend.db_services.ipchooser.constants import BkOsType, ModeType
+from backend.db_services.ipchooser.constants import BK_OS_CODE__TYPE, BkOsType, ModeType
 from backend.db_services.ipchooser.handlers.host_handler import HostHandler
 from backend.db_services.ipchooser.handlers.topo_handler import TopoHandler
 from backend.db_services.ipchooser.query.resource import ResourceQueryHelper
@@ -187,34 +187,42 @@ def query_dba_hosts(self, request):
     @action(detail=False, methods=["POST"], url_path="import", serializer_class=ResourceImportSerializer)
     def resource_import(self, request):
         validated_data = self.params_validate(self.get_serializer_class())
-        root_id = generate_root_id()
-
-        # 补充必要的单据参数
-        validated_data.update(
-            ticket_type=TicketType.RESOURCE_IMPORT,
-            created_by=request.user.username,
-            uid=None,
-            # 额外补充资源池导入的参数，用于记录操作日志
-            bill_id=None,
-            bill_type=None,
-            task_id=root_id,
-            operator=request.user.username,
-        )
-
-        # 资源导入记录
-        import_record = {"task_id": root_id, "operator": request.user.username, "hosts": validated_data["hosts"]}
-        DBResourceApi.import_operation_create(params=import_record)
-
-        # 执行资源导入的后台flow
-        BaseController(root_id=root_id, ticket_data=validated_data).import_resource_init_step()
+        import_hosts = validated_data.pop("hosts")
+        os_hosts = itertools.groupby(import_hosts, key=lambda x: x["os_type"])
+        # 按照集群类型分别导入
+        for os_type, hosts in os_hosts:
+            root_id = generate_root_id()
+            hosts = list(hosts)
+
+            # 补充必要的单据参数
+            validated_data.update(
+                ticket_type=TicketType.RESOURCE_IMPORT,
+                created_by=request.user.username,
+                uid=None,
+                hosts=hosts,
+                # 额外补充资源池导入的参数，用于记录操作日志
+                bill_id=None,
+                bill_type=None,
+                task_id=root_id,
+                operator=request.user.username,
+                os_type=BK_OS_CODE__TYPE[os_type],
+            )
 
-        # 缓存当前任务，并删除过期导入任务
-        now = int(time.time())
-        cache_key = RESOURCE_IMPORT_TASK_FIELD.format(user=request.user.username)
-        RedisConn.zadd(cache_key, {root_id: now})
-        expired_tasks = RedisConn.zrangebyscore(cache_key, "-inf", now - RESOURCE_IMPORT_EXPIRE_TIME)
-        if expired_tasks:
-            RedisConn.zrem(cache_key, *expired_tasks)
+            # 资源导入记录
+            import_record = {"task_id": root_id, "operator": request.user.username, "hosts": hosts}
+            DBResourceApi.import_operation_create(params=import_record)
+
+            # 执行资源导入的后台flow
+            validated_data.update(hosts=list(hosts), os_type=BK_OS_CODE__TYPE[os_type])
+            BaseController(root_id=root_id, ticket_data=validated_data).import_resource_init_step()
+
+            # 缓存当前任务，并删除过期导入任务
+            now = int(time.time())
+            cache_key = RESOURCE_IMPORT_TASK_FIELD.format(user=request.user.username)
+            RedisConn.zadd(cache_key, {root_id: now})
+            expired_tasks = RedisConn.zrangebyscore(cache_key, "-inf", now - RESOURCE_IMPORT_EXPIRE_TIME)
+            if expired_tasks:
+                RedisConn.zrem(cache_key, *expired_tasks)
 
         return Response()
 

dbm-ui/backend/flow/consts.py

@@ -68,6 +68,9 @@
 # window操作系统超级账号,标准运维调用
 WINDOW_ADMIN_USER_FOR_CHECK = "Administrator"
 
+# linux操作系统超级账号,标准运维调用
+LINUX_ADMIN_USER_FOR_CHECK = "root"
+
 # tendisplus默认kvstorecount
 DEFAULT_TENDISPLUS_KVSTORECOUNT = 10
 

dbm-ui/backend/flow/engine/bamboo/scene/common/machine_os_init.py

@@ -18,6 +18,8 @@
 from backend.components.dbresource.client import DBResourceApi
 from backend.configuration.constants import SystemSettingsEnum
 from backend.configuration.models import SystemSettings
+from backend.db_services.ipchooser.constants import BkOsType
+from backend.flow.consts import LINUX_ADMIN_USER_FOR_CHECK, WINDOW_ADMIN_USER_FOR_CHECK
 from backend.flow.engine.bamboo.scene.common.builder import Builder
 from backend.flow.plugins.components.collections.common.external_service import ExternalServiceComponent
 from backend.flow.plugins.components.collections.common.sa_idle_check import CheckMachineIdleComponent
@@ -39,12 +41,23 @@ def machine_init_flow(self):
         p = Builder(root_id=self.root_id, data=self.data)
         ip_list = self.data["hosts"]
         bk_biz_id = self.data["bk_biz_id"]
+
+        if self.data.get("os_type", BkOsType.LINUX.value) == BkOsType.WINDOWS.value:
+            # 如果是window类型机器，用administrator账号
+            account_name = WINDOW_ADMIN_USER_FOR_CHECK
+        else:
+            account_name = LINUX_ADMIN_USER_FOR_CHECK
+
         # 先执行空闲检查
         if env.SA_CHECK_TEMPLATE_ID:
             p.add_act(
                 act_name=_("执行sa空闲检查"),
                 act_component_code=CheckMachineIdleComponent.code,
-                kwargs=asdict(InitCheckForResourceKwargs(ips=[host["ip"] for host in ip_list], bk_biz_id=bk_biz_id)),
+                kwargs=asdict(
+                    InitCheckForResourceKwargs(
+                        ips=[host["ip"] for host in ip_list], bk_biz_id=bk_biz_id, account_name=account_name
+                    )
+                ),
             )
 
         # 在执行sa初始化
@@ -53,7 +66,7 @@ def machine_init_flow(self):
             p.add_act(
                 act_name=_("执行sa初始化"),
                 act_component_code=SaInitComponent.code,
-                kwargs={"ips": [host["ip"] for host in ip_list], "bk_biz_id": bk_biz_id},
+                kwargs={"ips": [host["ip"] for host in ip_list], "bk_biz_id": bk_biz_id, "account_name": account_name},
             )
 
         # 调用资源导入接口

dbm-ui/backend/flow/plugins/components/collections/common/sa_init.py

@@ -31,6 +31,7 @@ def _execute(self, data, parent_data) -> bool:
                 "${biz_cc_id}": bk_biz_id,
                 "${job_ip_list}": "\n".join(iplist),
                 "${bk_biz_id}": bk_biz_id,
+                "${job_account}": kwargs.get("account_name", "root"),
             },
         }
         rpdata = BkSopsApi.create_task(param)

dbm-ui/backend/flow/utils/mysql/mysql_act_dataclass.py

@@ -474,6 +474,7 @@ class InitCheckForResourceKwargs:
 
     ips: list
     bk_biz_id: int = env.DBA_APP_BK_BIZ_ID
+    account_name: str = "root"
 
 
 @dataclass()