test/DolphinScheduler
1
0
Fork 0
mirror of https://gitee.com/dolphinscheduler/DolphinScheduler.git synced 2024-11-30 19:27:38 +08:00
Code Issues Actions 2 Packages Projects Releases Wiki Activity

match frontend and backend for force success

Browse Source
This commit is contained in:
chengshiwen 2020-12-20 20:00:55 +08:00
parent 925d88dd5a
commit af720d3ef1
14 changed files with 50 additions and 39 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/controller/TaskInstanceController.java
View File

@ -130,28 +130,28 @@ public class TaskInstanceController extends BaseController {
} }
/** /**
* change one single task instance's state from FAILURE to FORCED_SUCCESS * change one task instance's state from FAILURE to FORCED_SUCCESS
* *
* @param loginUser login user * @param loginUser login user
* @param projectName project name * @param projectName project name
* @param taskInstanceId task instance id * @param taskInstanceId task instance id
* @return the result code and msg * @return the result code and msg
*/ */
@ApiOperation(value = "force-success", notes = "FORCE_SINGLE_TASK_SUCCESS") @ApiOperation(value = "force-success", notes = "FORCE_TASK_SUCCESS")
@ApiImplicitParams({ @ApiImplicitParams({
@ApiImplicitParam(name = "taskInstanceId", value = "TASK_INSTANCE_ID", required = true, dataType = "Int", example = "12") @ApiImplicitParam(name = "taskInstanceId", value = "TASK_INSTANCE_ID", required = true, dataType = "Int", example = "12")
}) })
@PostMapping(value = "/force-success") @PostMapping(value = "/force-success")
@ResponseStatus(HttpStatus.OK) @ResponseStatus(HttpStatus.OK)
@ApiException(FORCE_TASK_SUCCESS_ERROR) @ApiException(FORCE_TASK_SUCCESS_ERROR)
public Result<Object> forceSingleTaskSuccess(@ApiIgnore @RequestAttribute(value = Constants.SESSION_USER) User loginUser, public Result<Object> forceTaskSuccess(@ApiIgnore @RequestAttribute(value = Constants.SESSION_USER) User loginUser,
@ApiParam(name = "projectName", value = "PROJECT_NAME", required = true) @PathVariable String projectName, @ApiParam(name = "projectName", value = "PROJECT_NAME", required = true) @PathVariable String projectName,
@RequestParam(value = "taskInstanceId") Integer taskInstanceId) { @RequestParam(value = "taskInstanceId") Integer taskInstanceId) {
String userNameReplace = StringUtils.replaceNRTtoUnderline(loginUser.getUserName()); String userNameReplace = StringUtils.replaceNRTtoUnderline(loginUser.getUserName());
String projectNameReplace = StringUtils.replaceNRTtoUnderline(projectName); String projectNameReplace = StringUtils.replaceNRTtoUnderline(projectName);
logger.info("force task success, login user: {}, project:{}, task instance id:{}", logger.info("force task success, login user: {}, project:{}, task instance id:{}",
userNameReplace, projectNameReplace, taskInstanceId); userNameReplace, projectNameReplace, taskInstanceId);
Map<String, Object> result = taskInstanceService.forceSingleTaskSuccess(loginUser, projectName, taskInstanceId); Map<String, Object> result = taskInstanceService.forceTaskSuccess(loginUser, projectName, taskInstanceId);
return returnDataList(result); return returnDataList(result);
} }

7
dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/TaskInstanceService.java
View File

