Merge pull request #281 from TencentBlueKing/develop

release V1.11.0

Author: dengyh

.gitignore

@@ -142,4 +142,9 @@ staticfiles
 
 bkflow/interface/templates
 static/bkflow
-frontend/static
+frontend/static
+
+pre-*-bkcodeai
+.vscode/settings.json
+
+.idea

app_desc.yaml

@@ -1,5 +1,5 @@
 spec_version: 2
-app_version: "1.10.0"
+app_version: "1.11.0"
 app:
   region: default
   bk_app_code: &APP_CODE bk_flow_engine

bkcodeai.json

@@ -0,0 +1,30 @@
+{
+  "crignore_file_patterns": [
+    "out/",
+    "dist/",
+    "assets/",
+    "images/",
+    "package.json",
+    "package-lock.json",
+    "yarn.lock",
+    ".env.*",
+    "*.env",
+    "*.map",
+    "*.yml",
+    "*.yaml",
+    "*.min.js",
+    "libs/",
+    "static/",
+    "doc/",
+    "docs/",
+    "*.mo",
+    "*.po",
+    "*.ini",
+    "*.toml",
+    "*.json",
+    "*.xml",
+    "*.lock",
+    "*.md",
+    "*.txt"
+  ]
+}

bkflow/apigw/serializers/task.py

@@ -53,6 +53,9 @@ class CreateMockTaskBaseSerializer(serializers.Serializer):
 
 class CreateMockTaskWithPipelineTreeSerializer(CreateMockTaskBaseSerializer):
     pipeline_tree = serializers.JSONField(help_text=_("任务树"), required=True)
+    include_node_ids = serializers.ListField(
+        child=serializers.CharField(allow_blank=False), help_text=_("包含的节点ID列表"), required=False
+    )
 
 
 class CreateMockTaskWithTemplateIdSerializer(CreateMockTaskBaseSerializer):

bkflow/apigw/views/create_template.py

@@ -35,6 +35,8 @@
 from bkflow.space.utils import build_default_pipeline_tree_with_space_id
 from bkflow.template.models import Template, TemplateSnapshot
 from bkflow.utils import err_code
+from bkflow.utils.canvas import OperateType
+from bkflow.utils.pipeline import replace_pipeline_tree_node_ids
 
 logger = logging.getLogger("root")
 
@@ -70,7 +72,7 @@ def create_template(request, space_id):
         # 在序列化器中已经判断了存在，所以不需要处理异常
         source_template = Template.objects.get(id=source_template_id)
         pipeline_tree = copy.deepcopy(source_template.pipeline_tree)
-        recursive_replace_id(pipeline_tree)
+        replace_pipeline_tree_node_ids(pipeline_tree, OperateType.CREATE_TEMPLATE.value)
     elif pipeline_tree:
         recursive_replace_id(pipeline_tree)
     else:

bkflow/contrib/api/collections/task.py

@@ -108,6 +108,13 @@ def render_current_constants(self, task_id):
             method="get", url=self._get_task_url("task/{}/render_current_constants/".format(task_id)), data=None
         )
 
