package tech.powerjob.server.core.workflow;
|
|
import com.alibaba.fastjson.JSON;
|
import lombok.RequiredArgsConstructor;
|
import lombok.extern.slf4j.Slf4j;
|
import org.springframework.beans.BeanUtils;
|
import org.springframework.stereotype.Service;
|
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.response.WorkflowInstanceInfoDTO;
|
import tech.powerjob.common.enums.SwitchableStatus;
|
import tech.powerjob.server.common.utils.SpringUtils;
|
import tech.powerjob.server.core.instance.InstanceService;
|
import tech.powerjob.server.core.lock.UseCacheLock;
|
import tech.powerjob.server.core.workflow.algorithm.WorkflowDAGUtils;
|
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 tech.powerjob.server.remote.server.redirector.DesignateServer;
|
|
import java.util.Date;
|
import java.util.Objects;
|
import java.util.Optional;
|
|
/**
|
* 工作流实例服务
|
*
|
* @author tjq
|
* @author Echo009
|
* @since 2020/5/31
|
*/
|
@Slf4j
|
@Service
|
@RequiredArgsConstructor
|
public class WorkflowInstanceService {
|
|
private final InstanceService instanceService;
|
|
private final WorkflowInstanceInfoRepository wfInstanceInfoRepository;
|
|
private final WorkflowInstanceManager workflowInstanceManager;
|
|
private final WorkflowInfoRepository workflowInfoRepository;
|
|
/**
|
* 停止工作流实例（入口）
|
*
|
* @param wfInstanceId 工作流实例ID
|
* @param appId 所属应用ID
|
*/
|
public void stopWorkflowInstanceEntrance(Long wfInstanceId, Long appId) {
|
WorkflowInstanceInfoDO wfInstance = fetchWfInstance(wfInstanceId, appId);
|
if (!WorkflowInstanceStatus.GENERALIZED_RUNNING_STATUS.contains(wfInstance.getStatus())) {
|
throw new PowerJobException("workflow instance already stopped");
|
}
|
// 如果这是一个被嵌套的工作流，则终止父工作流
|
if (wfInstance.getParentWfInstanceId() != null) {
|
SpringUtils.getBean(this.getClass()).stopWorkflowInstance(wfInstance.getParentWfInstanceId(), appId);
|
return;
|
}
|
SpringUtils.getBean(this.getClass()).stopWorkflowInstance(wfInstanceId, appId);
|
}
|
|
/**
|
* 停止工作流实例
|
*
|
* @param wfInstanceId 工作流实例ID
|
* @param appId 所属应用ID
|
*/
|
@DesignateServer
|
@UseCacheLock(type = "processWfInstance", key = "#wfInstanceId", concurrencyLevel = 1024)
|
public void stopWorkflowInstance(Long wfInstanceId, Long appId) {
|
WorkflowInstanceInfoDO wfInstance = fetchWfInstance(wfInstanceId, appId);
|
if (!WorkflowInstanceStatus.GENERALIZED_RUNNING_STATUS.contains(wfInstance.getStatus())) {
|
throw new PowerJobException("workflow instance already stopped");
|
}
|
// 停止所有已启动且未完成的服务
|
PEWorkflowDAG dag = JSON.parseObject(wfInstance.getDag(), PEWorkflowDAG.class);
|
// 遍历所有节点，终止正在运行的
|
dag.getNodes().forEach(node -> {
|
try {
|
if (node.getInstanceId() != null && InstanceStatus.GENERALIZED_RUNNING_STATUS.contains(node.getStatus())) {
|
log.debug("[WfInstance-{}] instance({}) is running, try to stop it now.", wfInstanceId, node.getInstanceId());
|
node.setStatus(InstanceStatus.STOPPED.getV());
|
node.setResult(SystemInstanceResult.STOPPED_BY_USER);
|
// 特殊处理嵌套工作流节点
|
if (Objects.equals(node.getNodeType(), WorkflowNodeType.NESTED_WORKFLOW.getCode())) {
|
stopWorkflowInstance(node.getInstanceId(), appId);
|
//
|
} else {
|
// 注意，这里并不保证一定能终止正在运行的实例
|
instanceService.stopInstance(appId, node.getInstanceId());
|
}
|
}
|
} catch (Exception e) {
|
log.warn("[WfInstance-{}] stop instance({}) failed.", wfInstanceId, JSON.toJSONString(node), e);
|
}
|
});
|
|
// 修改数据库状态
|
wfInstance.setDag(JSON.toJSONString(dag));
|
wfInstance.setStatus(WorkflowInstanceStatus.STOPPED.getV());
|
wfInstance.setResult(SystemInstanceResult.STOPPED_BY_USER);
|
wfInstance.setGmtModified(new Date());
|
wfInstanceInfoRepository.saveAndFlush(wfInstance);
|
|
log.info("[WfInstance-{}] stop workflow instance successfully~", wfInstanceId);
|
}
|
|
/**
|
* Add by Echo009 on 2021/02/07
|
*
|
* @param wfInstanceId 工作流实例ID
|
* @param appId 应用ID
|
*/
|
@DesignateServer
|
@UseCacheLock(type = "processWfInstance", key = "#wfInstanceId", concurrencyLevel = 1024)
|
public void retryWorkflowInstance(Long wfInstanceId, Long appId) {
|
WorkflowInstanceInfoDO wfInstance = fetchWfInstance(wfInstanceId, appId);
|
// 仅允许重试 失败的工作流
|
if (WorkflowInstanceStatus.GENERALIZED_RUNNING_STATUS.contains(wfInstance.getStatus())) {
|
throw new PowerJobException("workflow instance is running");
|
}
|
if (wfInstance.getStatus() == WorkflowInstanceStatus.SUCCEED.getV()) {
|
throw new PowerJobException("workflow instance is already successful");
|
}
|
// 因为 DAG 非法 或者 因任务信息缺失 而失败的工作流实例无法重试
|
if (SystemInstanceResult.CAN_NOT_FIND_JOB.equals(wfInstance.getResult())) {
|
throw new PowerJobException("you can't retry the workflow instance which is missing job info!");
|
}
|
// 校验 DAG 信息
|
PEWorkflowDAG dag;
|
try {
|
dag = JSON.parseObject(wfInstance.getDag(), PEWorkflowDAG.class);
|
if (!WorkflowDAGUtils.valid(dag)) {
|
throw new PowerJobException(SystemInstanceResult.INVALID_DAG);
|
}
|
} catch (Exception e) {
|
throw new PowerJobException("you can't retry the workflow instance whose DAG is illegal!");
|
}
|
// 检查当前工作流信息
|
Optional<WorkflowInfoDO> workflowInfo = workflowInfoRepository.findById(wfInstance.getWorkflowId());
|
if (!workflowInfo.isPresent() || workflowInfo.get().getStatus() == SwitchableStatus.DISABLE.getV()) {
|
throw new PowerJobException("you can't retry the workflow instance whose metadata is unavailable!");
|
}
|
WorkflowDAGUtils.resetRetryableNode(dag);
|
wfInstance.setDag(JSON.toJSONString(dag));
|
// 更新工作流实例状态，不覆盖实际触发时间
|
wfInstance.setStatus(WorkflowInstanceStatus.WAITING.getV());
|
wfInstance.setGmtModified(new Date());
|
wfInstanceInfoRepository.saveAndFlush(wfInstance);
|
// 立即开始
|
workflowInstanceManager.start(workflowInfo.get(), wfInstanceId);
|
}
|
|
|
public WorkflowInstanceInfoDTO fetchWorkflowInstanceInfo(Long wfInstanceId, Long appId) {
|
WorkflowInstanceInfoDO wfInstance = fetchWfInstance(wfInstanceId, appId);
|
WorkflowInstanceInfoDTO dto = new WorkflowInstanceInfoDTO();
|
BeanUtils.copyProperties(wfInstance, dto);
|
return dto;
|
}
|
|
public WorkflowInstanceInfoDO fetchWfInstance(Long wfInstanceId, Long appId) {
|
WorkflowInstanceInfoDO wfInstance = wfInstanceInfoRepository.findByWfInstanceId(wfInstanceId).orElseThrow(() -> new IllegalArgumentException("can't find workflow instance by wfInstanceId: " + wfInstanceId));
|
if (!Objects.equals(appId, wfInstance.getAppId())) {
|
throw new PowerJobException("Permission Denied!");
|
}
|
return wfInstance;
|
}
|
|
/**
|
* Add by Echo009 on 2021/02/20
|
* 将节点标记成功
|
* 注意：这里仅能标记真正执行失败的且不允许跳过的节点
|
* 即处于 [失败且不允许跳过] 的节点
|
* 而且仅会操作工作流实例 DAG 中的节点信息（状态、result）
|
* 并不会改变对应任务实例中的任何信息
|
* <p>
|
* 还是加把锁保平安 ~
|
*
|
* @param wfInstanceId 工作流实例 ID
|
* @param nodeId 节点 ID
|
*/
|
@DesignateServer
|
@UseCacheLock(type = "processWfInstance", key = "#wfInstanceId", concurrencyLevel = 1024)
|
public void markNodeAsSuccess(Long appId, Long wfInstanceId, Long nodeId) {
|
|
WorkflowInstanceInfoDO wfInstance = fetchWfInstance(wfInstanceId, appId);
|
// 校验工作流实例状态，运行中的不允许处理
|
if (WorkflowInstanceStatus.GENERALIZED_RUNNING_STATUS.contains(wfInstance.getStatus())) {
|
throw new PowerJobException("you can't mark the node in a running workflow!");
|
}
|
// 这里一定能反序列化成功
|
PEWorkflowDAG dag = JSON.parseObject(wfInstance.getDag(), PEWorkflowDAG.class);
|
PEWorkflowDAG.Node targetNode = null;
|
for (PEWorkflowDAG.Node node : dag.getNodes()) {
|
if (node.getNodeId().equals(nodeId)) {
|
targetNode = node;
|
break;
|
}
|
}
|
if (targetNode == null) {
|
throw new PowerJobException("can't find the node in current DAG!");
|
}
|
boolean allowSkipWhenFailed = targetNode.getSkipWhenFailed() != null && targetNode.getSkipWhenFailed();
|
// 仅允许处理 执行失败的且不允许失败跳过的节点
|
if (targetNode.getInstanceId() != null
|
&& targetNode.getStatus() == InstanceStatus.FAILED.getV()
|
// 不允许失败跳过
|
&& !allowSkipWhenFailed) {
|
// 仅处理工作流实例中的节点信息
|
targetNode.setStatus(InstanceStatus.SUCCEED.getV())
|
.setResult(SystemInstanceResult.MARK_AS_SUCCESSFUL_NODE);
|
|
wfInstance.setDag(JSON.toJSONString(dag));
|
wfInstanceInfoRepository.saveAndFlush(wfInstance);
|
return;
|
}
|
// 其他情况均拒绝处理
|
throw new PowerJobException("you can only mark the node which is failed and not allow to skip!");
|
|
}
|
|
}
|
