feat: 订阅接口支持指定用户操作进程 (closed TencentBlueKing#2297)

apps/backend/components/collections/plugin.py

@@ -1168,6 +1168,12 @@ def _execute(self, data, parent_data, common_data: PluginCommonData):
         plugin = policy_step_adapter.plugin_desc
         group_id_instance_map = common_data.group_id_instance_map
         host_id_obj_map = common_data.host_id_obj_map
+        operate_info: List = common_data.subscription.operate_info
+        host_id_user_map = {}
+        system_account = {}
+        if operate_info:
+            host_id_user_map: Dict[int, str] = {info.get("bk_host_id"): info.get("user") for info in operate_info}
+            system_account: Dict[str, str] = operate_info[0]
 
         host_id__resource_policy_map = self.get_resource_policy(common_data.bk_host_ids, plugin.name)
         proc_operate_req = []
@@ -1178,6 +1184,7 @@ def _execute(self, data, parent_data, common_data: PluginCommonData):
         for process_status in process_statuses:
             bk_host_id = process_status.bk_host_id
             host = host_id_obj_map.get(bk_host_id)
+            operate_user = host_id_user_map.get(bk_host_id) or system_account.get(host.os_type)
             subscription_instance = group_id_instance_map.get(process_status.group_id)
             package = self.get_package_by_process_status(process_status, common_data)
             package_control = package.proc_control
@@ -1204,7 +1211,7 @@ def _execute(self, data, parent_data, common_data: PluginCommonData):
                         "proc_name": package_control.process_name or plugin.name,
                         "setup_path": process_status.setup_path,
                         "pid_path": process_status.pid_path,
-                        "user": constants.ACCOUNT_MAP.get(host.os_type, "root"),
+                        "user": operate_user or constants.ACCOUNT_MAP.get(host.os_type, "root"),
                     },
                     "control": gse_control,
                     "resource": host_id__resource_policy_map[bk_host_id]["resource"],

apps/backend/subscription/serializers.py

@@ -70,6 +70,11 @@ def validate(self, attrs):
         raise ValidationError("目前机器参数必须要有 bk_host_id 或者 (ip/bk_host_innerip + bk_cloud_id)")
 
 
+class HostUserInfoSerializer(serializers.Serializer):
+    bk_host_id = serializers.IntegerField(label=_("主机ID"))
+    user = serializers.CharField(label=_("操作用户"))
+
+
 class CreateSubscriptionSerializer(GatewaySerializer):
     class CreateStepSerializer(serializers.Serializer):
         id = serializers.CharField(label="步骤标识符", validators=[])
@@ -83,6 +88,8 @@ class CreateStepSerializer(serializers.Serializer):
     target_hosts = TargetHostSerializer(many=True, label="下发的目标机器列表", required=False, allow_empty=False)
     run_immediately = serializers.BooleanField(required=False, default=False, label="是否立即执行")
     is_main = serializers.BooleanField(required=False, default=False, label="是否为主配置")
+    operate_info = serializers.ListField(required=False, child=HostUserInfoSerializer(), default=[], label="操作信息")
+    system_account = serializers.DictField(required=False, label=_("操作系统对应账户"))
 
     # 策略新参数
     plugin_name = serializers.CharField(required=False, label="插件名")
@@ -102,6 +109,10 @@ def validate(self, attrs):
         ):
             raise ValidationError(_("订阅范围包含Gse2.0灰度业务"))
         step_types = {step["type"] for step in attrs["steps"]}
+        if attrs.get("system_account"):
+            for key in attrs["system_account"]:
+                if key not in constants.OS_TUPLE:
+                    raise ValidationError(_(f"操作系统类型只能为{constants.OS_TUPLE}"))
         if constants.SubStepType.AGENT not in step_types:
             return attrs
 
@@ -147,12 +158,21 @@ class UpdateStepSerializer(serializers.Serializer):
     scope = UpdateScopeSerializer()
     steps = serializers.ListField(child=UpdateStepSerializer())
     run_immediately = serializers.BooleanField(required=False, default=False)
+    operate_info = serializers.ListField(required=False, child=HostUserInfoSerializer(), default=[], label="操作信息")
+    system_account = serializers.DictField(required=False, label=_("操作系统对应账户"))
 
     # 策略新参数
     plugin_name = serializers.CharField(required=False)
     bk_biz_scope = serializers.ListField(child=serializers.IntegerField(), required=False, default=[])
     category = serializers.CharField(required=False)
 
+    def validate(self, attrs):
+        if attrs.get("system_account"):
+            for key in attrs["system_account"]:
+                if key not in constants.OS_TUPLE:
+                    raise ValidationError(_(f"操作系统类型只能为{constants.OS_TUPLE}"))
+        return attrs
+
 
 class DeleteSubscriptionSerializer(GatewaySerializer):
     subscription_id = serializers.IntegerField(label="订阅ID")

apps/backend/subscription/views.py

@@ -75,7 +75,8 @@ def create_subscription(self, request):
         if category == models.Subscription.CategoryType.POLICY:
             # 策略类型订阅默认开启
             enable = True
