feat: 新增字画布节点 --story=135278474

bkflow/apigw/serializers/task.py

@@ -228,6 +228,7 @@ class GetTaskListSerializer(serializers.Serializer):
     executor = serializers.CharField(help_text=_("执行者"), max_length=USER_NAME_MAX_LENGTH, required=False)
     template_id = serializers.IntegerField(help_text=_("流程ID"), required=False)
     label = serializers.CharField(help_text=_("标签名称"), required=False)
+    is_child_taskflow = serializers.CharField(help_text=_("过滤子任务"), required=False, default="false")
 
     def validate_task_id_list(self, value):
         return ",".join(str(i) for i in value)

bkflow/apigw/views/get_task_list.py

@@ -50,6 +50,7 @@ def get_task_list(request, space_id):
         "task_id_list": "id__in",
         "executor": "executor",
         "template_id": "template_id",
+        "is_child_taskflow": "is_child_taskflow",
     }
     for k, v in filter_map.items():
         if k in data:

bkflow/constants.py

@@ -94,6 +94,7 @@ class TaskTriggerMethod(Enum):
     manual = _("手动")
     timing = _("定时")
     subprocess = _("子流程")
+    sub_canvas = _("子画布")
 
 
 class TemplateOperationSource(Enum):

bkflow/contrib/api/collections/task.py

@@ -178,3 +178,20 @@ def upsert_engine_config(self, data):
 
     def delete_engine_config(self, data):
         return self._request(method="delete", url=self._get_task_url("task/delete_engine_config/"), data=data)
+
+    def list_children_taskflow(self, data):
+        """获取根任务下的所有子任务列表"""
+        return self._request(method="get", url=self._get_task_url("task/list_children_taskflow/"), data=data)
+
+    def root_task_info(self, data):
+        """批量查询任务是否包含子任务"""
+        return self._request(method="get", url=self._get_task_url("task/root_task_info/"), data=data)
+
+    def get_node_outputs(self, data):
+        return self._request(method="post", url=self._get_task_url("task/get_node_outputs/"), data=data)
+
+    def get_tasks_pipeline(self, data):
+        return self._request(method="get", url=self._get_task_url("task/get_tasks_pipeline/"), data=data)
+
+    def batch_get_task_states(self, data):
+        return self._request(method="get", url=self._get_task_url("task/batch_get_task_states/"), data=data)

bkflow/contrib/openapi/serializers.py

@@ -1,4 +1,3 @@
-# -*- coding: utf-8 -*-
 """
 TencentBlueKing is pleased to support the open source community by making
 蓝鲸流程引擎服务 (BlueKing Flow Engine Service) available.
@@ -56,6 +55,12 @@ class GetTasksStatesBodySerializer(serializers.Serializer):
     space_id = serializers.IntegerField(required=True)
 
 
+class GetNodeOutputsSerializer(serializers.Serializer):
+    task_id = serializers.IntegerField(required=True)
+    space_id = serializers.IntegerField(required=True)
+    node_ids = serializers.ListField(child=serializers.CharField())
+
+
 class TaskEngineAdminSerializer(serializers.Serializer):
     space_id = serializers.IntegerField(required=True)
     instance_id = serializers.CharField(required=True)

bkflow/interface/task/view.py

@@ -314,3 +314,44 @@ def render_stage_constants(self, request, task_id, *args, **kwargs):
         handler = StageConstantHandler(space_id, request.user.is_superuser)
         result = handler.process(task_id, node_ids, stage_constants)
         return Response(result)
+
+    @action(methods=["GET"], detail=False, url_path="list_children_taskflow/(?P<task_id>\\d+)")
+    def list_children_taskflow(self, request, task_id, *args, **kwargs):
+        """获取根任务下的所有子任务列表"""
+        space_id = self.get_space_id(request)
+        client = TaskComponentClient(space_id=space_id, from_superuser=request.user.is_superuser)
+        result = client.list_children_taskflow(data={"task_id": task_id, "space_id": space_id})
+        return Response(result)
+
+    @action(methods=["GET"], detail=False, url_path="root_task_info")
+    def root_task_info(self, request, *args, **kwargs):
+        """批量查询任务是否包含子任务"""
+        space_id = self.get_space_id(request)
+        client = TaskComponentClient(space_id=space_id, from_superuser=request.user.is_superuser)
+        result = client.root_task_info(data={"task_ids": request.query_params.get("task_ids"), "space_id": space_id})
+        return Response(result)
+
+    @action(methods=["POST"], detail=False, url_path="get_node_outputs")
+    def get_node_outputs(self, request, *args, **kwargs):
+        space_id = self.get_space_id(request)
+        client = TaskComponentClient(space_id=space_id, from_superuser=request.user.is_superuser)
+        result = client.get_node_outputs(data={"space_id": space_id, **request.data})
+        return Response(result)
+
+    @action(methods=["GET"], detail=False, url_path="get_tasks_pipeline")
+    def get_tasks_pipeline(self, request, *args, **kwargs):
+        space_id = self.get_space_id(request)
+        client = TaskComponentClient(space_id=space_id, from_superuser=request.user.is_superuser)
+        result = client.get_tasks_pipeline(
+            data={"space_id": space_id, "task_ids": request.query_params.get("task_ids")}
+        )
+        return Response(result)
+
+    @action(methods=["GET"], detail=False, url_path="batch_get_task_states")
+    def batch_get_task_states(self, request, *args, **kwargs):
+        space_id = self.get_space_id(request)
+        client = TaskComponentClient(space_id=space_id, from_superuser=request.user.is_superuser)
+        result = client.batch_get_task_states(
+            data={"space_id": space_id, "task_ids": request.query_params.get("task_ids")}
+        )
+        return Response(result)

bkflow/pipeline_plugins/components/collections/base.py

@@ -16,12 +16,24 @@
 
 to the current version of the project delivered to anyone in the future.
 """
