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", )