diff --git a/app.yml b/app.yml index 1dedbf048..83131947a 100644 --- a/app.yml +++ b/app.yml @@ -6,7 +6,7 @@ is_use_celery: True author: 蓝鲸智云 introduction: 标准运维是通过一套成熟稳定的任务调度引擎，把在多系统间的工作整合到一个流程，助力运维实现跨系统调度自动化的SaaS应用。 introduction_en: SOPS is a SaaS application that utilizes a set of mature and stable task scheduling engines to help realize cross-system scheduling automation, and integrates the work among multiple systems into a single process. -version: 3.32.1-p2 +version: 3.32.1-p4 category: 运维工具 language_support: 中文 desktop: diff --git a/app_desc.yaml b/app_desc.yaml index 4ebea3fc0..7954c0735 100644 --- a/app_desc.yaml +++ b/app_desc.yaml @@ -1,5 +1,5 @@ spec_version: 2 -app_version: "3.32.1-p2" +app_version: "3.32.1-p4" app: region: default bk_app_code: bk_sops diff --git a/config/default.py b/config/default.py index cac52bbc6..5c1599623 100644 --- a/config/default.py +++ b/config/default.py @@ -215,7 +215,7 @@ # mako模板中： # 如果静态资源修改了以后，上线前改这个版本号即可 -STATIC_VERSION = "3.32.1-p2" +STATIC_VERSION = "3.32.1-p4" DEPLOY_DATETIME = datetime.datetime.now().strftime("%Y%m%d%H%M%S") STATICFILES_DIRS = [os.path.join(BASE_DIR, "static")] diff --git a/frontend/desktop/src/pages/task/TaskExecute/ExecuteInfo.vue b/frontend/desktop/src/pages/task/TaskExecute/ExecuteInfo.vue index 0bd5be39e..3a82ba6ba 100644 --- a/frontend/desktop/src/pages/task/TaskExecute/ExecuteInfo.vue +++ b/frontend/desktop/src/pages/task/TaskExecute/ExecuteInfo.vue @@ -684,7 +684,7 @@ }, // 补充记录缺少的字段 async setFillRecordField (record) { - const { version, component_code: componentCode } = this.nodeDetailConfig + const { version, component_code: componentCode, componentData = {} } = this.nodeDetailConfig const { inputs, state } = record let outputs = record.outputs // 执行记录的outputs可能为Object格式，需要转为Array格式 @@ -721,7 +721,8 @@ const keys = Object.keys(inputs) this.renderConfig = renderConfig.filter(item => keys.includes(item.tag_code)) } else if (componentCode) { // 任务节点需要加载标准插件 - await this.getNodeConfig(componentCode, version, inputs.plugin_version) + const pluginVersion = componentData.plugin_version?.value + await this.getNodeConfig(componentCode, version, pluginVersion) } inputsInfo = Object.keys(inputs).reduce((acc, cur) => { const scheme = Array.isArray(this.renderConfig) ? this.renderConfig.find(item => item.tag_code === cur) : null diff --git a/frontend/desktop/src/pages/template/TemplateEdit/TemplateSetting/TabGlobalVariables/VariableEdit.vue b/frontend/desktop/src/pages/template/TemplateEdit/TemplateSetting/TabGlobalVariables/VariableEdit.vue index 5dc8fa80d..868231553 100644 --- a/frontend/desktop/src/pages/template/TemplateEdit/TemplateSetting/TabGlobalVariables/VariableEdit.vue +++ b/frontend/desktop/src/pages/template/TemplateEdit/TemplateSetting/TabGlobalVariables/VariableEdit.vue @@ -658,9 +658,10 @@ }, // 变量类型切换 onValTypeChange (val, oldValue) { - // 将上一个类型的填写的数据存起来("集群模块IP选择器"的code与"ip选择器"code相同,需要单独处理) - const valData = oldValue === 'set_module_ip_selector' - ? { set_module_ip_selector: tools.deepClone(this.renderData['ip_selector']) } + // 将上一个类型的填写的数据存起来("集群模块IP选择器"和"GSEKit IP选择器"的code与"ip选择器"code相同,需要单独处理) + const sameIpSelectorCode = ['set_module_ip_selector', 'gse_kit_ip_selector'] + const valData = sameIpSelectorCode.includes(oldValue) + ? { [oldValue]: tools.deepClone(this.renderData['ip_selector']) } : tools.deepClone(this.renderData) Object.assign(this.varTypeData, valData) // 将input textarea类型正则存起来 @@ -677,7 +678,7 @@ }) if (val in this.varTypeData) { const value = this.varTypeData[val] - this.renderData = { [val === 'set_module_ip_selector' ? 'ip_selector' : val]: value } + this.renderData = { [sameIpSelectorCode.includes(val) ? 'ip_selector' : val]: value } } else { this.renderData = {} } diff --git a/gcloud/apigw/views/create_task.py b/gcloud/apigw/views/create_task.py index 30335a038..e27b2b8c8 100644 --- a/gcloud/apigw/views/create_task.py +++ b/gcloud/apigw/views/create_task.py @@ -100,6 +100,7 @@ def create_task(request, template_id, project_id): "code": err_code.REQUEST_PARAM_INVALID.code, "message": f"callback_url format error, must match {CALLBACK_URL_PATTERN}", } + callback_version = params.get("callback_version", None) # 兼容老版本的接口调用 if template_source in NON_COMMON_TEMPLATE_TYPES: @@ -214,7 +215,13 @@ def create_task(request, template_id, project_id): # create callback url record if callback_url: - TaskCallBackRecord.objects.create(task_id=task.id, url=callback_url) + record_kwargs = { + "task_id": task.id, + "url": callback_url, + } + if callback_version: + record_kwargs["extra_info"] = json.dumps({"callback_version": callback_version}) + TaskCallBackRecord.objects.create(**record_kwargs) # crete auto retry strategy arn_creator = AutoRetryNodeStrategyCreator(taskflow_id=task.id, root_pipeline_id=task.pipeline_instance.instance_id) diff --git a/gcloud/taskflow3/domains/callback.py b/gcloud/taskflow3/domains/callback.py index 043b82966..842f279ff 100644 --- a/gcloud/taskflow3/domains/callback.py +++ b/gcloud/taskflow3/domains/callback.py @@ -32,7 +32,8 @@ class TaskCallBacker: def __init__(self, task_id, *args, **kwargs): self.task_id = task_id self.record = TaskCallBackRecord.objects.filter(task_id=self.task_id).first() - self.extra_info = {"task_id": self.task_id, **json.loads(self.record.extra_info), **kwargs} + self.record_extra_info = json.loads(self.record.extra_info) + self.extra_info = {"task_id": self.task_id, **self.record_extra_info, **kwargs} def check_record_existence(self): return True if self.record else False @@ -96,9 +97,13 @@ def _url_callback(self): logger.error(f"[TaskCallBacker _url_callback] get lock error: {err}") return None url = self.record.url + callback_version = self.record_extra_info.get("callback_version") response = None try: - response = requests.post(url, data=self.extra_info) + if callback_version == TaskCallBackRecord.CALLBACK_VERSION_V2: + response = requests.post(url, json=self.extra_info) + else: + response = requests.post(url, data=self.extra_info) response.raise_for_status() except HTTPError as e: message = ( diff --git a/gcloud/taskflow3/models.py b/gcloud/taskflow3/models.py index 20fb804e8..b0df2c954 100644 --- a/gcloud/taskflow3/models.py +++ b/gcloud/taskflow3/models.py @@ -1356,6 +1356,8 @@ class Meta: class TaskCallBackRecord(models.Model): + CALLBACK_VERSION_V2 = "v2" + id = models.BigAutoField(verbose_name="ID", primary_key=True) task_id = models.BigIntegerField(verbose_name=_("任务ID"), db_index=True) url = models.TextField(verbose_name=_("回调地址")) diff --git a/gcloud/taskflow3/signals/handlers.py b/gcloud/taskflow3/signals/handlers.py index 7788dea64..c21a6d3ba 100644 --- a/gcloud/taskflow3/signals/handlers.py +++ b/gcloud/taskflow3/signals/handlers.py @@ -47,25 +47,23 @@ def _finish_taskflow_and_send_signal(instance_id, sig, task_success=False): - qs = TaskFlowInstance.objects.filter(pipeline_instance__instance_id=instance_id).only("id") - if not qs: + task = TaskFlowInstance.objects.filter(pipeline_instance__instance_id=instance_id).first() + if not task: logger.error("pipeline archive handler get taskflow error, pipeline_instance_id={}".format(instance_id)) return - task_id = qs[0].id - - TaskFlowInstance.objects.filter(id=task_id).update(current_flow="finished") - sig.send(TaskFlowInstance, task_id=task_id) + TaskFlowInstance.objects.filter(id=task.id).update(current_flow="finished") + sig.send(TaskFlowInstance, task_id=task.id) if task_success: - _check_and_callback(task_id, task_success=task_success, task=qs[0]) + _check_and_callback(task, task_success=task_success) try: - send_taskflow_message.delay(task_id=task_id, msg_type=TASK_FINISHED) + send_taskflow_message.delay(task_id=task.id, msg_type=TASK_FINISHED) except Exception as e: - logger.exception("send_taskflow_message[taskflow_id=%s] task delay error: %s" % (task_id, e)) + logger.exception("send_taskflow_message[taskflow_id=%s] task delay error: %s" % (task.id, e)) if sig is taskflow_revoked: - _check_and_callback(task_id, task_success=False, task=qs[0]) + _check_and_callback(task, task_success=False) def _send_node_fail_message(node_id, pipeline_id): @@ -74,7 +72,7 @@ def _send_node_fail_message(node_id, pipeline_id): except TaskFlowInstance.DoesNotExist: logger.error("pipeline finished handler get taskflow error, pipeline_instance_id=%s" % pipeline_id) return - _check_and_callback(taskflow.id, task_success=False, task=taskflow) + _check_and_callback(taskflow, task_success=False) if taskflow.is_child_taskflow is False: try: @@ -85,15 +83,24 @@ def _send_node_fail_message(node_id, pipeline_id): logger.exception("pipeline_fail_handler[taskflow_id=%s] task delay error: %s" % (taskflow.id, e)) -def _check_and_callback(taskflow_id, *args, **kwargs): - if not TaskCallBackRecord.objects.filter(task_id=taskflow_id).exists(): +def _check_and_callback(task, *args, **kwargs): + record = TaskCallBackRecord.objects.filter(task_id=task.id).first() + if not record: return try: - if kwargs.get("task"): - task = kwargs.pop("task") - kwargs["task_outputs"] = json.dumps(task.get_task_detail()["outputs"]) + if ( + record.url + and json.loads(record.extra_info).get("callback_version") == TaskCallBackRecord.CALLBACK_VERSION_V2 + ): + # 检查任务的输出是否可以被json序列化，如果可以则将输出作为参数传给回调函数，否则不做处理 + try: + task_outputs = task.get_task_detail()["outputs"] + json.dumps(task_outputs) + kwargs["task_outputs"] = task_outputs + except Exception as e: + logger.exception(f"[task {task.id}] outputs data serialize error: {e}") task_callback.apply_async( - kwargs=dict(task_id=taskflow_id, **kwargs), + kwargs=dict(task_id=task.id, **kwargs), queue="task_callback", routing_key="task_callback", )