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

[guohelu]

No description provided.

[dengyh]

bamboo_engine/version.py没有改

[dengyh]

pipeline/init.py也没有改

[dengyh]

如果在引擎侧抽象了SubCanvas这个概念，建议相关的方法都在引擎侧实现，然后bkflow调用，否则对于bamboo-engine而言，功能上就不是闭环的了，需要耦合bkflow的逻辑才能解析SubCanvas。

或者另一个改法是：SubCanvas这个概念只放到bkflow里，不在bamboo-engine里抽象这个概念

参考：

bamboo-engine/runtime/bamboo-pipeline/pipeline/builder/builder.py

Lines 35 to 58 in 45c2e44

	__node_type = {
	PE.ServiceActivity: PE.activities,
	PE.SubProcess: PE.activities,
	PE.EmptyEndEvent: PE.end_event,
	PE.EmptyStartEvent: PE.start_event,
	PE.ParallelGateway: PE.gateways,
	PE.ConditionalParallelGateway: PE.gateways,
	PE.ExclusiveGateway: PE.gateways,
	PE.ConvergeGateway: PE.gateways,
	}
	__start_elem = {PE.EmptyStartEvent}
	__end_elem = {PE.EmptyEndEvent}
	__multiple_incoming_type = {
	PE.ServiceActivity,
	PE.ConvergeGateway,
	PE.EmptyEndEvent,
	PE.ParallelGateway,
	PE.ConditionalParallelGateway,
	PE.ExclusiveGateway,
	PE.SubProcess,
	}

bamboo-engine/runtime/bamboo-pipeline/pipeline/parser/pipeline_parser.py

Lines 142 to 185 in 45c2e44

	for act in list(acts.values()):
	act_cls = FlowNodeClsFactory.get_node_cls(act[PE.type])
	if act[PE.type] == PE.ServiceActivity:
	component = ComponentLibrary.get_component(
	component_code=act[PE.component][PE.code],
	data_dict=act[PE.component][PE.inputs],
	version=act[PE.component].get(PE.version),
	)
	service = component.service()
	data = component.data_for_execution(context, pipeline_data)
	handler_path = act.get("failure_handler")
	failure_handler = import_string(handler_path) if handler_path else None
	act_objs.append(
	act_cls(
	id=act[PE.id],
	service=service,
	name=act[PE.name],
	data=data,
	error_ignorable=act.get(PE.error_ignorable, False),
	skippable=act[PE.skippable] if PE.skippable in act else act.get(PE.skippable_old, True),
	retryable=act[PE.retryable] if PE.retryable in act else act.get(PE.retryable_old, True),
	timeout=act.get(PE.timeout),
	failure_handler=failure_handler,
	)
	)
	elif act[PE.type] == PE.SubProcess:
	sub_tree = act[PE.pipeline]
	params = act[PE.params]
	sub_parser = PipelineParser(pipeline_tree=sub_tree, cycle_tolerate=self.cycle_tolerate)
	act_objs.append(
	act_cls(
	id=act[PE.id],
	pipeline=sub_parser._parse(
	root_pipeline_data=root_pipeline_data,
	root_pipeline_params=root_pipeline_params,
	params=params,
	is_subprocess=True,
	parent_context=context,
	),
	name=act[PE.name],
	)
	)
	else:
	raise exceptions.FlowTypeError("Unknown Activity type: %s" % act[PE.type])

[dengyh]

bamboo-engine/runtime/bamboo-pipeline/pipeline/engine/core/handlers/__init__.py