-
+        if params.get("system_account"):
+            params["operate_info"].insert(0, params["system_account"])
         with transaction.atomic():
             # 创建订阅
             subscription = models.Subscription.objects.create(
@@ -95,6 +96,8 @@ def create_subscription(self, request):
                 category=params.get("category"),
                 plugin_name=params.get("plugin_name"),
                 pid=params.get("pid", models.Subscription.ROOT),
+                # 指定操作进程用户新增
+                operate_info=params.get("operate_info"),
             )
 
             # 创建订阅步骤
@@ -212,6 +215,10 @@ def update_subscription(self, request):
             # 策略部署新增
             subscription.plugin_name = params.get("plugin_name")
             subscription.bk_biz_scope = params.get("bk_biz_scope")
+            # 指定操作进程用户新增
+            if params.get("system_account"):
+                params["operate_info"].insert(0, params["system_account"])
+            subscription.operate_info = params["operate_info"]
             subscription.save()
 
             step_ids: Set[str] = set()

apps/node_man/handlers/plugin_v2.py

@@ -152,7 +152,14 @@ def history(query_params: Dict):
         return packages
 
     @staticmethod
-    def operate(job_type: str, plugin_name: str, scope: Dict, steps: List[Dict]):
+    def operate(
+        job_type: str,
+        plugin_name: str,
+        scope: Dict,
+        steps: List[Dict],
+        operate_info: List[Dict] = None,
+        system_account: Dict = None,
+    ):
         bk_biz_scope = list(set([node["bk_biz_id"] for node in scope["nodes"]]))
 
         CmdbHandler().check_biz_permission(bk_biz_scope, IamActionType.plugin_operate)
@@ -166,6 +173,10 @@ def operate(job_type: str, plugin_name: str, scope: Dict, steps: List[Dict]):
             "scope": scope,
             "bk_biz_scope": bk_biz_scope,
         }
+        if operate_info:
+            base_create_kwargs["operate_info"] = operate_info
+        if system_account:
+            base_create_kwargs["system_account"] = system_account
 
         if job_type == constants.JobType.MAIN_INSTALL_PLUGIN:
             create_data = {**base_create_kwargs, "steps": steps}

apps/node_man/migrations/0083_subscription_operate_info.py

@@ -0,0 +1,19 @@
+# Generated by Django 3.2.4 on 2024-09-10 10:32
+
+import django_mysql.models
+from django.db import migrations
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ("node_man", "0082_host_dept_name"),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name="subscription",
+            name="operate_info",
+            field=django_mysql.models.JSONField(default=list, verbose_name="操作信息"),
+        ),
+    ]

apps/node_man/models.py

@@ -1859,6 +1859,7 @@ class CategoryType(object):
     category = models.CharField(_("订阅类别"), max_length=32, null=True, blank=True, db_index=True)
     plugin_name = models.CharField(_("插件名称"), max_length=64, null=True, blank=True, db_index=True)
     bk_biz_scope = JSONField(_("业务范围"), default=list)
+    operate_info = JSONField(_("操作信息"), default=list)
 
     pid = models.BigIntegerField(_("父订阅ID"), default=ROOT, db_index=True)
 

apps/node_man/serializers/plugin_v2.py

@@ -189,20 +189,31 @@ def validate(self, data):
         return data
 
 
+class HostUserInfoSerializer(serializers.Serializer):
+    bk_host_id = serializers.IntegerField(label=_("主机ID"))
+    user = serializers.CharField(label=_("操作用户"))
+
+
 class PluginOperateSerializer(serializers.Serializer):
     job_type = serializers.ChoiceField(label=_("任务类型"), choices=list(constants.PLUGIN_JOB_TUPLE))
     plugin_name = serializers.CharField(label=_("插件名称"))
     # 一次性任务范围
     scope = base.ScopeSerializer()
     # 参数配置
     steps = serializers.ListField(child=base.StepSerializer(), required=False, default=[])
+    operate_info = serializers.ListField(label=_("操作信息"), child=HostUserInfoSerializer(), default=[], required=False)
+    system_account = serializers.DictField(required=False, label=_("操作系统对应账户"))
 
     def validate(self, data):
         if models.GsePluginDesc.objects.filter(name=data["plugin_name"]).first() is None:
             raise exceptions.PluginNotExistError(_("不存在名称为: {name} 的插件").format(name=data["plugin_name"]))
 
         if data["job_type"] == constants.JobType.MAIN_INSTALL_PLUGIN and not data["steps"]:
             raise ValidationError(_("插件安装/更新需要传递steps"))
+        if data.get("system_account"):
+            for key in data["system_account"]:
+                if key not in constants.OS_TUPLE:
+                    raise ValidationError(_(f"操作系统类型只能为{constants.OS_TUPLE}"))
 
         return data
 

apps/node_man/views/plugin_v2.py

@@ -749,6 +749,8 @@ def operate(self, request):
                 plugin_name=params["plugin_name"],
                 scope=params["scope"],
                 steps=params.get("steps"),
+                operate_info=params.get("operate_info"),
+                system_account=params.get("system_account"),
             )
         )