feature: 插件包管理支持对象存储模式，公共方法优化整合 (TencentBlueKing#2)

apps/backend/agent/manager.py

@@ -358,7 +358,7 @@ def push_files_to_proxy(self, file):
             type=Var.PLAIN,
             value=[{"ip": self.host_info["bk_host_innerip"], "bk_cloud_id": self.host_info["bk_cloud_id"]}],
         )
-        act.component.inputs.file_target_path = Var(type=Var.PLAIN, value=settings.NGINX_DOWNLOAD_PATH)
+        act.component.inputs.file_target_path = Var(type=Var.PLAIN, value=settings.DOWNLOAD_PATH)
         act.component.inputs.files = Var(type=Var.PLAIN, value=file["files"])
         act.component.inputs.from_type = Var(type=Var.PLAIN, value=file.get("from_type", ""))
         act.component.inputs.context = Var(type=Var.PLAIN, value="")
@@ -369,7 +369,7 @@ def start_nginx(self):
         path = os.path.join(settings.PROJECT_ROOT, "script_tools", "start_nginx.sh.tpl")
         with open(path, encoding="utf-8") as fh:
             script = fh.read()
-        script_content = script % {"nginx_path": settings.NGINX_DOWNLOAD_PATH}
+        script_content = script % {"nginx_path": settings.DOWNLOAD_PATH}
         act = AgentServiceActivity(component_code=JobFastExecuteScriptComponent.code, name=_("启动 NGINX 服务"))
         act.component.inputs.job_client = Var(
             type=Var.PLAIN,

apps/backend/agent/tools.py

@@ -130,7 +130,7 @@ def gen_commands(host, pipeline_id, is_uninstall, batch_install=False):
         download_cmd = f"curl {package_url}/{shell_file_name} -o {dest_dir}{shell_file_name} --connect-timeout 5 -sSf"
 
     else:
-        run_cmd_params.append(f"-L {settings.NGINX_DOWNLOAD_PATH}")
+        run_cmd_params.append(f"-L {settings.DOWNLOAD_PATH}")
         # 云区域自动安装
         dest_host = host
         # 手动批量安装均走该分支

apps/backend/components/collections/agent.py

@@ -1005,7 +1005,7 @@ def execute(self, data, parent_data):
         self.logger.info(_("开始下发升级包"))
         host_info = data.get_one_of_inputs("host_info")
         host = Host.get_by_host_info(host_info)
-        nginx_path = host.ap.nginx_path or settings.NGINX_DOWNLOAD_PATH
+        nginx_path = host.ap.nginx_path or settings.DOWNLOAD_PATH
         data.inputs.file_target_path = host.agent_config["temp_path"]
 
         os_type = host.os_type.lower()

apps/backend/components/collections/bulk_job.py

@@ -423,7 +423,7 @@ def execute(self, data, parent_data):
         self.logger.info(_("开始下发升级包"))
         host_info = data.get_one_of_inputs("host_info")
         host = models.Host.get_by_host_info(host_info)
-        nginx_path = host.ap.nginx_path or settings.NGINX_DOWNLOAD_PATH
+        nginx_path = host.ap.nginx_path or settings.DOWNLOAD_PATH
         data.inputs.file_target_path = host.agent_config["temp_path"]
 
         os_type = host.os_type.lower()

apps/backend/components/collections/bulk_job_redis.py

@@ -412,7 +412,7 @@ def execute(self, data, parent_data):
         self.logger.info(_("开始下发升级包"))
         host_info = data.get_one_of_inputs("host_info")
         host = models.Host.get_by_host_info(host_info)
-        nginx_path = host.ap.nginx_path or settings.NGINX_DOWNLOAD_PATH
+        nginx_path = host.ap.nginx_path or settings.DOWNLOAD_PATH
         data.inputs.file_target_path = host.agent_config["temp_path"]
 
         os_type = host.os_type.lower()

apps/backend/components/collections/job.py

@@ -573,7 +573,7 @@ def execute(self, data, parent_data):
 
         data.inputs.file_source = [
             {
-                "files": [f"{settings.NGINX_DOWNLOAD_PATH}/{file}" for file in files],
+                "files": [f"{settings.DOWNLOAD_PATH}/{file}" for file in files],
                 "account": "root",
                 "ip_list": [{"ip": settings.BKAPP_LAN_IP, "bk_cloud_id": 0}],
             }

apps/backend/components/collections/plugin.py

@@ -386,7 +386,7 @@ def _execute(self, data, parent_data, common_data: CommonData):
         # 如 linux-arm、linux-x86、windows-x86 的插件，需分为三组
         # 把多个IP合并为一个任务，可以利用GSE文件管道的BT能力，提高传输效率
         jobs: Dict[str, Dict[str, Union[list, str]]] = defaultdict(lambda: defaultdict(list))
-        nginx_path = settings.NGINX_DOWNLOAD_PATH
+        nginx_path = settings.DOWNLOAD_PATH
         for process_status in process_statuses:
             bk_host_id = process_status.bk_host_id
             subscription_instance = group_id_instance_map.get(process_status.group_id)

apps/backend/exceptions.py

@@ -51,3 +51,8 @@ class PackageVersionValidationError(BackendBaseException):
 class GenCommandsError(BackendBaseException):
     MESSAGE = _("安装命令生成失败")
     ERROR_CODE = 7
+
+
+class PluginParseError(BackendBaseException):
+    MESSAGE = _("插件解析错误")
+    ERROR_CODE = 8

apps/backend/management/commands/copy_file_to_nginx.py

@@ -29,7 +29,7 @@ def handle(self, *args, **options):
         # 接入点配置的nginx路径
         nginx_paths = [ap.nginx_path for ap in AccessPoint.objects.all() if ap.nginx_path]
         # 默认nginx路径
-        nginx_paths.append(settings.NGINX_DOWNLOAD_PATH)
+        nginx_paths.append(settings.DOWNLOAD_PATH)
         # 去重
         nginx_paths = list(set(nginx_paths))
         for _path in os.listdir(settings.BK_SCRIPTS_PATH):

apps/backend/management/commands/init_official_plugins.py

@@ -19,8 +19,9 @@
 from django.core.management.base import BaseCommand
 from django.db.transaction import atomic
 
+from apps.backend.plugin import tools
 from apps.node_man import models
-from apps.utils.basic import md5
+from apps.utils.files import md5sum
 from common.log import logger
 
 
@@ -59,7 +60,7 @@ def handle(self, *args, **options):
                         # 后续可以考虑通过路径来判断
                         module="gse_plugin",
                         file_path=file_abs_path,
-                        md5=md5(file_abs_path),
+                        md5=md5sum(name=file_abs_path),
                         operator="system",
                         source_app_code="bk_nodeman",
                         file_name=file_name,
@@ -68,7 +69,9 @@ def handle(self, *args, **options):
 
                     try:
                         # 如果是官方内置的插件，那么应该是直接发布的
-                        package_list = upload_record.create_package_records(
+                        package_list = tools.create_package_records(
+                            file_path=upload_record.file_path,
+                            file_name=upload_record.file_name,
                             is_release=True,
                             is_template_load=True,
                             is_template_overwrite=True,

apps/backend/plugin/serializers.py

@@ -13,11 +13,12 @@
 import base64
 import hashlib
 
+from django.utils.translation import ugettext_lazy as _
 from rest_framework import serializers
 from rest_framework.exceptions import ValidationError
 
 from apps.node_man import constants
-from apps.node_man.models import GsePluginDesc, Packages
+from apps.node_man.models import DownloadRecord, GsePluginDesc, Packages
 
 
 class GatewaySerializer(serializers.Serializer):
@@ -202,14 +203,28 @@ def validate(self, attrs):
         return attrs
 
 
-class UploadInfoSerializer(GatewaySerializer):
+class UploadInfoBaseSerializer(GatewaySerializer):
+    md5 = serializers.CharField(help_text=_("上传端计算的文件md5"), max_length=32)
+    file_name = serializers.CharField(help_text=_("上传端提供的文件名"), min_length=1)
+    module = serializers.CharField(max_length=32, required=False, default="gse_plugin")
+
+
+class UploadInfoSerializer(UploadInfoBaseSerializer):
     """上传插件包接口序列化器"""
 
-    module = serializers.CharField(max_length=32, required=False, default="gse_plugin")
-    md5 = serializers.CharField(max_length=32)
-    file_name = serializers.CharField()
-    file_local_path = serializers.CharField(max_length=512)
-    file_local_md5 = serializers.CharField(max_length=32)
+    file_local_path = serializers.CharField(help_text=_("本地文件路径"), max_length=512)
+    file_local_md5 = serializers.CharField(help_text=_("Nginx所计算的文件md5"), max_length=32)
+
+
+class CosUploadInfoSerializer(UploadInfoBaseSerializer):
+    download_url = serializers.URLField(help_text=_("对象存储文件下载url"), required=False)
+    file_path = serializers.CharField(help_text=_("文件保存路径"), min_length=1, required=False)
+
+    def validate(self, attrs):
+        # 两种参数模式少要有一种满足
+        if not ("download_url" in attrs or "file_path" in attrs):
+            raise ValidationError("at least has download_url or file_path")
+        return attrs
 
 
 class PluginStartDebugSerializer(GatewaySerializer):
@@ -287,6 +302,15 @@ def validate(self, data):
     creator = serializers.CharField()
     bk_app_code = serializers.CharField()
 
+    def validate(self, attrs):
+        if attrs["category"] not in DownloadRecord.CATEGORY_TASK_DICT:
+            raise ValidationError(
+                "请求下载类型 -> {category} 暂不支持，可选项 -> {choices}".format(
+                    category=attrs["category"], choices=DownloadRecord.CATEGORY_CHOICES
+                )
+            )
+        return attrs
+
 
 class DeletePluginSerializer(GatewaySerializer):
     name = serializers.CharField()

apps/backend/plugin/tasks.py

@@ -18,6 +18,7 @@
 from django.utils.translation import ugettext as _
 
 from apps.backend.celery import app
+from apps.backend.plugin import tools
 from apps.backend.utils.pipeline_parser import PipelineParser as CustomPipelineParser
 from apps.node_man import constants as const
 from apps.node_man import models
@@ -41,7 +42,7 @@ def package_task(job_id, task_params):
         job = models.Job.objects.get(id=job_id, job_type=const.JobType.PACKING_PLUGIN)
 
     except models.Job.DoesNotExist:
-        logger.error("try to execute job->[%s] but is not exists")
+        logger.error("try to execute job-> {job_id} but is not exists".format(job_id=job_id))
         return False
 
     try:
@@ -52,10 +53,12 @@ def package_task(job_id, task_params):
         upload_package_object = models.UploadPackage.objects.filter(file_name=file_name).order_by("-upload_time")[0]
 
         # 2. 执行任务
-        upload_package_object.create_package_records(
+        tools.create_package_records(
+            file_path=upload_package_object.file_path,
+            file_name=upload_package_object.file_name,
             is_release=is_release,
             creator=task_params["bk_username"],
-            select_pkg_abs_paths=select_pkg_abs_paths,
+            select_pkg_relative_paths=select_pkg_abs_paths,
             is_template_load=task_params.get("is_template_load", False),
             is_template_overwrite=task_params.get("is_template_overwrite", False),
         )
@@ -85,7 +88,7 @@ def package_task(job_id, task_params):
     job.save()
 
     if job.status == const.JobStatusType.SUCCESS:
-        logger.info("task->[%s] has finish all job." % job.id)
+        logger.info("task -> {job_id} has finish all job.".format(job_id=job.id))
 
 
 @app.task(queue="backend")