From f8e9f0418b6ad70c5dabe629689545c175738e14 Mon Sep 17 00:00:00 2001 From: guohelu <19503896967@163.com> Date: Mon, 25 Nov 2024 15:33:00 +0800 Subject: [PATCH 1/2] =?UTF-8?q?feat:=20sops=20=E5=9B=9E=E8=B0=83=E5=8A=9F?= =?UTF-8?q?=E8=83=BD=E5=90=8C=E6=AD=A5=E4=BB=A3=E7=A0=81=20--story=3D12088?= =?UTF-8?q?3063?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- gcloud/apigw/views/create_task.py | 9 ++++++++- gcloud/taskflow3/domains/callback.py | 9 +++++++-- gcloud/taskflow3/models.py | 2 ++ gcloud/taskflow3/signals/handlers.py | 3 +-- 4 files changed, 18 insertions(+), 5 deletions(-) diff --git a/gcloud/apigw/views/create_task.py b/gcloud/apigw/views/create_task.py index 30335a038c..e27b2b8c83 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 043b829669..842f279ffb 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 20fb804e8e..b0df2c9544 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 7788dea646..8f65353284 100644 --- a/gcloud/taskflow3/signals/handlers.py +++ b/gcloud/taskflow3/signals/handlers.py @@ -12,7 +12,6 @@ """ import datetime -import json import logging from bamboo_engine import states as bamboo_engine_states @@ -91,7 +90,7 @@ def _check_and_callback(taskflow_id, *args, **kwargs): try: if kwargs.get("task"): task = kwargs.pop("task") - kwargs["task_outputs"] = json.dumps(task.get_task_detail()["outputs"]) + kwargs["task_outputs"] = task.get_task_detail()["outputs"] task_callback.apply_async( kwargs=dict(task_id=taskflow_id, **kwargs), queue="task_callback", From 7bad91bb69d6d66e024f06f4a93298947db195f8 Mon Sep 17 00:00:00 2001 From: guohelu <19503896967@163.com> Date: Thu, 28 Nov 2024 18:04:42 +0800 Subject: [PATCH 2/2] =?UTF-8?q?fix:=20=E4=BF=AE=E5=A4=8Djson=E5=BA=8F?= =?UTF-8?q?=E5=88=97=E5=8C=96=E9=97=AE=E9=A2=98=20--story=3D120883063?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- gcloud/taskflow3/signals/handlers.py | 42 +++++++++++++++++----------- 1 file changed, 25 insertions(+), 17 deletions(-) diff --git a/gcloud/taskflow3/signals/handlers.py b/gcloud/taskflow3/signals/handlers.py index 8f65353284..c21a6d3baf 100644 --- a/gcloud/taskflow3/signals/handlers.py +++ b/gcloud/taskflow3/signals/handlers.py @@ -12,6 +12,7 @@ """ import datetime +import json import logging from bamboo_engine import states as bamboo_engine_states @@ -46,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): @@ -73,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: @@ -84,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"] = 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", )