@ -146,14 +146,14 @@ public class TaskInstanceService extends BaseService {
} }
/** /**
* change one single task instance's state from failure to forced success * change one task instance's state from failure to forced success
* *
* @param loginUser login user * @param loginUser login user
* @param projectName project name * @param projectName project name
* @param taskInstanceId task instance id * @param taskInstanceId task instance id
* @return the result code and msg * @return the result code and msg
*/ */
public Map<String, Object> forceSingleTaskSuccess(User loginUser, String projectName, Integer taskInstanceId) { public Map<String, Object> forceTaskSuccess(User loginUser, String projectName, Integer taskInstanceId) {
Map<String, Object> result = new HashMap<>(5); Map<String, Object> result = new HashMap<>(5);
Project project = projectMapper.queryByName(projectName); Project project = projectMapper.queryByName(projectName);
@ -182,8 +182,7 @@ public class TaskInstanceService extends BaseService {
int changedNum = taskInstanceMapper.updateById(task); int changedNum = taskInstanceMapper.updateById(task);
if (changedNum > 0) { if (changedNum > 0) {
putMsg(result, Status.SUCCESS); putMsg(result, Status.SUCCESS);
} } else {
else {
putMsg(result, Status.FORCE_TASK_SUCCESS_ERROR); putMsg(result, Status.FORCE_TASK_SUCCESS_ERROR);
} }

4
dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/controller/TaskInstanceControllerTest.java
View File

@ -84,14 +84,14 @@ public class TaskInstanceControllerTest extends AbstractControllerTest {
} }
@Test @Test
public void testForceSingleTaskSuccess() throws Exception { public void testForceTaskSuccess() throws Exception {
MultiValueMap<String, String> paramsMap = new LinkedMultiValueMap<>(); MultiValueMap<String, String> paramsMap = new LinkedMultiValueMap<>();
paramsMap.add("taskInstanceId","104"); paramsMap.add("taskInstanceId","104");
Map<String, Object> mockResult = new HashMap<>(5); Map<String, Object> mockResult = new HashMap<>(5);
mockResult.put(Constants.STATUS, Status.SUCCESS); mockResult.put(Constants.STATUS, Status.SUCCESS);
mockResult.put(Constants.MSG, Status.SUCCESS.getMsg()); mockResult.put(Constants.MSG, Status.SUCCESS.getMsg());
when(taskInstanceService.forceSingleTaskSuccess(any(User.class), anyString(), anyInt())).thenReturn(mockResult); when(taskInstanceService.forceTaskSuccess(any(User.class), anyString(), anyInt())).thenReturn(mockResult);
MvcResult mvcResult = mockMvc.perform(post("/projects/{projectName}/task-instance/force-success","test") MvcResult mvcResult = mockMvc.perform(post("/projects/{projectName}/task-instance/force-success","test")
.header(SESSION_ID, sessionId) .header(SESSION_ID, sessionId)

12
dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/service/TaskInstanceServiceTest.java
View File

@ -216,7 +216,7 @@ public class TaskInstanceServiceTest {
} }
@Test @Test
public void forceSingleTaskSuccess() { public void forceTaskSuccess() {
User user = getAdminUser(); User user = getAdminUser();
String projectName = "test"; String projectName = "test";
Project project = getProject(projectName); Project project = getProject(projectName);
@ -231,31 +231,31 @@ public class TaskInstanceServiceTest {
Map<String, Object> mockFailure = new HashMap<>(5); Map<String, Object> mockFailure = new HashMap<>(5);
putMsg(mockFailure, Status.USER_NO_OPERATION_PROJECT_PERM, user.getUserName(), projectName); putMsg(mockFailure, Status.USER_NO_OPERATION_PROJECT_PERM, user.getUserName(), projectName);
when(projectService.checkProjectAndAuth(user, project, projectName)).thenReturn(mockFailure); when(projectService.checkProjectAndAuth(user, project, projectName)).thenReturn(mockFailure);
Map<String, Object> authFailRes = taskInstanceService.forceSingleTaskSuccess(user, projectName, taskId); Map<String, Object> authFailRes = taskInstanceService.forceTaskSuccess(user, projectName, taskId);
Assert.assertNotSame(Status.SUCCESS, authFailRes.get(Constants.STATUS)); Assert.assertNotSame(Status.SUCCESS, authFailRes.get(Constants.STATUS));
// test task not found // test task not found
when(projectService.checkProjectAndAuth(user, project, projectName)).thenReturn(mockSuccess); when(projectService.checkProjectAndAuth(user, project, projectName)).thenReturn(mockSuccess);
when(taskInstanceMapper.selectById(Mockito.anyInt())).thenReturn(null); when(taskInstanceMapper.selectById(Mockito.anyInt())).thenReturn(null);
Map<String, Object> taskNotFoundRes = taskInstanceService.forceSingleTaskSuccess(user, projectName, taskId); Map<String, Object> taskNotFoundRes = taskInstanceService.forceTaskSuccess(user, projectName, taskId);
Assert.assertEquals(Status.TASK_INSTANCE_NOT_FOUND, taskNotFoundRes.get(Constants.STATUS)); Assert.assertEquals(Status.TASK_INSTANCE_NOT_FOUND, taskNotFoundRes.get(Constants.STATUS));
// test task instance state error // test task instance state error
task.setState(ExecutionStatus.SUCCESS); task.setState(ExecutionStatus.SUCCESS);
when(taskInstanceMapper.selectById(1)).thenReturn(task); when(taskInstanceMapper.selectById(1)).thenReturn(task);
Map<String, Object> taskStateErrorRes = taskInstanceService.forceSingleTaskSuccess(user, projectName, taskId); Map<String, Object> taskStateErrorRes = taskInstanceService.forceTaskSuccess(user, projectName, taskId);
Assert.assertEquals(Status.TASK_INSTANCE_STATE_OPERATION_ERROR, taskStateErrorRes.get(Constants.STATUS)); Assert.assertEquals(Status.TASK_INSTANCE_STATE_OPERATION_ERROR, taskStateErrorRes.get(Constants.STATUS));
// test error // test error
task.setState(ExecutionStatus.FAILURE); task.setState(ExecutionStatus.FAILURE);
when(taskInstanceMapper.updateById(task)).thenReturn(0); when(taskInstanceMapper.updateById(task)).thenReturn(0);
Map<String, Object> errorRes = taskInstanceService.forceSingleTaskSuccess(user, projectName, taskId); Map<String, Object> errorRes = taskInstanceService.forceTaskSuccess(user, projectName, taskId);
Assert.assertEquals(Status.FORCE_TASK_SUCCESS_ERROR, errorRes.get(Constants.STATUS)); Assert.assertEquals(Status.FORCE_TASK_SUCCESS_ERROR, errorRes.get(Constants.STATUS));
// test success // test success
task.setState(ExecutionStatus.FAILURE); task.setState(ExecutionStatus.FAILURE);
when(taskInstanceMapper.updateById(task)).thenReturn(1); when(taskInstanceMapper.updateById(task)).thenReturn(1);
Map<String, Object> successRes = taskInstanceService.forceSingleTaskSuccess(user, projectName, taskId); Map<String, Object> successRes = taskInstanceService.forceTaskSuccess(user, projectName, taskId);
Assert.assertEquals(Status.SUCCESS, successRes.get(Constants.STATUS)); Assert.assertEquals(Status.SUCCESS, successRes.get(Constants.STATUS));
} }
} }

2
dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/utils/VarPoolUtils.java
View File

@ -63,7 +63,7 @@ public class VarPoolUtils {
* @throws ParseException ParseException * @throws ParseException ParseException
*/ */
public static void convertVarPoolToMap(Map<String, Object> propToValue, String varPool) throws ParseException { public static void convertVarPoolToMap(Map<String, Object> propToValue, String varPool) throws ParseException {
if (varPool == null || propToValue == null) { if (propToValue == null || StringUtils.isEmpty(varPool)) {
return; return;
} }
String[] splits = varPool.split("\\$VarPool\\$"); String[] splits = varPool.split("\\$VarPool\\$");

12
dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/mapper/TaskInstanceMapper.java
View File

@ -73,11 +73,11 @@ public interface TaskInstanceMapper extends BaseMapper<TaskInstance> {
@Param("endTime") Date endTime @Param("endTime") Date endTime
); );
List<Integer> queryTaskByPIdAndStatusAndType(@Param("processInstanceId") Integer processInstanceId, List<Integer> queryTaskByProcessIdAndStateAndType(@Param("processInstanceId") Integer processInstanceId,
@Param("states") int[] stateArray, @Param("states") int[] stateArray,
@Param("taskType") String taskType); @Param("taskType") String taskType);
List<Integer> queryTaskBySubProcessTaskIdAndStatusAndType(@Param("subProcessTaskId") Integer subProcessTaskId, List<Integer> queryTaskBySubProcessTaskIdAndStateAndType(@Param("subProcessTaskId") Integer subProcessTaskId,
@Param("states") int[] stateArray, @Param("states") int[] stateArray,
@Param("taskType") String taskType); @Param("taskType") String taskType);
} }

4
dolphinscheduler-dao/src/main/resources/org/apache/dolphinscheduler/dao/mapper/TaskInstanceMapper.xml
View File

@ -152,7 +152,7 @@
</if> </if>
order by instance.start_time desc order by instance.start_time desc
</select> </select>
<select id="queryTaskByPIdAndStatusAndType" resultType="java.lang.Integer"> <select id="queryTaskByProcessIdAndStateAndType" resultType="java.lang.Integer">
select id from t_ds_task_instance select id from t_ds_task_instance
where process_instance_id = #{processInstanceId} where process_instance_id = #{processInstanceId}
and task_type = #{taskType} and task_type = #{taskType}
@ -162,7 +162,7 @@
</foreach> </foreach>
and flag = 1 and flag = 1
</select> </select>
<select id="queryTaskBySubProcessTaskIdAndStatusAndType" resultType="java.lang.Integer"> <select id="queryTaskBySubProcessTaskIdAndStateAndType" resultType="java.lang.Integer">
select id from t_ds_task_instance select id from t_ds_task_instance
where process_instance_id = where process_instance_id =
(select process_instance_id from t_ds_relation_process_instance where parent_task_instance_id = #{subProcessTaskId}) (select process_instance_id from t_ds_relation_process_instance where parent_task_instance_id = #{subProcessTaskId})

8
dolphinscheduler-dao/src/test/java/org/apache/dolphinscheduler/dao/mapper/TaskInstanceMapperTest.java
View File

@ -320,7 +320,7 @@ public class TaskInstanceMapperTest {
} }
@Test @Test
public void testQueryTaskByPIdAndStatusAndType() { public void testQueryTaskByProcessIdAndStateAndType() {
// insert three task instances with the same process instance id // insert three task instances with the same process instance id
List<TaskInstance> taskList = new ArrayList<>(); List<TaskInstance> taskList = new ArrayList<>();
for (int i = 0; i < 3; i++) { for (int i = 0; i < 3; i++) {
@ -329,7 +329,7 @@ public class TaskInstanceMapperTest {
} }
// test query result // test query result
List<Integer> resultArray = taskInstanceMapper.queryTaskByPIdAndStatusAndType(66, List<Integer> resultArray = taskInstanceMapper.queryTaskByProcessIdAndStateAndType(66,
new int[] {ExecutionStatus.FAILURE.ordinal(), ExecutionStatus.KILL.ordinal(), ExecutionStatus.NEED_FAULT_TOLERANCE.ordinal()}, new int[] {ExecutionStatus.FAILURE.ordinal(), ExecutionStatus.KILL.ordinal(), ExecutionStatus.NEED_FAULT_TOLERANCE.ordinal()},
TaskType.SUB_PROCESS.toString()); TaskType.SUB_PROCESS.toString());
Assert.assertEquals(3, resultArray.size()); Assert.assertEquals(3, resultArray.size());
@ -341,7 +341,7 @@ public class TaskInstanceMapperTest {
} }
@Test @Test
public void testQueryTaskBySubProcessTaskIdAndStatusAndType() { public void testQueryTaskBySubProcessTaskIdAndStateAndType() {
TaskInstance parentTask = insertOne("parent-task", 66, ExecutionStatus.FAILURE, TaskType.SUB_PROCESS.toString()); TaskInstance parentTask = insertOne("parent-task", 66, ExecutionStatus.FAILURE, TaskType.SUB_PROCESS.toString());
ProcessInstanceMap processInstanceMap = new ProcessInstanceMap(); ProcessInstanceMap processInstanceMap = new ProcessInstanceMap();
@ -354,7 +354,7 @@ public class TaskInstanceMapperTest {
TaskInstance subTask2 = insertOne("sub2", 67, ExecutionStatus.FORCED_SUCCESS, TaskType.SHELL.toString()); TaskInstance subTask2 = insertOne("sub2", 67, ExecutionStatus.FORCED_SUCCESS, TaskType.SHELL.toString());
// test query result // test query result
List<Integer> resultList = taskInstanceMapper.queryTaskBySubProcessTaskIdAndStatusAndType(parentTask.getId(), List<Integer> resultList = taskInstanceMapper.queryTaskBySubProcessTaskIdAndStateAndType(parentTask.getId(),
new int[] {ExecutionStatus.FORCED_SUCCESS.ordinal()}, new int[] {ExecutionStatus.FORCED_SUCCESS.ordinal()},
null); null);

6
dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/runner/MasterExecThread.java
View File

@ -1115,9 +1115,9 @@ public class MasterExecThread implements Runnable {
TaskInstance task = readyToSubmitTaskQueue.peek(); TaskInstance task = readyToSubmitTaskQueue.peek();
// stop tasks which is retrying if forced success happens // stop tasks which is retrying if forced success happens
if (task.taskCanRetry()) { if (task.taskCanRetry()) {
TaskInstance tmpTask = processService.findTaskInstanceById(task.getId()); TaskInstance retryTask = processService.findTaskInstanceById(task.getId());
if (tmpTask != null && tmpTask.getState().equals(ExecutionStatus.FORCED_SUCCESS)) { if (retryTask != null && retryTask.getState().equals(ExecutionStatus.FORCED_SUCCESS)) {
task.setState(tmpTask.getState()); task.setState(retryTask.getState());
logger.info("task: {} has been forced success, put it into complete task list and stop retrying", task.getName()); logger.info("task: {} has been forced success, put it into complete task list and stop retrying", task.getName());
removeTaskFromStandbyList(task); removeTaskFromStandbyList(task);
completeTaskList.put(task.getName(), task); completeTaskList.put(task.getName(), task);

6
dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/process/ProcessService.java
View File

@ -1426,7 +1426,7 @@ public class ProcessService {
for (int i = 0; i < states.length; i++) { for (int i = 0; i < states.length; i++) {
statesArray[i] = states[i].ordinal(); statesArray[i] = states[i].ordinal();
} }
return taskInstanceMapper.queryTaskByPIdAndStatusAndType(processInstanceId, statesArray, taskType.toString()); return taskInstanceMapper.queryTaskByProcessIdAndStateAndType(processInstanceId, statesArray, taskType.toString());
} }
/** /**
@ -1443,9 +1443,9 @@ public class ProcessService {
statesArray[i] = states[i].ordinal(); statesArray[i] = states[i].ordinal();
} }
if (taskType == null) { if (taskType == null) {
return taskInstanceMapper.queryTaskBySubProcessTaskIdAndStatusAndType(taskId, statesArray, null); return taskInstanceMapper.queryTaskBySubProcessTaskIdAndStateAndType(taskId, statesArray, null);
} }
return taskInstanceMapper.queryTaskBySubProcessTaskIdAndStatusAndType(taskId, statesArray, taskType.toString()); return taskInstanceMapper.queryTaskBySubProcessTaskIdAndStateAndType(taskId, statesArray, taskType.toString());
} }
/** /**

4
dolphinscheduler-ui/src/js/conf/home/pages/dag/_source/config.js
View File

@ -133,6 +133,10 @@ const runningType = [
{ {
desc: `${i18n.$t('Recovery waiting thread')}`, desc: `${i18n.$t('Recovery waiting thread')}`,
code: 'RECOVER_WAITTING_THREAD' code: 'RECOVER_WAITTING_THREAD'
},
{
desc: `${i18n.$t('Resume from forced success')}`,
code: 'RESUME_FROM_FORCED_SUCCESS'
} }
] ]

2
dolphinscheduler-ui/src/js/conf/home/pages/projects/pages/instance/pages/list/_source/list.vue
View File

@ -79,7 +79,7 @@
</el-tooltip> </el-tooltip>
<el-tooltip :content="$t('Resume From Forced Success')" placement="top" :enterable="false"> <el-tooltip :content="$t('Resume From Forced Success')" placement="top" :enterable="false">
<span> <span>
<el-button type="success" size="mini" icon="el-icon-stopwatch" :disabled="scope.row.state !== 'FORCED_SUCCESS'" @click="_resumeFromForcedSuccess(scope.row,scope.$index)" circle></el-button> <el-button type="success" size="mini" icon="el-icon-stopwatch" :disabled="!(scope.row.state === 'FAILURE' || scope.row.state === 'NEED_FAULT_TOLERANCE' || scope.row.state === 'KILL')" @click="_resumeFromForcedSuccess(scope.row,scope.$index)" circle></el-button>
</span> </span>
</el-tooltip> </el-tooltip>
<el-tooltip :content="scope.row.state === 'STOP' ? $t('Recovery Suspend') : $t('Stop')" placement="top" :enterable="false"> <el-tooltip :content="scope.row.state === 'STOP' ? $t('Recovery Suspend') : $t('Stop')" placement="top" :enterable="false">

6
dolphinscheduler-ui/src/js/conf/home/pages/projects/pages/taskInstance/_source/list.vue
View File

@ -116,15 +116,14 @@
this.logDialog = true this.logDialog = true
}, },
ok () {}, ok () {},
close () { close () {
this.logDialog = false this.logDialog = false
}, },
_forceSuccess (item) { _forceSuccess (item) {
this.forceTaskSuccess({ taskInstanceId: item.id }).then(res => { this.forceTaskSuccess({ taskInstanceId: item.id }).then(res => {
if (res.code === 0) { if (res.code === 0) {
this.$message.success(res.msg) this.$message.success(res.msg)
setTimeout(this._onUpdate, 1000)
} else { } else {
this.$message.error(res.msg) this.$message.error(res.msg)
} }
@ -132,6 +131,9 @@
this.$message.error(e.msg) this.$message.error(e.msg)
}) })
}, },
_onUpdate () {
this.$emit('on-update')
},
_go (item) { _go (item) {
this.$router.push({ path: `/projects/instance/list/${item.processInstanceId}` }) this.$router.push({ path: `/projects/instance/list/${item.processInstanceId}` })
} }

8
dolphinscheduler-ui/src/js/conf/home/pages/projects/pages/taskInstance/index.vue
View File

@ -23,7 +23,7 @@
<template slot="content"> <template slot="content">
<template v-if="taskInstanceList.length"> <template v-if="taskInstanceList.length">
<m-list :task-instance-list="taskInstanceList" :page-no="searchParams.pageNo" :page-size="searchParams.pageSize"> <m-list :task-instance-list="taskInstanceList" @on-update="_onUpdate" :page-no="searchParams.pageNo" :page-size="searchParams.pageSize">
</m-list> </m-list>
<div class="page-box"> <div class="page-box">
<el-pagination <el-pagination
@ -126,6 +126,12 @@
this.isLoading = false this.isLoading = false
}) })
}, },
/**
* update
*/
_onUpdate () {
this._debounceGET()
},
/** /**
* Anti shake request interface * Anti shake request interface
* @desc Prevent functions from being called multiple times * @desc Prevent functions from being called multiple times