+import copy
+import datetime
 
+from bamboo_engine.context import Context
+from bamboo_engine.eri import ContextValue, ContextValueType
+from bamboo_engine.template import Template
 from django.apps import apps
 from django.conf import settings
+from django.db import transaction
+from django.utils import timezone
 from pipeline.core.flow import AbstractIntervalGenerator, StaticIntervalGenerator
 from pipeline.core.flow.activity import Service
+from pipeline.core.flow.io import ArrayItemSchema, IntItemSchema, ObjectItemSchema
+from pipeline.eri.runtime import BambooDjangoRuntime
 
+from bkflow.constants import TaskOperationSource, TaskOperationType
+from bkflow.exceptions import ValidationError
+from bkflow.utils.handlers import mask_sensitive_data_for_display
 from bkflow.utils.trace import (
     PLUGIN_SCHEDULE_COUNT_KEY,
     PLUGIN_SPAN_ENDED_KEY,
@@ -258,3 +270,248 @@ def next(self):
 
     def reach_limit(self):
         return self.count >= self.max_count
+
+
+class LoopBaseService(BKFlowBaseService):
+    """循环基类服务，提供循环执行相关的基础方法"""
+
+    runtime = BambooDjangoRuntime()
+
+    def outputs_format(self):
+        return [
+            self.OutputItem(name="任务ID", key="task_id", type="int", schema=IntItemSchema(description="Task ID")),
+            self.OutputItem(
+                name="循环输出",
+                key=settings.PLUGIN_LOOP_OUTPUTS_KEY,
+                type="array",
+                schema=ArrayItemSchema(
+                    description="循环输出", item_schema=ObjectItemSchema(description="循环输出", property_schemas={})
+                ),
+            ),
+        ]
+
+    def _render_parent_parameters(self, pipeline_tree, parent_task):
+        """渲染父任务参数到子流程常量"""
+
+        # 渲染父任务中的参数
+        constants = pipeline_tree.get("constants", {})
+        subprocess_inputs = {
+            key: constant["value"]
+            for key, constant in constants.items()
+            if constant.get("show_type") == "show" and constant.get("need_render", True)
+        }
+        raw_subprocess_inputs = copy.deepcopy(subprocess_inputs)
+        inputs_refs = Template(subprocess_inputs).get_reference()
+        self.logger.info(f"subprocess original refs: {inputs_refs}")
+        additional_refs = self.runtime.get_context_key_references(pipeline_id=self.top_pipeline_id, keys=inputs_refs)
+        inputs_refs = inputs_refs.union(additional_refs)
+        self.logger.info(f"subprocess final refs: {inputs_refs}")
+        context_values = self.runtime.get_context_values(pipeline_id=self.top_pipeline_id, keys=inputs_refs)
+        node = self.runtime.get_node(self.id)
+        if node.loop_enabled:
+            loop_params = (
+                parent_task.pipeline_tree["activities"][self.id].get("loop_config", {}).get("loop_params") or {}
+            )
+            min_loop_times = None
+            for param_key, param_value in loop_params.items():
+                param_refs = Template(param_value).get_reference()
+                if param_refs:
+                    param_context_values = self.runtime.get_context_values(
+                        pipeline_id=self.top_pipeline_id, keys=param_refs
+                    )
+
+                    hydrated_context = Context(self.runtime, param_context_values, {}).hydrate(deformat=True)
+                    inputs = Template(param_value).render(hydrated_context)
+
+                    # 判断渲染后的值是否为可迭代对象（列表/元组/字典），若不是则抛出异常
+                    if not isinstance(inputs, (list, tuple, dict)):
+                        raise ValidationError(
+                            f"循环参数 {param_key} 的值必须是可迭代对象，" f"当前值类型为 {type(inputs).__name__}，值为：{inputs}"
+                        )
+
+                    if len(inputs) > settings.MAX_LOOP_TIMES:
+                        raise ValidationError(f"循环参数 {param_key} 的值超过最大循环次数 {settings.MAX_LOOP_TIMES}")
+
+                    current_len = len(inputs)
+                    loop_item_value = list(inputs)[self.inner_loop - 1]
+                else:
+                    items = [item.strip() for item in param_value.split(",") if item.strip()]
+                    current_len = len(items)
+                    loop_item_value = items[self.inner_loop - 1]
+
+                min_loop_times = current_len if min_loop_times is None else min(min_loop_times, current_len)
+
+                context_value = ContextValue(
+                    key=param_key, type=ContextValueType.PLAIN, value=loop_item_value, code=None
+                )
+                context_values.append(context_value)
+
+            if not node.loop_times:
+                self.runtime.update_node_loop_times(node_id=self.id, loop_times=min_loop_times)
+
+        context_mappings = {c.key: c for c in context_values}
+        root_pipeline_inputs = {
+            key: inputs.value for key, inputs in self.runtime.get_data_inputs(self.top_pipeline_id).items()
+        }
+        context = Context(self.runtime, context_values, root_pipeline_inputs)
+        hydrated_context = context.hydrate(deformat=True)
+        # 对上下文进行脱敏后再打印日志，避免泄露 credentials 等敏感信息
+        self.logger.info(f"subprocess parent hydrated context: {mask_sensitive_data_for_display(hydrated_context)}")
+
+        parsed_subprocess_inputs = Template(subprocess_inputs).render(hydrated_context)
+        parent_constants = parent_task.pipeline_tree["constants"]
+        for key, constant in pipeline_tree.get("constants", {}).items():
+            # 如果父流程直接勾选，则直接使用父流程对应变量的值
+            raw_constant_value = raw_subprocess_inputs.get(key)
+            if (
+                raw_constant_value
+                and isinstance(raw_constant_value, str)
+                and parent_constants.get(raw_constant_value)
+                and self.id in parent_constants[raw_constant_value]["source_info"]
+                and key in parent_constants[raw_constant_value]["source_info"][self.id]
+            ):
+                constant["value"] = context_mappings[raw_subprocess_inputs[key]].value
+            elif constant.get("need_render", True) and key in parsed_subprocess_inputs:
+                constant["value"] = parsed_subprocess_inputs[key]
+        # 对 constants 进行脱敏后再打印日志，避免泄露 credentials 等敏感信息
+        self.logger.info(
+            f'subprocess parsed constants: {mask_sensitive_data_for_display(pipeline_tree.get("constants", {}))}'
+        )
+        return context_values
+
+    def _create_subprocess_task_instance(
+        self, template_name, pipeline_tree, parent_task, trigger_method, template_id=None, notify_config=None
+    ):
+        """创建子任务实例和关系记录"""
+        from bkflow.task.models import (
+            TaskFlowRelation,
+            TaskInstance,
+            TaskOperationRecord,
+        )
+        from bkflow.task.utils import extract_extra_info
+
+        with transaction.atomic():
+            time_zone = timezone.pytz.timezone(settings.TIME_ZONE) or "Asia/Shanghai"
+            time_stamp = datetime.datetime.now(tz=time_zone).strftime("%Y%m%d%H%M%S")
+            create_task_data = {
+                "name": f"{template_name}_子流程_{time_stamp}",
+                "template_id": template_id,
+                "creator": parent_task.creator,
+                "scope_type": parent_task.scope_type,
+                "scope_value": parent_task.scope_value,
+                "space_id": parent_task.space_id,
+                "pipeline_tree": pipeline_tree,
+                "trigger_method": trigger_method,
+                "mock_data": {},
+            }
+            DEFAULT_NOTIFY_CONFIG = {
+                "notify_type": {"fail": [], "success": []},
+                "notify_receivers": {"more_receiver": "", "receiver_group": []},
+            }
+            create_task_data.setdefault("extra_info", {}).update(
+                {"notify_config": notify_config or DEFAULT_NOTIFY_CONFIG}
+            )
+
+            task_instance = TaskInstance.objects.create_instance(**create_task_data)
+
+            # 记录操作流水
+            pipeline_constants = task_instance.pipeline_tree.get("constants")
+            extra_info = extract_extra_info(pipeline_constants)
+            TaskOperationRecord.objects.create(
+                instance_id=task_instance.id,
+                operate_type=TaskOperationType.create.name,
+                operate_source=TaskOperationSource.api.name,
+                operator=parent_task.creator,
+                extra_info=extra_info,
+            )
+
+            try:
+                root_task_id = TaskFlowRelation.objects.get(task_id=parent_task.id).root_task_id
+            except TaskFlowRelation.DoesNotExist:
+                root_task_id = parent_task.id
+
+            relate_info = {"node_id": self.id, "node_version": self.version, "trigger_method": trigger_method}
+            TaskFlowRelation.objects.create(
+                task_id=task_instance.id,
+                parent_task_id=parent_task.id,
+                root_task_id=root_task_id,
+                extra_info=relate_info,
+            )
+
+            return task_instance
+
+    def plugin_execute(self, data, parent_data):
+        pass
+
+    def plugin_schedule(self, data, parent_data, callback_data=None):
+        from bkflow.task.models import TaskInstance
+
+        task_success = callback_data.get("task_success", False)
+        task_id = data.get_one_of_outputs("task_id")
+
+        try:
+            subprocess_task = TaskInstance.objects.get(id=task_id)
+        except TaskInstance.DoesNotExist:
+            message = f"子任务[{task_id}]不存在"
+            self.logger.error(message)
+            data.set_outputs("ex_data", message)
+            return False
+
+        subprocess_pipeline_id = subprocess_task.instance_id
+        self.logger.info(f"subprocess pipeline id: {subprocess_pipeline_id}")
+        subprocess_execution_data_outputs = self.runtime.get_execution_data_outputs(node_id=subprocess_pipeline_id)
+        self.logger.info(f"subprocess execution data outputs: {subprocess_execution_data_outputs}")
+        node_outputs = self.runtime.get_data_outputs(self.id)
+        self.logger.info(f"node outputs: {node_outputs}")
+
+        node = self.runtime.get_node(self.id)
+        loop_outputs_key = node.loop_outputs_key
+        self.finish_schedule()
+        if not node.loop_enabled:
+            if not task_success:
+                data.set_outputs("ex_data", "子任务执行失败，请检查失败节点")
+                return False
+
+            for key in filter(lambda x: x in subprocess_execution_data_outputs, node_outputs.keys()):
+                data.set_outputs(key, subprocess_execution_data_outputs[key])
+        else:
+            # 先剔除上下文循环输出列表中与当前循环次数（inner_loop）相同的旧记录，
+            # 避免同一次循环被重复回调时（如节点先失败跳过、后重试成功的场景）
+            # 导致上下文输出列表中出现重复记录。之后再由 extract_outputs 统一追加本次的 outputs。
+            parent_task_id = parent_data.get_one_of_inputs("task_id")
+            parent_pipeline_id = TaskInstance.objects.get(id=parent_task_id).instance_id
+
+            loop_context_values = self.runtime.get_context_values(parent_pipeline_id, {loop_outputs_key})
+            if loop_context_values:
+                current_value = loop_context_values[0].value
+                if isinstance(current_value, list):
+                    filtered_value = [
+                        item
+                        for item in current_value
+                        if not (isinstance(item, dict) and item.get("inner_loop") == self.inner_loop)
+                    ]
+                    if len(filtered_value) != len(current_value):
+                        updated_context_values = [
+                            ContextValue(
+                                key=loop_outputs_key,
+                                type=ContextValueType.PLAIN,
+                                value=filtered_value,
+                            )
+                        ]
+                        self.runtime.update_context_values(parent_pipeline_id, updated_context_values)
+
+            outputs = {"task_id": task_id, "inner_loop": self.inner_loop}
+            if not task_success:
+                outputs["ex_data"] = "子任务执行失败，请检查失败节点"
+                data.set_outputs("ex_data", "子任务执行失败，请检查失败节点")
+            else:
+                # 遍历子流程的输出，判断该输出是否在节点的输出变量中，在则加入
+                for key, value in subprocess_execution_data_outputs.items():
+                    data.set_outputs(key, subprocess_execution_data_outputs[key])
+                    key = key.removeprefix("${").removesuffix("}")
+                    outputs[key] = value
+            # 无论成功失败，都将 outputs 字典设置到输出中，由 extract_outputs 统一追加到列表
+            data.set_outputs(settings.LOOP_OUTPUTS_INNER_KEY, outputs)
+            if not task_success:
+                return False
+        return True