first commit

WangHan

2024-09-12 d5855a4926926698b740bc6c7ba489de47adb68b

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
package tech.powerjob.server.core.workflow.hanlder.impl;
 
import com.alibaba.fastjson.JSON;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;
import tech.powerjob.common.SystemInstanceResult;
import tech.powerjob.common.enums.InstanceStatus;
import tech.powerjob.common.enums.WorkflowInstanceStatus;
import tech.powerjob.common.enums.WorkflowNodeType;
import tech.powerjob.common.exception.PowerJobException;
import tech.powerjob.common.model.PEWorkflowDAG;
import tech.powerjob.common.utils.CommonUtils;
import tech.powerjob.common.enums.SwitchableStatus;
import tech.powerjob.server.common.utils.SpringUtils;
import tech.powerjob.server.core.workflow.WorkflowInstanceManager;
import tech.powerjob.server.core.workflow.algorithm.WorkflowDAGUtils;
import tech.powerjob.server.core.workflow.hanlder.TaskNodeHandler;
import tech.powerjob.server.persistence.remote.model.WorkflowInfoDO;
import tech.powerjob.server.persistence.remote.model.WorkflowInstanceInfoDO;
import tech.powerjob.server.persistence.remote.repository.WorkflowInfoRepository;
import tech.powerjob.server.persistence.remote.repository.WorkflowInstanceInfoRepository;
 
import java.util.Date;
 
/**
 * @author Echo009
 * @since 2021/12/13
 */
@Component
@Slf4j
@RequiredArgsConstructor
public class NestedWorkflowNodeHandler implements TaskNodeHandler {
 
    private final WorkflowInfoRepository workflowInfoRepository;
 
    private final WorkflowInstanceInfoRepository workflowInstanceInfoRepository;
 
    @Override
    public void createTaskInstance(PEWorkflowDAG.Node node, PEWorkflowDAG dag, WorkflowInstanceInfoDO wfInstanceInfo) {
        // check
        Long wfId = node.getJobId();
        WorkflowInfoDO targetWf = workflowInfoRepository.findById(wfId).orElse(null);
        if (targetWf == null || targetWf.getStatus() == SwitchableStatus.DELETED.getV()) {
            if (targetWf == null) {
                log.error("[Workflow-{}|{}] invalid nested workflow node({}),target workflow({}) is not exist!", wfInstanceInfo.getWorkflowId(), wfInstanceInfo.getWfInstanceId(), node.getNodeId(), node.getJobId());
            } else {
                log.error("[Workflow-{}|{}] invalid nested workflow node({}),target workflow({}) has been deleted!", wfInstanceInfo.getWorkflowId(), wfInstanceInfo.getWfInstanceId(), node.getNodeId(), node.getJobId());
            }
            throw new PowerJobException("invalid nested workflow node," + node.getNodeId());
        }
        if (node.getInstanceId() != null) {
            // 处理重试的情形，不需要创建实例，仅需要更改对应实例的状态，以及相应的节点状态
            WorkflowInstanceInfoDO wfInstance = workflowInstanceInfoRepository.findByWfInstanceId(node.getInstanceId()).orElse(null);
            if (wfInstance == null) {
                log.error("[Workflow-{}|{}] invalid nested workflow node({}),target workflow instance({}) is not exist!", wfInstanceInfo.getWorkflowId(), wfInstanceInfo.getWfInstanceId(), node.getNodeId(), node.getInstanceId());
                throw new PowerJobException("invalid nested workflow instance id " + node.getInstanceId());
            }
            // 不用考虑状态，只有失败的工作流嵌套节点状态会被重置
            // 需要将子工作流中失败的节点状态重置为 等待 派发
            try {
                PEWorkflowDAG nodeDag = JSON.parseObject(wfInstance.getDag(), PEWorkflowDAG.class);
                if (!WorkflowDAGUtils.valid(nodeDag)) {
                    throw new PowerJobException(SystemInstanceResult.INVALID_DAG);
                }
                WorkflowDAGUtils.resetRetryableNode(nodeDag);
                wfInstance.setDag(JSON.toJSONString(nodeDag));
                wfInstance.setStatus(WorkflowInstanceStatus.WAITING.getV());
                wfInstance.setGmtModified(new Date());
                workflowInstanceInfoRepository.saveAndFlush(wfInstance);
            } catch (Exception e) {
                log.error("[Workflow-{}|{}] invalid nested workflow node({}),target workflow instance({})'s DAG is illegal!", wfInstanceInfo.getWorkflowId(), wfInstanceInfo.getWfInstanceId(), node.getNodeId(), node.getInstanceId(),e);
                throw new PowerJobException("illegal nested workflow instance, id : "+ node.getInstanceId());
            }
        } else {
            // 透传当前的上下文创建新的工作流实例
            String wfContext = wfInstanceInfo.getWfContext();
            Long instanceId = SpringUtils.getBean(WorkflowInstanceManager.class).create(targetWf, wfContext, System.currentTimeMillis(), wfInstanceInfo.getWfInstanceId());
            node.setInstanceId(instanceId);
        }
        node.setStartTime(CommonUtils.formatTime(System.currentTimeMillis()));
        node.setStatus(InstanceStatus.RUNNING.getV());
    }
 
    @Override
    public void startTaskInstance(PEWorkflowDAG.Node node) {
        Long wfId = node.getJobId();
        WorkflowInfoDO targetWf = workflowInfoRepository.findById(wfId).orElse(null);
        SpringUtils.getBean(WorkflowInstanceManager.class).start(targetWf, node.getInstanceId());
    }
 
    @Override
    public WorkflowNodeType matchingType() {
        return WorkflowNodeType.NESTED_WORKFLOW;
    }
}