diff --git a/escheduler-api/src/main/java/cn/escheduler/api/controller/ProcessInstanceController.java b/escheduler-api/src/main/java/cn/escheduler/api/controller/ProcessInstanceController.java index b05aa656f13f7f926b202279821cc7b82f9369f8..cea47be2c6b29060b2741fe911602207502f65ea 100644 --- a/escheduler-api/src/main/java/cn/escheduler/api/controller/ProcessInstanceController.java +++ b/escheduler-api/src/main/java/cn/escheduler/api/controller/ProcessInstanceController.java @@ -22,6 +22,8 @@ import cn.escheduler.api.utils.Constants; import cn.escheduler.api.utils.Result; import cn.escheduler.common.enums.ExecutionStatus; import cn.escheduler.common.enums.Flag; +import cn.escheduler.common.queue.ITaskQueue; +import cn.escheduler.common.queue.TaskQueueFactory; import cn.escheduler.common.utils.ParameterUtils; import cn.escheduler.dao.model.User; import org.slf4j.Logger; @@ -189,7 +191,9 @@ public class ProcessInstanceController extends BaseController{ try{ logger.info("delete process instance by id, login user:{}, project name:{}, process instance id:{}", loginUser.getUserName(), projectName, processInstanceId); - Map result = processInstanceService.deleteProcessInstanceById(loginUser, projectName, processInstanceId); + // task queue + ITaskQueue tasksQueue = TaskQueueFactory.getTaskQueueInstance(); + Map result = processInstanceService.deleteProcessInstanceById(loginUser, projectName, processInstanceId,tasksQueue); return returnDataList(result); }catch (Exception e){ logger.error(DELETE_PROCESS_INSTANCE_BY_ID_ERROR.getMsg(),e); diff --git a/escheduler-api/src/main/java/cn/escheduler/api/service/ProcessInstanceService.java b/escheduler-api/src/main/java/cn/escheduler/api/service/ProcessInstanceService.java index 70abaf07eb24fe44be3644b601e4db4f05dd72c5..6475b42cee4121ab7812bc4ef70f6fa6ffd2fb71 100644 --- a/escheduler-api/src/main/java/cn/escheduler/api/service/ProcessInstanceService.java +++ b/escheduler-api/src/main/java/cn/escheduler/api/service/ProcessInstanceService.java @@ -30,6 +30,8 @@ import cn.escheduler.common.graph.DAG; import cn.escheduler.common.model.TaskNode; import cn.escheduler.common.model.TaskNodeRelation; import cn.escheduler.common.process.Property; +import cn.escheduler.common.queue.ITaskQueue; +import cn.escheduler.common.queue.TaskQueueFactory; import cn.escheduler.common.utils.CollectionUtils; import cn.escheduler.common.utils.DateUtils; import cn.escheduler.common.utils.JSONUtils; @@ -446,13 +448,13 @@ public class ProcessInstanceService extends BaseDAGService { /** * delete process instance by id, at the same time,delete task instance and their mapping relation data - * * @param loginUser * @param projectName - * @param workflowId + * @param processInstanceId + * @param tasksQueue * @return */ - public Map deleteProcessInstanceById(User loginUser, String projectName, Integer workflowId) { + public Map deleteProcessInstanceById(User loginUser, String projectName, Integer processInstanceId,ITaskQueue tasksQueue) { Map result = new HashMap<>(5); Project project = projectMapper.queryByName(projectName); @@ -462,17 +464,34 @@ public class ProcessInstanceService extends BaseDAGService { if (resultEnum != Status.SUCCESS) { return checkResult; } - ProcessInstance processInstance = processDao.findProcessInstanceDetailById(workflowId); + ProcessInstance processInstance = processDao.findProcessInstanceDetailById(processInstanceId); + List taskInstanceList = processDao.findValidTaskListByProcessId(processInstanceId); + //process instance priority + int processInstancePriority = processInstance.getProcessInstancePriority().ordinal(); if (processInstance == null) { - putMsg(result, Status.PROCESS_INSTANCE_NOT_EXIST, workflowId); + putMsg(result, Status.PROCESS_INSTANCE_NOT_EXIST, processInstanceId); return result; } - int delete = processDao.deleteWorkProcessInstanceById(workflowId); - processDao.deleteAllSubWorkProcessByParentId(workflowId); - processDao.deleteWorkProcessMapByParentId(workflowId); + int delete = processDao.deleteWorkProcessInstanceById(processInstanceId); + processDao.deleteAllSubWorkProcessByParentId(processInstanceId); + processDao.deleteWorkProcessMapByParentId(processInstanceId); if (delete > 0) { + if (CollectionUtils.isNotEmpty(taskInstanceList)){ + for (TaskInstance taskInstance : taskInstanceList){ + // task instance priority + int taskInstancePriority = taskInstance.getTaskInstancePriority().ordinal(); + String nodeValue=processInstancePriority + "_" + processInstanceId + "_" +taskInstancePriority + "_" + taskInstance.getId(); + try { + logger.info("delete task queue node : {}",nodeValue); + tasksQueue.removeNode(cn.escheduler.common.Constants.SCHEDULER_TASKS_QUEUE, nodeValue); + }catch (Exception e){ + logger.error("delete task queue node : {}", nodeValue); + } + } + } + putMsg(result, Status.SUCCESS); } else { putMsg(result, Status.DELETE_PROCESS_INSTANCE_BY_ID_ERROR); @@ -489,6 +508,8 @@ public class ProcessInstanceService extends BaseDAGService { * @return */ public Map batchDeleteProcessInstanceByIds(User loginUser, String projectName, String processInstanceIds) { + // task queue + ITaskQueue tasksQueue = TaskQueueFactory.getTaskQueueInstance(); Map result = new HashMap<>(5); List deleteFailedIdList = new ArrayList(); @@ -507,7 +528,7 @@ public class ProcessInstanceService extends BaseDAGService { for (String strProcessInstanceId:processInstanceIdArray) { int processInstanceId = Integer.parseInt(strProcessInstanceId); try { - deleteProcessInstanceById(loginUser, projectName, processInstanceId); + deleteProcessInstanceById(loginUser, projectName, processInstanceId,tasksQueue); } catch (Exception e) { deleteFailedIdList.add(processInstanceId); } diff --git a/escheduler-dao/src/main/java/cn/escheduler/dao/ProcessDao.java b/escheduler-dao/src/main/java/cn/escheduler/dao/ProcessDao.java index b18fb5e974963228e18d17b0bd678e9efadfa930..1c7fc969aee4daa2db0049af08fb0df121afcd53 100644 --- a/escheduler-dao/src/main/java/cn/escheduler/dao/ProcessDao.java +++ b/escheduler-dao/src/main/java/cn/escheduler/dao/ProcessDao.java @@ -1526,6 +1526,14 @@ public class ProcessDao extends AbstractBaseDao { } + public void selfFaultTolerant(int ... states){ + List processInstanceList = processInstanceMapper.listByStatus(states); + for (ProcessInstance processInstance:processInstanceList){ + selfFaultTolerant(processInstance); + } + + } + @Transactional(value = "TransactionManager",rollbackFor = Exception.class) public void selfFaultTolerant(ProcessInstance processInstance){ diff --git a/escheduler-server/src/main/java/cn/escheduler/server/utils/AlertManager.java b/escheduler-server/src/main/java/cn/escheduler/server/utils/AlertManager.java index fc62bcf73d4181f7581c984b3d0a5bd28f52c458..147f538bcaf80de05bbbe0e9b0392169a81cc2bb 100644 --- a/escheduler-server/src/main/java/cn/escheduler/server/utils/AlertManager.java +++ b/escheduler-server/src/main/java/cn/escheduler/server/utils/AlertManager.java @@ -175,7 +175,7 @@ public class AlertManager { alert.setContent(content); alert.setAlertType(AlertType.EMAIL); alert.setCreateTime(new Date()); - alert.setAlertGroupId(processInstance.getWarningGroupId()); + alert.setAlertGroupId(processInstance.getWarningGroupId() == null ? 1:processInstance.getWarningGroupId()); alert.setReceivers(processInstance.getProcessDefinition().getReceivers()); alert.setReceiversCc(processInstance.getProcessDefinition().getReceiversCc()); diff --git a/escheduler-server/src/main/java/cn/escheduler/server/zk/ZKMasterClient.java b/escheduler-server/src/main/java/cn/escheduler/server/zk/ZKMasterClient.java index 624d0193be88aa101e1561d694ae5c53bbdd116f..fe3360484c47aa0717522c1ebdf568474d6480b9 100644 --- a/escheduler-server/src/main/java/cn/escheduler/server/zk/ZKMasterClient.java +++ b/escheduler-server/src/main/java/cn/escheduler/server/zk/ZKMasterClient.java @@ -123,9 +123,9 @@ public class ZKMasterClient extends AbstractZKClient { // register master this.registMaster(); - // check if fault tolerance is required + // check if fault tolerance is required,failure and tolerance if (getActiveMasterNum() == 1) { - processDao.selfFaultTolerant(ExecutionStatus.RUNNING_EXEUTION.ordinal()); + processDao.selfFaultTolerant(ExecutionStatus.RUNNING_EXEUTION.ordinal(),ExecutionStatus.NEED_FAULT_TOLERANCE.ordinal()); } } diff --git a/escheduler-ui/.env b/escheduler-ui/.env index 5cd5ba736da46246570a0b8fa2e32f48eeddab3e..fc1b619f694fb0e881d2f888b7095a6f01bfc7be 100644 --- a/escheduler-ui/.env +++ b/escheduler-ui/.env @@ -1,5 +1,5 @@ -# 后端接口地址 +# 后端接口地址11 API_BASE = http://192.168.xx.xx:12345 # 本地开发如需ip访问项目把"#"号去掉 diff --git a/escheduler-ui/src/js/conf/home/store/dag/actions.js b/escheduler-ui/src/js/conf/home/store/dag/actions.js index 0872249d9c9db20c4b50029797400b88e3b3ec49..929c4f6df57c8dfcd16af8099644857d3336f6b6 100644 --- a/escheduler-ui/src/js/conf/home/store/dag/actions.js +++ b/escheduler-ui/src/js/conf/home/store/dag/actions.js @@ -538,7 +538,7 @@ export default { */ getReceiver ({ state }, payload) { return new Promise((resolve, reject) => { - io.get(`projects/{projectName}/executors/get-receiver-cc`, payload, res => { + io.get(`projects/${state.projectName}/executors/get-receiver-cc`, payload, res => { resolve(res.data) }).catch(e => { reject(e) @@ -547,7 +547,7 @@ export default { }, getTaskListDefIdAll ({ state }, payload) { return new Promise((resolve, reject) => { - io.get(`projects/{projectName}/process/get-task-list`, payload, res => { + io.get(`projects/${state.projectName}/process/get-task-list`, payload, res => { resolve(res.data) }).catch(e => { reject(e)