+    def render_context_with_node_outputs(self, task_id, data=None):
+        return self._request(
+            method="post",
+            url=self._get_task_url("task/{}/render_context_with_node_outputs/".format(task_id)),
+            data=data,
+        )
+
     def get_task_operation_record(self, task_id, data=None):
         return self._request(
             method="get", url=self._get_task_url("task/{}/get_task_operation_record/".format(task_id)), data=data
@@ -130,3 +137,12 @@ def trigger_engine_admin_action(self, instance_id, action, data=None):
 
     def batch_delete_tasks(self, data):
         return self._request(method="post", url=self._get_task_url("task/batch_delete_tasks/"), data=data)
+
+    def get_engine_config(self, data):
+        return self._request(method="get", url=self._get_task_url("task/get_engine_config/"), data=data)
+
+    def upsert_engine_config(self, data):
+        return self._request(method="post", url=self._get_task_url("task/upsert_engine_config/"), data=data)
+
+    def delete_engine_config(self, data):
+        return self._request(method="delete", url=self._get_task_url("task/delete_engine_config/"), data=data)

bkflow/contrib/openapi/serializers.py

@@ -74,3 +74,8 @@ def validate(self, attrs):
             raise serializers.ValidationError("is_mock must exist when delete all tasks")
 
         return attrs
+
+
+class RenderConstantsBodySerializer(serializers.Serializer):
+    node_ids = serializers.ListField(required=True, child=serializers.CharField())
+    to_render_constants = serializers.ListField(required=True, child=serializers.DictField())

bkflow/interface/task/utils.py

@@ -0,0 +1,225 @@
+# -*- coding: utf-8 -*-
+"""
+TencentBlueKing is pleased to support the open source community by making
+蓝鲸流程引擎服务 (BlueKing Flow Engine Service) available.
+Copyright (C) 2024 THL A29 Limited,
+a Tencent company. All rights reserved.
+Licensed under the MIT License (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at http://opensource.org/licenses/MIT
+Unless required by applicable law or agreed to in writing,
+software distributed under the License is distributed on
+an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND,
+either express or implied. See the License for the
+specific language governing permissions and limitations under the License.
+
+We undertake not to change the open source license (MIT license) applicable
+
+to the current version of the project delivered to anyone in the future.
+"""
+import logging
+from typing import Dict
+
+from bkflow.contrib.api.collections.task import TaskComponentClient
+
+logger = logging.getLogger(__name__)
+
+
+class StageJobStateHandler:
+    """处理Stage和Job状态的业务逻辑处理器"""
+
+    def __init__(self, space_id: int, is_superuser: bool = False):
+        self.space_id = space_id
+        self.is_superuser = is_superuser
+        self.client = TaskComponentClient(space_id=space_id, from_superuser=is_superuser)
+
+    def get_task_data(self, task_id: str) -> dict:
+        """获取任务相关的基础数据
+
+        Args: task_id: 任务ID
+        Returns: dict: 包含任务相关数据的字典，如果获取失败则包含默认值
+        """
+        try:
+            task_detail = self.client.get_task_detail(task_id)
+
+            if "data" not in task_detail:
+                logger.warning(f"Task {task_id} response data missing: detail={task_detail}")
+
+            pipeline_tree = task_detail.get("data", {}).get("pipeline_tree", {})
+            return {
+                "task_detail": task_detail,
+                "pipeline_tree": pipeline_tree,
+            }
+        except Exception as e:
+            logger.exception(f"Failed to get task data for task {task_id}: {str(e)}")
+            raise
+
+    def get_node_states(self, task_id: str) -> Dict:
+        """获取节点状态信息
+
+        Args: task_id: 任务ID
+        Returns: 节点状态信息字典
+        """
+        data = {"space_id": self.space_id}
+
+        try:
+            response = self.client.get_task_states(task_id, data=data)
+            if "data" not in response:
+                logger.warning(f"Task {task_id} states response data missing: {response}")
+
+            return response.get("data", {}).get("children", {})
+        except Exception as e:
+            logger.exception(f"Failed to get node states for task {task_id}: {str(e)}")
+            return {}
+
+    def build_template_task_mapping(self, activities: dict) -> dict:
+        """构建模板节点ID到任务节点ID的映射"""
+        return {
+            activity.get("template_node_id"): task_node_id
+            for task_node_id, activity in activities.items()
+            if activity.get("template_node_id")
+        }
+
+    def build_node_info_map(self, template_to_task_id: dict, node_states: dict) -> dict:
+        """构建节点信息映射"""
+        return {
+            template_id: {
+                "state": node_states[task_id].get("state", "READY"),
+                "start_time": node_states[task_id].get("start_time", ""),
+                "finish_time": node_states[task_id].get("finish_time", ""),
+                "loop": node_states[task_id].get("loop", 1),
+                "retry": node_states[task_id].get("retry", 0),
+                "skip": node_states[task_id].get("skip", False),
+                "error_ignorable": node_states[task_id].get("error_ignorable", False),
+                "error_ignored": node_states[task_id].get("error_ignored", False),
+            }
+            for template_id, task_id in template_to_task_id.items()
+            if task_id in node_states
+        }
+
+    @staticmethod
+    def calculate_job_state(node_states: list) -> str:
+        """计算Job状态"""
+        if not node_states:
+            return "READY"
+
+        if "FAILED" in node_states:
+            return "FAILED"
+
+        if "RUNNING" in node_states:
+            return "RUNNING"
+
+        if all(state == "READY" for state in node_states):
+            return "READY"
+
+        if all(state == "FINISHED" for state in node_states):
+            return "FINISHED"
+
+        return "RUNNING"
+
+    @staticmethod
+    def calculate_stage_state(job_states: list) -> str:
+        """计算Stage状态"""
+        if not job_states:
+            return "READY"
+
+        if "FAILED" in job_states:
+            return "FAILED"
+
+        if "RUNNING" in job_states:
+            return "RUNNING"
+
+        if all(state == "READY" for state in job_states):
+            return "READY"
+
+        if all(state == "FINISHED" for state in job_states):
+            return "FINISHED"
+
+        return "RUNNING"
+
+    def update_states(self, stage_struct: list, node_info_map: dict) -> None:
+        """更新状态信息"""
+        # 构建job到nodes的映射，避免深层嵌套
+        job_nodes_map = {}
+        stage_jobs_map = {}
+
+        # 第一次遍历：构建映射关系
+        for stage in stage_struct:
+            stage_jobs = []
+            for job in stage["jobs"]:
+                job_id = job["id"]
+                job_nodes_map[job_id] = job["nodes"]
+                stage_jobs.append(job_id)
+            stage_jobs_map[stage["id"]] = stage_jobs
+
+        # 第二次遍历：更新节点状态
+        job_states_map = {}
+        for job_id, nodes in job_nodes_map.items():
+            node_states = []
+            for node in nodes:
+                template_node_id = node["id"]
+                info = node_info_map.get(
+                    template_node_id,
+                    {
+                        "state": "READY",
+                        "start_time": "",
+                        "finish_time": "",
+                        "loop": 1,
+                        "retry": 0,
+                        "skip": False,
+                        "error_ignorable": False,
+                        "error_ignored": False,
+                    },
+                )
+                node.update(info)
+                node_states.append(info["state"])
+
+            # 计算job状态
+            job_state = self.calculate_job_state(node_states)
+            job_states_map[job_id] = job_state
+
+        # 第三次遍历：更新stage和job状态
+        for stage in stage_struct:
+            stage_jobs = stage_jobs_map[stage["id"]]
+            job_states = [job_states_map[job_id] for job_id in stage_jobs]
+
+            # 更新job状态
+            for job in stage["jobs"]:
+                job["state"] = job_states_map[job["id"]]
+
+            # 更新stage状态
+            stage["state"] = self.calculate_stage_state(job_states)
+
+    def process(self, task_id: str) -> list:
+        """处理完整的状态更新流程"""
+        # 1. 获取基础数据
+        task_data = self.get_task_data(task_id)
+        activities = task_data.get("pipeline_tree", {}).get("activities", {})
+        stage_struct = task_data.get("pipeline_tree", {}).get("stage_canvas_data", [])
+
+        # 2. 获取节点状态
+        node_states = self.get_node_states(task_id)
+
+        # 3. 构建映射关系
+        template_to_task_id = self.build_template_task_mapping(activities)
+        node_info_map = self.build_node_info_map(template_to_task_id, node_states)
+
+        # 4. 更新状态
+        self.update_states(stage_struct, node_info_map)
+
+        return stage_struct
+
+
+class StageConstantHandler:
+    def __init__(self, space_id: int, is_superuser: bool = False):
+        self.space_id = space_id
+        self.is_superuser = is_superuser
+        self.client = TaskComponentClient(space_id=space_id, from_superuser=is_superuser)
+
+    def process(self, task_id: str, node_ids: list, stage_constants: list) -> dict:
+        to_render_constants = [constant["key"] for constant in stage_constants]
+        rendered_constants = self.client.render_context_with_node_outputs(
+            task_id, data={"node_ids": node_ids, "to_render_constants": to_render_constants}
+        )
+
+        return rendered_constants

bkflow/interface/task/view.py

@@ -29,6 +29,7 @@
 from bkflow.contrib.api.collections.task import TaskComponentClient
 from bkflow.contrib.openapi.serializers import (
     GetTasksStatesBodySerializer,
+    RenderConstantsBodySerializer,
     TaskBatchDeleteSerializer,
     TaskEngineAdminSerializer,
 )
@@ -37,6 +38,7 @@
     TaskMockTokenPermission,
     TaskTokenPermission,
 )
