feat: sops 回调功能同步代码 --story=120883063

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)

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 = (

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=_("回调地址"))

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