Lines 24 to 51 in 6e5e357

	class HandlersFactory(object):
	_handlers = {
	EmptyStartEventHandler.element_cls(): EmptyStartEventHandler(),
	EmptyEndEventHandler.element_cls(): EmptyEndEventHandler(),
	ServiceActivityHandler.element_cls(): ServiceActivityHandler(),
	SubprocessHandler.element_cls(): SubprocessHandler(),
	ExclusiveGatewayHandler.element_cls(): ExclusiveGatewayHandler(),
	ParallelGatewayHandler.element_cls(): ParallelGatewayHandler(),
	ConditionalParallelGatewayHandler.element_cls(): ConditionalParallelGatewayHandler(),
	ConvergeGatewayHandler.element_cls(): ConvergeGatewayHandler(),
	ExecutableEndEventHandler.element_cls(): ExecutableEndEventHandler(),
	}
	_cluster_roots = [ExecutableEndEventHandler.element_cls()]
	@classmethod
	def find_cluster_root_cls(cls, element):
	for root in cls._cluster_roots:
	if issubclass(type(element), root):
	return root
	return type(element)
	@classmethod
	def handlers_for(cls, element):
	handler = cls._handlers.get(cls.find_cluster_root_cls(element))
	if not handler:
	raise KeyError("handler for element({element}) not found.".format(element=element))

这里没有把handler注册进去吧

[dengyh]

bamboo-engine/runtime/bamboo-pipeline/pipeline/models.py

Lines 233 to 270 in 6e5e357

	def unfold_subprocess(self, pipeline_data):
	"""
	展开 pipeline 数据中所有的子流程
	@param pipeline_data: pipeline 数据
	@return:
	"""
	id_maps = replace_all_id(pipeline_data)
	activities = pipeline_data[PE.activities]
	for act_id, act in list(activities.items()):
	if act[PE.type] == PE.SubProcess:
	subproc_data = self.get(template_id=act[PE.template_id]).data_for_version(act.get(PE.version))
	sub_id_maps = self.unfold_subprocess(subproc_data)
	# act_id is new id
	id_maps[PE.subprocess_detail].update({act_id: sub_id_maps})
	subproc_data[PE.id] = act_id
	act[PE.pipeline] = subproc_data
	return id_maps
	def replace_id(self, pipeline_data):
	"""
	替换 pipeline 中所有 ID
	@param pipeline_data: pipeline 数据
	@return:
	"""
	id_maps = replace_all_id(pipeline_data)
	activities = pipeline_data[PE.activities]
	for act_id, act in list(activities.items()):
	if act[PE.type] == PE.SubProcess:
	subproc_data = act[PE.pipeline]
	sub_id_maps = self.replace_id(subproc_data)
	# act_id is new id
	id_maps[PE.subprocess_detail].update({act_id: sub_id_maps})
	subproc_data[PE.id] = act_id
	act[PE.pipeline] = subproc_data
	return id_maps

在对流程进行unfold的时候，有一个替换节点id的操作，可能要看下是否要考虑子画布里的节点

[dengyh]

bamboo-engine/bamboo_engine/api.py

Lines 640 to 662 in 6e5e357

	if subprocess_stack:
	subprocess = subprocess_stack[0]
	child_pipeline = pipeline["activities"][subprocess]["pipeline"]
	param_data = {key: info["value"] for key, info in pipeline["activities"][subprocess]["params"].items()}
	hydrated_context = context.hydrate(deformat=True)
	hydrated_param_data = Template(param_data).render(hydrated_context)
	formatted_param_data = {key: {"value": value, "type": "plain"} for key, value in hydrated_param_data.items()}
	return preview_node_inputs(
	runtime=runtime,
	pipeline=child_pipeline,
	node_id=node_id,
	subprocess_stack=subprocess_stack[1:],
	root_pipeline_data=root_pipeline_data,
	parent_params=formatted_param_data,
	)
	node_type = pipeline["activities"][node_id]["type"]
	if node_type == NodeType.ServiceActivity.value:
	raw_inputs = pipeline["activities"][node_id]["component"]["inputs"]
	elif node_type == NodeType.SubProcess.value:
	raw_inputs = pipeline["activities"][node_id]["params"]
	else:
	raise InvalidOperationError(f"can not preview inputs for node type: {node_type}")

关于这个处理函数，得看下逻辑上是否兼容