+from bkflow.interface.task.utils import StageConstantHandler, StageJobStateHandler
 from bkflow.permission.models import TASK_PERMISSION_TYPE, Token
 from bkflow.space.configs import SuperusersConfig
 from bkflow.space.models import SpaceConfig
@@ -229,3 +231,24 @@ def get_node_snapshot_config(self, request, task_id, node_id, *args, **kwargs):
         data = {"node_id": node_id}
         result = client.get_node_snapshot_config(task_id, data)
         return Response(result)
+
+    @action(methods=["GET"], detail=False, url_path="get_stage_job_states/(?P<task_id>\\d+)")
+    def get_stage_and_job_states(self, request, task_id, *args, **kwargs):
+        """获取stage和job状态的视图函数"""
+        space_id = self.get_space_id(request)
+        handler = StageJobStateHandler(space_id, request.user.is_superuser)
+        result = handler.process(task_id)
+        return Response(result)
+
+    @action(methods=["POST"], detail=False, url_path="rendered_stage_constants/(?P<task_id>\\d+)")
+    @swagger_auto_schema(operation_description="渲染stage画布变量", request_body=RenderConstantsBodySerializer)
+    def render_stage_constants(self, request, task_id, *args, **kwargs):
+        """渲染stage画布变量"""
+        space_id = self.get_space_id(request)
+        serializer = RenderConstantsBodySerializer(data=request.data)
+        serializer.is_valid(raise_exception=True)
+        node_ids = serializer.validated_data.get("node_ids", [])
+        stage_constants = serializer.validated_data.get("to_render_constants", {})
+        handler = StageConstantHandler(space_id, request.user.is_superuser)
+        result = handler.process(task_id, node_ids, stage_constants)
+        return Response(result)