diff --git a/docs/docs/en/guide/project/task-definition.md b/docs/docs/en/guide/project/task-definition.md index c30948987a..a08d3a3da8 100644 --- a/docs/docs/en/guide/project/task-definition.md +++ b/docs/docs/en/guide/project/task-definition.md @@ -1,13 +1,23 @@ # Task Definition +## Batch Task Definition Task definition allows to modify or operate tasks at the task level rather than modifying them in the workflow definition. We already have workflow level task editor in [workflow definition](workflow-definition.md) which you can click the specific workflow and then edit its task definition. It is depressing when you want to edit the task definition but do not remember which workflow it belongs to. So we decide to add `Task Definition` view under `Task` menu. -![task-definition](../../../../img/new_ui/dev/project/task-definition.jpg) +![task-definition](../../../../img/new_ui/dev/project/batch-task-definition.png) In this view, you can create, query, update, delete task definition by click the related button in `operation` column. The most exciting thing is you could query task by task name in the wildcard, and it is useful when you only remember the task name but forget which workflow it belongs to. It is also supported query by the task name alone with `Task Type` or `Workflow Name` + +## Stream Task Definition +Stream task definitions are created in the workflow definition, and can be modified and executed. + +![task-definition](../../../../img/new_ui/dev/project/stream-task-definition.png) + +Click the execute button, check the execution parameters and click Confirm to submit the stream task. + +![task-definition](../../../../img/new_ui/dev/project/stream-task-execute.png) \ No newline at end of file diff --git a/docs/docs/en/guide/project/task-instance.md b/docs/docs/en/guide/project/task-instance.md index 2d7651732c..0371ed9d79 100644 --- a/docs/docs/en/guide/project/task-instance.md +++ b/docs/docs/en/guide/project/task-instance.md @@ -1,13 +1,23 @@ # Task Instance -## Create Task Instance +## Batch Task Instance +### Create Task Instance Click `Project Management -> Workflow -> Task Instance` to enter the task instance page, as shown in the figure below, click the name of the workflow instance to jump to the DAG diagram of the workflow instance to view the task status. -![task-instance](../../../../img/new_ui/dev/project/task-instance.png) +![task-instance](../../../../img/new_ui/dev/project/batch-task-instance.png) -## View Log +### View Log Click the `View Log` button in the operation column to view the log of the task execution ![task-log](../../../../img/new_ui/dev/project/task-log.png) + +## Stream Task Instance + +-Switch to the stream task instance page, as shown in the following figure: + +![task-instance](../../../../img/new_ui/dev/project/stream-task-instance.png) + +- SavePoint: Click the `SavePoint` button in the operation column to do stream task savepoint. +- Stop: Click the `Stop` button in the operation column to stop the stream task. diff --git a/docs/docs/en/guide/project/workflow-definition.md b/docs/docs/en/guide/project/workflow-definition.md index 443cedbed7..fe8e86dd3c 100644 --- a/docs/docs/en/guide/project/workflow-definition.md +++ b/docs/docs/en/guide/project/workflow-definition.md @@ -34,6 +34,11 @@ Click the plus sign on the right of the task node to connect the task; as shown ![workflow-dependent](../../../../img/new_ui/dev/project/workflow-dependent.png) +### Dependencies with stream task +If the DAG contains stream tasks, the relationship between stream tasks is displayed as a dotted line, and the execution of stream tasks will be skipped when the workflow instance is executed. + + ![workflow-dependent](../../../../img/new_ui/dev/project/workflow-definition-with-stream-task.png) + **Delete dependencies:** Using your mouse to select the connection line, and click the "Delete" icon in the upper right corner , delete dependencies between tasks. ![workflow-delete](../../../../img/new_ui/dev/project/workflow-delete.png) diff --git a/docs/docs/zh/guide/project/task-definition.md b/docs/docs/zh/guide/project/task-definition.md index 12b775ec49..6b1ffe55f7 100644 --- a/docs/docs/zh/guide/project/task-definition.md +++ b/docs/docs/zh/guide/project/task-definition.md @@ -1,9 +1,22 @@ # 任务定义 -任务定义允许您在基于任务级别而不是在工作流中操作修改任务。再此之前，我们已经有了工作流级别的任务编辑器，你可以在[工作流定义](workflow-definition.md) +## 批量任务定义 +批量任务定义允许您在基于任务级别而不是在工作流中操作修改任务。再此之前，我们已经有了工作流级别的任务编辑器，你可以在[工作流定义](workflow-definition.md) 单击特定的工作流，然后编辑任务的定义。当您想编辑特定的任务定义但不记得它属于哪个工作流时，这是令人沮丧的。所以我们决定在 `任务` 菜单下添加 `任务定义` 视图。 -![task-definition](../../../../img/new_ui/dev/project/task-definition.jpg) +![task-definition](../../../../img/new_ui/dev/project/batch-task-definition.png) 在该视图中，您可以通过单击 `操作` 列中的相关按钮来进行创建、查询、更新、删除任务定义。最令人兴奋的是您可以通过通配符进行全部任务查询，当您只 记得任务名称但忘记它属于哪个工作流时是非常有用的。也支持通过任务名称结合使用 `任务类型` 或 `工作流程名称` 进行查询。 + + +## 实时任务定义 +实时任务定义在工作流定义中创建，在任务定义页面可以进行修改和执行。 + +![task-definition](../../../../img/new_ui/dev/project/stream-task-definition.png) + +点击实时任务执行，检查执行参数后点击确认，即可提交实时任务。 + +![task-definition](../../../../img/new_ui/dev/project/stream-task-execute.png) + + diff --git a/docs/docs/zh/guide/project/task-instance.md b/docs/docs/zh/guide/project/task-instance.md index 205086d02a..b7a928b7e2 100644 --- a/docs/docs/zh/guide/project/task-instance.md +++ b/docs/docs/zh/guide/project/task-instance.md @@ -1,9 +1,19 @@ # 任务实例 +## 批量任务实例 - 点击项目管理->工作流->任务实例，进入任务实例页面，如下图所示，点击工作流实例名称，可跳转到工作流实例DAG图查看任务状态。 -![task-instance](../../../../img/new_ui/dev/project/task-instance.png) +![task-instance](../../../../img/new_ui/dev/project/batch-task-instance.png) - 查看日志：点击操作列中的“查看日志”按钮，可以查看任务执行的日志情况。 ![task-log](../../../../img/new_ui/dev/project/task-log.png) + +## 实时任务实例 + +- 切换到实时任务实例页面，如下图所示： + +![task-instance](../../../../img/new_ui/dev/project/stream-task-instance.png) + +- SavePoint：点击操作列中的SavePoint按钮，可以进行实时任务的SavePoint。 +- Stop：点击操作列中的Stop按钮，可以停止该实时任务。 diff --git a/docs/docs/zh/guide/project/workflow-definition.md b/docs/docs/zh/guide/project/workflow-definition.md index 65fbbdebf9..63c8e79d4b 100644 --- a/docs/docs/zh/guide/project/workflow-definition.md +++ b/docs/docs/zh/guide/project/workflow-definition.md @@ -24,6 +24,10 @@ ![workflow-dependent](../../../../img/new_ui/dev/project/workflow-dependent.png) +- **实时任务的依赖关系：** 若DAG中包含了实时任务的组件，则实时任务的关联关系显示为虚线，在执行工作流实例的时候会跳过实时任务的执行 + + ![workflow-dependent](../../../../img/new_ui/dev/project/workflow-definition-with-stream-task.png) + - **删除依赖关系：** 点击右上角"箭头"图标，选中连接线，点击右上角"删除"图标，删除任务间的依赖关系。 ![workflow-delete](../../../../img/new_ui/dev/project/workflow-delete.png) diff --git a/docs/img/new_ui/dev/project/batch-task-definition.png b/docs/img/new_ui/dev/project/batch-task-definition.png new file mode 100644 index 0000000000..ca1608477e Binary files /dev/null and b/docs/img/new_ui/dev/project/batch-task-definition.png differ diff --git a/docs/img/new_ui/dev/project/batch-task-instance.png b/docs/img/new_ui/dev/project/batch-task-instance.png new file mode 100644 index 0000000000..cdbbe3d38d Binary files /dev/null and b/docs/img/new_ui/dev/project/batch-task-instance.png differ diff --git a/docs/img/new_ui/dev/project/stream-task-definition.png b/docs/img/new_ui/dev/project/stream-task-definition.png new file mode 100644 index 0000000000..b6c0629a45 Binary files /dev/null and b/docs/img/new_ui/dev/project/stream-task-definition.png differ diff --git a/docs/img/new_ui/dev/project/stream-task-execute.png b/docs/img/new_ui/dev/project/stream-task-execute.png new file mode 100644 index 0000000000..2b49cd8df7 Binary files /dev/null and b/docs/img/new_ui/dev/project/stream-task-execute.png differ diff --git a/docs/img/new_ui/dev/project/stream-task-instance.png b/docs/img/new_ui/dev/project/stream-task-instance.png new file mode 100644 index 0000000000..4b3693d209 Binary files /dev/null and b/docs/img/new_ui/dev/project/stream-task-instance.png differ diff --git a/docs/img/new_ui/dev/project/task-definition.jpg b/docs/img/new_ui/dev/project/task-definition.jpg deleted file mode 100644 index cd2db4e8ce..0000000000 Binary files a/docs/img/new_ui/dev/project/task-definition.jpg and /dev/null differ diff --git a/docs/img/new_ui/dev/project/task-instance.png b/docs/img/new_ui/dev/project/task-instance.png deleted file mode 100644 index 4026736573..0000000000 Binary files a/docs/img/new_ui/dev/project/task-instance.png and /dev/null differ diff --git a/docs/img/new_ui/dev/project/workflow-definition-with-stream-task.png b/docs/img/new_ui/dev/project/workflow-definition-with-stream-task.png new file mode 100644 index 0000000000..298df8b4ac Binary files /dev/null and b/docs/img/new_ui/dev/project/workflow-definition-with-stream-task.png differ diff --git a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/controller/ExecutorController.java b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/controller/ExecutorController.java index aa48c67448..dfe9396089 100644 --- a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/controller/ExecutorController.java +++ b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/controller/ExecutorController.java @@ -380,4 +380,51 @@ public class ExecutorController extends BaseController { WorkflowExecuteDto workflowExecuteDto = execService.queryExecutingWorkflowByProcessInstanceId(processInstanceId); return Result.success(workflowExecuteDto); } + + /** + * execute task instance + * + * @param loginUser login user + * @param projectCode project code + * @param code taskDefinitionCode + * @param version taskDefinitionVersion + * @param warningGroupId warning group id + * @param workerGroup worker group + * @return start task result code + */ + @ApiOperation(value = "startTaskInstance", notes = "RUN_TASK_INSTANCE_NOTES") + @ApiImplicitParams({ + @ApiImplicitParam(name = "version", value = "VERSION", dataType = "Int", example = "1"), + @ApiImplicitParam(name = "failureStrategy", value = "FAILURE_STRATEGY", required = true, dataType = "FailureStrategy"), + @ApiImplicitParam(name = "execType", value = "COMMAND_TYPE", dataType = "CommandType"), + @ApiImplicitParam(name = "warningType", value = "WARNING_TYPE", required = true, dataType = "WarningType"), + @ApiImplicitParam(name = "warningGroupId", value = "WARNING_GROUP_ID", dataType = "Int", example = "100"), + @ApiImplicitParam(name = "workerGroup", value = "WORKER_GROUP", dataType = "String", example = "default"), + @ApiImplicitParam(name = "environmentCode", value = "ENVIRONMENT_CODE", dataType = "Long", example = "-1"), + @ApiImplicitParam(name = "timeout", value = "TIMEOUT", dataType = "Int", example = "100"), + @ApiImplicitParam(name = "dryRun", value = "DRY_RUN", dataType = "Int", example = "0"), + }) + @PostMapping(value = "/task-instance/{code}/start") + @ResponseStatus(HttpStatus.OK) + @ApiException(START_PROCESS_INSTANCE_ERROR) + @AccessLogAnnotation(ignoreRequestArgs = "loginUser") + public Result startStreamTaskInstance(@ApiIgnore @RequestAttribute(value = Constants.SESSION_USER) User loginUser, + @ApiParam(name = "projectCode", value = "PROJECT_CODE", required = true) @PathVariable long projectCode, + @ApiParam(name = "code", value = "TASK_CODE", required = true) @PathVariable long code, + @RequestParam(value = "version", required = true) int version, + @RequestParam(value = "warningGroupId", required = false, defaultValue = "0") Integer warningGroupId, + @RequestParam(value = "workerGroup", required = false, defaultValue = "default") String workerGroup, + @RequestParam(value = "environmentCode", required = false, defaultValue = "-1") Long environmentCode, + @RequestParam(value = "startParams", required = false) String startParams, + @RequestParam(value = "dryRun", defaultValue = "0", required = false) int dryRun) { + + Map startParamMap = null; + if (startParams != null) { + startParamMap = JSONUtils.toMap(startParams); + } + + Map result = execService.execStreamTaskInstance(loginUser, projectCode, code, version, + warningGroupId, workerGroup, environmentCode, startParamMap, dryRun); + return returnDataList(result); + } } diff --git a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/controller/TaskDefinitionController.java b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/controller/TaskDefinitionController.java index a5d1b2f762..64269ddfa3 100644 --- a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/controller/TaskDefinitionController.java +++ b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/controller/TaskDefinitionController.java @@ -34,6 +34,7 @@ import org.apache.dolphinscheduler.api.service.TaskDefinitionService; import org.apache.dolphinscheduler.api.utils.Result; import org.apache.dolphinscheduler.common.Constants; import org.apache.dolphinscheduler.common.enums.ReleaseState; +import org.apache.dolphinscheduler.common.enums.TaskExecuteType; import org.apache.dolphinscheduler.common.utils.ParameterUtils; import org.apache.dolphinscheduler.dao.entity.User; @@ -322,6 +323,7 @@ public class TaskDefinitionController extends BaseController { * @param searchWorkflowName searchWorkflowName * @param searchTaskName searchTaskName * @param taskType taskType + * @param taskExecuteType taskExecuteType * @param pageNo page number * @param pageSize page size * @return task definition page @@ -332,6 +334,7 @@ public class TaskDefinitionController extends BaseController { @ApiImplicitParam(name = "searchWorkflowName", value = "SEARCH_WORKFLOW_NAME", required = false, type = "String"), @ApiImplicitParam(name = "searchTaskName", value = "SEARCH_TASK_NAME", required = false, type = "String"), @ApiImplicitParam(name = "taskType", value = "TASK_TYPE", required = false, dataType = "TaskType", example = "SHELL"), + @ApiImplicitParam(name = "taskExecuteType", value = "TASK_EXECUTE_TYPE", required = false, dataType = "TaskExecuteType", example = "STREAM"), @ApiImplicitParam(name = "pageNo", value = "PAGE_NO", required = true, dataType = "Int", example = "1"), @ApiImplicitParam(name = "pageSize", value = "PAGE_SIZE", required = true, dataType = "Int", example = "10") }) @@ -344,6 +347,7 @@ public class TaskDefinitionController extends BaseController { @RequestParam(value = "searchWorkflowName", required = false) String searchWorkflowName, @RequestParam(value = "searchTaskName", required = false) String searchTaskName, @RequestParam(value = "taskType", required = false) String taskType, + @RequestParam(value = "taskExecuteType", required = false, defaultValue = "BATCH") TaskExecuteType taskExecuteType, @RequestParam("pageNo") Integer pageNo, @RequestParam("pageSize") Integer pageSize) { Result result = checkPageParams(pageNo, pageSize); @@ -352,7 +356,7 @@ public class TaskDefinitionController extends BaseController { } searchWorkflowName = ParameterUtils.handleEscapes(searchWorkflowName); searchTaskName = ParameterUtils.handleEscapes(searchTaskName); - return taskDefinitionService.queryTaskDefinitionListPaging(loginUser, projectCode, searchWorkflowName, searchTaskName, taskType, pageNo, pageSize); + return taskDefinitionService.queryTaskDefinitionListPaging(loginUser, projectCode, searchWorkflowName, searchTaskName, taskType, taskExecuteType, pageNo, pageSize); } /** diff --git a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/controller/TaskInstanceController.java b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/controller/TaskInstanceController.java index 411fd417e6..e7a707af28 100644 --- a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/controller/TaskInstanceController.java +++ b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/controller/TaskInstanceController.java @@ -19,12 +19,15 @@ package org.apache.dolphinscheduler.api.controller; import static org.apache.dolphinscheduler.api.enums.Status.FORCE_TASK_SUCCESS_ERROR; import static org.apache.dolphinscheduler.api.enums.Status.QUERY_TASK_LIST_PAGING_ERROR; +import static org.apache.dolphinscheduler.api.enums.Status.TASK_SAVEPOINT_ERROR; +import static org.apache.dolphinscheduler.api.enums.Status.TASK_STOP_ERROR; import org.apache.dolphinscheduler.api.aspect.AccessLogAnnotation; import org.apache.dolphinscheduler.api.exceptions.ApiException; import org.apache.dolphinscheduler.api.service.TaskInstanceService; import org.apache.dolphinscheduler.api.utils.Result; import org.apache.dolphinscheduler.common.Constants; +import org.apache.dolphinscheduler.common.enums.TaskExecuteType; import org.apache.dolphinscheduler.common.utils.ParameterUtils; import org.apache.dolphinscheduler.dao.entity.User; @@ -74,6 +77,7 @@ public class TaskInstanceController extends BaseController { * @param endTime end time * @param pageNo page number * @param pageSize page size + * @param taskExecuteType task execute type * @return task list page */ @ApiOperation(value = "queryTaskListPaging", notes = "QUERY_TASK_INSTANCE_LIST_PAGING_NOTES") @@ -87,8 +91,9 @@ public class TaskInstanceController extends BaseController { @ApiImplicitParam(name = "host", value = "HOST", type = "String"), @ApiImplicitParam(name = "startDate", value = "START_DATE", type = "String"), @ApiImplicitParam(name = "endDate", value = "END_DATE", type = "String"), + @ApiImplicitParam(name = "taskExecuteType", value = "TASK_EXECUTE_TYPE", required = false, dataType = "TaskExecuteType", example = "STREAM"), @ApiImplicitParam(name = "pageNo", value = "PAGE_NO", required = true, dataType = "Int", example = "1"), - @ApiImplicitParam(name = "pageSize", value = "PAGE_SIZE", required = true, dataType = "Int", example = "20") + @ApiImplicitParam(name = "pageSize", value = "PAGE_SIZE", required = true, dataType = "Int", example = "20"), }) @GetMapping() @ResponseStatus(HttpStatus.OK) @@ -98,6 +103,7 @@ public class TaskInstanceController extends BaseController { @ApiParam(name = "projectCode", value = "PROJECT_CODE", required = true) @PathVariable long projectCode, @RequestParam(value = "processInstanceId", required = false, defaultValue = "0") Integer processInstanceId, @RequestParam(value = "processInstanceName", required = false) String processInstanceName, + @RequestParam(value = "processDefinitionName", required = false) String processDefinitionName, @RequestParam(value = "searchVal", required = false) String searchVal, @RequestParam(value = "taskName", required = false) String taskName, @RequestParam(value = "executorName", required = false) String executorName, @@ -105,6 +111,7 @@ public class TaskInstanceController extends BaseController { @RequestParam(value = "host", required = false) String host, @RequestParam(value = "startDate", required = false) String startTime, @RequestParam(value = "endDate", required = false) String endTime, + @RequestParam(value = "taskExecuteType", required = false, defaultValue = "BATCH") TaskExecuteType taskExecuteType, @RequestParam("pageNo") Integer pageNo, @RequestParam("pageSize") Integer pageSize) { Result result = checkPageParams(pageNo, pageSize); @@ -112,8 +119,8 @@ public class TaskInstanceController extends BaseController { return result; } searchVal = ParameterUtils.handleEscapes(searchVal); - result = taskInstanceService.queryTaskListPaging(loginUser, projectCode, processInstanceId, processInstanceName, - taskName, executorName, startTime, endTime, searchVal, stateType, host, pageNo, pageSize); + result = taskInstanceService.queryTaskListPaging(loginUser, projectCode, processInstanceId, processInstanceName, processDefinitionName, + taskName, executorName, startTime, endTime, searchVal, stateType, host, taskExecuteType, pageNo, pageSize); return result; } @@ -140,4 +147,47 @@ public class TaskInstanceController extends BaseController { return returnDataList(result); } + /** + * task savepoint, for stream task + * + * @param loginUser login user + * @param projectCode project code + * @param id task instance id + * @return the result code and msg + */ + @ApiOperation(value = "savepoint", notes = "TASK_SAVEPOINT") + @ApiImplicitParams({ + @ApiImplicitParam(name = "id", value = "TASK_INSTANCE_ID", required = true, dataType = "Int", example = "12") + }) + @PostMapping(value = "/{id}/savepoint") + @ResponseStatus(HttpStatus.OK) + @ApiException(TASK_SAVEPOINT_ERROR) + @AccessLogAnnotation(ignoreRequestArgs = "loginUser") + public Result taskSavePoint(@ApiIgnore @RequestAttribute(value = Constants.SESSION_USER) User loginUser, + @ApiParam(name = "projectCode", value = "PROJECT_CODE", required = true) @PathVariable long projectCode, + @PathVariable(value = "id") Integer id) { + return taskInstanceService.taskSavePoint(loginUser, projectCode, id); + } + + /** + * task stop, for stream task + * + * @param loginUser login user + * @param projectCode project code + * @param id task instance id + * @return the result code and msg + */ + @ApiOperation(value = "stop", notes = "TASK_STOP") + @ApiImplicitParams({ + @ApiImplicitParam(name = "id", value = "TASK_INSTANCE_ID", required = true, dataType = "Int", example = "12") + }) + @PostMapping(value = "/{id}/stop") + @ResponseStatus(HttpStatus.OK) + @ApiException(TASK_STOP_ERROR) + @AccessLogAnnotation(ignoreRequestArgs = "loginUser") + public Result stopTask(@ApiIgnore @RequestAttribute(value = Constants.SESSION_USER) User loginUser, + @ApiParam(name = "projectCode", value = "PROJECT_CODE", required = true) @PathVariable long projectCode, + @PathVariable(value = "id") Integer id) { + return taskInstanceService.stopTask(loginUser, projectCode, id); + } } diff --git a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/enums/Status.java b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/enums/Status.java index e54c23e906..53501a3c5f 100644 --- a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/enums/Status.java +++ b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/enums/Status.java @@ -220,6 +220,8 @@ public enum Status { DELETE_PROCESS_DEFINITION_USE_BY_OTHER_FAIL(10193, "delete process definition fail, cause used by other tasks: {0}", "删除工作流定时失败，被其他任务引用：{0}"), DELETE_TASK_USE_BY_OTHER_FAIL(10194, "delete task {0} fail, cause used by other tasks: {1}", "删除任务 {0} 失败，被其他任务引用：{1}"), TASK_WITH_DEPENDENT_ERROR(10195, "task used in other tasks", "删除被其他任务引用"), + TASK_SAVEPOINT_ERROR(10196, "task savepoint error", "任务实例savepoint错误"), + TASK_STOP_ERROR(10197, "task stop error", "任务实例停止错误"), UDF_FUNCTION_NOT_EXIST(20001, "UDF function not found", "UDF函数不存在"), UDF_FUNCTION_EXISTS(20002, "UDF function already exists", "UDF函数已存在"), @@ -305,6 +307,8 @@ public enum Status { NOT_SUPPORT_UPDATE_TASK_DEFINITION(50056, "task state does not support modification", "当前任务不支持修改"), NOT_SUPPORT_COPY_TASK_TYPE(50057, "task type [{0}] does not support copy", "不支持复制的任务类型[{0}]"), BATCH_EXECUTE_PROCESS_INSTANCE_ERROR(50058, "change process instance status error: {0}", "修改工作实例状态错误: {0}"), + START_TASK_INSTANCE_ERROR(50059, "start task instance error", "运行任务流实例错误"), + HDFS_NOT_STARTUP(60001, "hdfs not startup", "hdfs未启用"), STORAGE_NOT_STARTUP(60002, "storage not startup", "存储未启用"), S3_CANNOT_RENAME(60003, "directory cannot be renamed", "S3无法重命名文件夹"), diff --git a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/ExecutorService.java b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/ExecutorService.java index 1aefd35471..8146f96e1c 100644 --- a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/ExecutorService.java +++ b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/ExecutorService.java @@ -119,4 +119,22 @@ public interface ExecutorService { * @return */ WorkflowExecuteDto queryExecutingWorkflowByProcessInstanceId(Integer processInstanceId); + + /** + * execute stream task instance + * + * @param loginUser login user + * @param projectCode project code + * @param warningGroupId notify group id + * @param workerGroup worker group name + * @param environmentCode environment code + * @param startParams the global param values which pass to new process instance + * @return execute process instance code + */ + Map execStreamTaskInstance(User loginUser, long projectCode, + long taskDefinitionCode, int taskDefinitionVersion, + int warningGroupId, + String workerGroup, Long environmentCode, + Map startParams, + int dryRun); } diff --git a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/TaskDefinitionService.java b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/TaskDefinitionService.java index a715b68e97..0a84ec7553 100644 --- a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/TaskDefinitionService.java +++ b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/TaskDefinitionService.java @@ -19,6 +19,7 @@ package org.apache.dolphinscheduler.api.service; import org.apache.dolphinscheduler.api.utils.Result; import org.apache.dolphinscheduler.common.enums.ReleaseState; +import org.apache.dolphinscheduler.common.enums.TaskExecuteType; import org.apache.dolphinscheduler.dao.entity.User; import java.util.Map; @@ -171,6 +172,7 @@ public interface TaskDefinitionService { * @param searchWorkflowName searchWorkflowName * @param searchTaskName searchTaskName * @param taskType taskType + * @param taskExecuteType taskExecuteType * @param pageNo page number * @param pageSize page size * @return task definition page @@ -180,6 +182,7 @@ public interface TaskDefinitionService { String searchWorkflowName, String searchTaskName, String taskType, + TaskExecuteType taskExecuteType, Integer pageNo, Integer pageSize); diff --git a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/TaskInstanceService.java b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/TaskInstanceService.java index f532b7d046..b86d4d757e 100644 --- a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/TaskInstanceService.java +++ b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/TaskInstanceService.java @@ -18,6 +18,7 @@ package org.apache.dolphinscheduler.api.service; import org.apache.dolphinscheduler.api.utils.Result; +import org.apache.dolphinscheduler.common.enums.TaskExecuteType; import org.apache.dolphinscheduler.dao.entity.User; import org.apache.dolphinscheduler.plugin.task.api.enums.TaskExecutionStatus; @@ -40,6 +41,7 @@ public interface TaskInstanceService { * @param host host * @param startDate start time * @param endDate end time + * @param taskExecuteType task execute type * @param pageNo page number * @param pageSize page size * @return task list page @@ -48,6 +50,7 @@ public interface TaskInstanceService { long projectCode, Integer processInstanceId, String processInstanceName, + String processDefinitionName, String taskName, String executorName, String startDate, @@ -55,6 +58,7 @@ public interface TaskInstanceService { String searchVal, TaskExecutionStatus stateType, String host, + TaskExecuteType taskExecuteType, Integer pageNo, Integer pageSize); @@ -69,4 +73,22 @@ public interface TaskInstanceService { Map forceTaskSuccess(User loginUser, long projectCode, Integer taskInstanceId); + + /** + * task savepoint + * @param loginUser + * @param projectCode + * @param taskInstanceId + * @return + */ + Result taskSavePoint(User loginUser, long projectCode, Integer taskInstanceId); + + /** + * stop task + * @param loginUser + * @param projectCode + * @param taskInstanceId + * @return + */ + Result stopTask(User loginUser, long projectCode, Integer taskInstanceId); } diff --git a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/ExecutorServiceImpl.java b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/ExecutorServiceImpl.java index 7cf164aaaa..4502fcd8ec 100644 --- a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/ExecutorServiceImpl.java +++ b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/ExecutorServiceImpl.java @@ -57,6 +57,7 @@ import org.apache.dolphinscheduler.dao.mapper.ProjectMapper; import org.apache.dolphinscheduler.dao.mapper.TaskDefinitionMapper; import org.apache.dolphinscheduler.dao.mapper.TaskGroupQueueMapper; import org.apache.dolphinscheduler.plugin.task.api.TaskConstants; +import org.apache.dolphinscheduler.remote.command.TaskExecuteStartCommand; import org.apache.dolphinscheduler.remote.command.WorkflowStateEventChangeCommand; import org.apache.dolphinscheduler.remote.command.WorkflowExecutingDataRequestCommand; import org.apache.dolphinscheduler.remote.command.WorkflowExecutingDataResponseCommand; @@ -1013,4 +1014,45 @@ public class ExecutorServiceImpl extends BaseServiceImpl implements ExecutorServ JSONUtils.parseObject(command.getBody(), WorkflowExecutingDataResponseCommand.class); return responseCommand.getWorkflowExecuteDto(); } + + @Override + public Map execStreamTaskInstance(User loginUser, long projectCode, long taskDefinitionCode, int taskDefinitionVersion, + int warningGroupId, String workerGroup, Long environmentCode, Map startParams, int dryRun) { + Project project = projectMapper.queryByCode(projectCode); + //check user access for project + Map result = + projectService.checkProjectAndAuth(loginUser, project, projectCode, WORKFLOW_START); + if (result.get(Constants.STATUS) != Status.SUCCESS) { + return result; + } + + // check master exists + if (!checkMasterExists(result)) { + return result; + } + + // todo dispatch improvement + List masterServerList = monitorService.getServerListFromRegistry(true); + Host host = new Host(masterServerList.get(0).getHost(), masterServerList.get(0).getPort()); + + TaskExecuteStartCommand taskExecuteStartCommand = new TaskExecuteStartCommand(); + taskExecuteStartCommand.setExecutorId(loginUser.getId()); + taskExecuteStartCommand.setExecutorName(loginUser.getUserName()); + taskExecuteStartCommand.setProjectCode(projectCode); + taskExecuteStartCommand.setTaskDefinitionCode(taskDefinitionCode); + taskExecuteStartCommand.setTaskDefinitionVersion(taskDefinitionVersion); + taskExecuteStartCommand.setWorkerGroup(workerGroup); + taskExecuteStartCommand.setWarningGroupId(warningGroupId); + taskExecuteStartCommand.setEnvironmentCode(environmentCode); + taskExecuteStartCommand.setStartParams(startParams); + taskExecuteStartCommand.setDryRun(dryRun); + + org.apache.dolphinscheduler.remote.command.Command response = stateEventCallbackService.sendSync(host, taskExecuteStartCommand.convert2Command()); + if (response != null) { + putMsg(result, Status.SUCCESS); + } else { + putMsg(result, Status.START_TASK_INSTANCE_ERROR); + } + return result; + } } diff --git a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/TaskDefinitionServiceImpl.java b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/TaskDefinitionServiceImpl.java index 338d6644e2..87ba643188 100644 --- a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/TaskDefinitionServiceImpl.java +++ b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/TaskDefinitionServiceImpl.java @@ -28,6 +28,7 @@ import org.apache.dolphinscheduler.common.enums.AuthorizationType; import org.apache.dolphinscheduler.common.enums.ConditionType; import org.apache.dolphinscheduler.common.enums.Flag; import org.apache.dolphinscheduler.common.enums.ReleaseState; +import org.apache.dolphinscheduler.common.enums.TaskExecuteType; import org.apache.dolphinscheduler.common.utils.CodeGenerateUtils; import org.apache.dolphinscheduler.common.utils.CodeGenerateUtils.CodeGenerateException; import org.apache.dolphinscheduler.common.utils.JSONUtils; @@ -413,8 +414,11 @@ public class TaskDefinitionServiceImpl extends BaseServiceImpl implements TaskDe return null; } if (processService.isTaskOnline(taskCode) && taskDefinition.getFlag() == Flag.YES) { - putMsg(result, Status.NOT_SUPPORT_UPDATE_TASK_DEFINITION); - return null; + // if stream, can update task definition without online check + if (taskDefinition.getTaskExecuteType() != TaskExecuteType.STREAM) { + putMsg(result, Status.NOT_SUPPORT_UPDATE_TASK_DEFINITION); + return null; + } } TaskDefinitionLog taskDefinitionToUpdate = JSONUtils.parseObject(taskDefinitionJsonObj, TaskDefinitionLog.class); if (taskDefinition.equals(taskDefinitionToUpdate)) { @@ -662,6 +666,7 @@ public class TaskDefinitionServiceImpl extends BaseServiceImpl implements TaskDe String searchWorkflowName, String searchTaskName, String taskType, + TaskExecuteType taskExecuteType, Integer pageNo, Integer pageSize) { Result result = new Result(); @@ -673,9 +678,10 @@ public class TaskDefinitionServiceImpl extends BaseServiceImpl implements TaskDe putMsg(result, resultStatus); return result; } + taskType = taskType == null ? StringUtils.EMPTY : taskType; Page page = new Page<>(pageNo, pageSize); IPage taskMainInfoIPage = taskDefinitionMapper.queryDefineListPaging(page, projectCode, searchWorkflowName, - searchTaskName, taskType == null ? StringUtils.EMPTY : taskType); + searchTaskName, taskType, taskExecuteType); List records = taskMainInfoIPage.getRecords(); if (!records.isEmpty()) { Map taskMainInfoMap = new HashMap<>(); diff --git a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/TaskInstanceServiceImpl.java b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/TaskInstanceServiceImpl.java index 73a2e3f96a..67da7c95ed 100644 --- a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/TaskInstanceServiceImpl.java +++ b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/TaskInstanceServiceImpl.java @@ -28,6 +28,7 @@ import org.apache.dolphinscheduler.api.service.UsersService; import org.apache.dolphinscheduler.api.utils.PageInfo; import org.apache.dolphinscheduler.api.utils.Result; import org.apache.dolphinscheduler.common.Constants; +import org.apache.dolphinscheduler.common.enums.TaskExecuteType; import org.apache.dolphinscheduler.common.utils.CollectionUtils; import org.apache.dolphinscheduler.common.utils.DateUtils; import org.apache.dolphinscheduler.dao.entity.Project; @@ -37,6 +38,10 @@ import org.apache.dolphinscheduler.dao.entity.User; import org.apache.dolphinscheduler.dao.mapper.ProjectMapper; import org.apache.dolphinscheduler.dao.mapper.TaskDefinitionMapper; import org.apache.dolphinscheduler.dao.mapper.TaskInstanceMapper; +import org.apache.dolphinscheduler.remote.command.TaskKillRequestCommand; +import org.apache.dolphinscheduler.remote.command.TaskSavePointRequestCommand; +import org.apache.dolphinscheduler.remote.processor.StateEventCallbackService; +import org.apache.dolphinscheduler.remote.utils.Host; import org.apache.dolphinscheduler.plugin.task.api.enums.TaskExecutionStatus; import org.apache.dolphinscheduler.service.process.ProcessService; @@ -81,6 +86,9 @@ public class TaskInstanceServiceImpl extends BaseServiceImpl implements TaskInst @Autowired TaskDefinitionMapper taskDefinitionMapper; + @Autowired + private StateEventCallbackService stateEventCallbackService; + /** * query task list by project, process instance, task name, task start time, task end time, task status, keyword paging * @@ -102,6 +110,7 @@ public class TaskInstanceServiceImpl extends BaseServiceImpl implements TaskInst long projectCode, Integer processInstanceId, String processInstanceName, + String processDefinitionName, String taskName, String executorName, String startDate, @@ -109,6 +118,7 @@ public class TaskInstanceServiceImpl extends BaseServiceImpl implements TaskInst String searchVal, TaskExecutionStatus stateType, String host, + TaskExecuteType taskExecuteType, Integer pageNo, Integer pageSize) { Result result = new Result(); @@ -136,9 +146,17 @@ public class TaskInstanceServiceImpl extends BaseServiceImpl implements TaskInst Page page = new Page<>(pageNo, pageSize); PageInfo> pageInfo = new PageInfo<>(pageNo, pageSize); int executorId = usersService.getUserIdByName(executorName); - IPage taskInstanceIPage = taskInstanceMapper.queryTaskInstanceListPaging( - page, project.getCode(), processInstanceId, processInstanceName, searchVal, taskName, executorId, - statusArray, host, start, end); + IPage taskInstanceIPage; + if (taskExecuteType == TaskExecuteType.STREAM) { + // stream task without process instance + taskInstanceIPage = taskInstanceMapper.queryStreamTaskInstanceListPaging( + page, project.getCode(), processDefinitionName, searchVal, taskName, executorId, statusArray, host, taskExecuteType, start, end + ); + } else { + taskInstanceIPage = taskInstanceMapper.queryTaskInstanceListPaging( + page, project.getCode(), processInstanceId, processInstanceName, searchVal, taskName, executorId, statusArray, host, taskExecuteType, start, end + ); + } Set exclusionSet = new HashSet<>(); exclusionSet.add(Constants.CLASS); exclusionSet.add("taskJson"); @@ -210,4 +228,59 @@ public class TaskInstanceServiceImpl extends BaseServiceImpl implements TaskInst } return result; } + + @Override + public Result taskSavePoint(User loginUser, long projectCode, Integer taskInstanceId) { + Result result = new Result(); + + Project project = projectMapper.queryByCode(projectCode); + //check user access for project + Map checkResult = projectService.checkProjectAndAuth(loginUser, project, projectCode,FORCED_SUCCESS); + Status status = (Status) checkResult.get(Constants.STATUS); + if (status != Status.SUCCESS) { + putMsg(result,status); + return result; + } + + TaskInstance taskInstance = taskInstanceMapper.selectById(taskInstanceId); + if (taskInstance == null) { + putMsg(result, Status.TASK_INSTANCE_NOT_FOUND); + return result; + } + + TaskSavePointRequestCommand command = new TaskSavePointRequestCommand(taskInstanceId); + + Host host = new Host(taskInstance.getHost()); + stateEventCallbackService.sendResult(host, command.convert2Command()); + putMsg(result, Status.SUCCESS); + + return result; + } + + @Override + public Result stopTask(User loginUser, long projectCode, Integer taskInstanceId) { + Result result = new Result(); + + Project project = projectMapper.queryByCode(projectCode); + //check user access for project + Map checkResult = projectService.checkProjectAndAuth(loginUser, project, projectCode,FORCED_SUCCESS); + Status status = (Status) checkResult.get(Constants.STATUS); + if (status != Status.SUCCESS) { + putMsg(result,status); + return result; + } + + TaskInstance taskInstance = taskInstanceMapper.selectById(taskInstanceId); + if (taskInstance == null) { + putMsg(result, Status.TASK_INSTANCE_NOT_FOUND); + return result; + } + + TaskKillRequestCommand command = new TaskKillRequestCommand(taskInstanceId); + Host host = new Host(taskInstance.getHost()); + stateEventCallbackService.sendResult(host, command.convert2Command()); + putMsg(result, Status.SUCCESS); + + return result; + } } diff --git a/dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/controller/TaskInstanceControllerTest.java b/dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/controller/TaskInstanceControllerTest.java index 852520f1f3..7090a2f372 100644 --- a/dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/controller/TaskInstanceControllerTest.java +++ b/dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/controller/TaskInstanceControllerTest.java @@ -31,6 +31,7 @@ import org.apache.dolphinscheduler.api.service.TaskInstanceService; import org.apache.dolphinscheduler.api.utils.PageInfo; import org.apache.dolphinscheduler.api.utils.Result; import org.apache.dolphinscheduler.common.Constants; +import org.apache.dolphinscheduler.common.enums.TaskExecuteType; import org.apache.dolphinscheduler.common.utils.JSONUtils; import org.apache.dolphinscheduler.dao.entity.TaskInstance; import org.apache.dolphinscheduler.dao.entity.User; @@ -69,11 +70,10 @@ public class TaskInstanceControllerTest extends AbstractControllerTest { result.setCode(Status.SUCCESS.getCode()); result.setMsg(Status.SUCCESS.getMsg()); - when(taskInstanceService.queryTaskListPaging(any(), eq(1L), eq(1), eq(""), eq(""), eq(""), any(), any(), - eq(""), Mockito.any(), eq("192.168.xx.xx"), any(), any())).thenReturn(result); - Result taskResult = taskInstanceController.queryTaskListPaging(null, 1L, 1, "", "", - "", "", TaskExecutionStatus.SUCCESS, "192.168.xx.xx", "2020-01-01 00:00:00", "2020-01-02 00:00:00", - pageNo, pageSize); + when(taskInstanceService.queryTaskListPaging(any(), eq(1L), eq(1), eq(""), eq(""), eq(""), eq(""), any(), any(), + eq(""), Mockito.any(), eq("192.168.xx.xx"), eq(TaskExecuteType.BATCH), any(), any())).thenReturn(result); + Result taskResult = taskInstanceController.queryTaskListPaging(null, 1L, 1, "", "", "", + "", "", TaskExecutionStatus.SUCCESS, "192.168.xx.xx", "2020-01-01 00:00:00", "2020-01-02 00:00:00", TaskExecuteType.BATCH, pageNo, pageSize); Assert.assertEquals(Integer.valueOf(Status.SUCCESS.getCode()), taskResult.getCode()); } @@ -91,9 +91,9 @@ public class TaskInstanceControllerTest extends AbstractControllerTest { MvcResult mvcResult = mockMvc.perform(post("/projects/{projectName}/task-instance/force-success", "cxc_1113") .header(SESSION_ID, sessionId) .params(paramsMap)) - .andExpect(status().isOk()) - .andExpect(content().contentType(MediaType.APPLICATION_JSON)) - .andReturn(); + .andExpect(status().isOk()) + .andExpect(content().contentType(MediaType.APPLICATION_JSON)) + .andReturn(); Result result = JSONUtils.parseObject(mvcResult.getResponse().getContentAsString(), Result.class); Assert.assertEquals(Status.SUCCESS.getCode(), result.getCode().intValue()); diff --git a/dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/service/TaskInstanceServiceTest.java b/dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/service/TaskInstanceServiceTest.java index c748838bfc..e1e783c2ac 100644 --- a/dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/service/TaskInstanceServiceTest.java +++ b/dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/service/TaskInstanceServiceTest.java @@ -25,11 +25,13 @@ import static org.mockito.ArgumentMatchers.eq; import static org.mockito.Mockito.when; import org.apache.dolphinscheduler.api.ApiApplicationServer; +import org.apache.dolphinscheduler.api.dto.gantt.Task; import org.apache.dolphinscheduler.api.enums.Status; import org.apache.dolphinscheduler.api.service.impl.ProjectServiceImpl; import org.apache.dolphinscheduler.api.service.impl.TaskInstanceServiceImpl; import org.apache.dolphinscheduler.api.utils.Result; import org.apache.dolphinscheduler.common.Constants; +import org.apache.dolphinscheduler.common.enums.TaskExecuteType; import org.apache.dolphinscheduler.common.enums.UserType; import org.apache.dolphinscheduler.common.utils.DateUtils; import org.apache.dolphinscheduler.dao.entity.ProcessInstance; @@ -101,18 +103,20 @@ public class TaskInstanceServiceTest { when(projectMapper.queryByCode(projectCode)).thenReturn(project); when(projectService.checkProjectAndAuth(loginUser, project, projectCode, TASK_INSTANCE)).thenReturn(result); Result projectAuthFailRes = taskInstanceService.queryTaskListPaging(loginUser, - projectCode, - 0, - "", - "", - "test_user", - "2019-02-26 19:48:00", - "2019-02-26 19:48:22", - "", - null, - "", - 1, - 20); + projectCode, + 0, + "", + "", + "", + "test_user", + "2019-02-26 19:48:00", + "2019-02-26 19:48:22", + "", + null, + "", + TaskExecuteType.BATCH, + 1, + 20); Assert.assertEquals(Status.PROJECT_NOT_FOUND.getCode(), (int) projectAuthFailRes.getCode()); // data parameter check @@ -120,18 +124,20 @@ public class TaskInstanceServiceTest { when(projectMapper.queryByCode(projectCode)).thenReturn(project); when(projectService.checkProjectAndAuth(loginUser, project, projectCode, TASK_INSTANCE)).thenReturn(result); Result dataParameterRes = taskInstanceService.queryTaskListPaging(loginUser, - projectCode, - 1, - "", - "", - "test_user", - "20200101 00:00:00", - "2020-01-02 00:00:00", - "", - TaskExecutionStatus.SUCCESS, - "192.168.xx.xx", - 1, - 20); + projectCode, + 1, + "", + "", + "", + "test_user", + "20200101 00:00:00", + "2020-01-02 00:00:00", + "", + TaskExecutionStatus.SUCCESS, + "192.168.xx.xx", + TaskExecuteType.BATCH, + 1, + 20); Assert.assertEquals(Status.REQUEST_PARAMS_NOT_VALID_ERROR.getCode(), (int) dataParameterRes.getCode()); // project @@ -148,53 +154,47 @@ public class TaskInstanceServiceTest { when(projectService.checkProjectAndAuth(loginUser, project, projectCode, TASK_INSTANCE)).thenReturn(result); when(usersService.queryUser(loginUser.getId())).thenReturn(loginUser); when(usersService.getUserIdByName(loginUser.getUserName())).thenReturn(loginUser.getId()); - when(taskInstanceMapper.queryTaskInstanceListPaging(Mockito.any(Page.class), eq(project.getCode()), eq(1), - eq(""), eq(""), eq(""), - eq(0), Mockito.any(), eq("192.168.xx.xx"), eq(start), eq(end))).thenReturn(pageReturn); + when(taskInstanceMapper.queryTaskInstanceListPaging(Mockito.any(Page.class), eq(project.getCode()), eq(1), eq(""), eq(""), eq(""), + eq(0), Mockito.any(), eq("192.168.xx.xx"), eq(TaskExecuteType.BATCH), eq(start), eq(end))).thenReturn(pageReturn); when(usersService.queryUser(processInstance.getExecutorId())).thenReturn(loginUser); when(processService.findProcessInstanceDetailById(taskInstance.getProcessInstanceId())) .thenReturn(processInstance); - Result successRes = taskInstanceService.queryTaskListPaging(loginUser, projectCode, 1, "", "", - "test_user", "2020-01-01 00:00:00", "2020-01-02 00:00:00", "", TaskExecutionStatus.SUCCESS, - "192.168.xx.xx", 1, 20); + Result successRes = taskInstanceService.queryTaskListPaging(loginUser, projectCode, 1, "", "", "", + "test_user", "2020-01-01 00:00:00", "2020-01-02 00:00:00", "", TaskExecutionStatus.SUCCESS, "192.168.xx.xx", TaskExecuteType.BATCH, 1, 20); Assert.assertEquals(Status.SUCCESS.getCode(), (int) successRes.getCode()); - // executor name empty - when(taskInstanceMapper.queryTaskInstanceListPaging(Mockito.any(Page.class), eq(project.getCode()), eq(1), - eq(""), eq(""), eq(""), - eq(0), Mockito.any(), eq("192.168.xx.xx"), eq(start), eq(end))).thenReturn(pageReturn); - Result executorEmptyRes = taskInstanceService.queryTaskListPaging(loginUser, projectCode, 1, "", "", - "", "2020-01-01 00:00:00", "2020-01-02 00:00:00", "", TaskExecutionStatus.SUCCESS, "192.168.xx.xx", 1, - 20); + //executor name empty + when(taskInstanceMapper.queryTaskInstanceListPaging(Mockito.any(Page.class), eq(project.getCode()), eq(1), eq(""), eq(""), eq(""), + eq(0), Mockito.any(), eq("192.168.xx.xx"), eq(TaskExecuteType.BATCH), eq(start), eq(end))).thenReturn(pageReturn); + Result executorEmptyRes = taskInstanceService.queryTaskListPaging(loginUser, projectCode, 1, "", "", "", + "", "2020-01-01 00:00:00", "2020-01-02 00:00:00", "", TaskExecutionStatus.SUCCESS, "192.168.xx.xx", TaskExecuteType.BATCH, 1, 20); Assert.assertEquals(Status.SUCCESS.getCode(), (int) executorEmptyRes.getCode()); // executor null when(usersService.queryUser(loginUser.getId())).thenReturn(null); when(usersService.getUserIdByName(loginUser.getUserName())).thenReturn(-1); - Result executorNullRes = taskInstanceService.queryTaskListPaging(loginUser, projectCode, 1, "", "", - "test_user", "2020-01-01 00:00:00", "2020-01-02 00:00:00", "", TaskExecutionStatus.SUCCESS, - "192.168.xx.xx", 1, 20); + + Result executorNullRes = taskInstanceService.queryTaskListPaging(loginUser, projectCode, 1, "", "", "", + "test_user", "2020-01-01 00:00:00", "2020-01-02 00:00:00", "", TaskExecutionStatus.SUCCESS, "192.168.xx.xx", TaskExecuteType.BATCH, 1, 20); Assert.assertEquals(Status.SUCCESS.getCode(), (int) executorNullRes.getCode()); - // start/end date null - when(taskInstanceMapper.queryTaskInstanceListPaging(Mockito.any(Page.class), eq(project.getCode()), eq(1), - eq(""), eq(""), eq(""), - eq(0), Mockito.any(), eq("192.168.xx.xx"), any(), any())).thenReturn(pageReturn); - Result executorNullDateRes = taskInstanceService.queryTaskListPaging(loginUser, projectCode, 1, "", "", - "", null, null, "", TaskExecutionStatus.SUCCESS, "192.168.xx.xx", 1, 20); + //start/end date null + when(taskInstanceMapper.queryTaskInstanceListPaging(Mockito.any(Page.class), eq(project.getCode()), eq(1), eq(""), eq(""), eq(""), + eq(0), Mockito.any(), eq("192.168.xx.xx"), eq(TaskExecuteType.BATCH), any(), any())).thenReturn(pageReturn); + Result executorNullDateRes = taskInstanceService.queryTaskListPaging(loginUser, projectCode, 1, "", "", "", + "", null, null, "", TaskExecutionStatus.SUCCESS, "192.168.xx.xx", TaskExecuteType.BATCH, 1, 20); Assert.assertEquals(Status.SUCCESS.getCode(), (int) executorNullDateRes.getCode()); - // start date error format - when(taskInstanceMapper.queryTaskInstanceListPaging(Mockito.any(Page.class), eq(project.getCode()), eq(1), - eq(""), eq(""), eq(""), - eq(0), Mockito.any(), eq("192.168.xx.xx"), any(), any())).thenReturn(pageReturn); + //start date error format + when(taskInstanceMapper.queryTaskInstanceListPaging(Mockito.any(Page.class), eq(project.getCode()), eq(1), eq(""), eq(""), eq(""), + eq(0), Mockito.any(), eq("192.168.xx.xx"), eq(TaskExecuteType.BATCH), any(), any())).thenReturn(pageReturn); - Result executorErrorStartDateRes = taskInstanceService.queryTaskListPaging(loginUser, projectCode, 1, "", "", - "", "error date", null, "", TaskExecutionStatus.SUCCESS, "192.168.xx.xx", 1, 20); + Result executorErrorStartDateRes = taskInstanceService.queryTaskListPaging(loginUser, projectCode, 1, "", "", "", + "", "error date", null, "", TaskExecutionStatus.SUCCESS, "192.168.xx.xx", TaskExecuteType.BATCH, 1, 20); Assert.assertEquals(Status.REQUEST_PARAMS_NOT_VALID_ERROR.getCode(), (int) executorErrorStartDateRes.getCode()); - Result executorErrorEndDateRes = taskInstanceService.queryTaskListPaging(loginUser, projectCode, 1, "", "", - "", null, "error date", "", TaskExecutionStatus.SUCCESS, "192.168.xx.xx", 1, 20); + Result executorErrorEndDateRes = taskInstanceService.queryTaskListPaging(loginUser, projectCode, 1, "", "", "", + "", null, "error date", "", TaskExecutionStatus.SUCCESS, "192.168.xx.xx", TaskExecuteType.BATCH, 1, 20); Assert.assertEquals(Status.REQUEST_PARAMS_NOT_VALID_ERROR.getCode(), (int) executorErrorEndDateRes.getCode()); } @@ -253,6 +253,8 @@ public class TaskInstanceServiceTest { taskInstance.setStartTime(new Date()); taskInstance.setEndTime(new Date()); taskInstance.setExecutorId(-1); + taskInstance.setTaskExecuteType(TaskExecuteType.BATCH); + taskInstance.setState(TaskExecutionStatus.SUBMITTED_SUCCESS); return taskInstance; } diff --git a/dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/enums/TaskExecuteType.java b/dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/enums/TaskExecuteType.java new file mode 100644 index 0000000000..4694cec26d --- /dev/null +++ b/dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/enums/TaskExecuteType.java @@ -0,0 +1,49 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.dolphinscheduler.common.enums; + +import com.baomidou.mybatisplus.annotation.EnumValue; + +/** + * task execute type + */ +public enum TaskExecuteType { + /** + * 0 batch + * 1 stream + */ + BATCH(0, "batch"), + STREAM(1, "stream"); + + TaskExecuteType(int code, String desc) { + this.code = code; + this.desc = desc; + } + + @EnumValue + private final int code; + private final String desc; + + public int getCode() { + return code; + } + + public String getDesc() { + return desc; + } +} diff --git a/dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/model/TaskNode.java b/dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/model/TaskNode.java index b5f975b4d2..227c78b63f 100644 --- a/dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/model/TaskNode.java +++ b/dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/model/TaskNode.java @@ -24,6 +24,7 @@ import static org.apache.dolphinscheduler.plugin.task.api.TaskConstants.TASK_TYP import com.fasterxml.jackson.core.type.TypeReference; import org.apache.dolphinscheduler.common.Constants; import org.apache.dolphinscheduler.common.enums.Priority; +import org.apache.dolphinscheduler.common.enums.TaskExecuteType; import org.apache.dolphinscheduler.plugin.task.api.enums.TaskTimeoutStrategy; import org.apache.dolphinscheduler.plugin.task.api.parameters.TaskTimeoutParameter; import org.apache.dolphinscheduler.common.utils.CollectionUtils; @@ -188,6 +189,11 @@ public class TaskNode { */ private Integer memoryMax; + /** + * task execute type + */ + private TaskExecuteType taskExecuteType; + public String getId() { return id; } @@ -273,8 +279,11 @@ public class TaskNode { } public boolean isForbidden() { - return (!StringUtils.isEmpty(this.runFlag) - && this.runFlag.equals(Constants.FLOWNODE_RUN_FLAG_FORBIDDEN)); + // skip stream task when run DAG + if (taskExecuteType == TaskExecuteType.STREAM) { + return true; + } + return StringUtils.isNotEmpty(this.runFlag) && this.runFlag.equals(Constants.FLOWNODE_RUN_FLAG_FORBIDDEN); } @Override @@ -297,7 +306,8 @@ public class TaskNode { && Objects.equals(workerGroup, taskNode.workerGroup) && Objects.equals(environmentCode, taskNode.environmentCode) && Objects.equals(conditionResult, taskNode.conditionResult) - && CollectionUtils.equalLists(depList, taskNode.depList); + && CollectionUtils.equalLists(depList, taskNode.depList) + && Objects.equals(taskExecuteType, taskNode.taskExecuteType); } @Override @@ -464,7 +474,8 @@ public class TaskNode { + ", workerGroup='" + workerGroup + '\'' + ", environmentCode=" + environmentCode + ", timeout='" + timeout + '\'' - + ", delayTime=" + delayTime + + ", delayTime=" + delayTime + '\'' + + ", taskExecuteType=" + taskExecuteType + '}'; } @@ -523,4 +534,12 @@ public class TaskNode { public void setMemoryMax(Integer memoryMax) { this.memoryMax = memoryMax; } + + public TaskExecuteType getTaskExecuteType() { + return taskExecuteType; + } + + public void setTaskExecuteType(TaskExecuteType taskExecuteType) { + this.taskExecuteType = taskExecuteType; + } } diff --git a/dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/utils/DateUtils.java b/dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/utils/DateUtils.java index 81af098f8b..f883d5ca68 100644 --- a/dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/utils/DateUtils.java +++ b/dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/utils/DateUtils.java @@ -340,17 +340,20 @@ public final class DateUtils { } /** - * format time to duration + * format time to duration, if end date is null, use current time as end time * - * @param d1 d1 - * @param d2 d2 + * @param start start + * @param end end * @return format time */ - public static String format2Duration(Date d1, Date d2) { - if (d1 == null || d2 == null) { + public static String format2Duration(Date start, Date end) { + if (start == null) { return null; } - return format2Duration(differMs(d1, d2)); + if (end == null) { + end = new Date(); + } + return format2Duration(differMs(start, end)); } /** @@ -366,6 +369,11 @@ public final class DateUtils { long minutes = MILLISECONDS.toDurationMinutes(ms); long seconds = MILLISECONDS.toDurationSeconds(ms); + if (days == 0 && hours == 0 && minutes == 0 && seconds == 0) { + // if duration is zero, set 1s + seconds = 1; + } + StringBuilder strBuilder = new StringBuilder(); strBuilder = days > 0 ? strBuilder.append(days).append("d").append(" ") : strBuilder; strBuilder = hours > 0 ? strBuilder.append(hours).append("h").append(" ") : strBuilder; diff --git a/dolphinscheduler-common/src/test/java/org/apache/dolphinscheduler/common/utils/DateUtilsTest.java b/dolphinscheduler-common/src/test/java/org/apache/dolphinscheduler/common/utils/DateUtilsTest.java index 2042fa2d80..0233fcbd91 100644 --- a/dolphinscheduler-common/src/test/java/org/apache/dolphinscheduler/common/utils/DateUtilsTest.java +++ b/dolphinscheduler-common/src/test/java/org/apache/dolphinscheduler/common/utils/DateUtilsTest.java @@ -171,47 +171,49 @@ public class DateUtilsTest { public void testFormat2Duration() { // days hours minutes seconds - Date d1 = DateUtils.stringToDate("2020-01-20 11:00:00"); - Date d2 = DateUtils.stringToDate("2020-01-21 12:10:10"); - String duration = DateUtils.format2Duration(d2, d1); + Date start = DateUtils.stringToDate("2020-01-20 11:00:00"); + Date end = DateUtils.stringToDate("2020-01-21 12:10:10"); + String duration = DateUtils.format2Duration(start, end); Assert.assertEquals("1d 1h 10m 10s", duration); // hours minutes seconds - d1 = DateUtils.stringToDate("2020-01-20 11:00:00"); - d2 = DateUtils.stringToDate("2020-01-20 12:10:10"); - duration = DateUtils.format2Duration(d2, d1); + start = DateUtils.stringToDate("2020-01-20 11:00:00"); + end = DateUtils.stringToDate("2020-01-20 12:10:10"); + duration = DateUtils.format2Duration(start, end); Assert.assertEquals("1h 10m 10s", duration); // minutes seconds - d1 = DateUtils.stringToDate("2020-01-20 11:00:00"); - d2 = DateUtils.stringToDate("2020-01-20 11:10:10"); - duration = DateUtils.format2Duration(d2, d1); + start = DateUtils.stringToDate("2020-01-20 11:00:00"); + end = DateUtils.stringToDate("2020-01-20 11:10:10"); + duration = DateUtils.format2Duration(start, end); Assert.assertEquals("10m 10s", duration); // minutes seconds - d1 = DateUtils.stringToDate("2020-01-20 11:10:00"); - d2 = DateUtils.stringToDate("2020-01-20 11:10:10"); - duration = DateUtils.format2Duration(d2, d1); + start = DateUtils.stringToDate("2020-01-20 11:10:00"); + end = DateUtils.stringToDate("2020-01-20 11:10:10"); + duration = DateUtils.format2Duration(start, end); Assert.assertEquals("10s", duration); - d1 = DateUtils.stringToDate("2020-01-20 11:10:00"); - d2 = DateUtils.stringToDate("2020-01-21 11:10:10"); - duration = DateUtils.format2Duration(d2, d1); + start = DateUtils.stringToDate("2020-01-20 11:10:00"); + end = DateUtils.stringToDate("2020-01-21 11:10:10"); + duration = DateUtils.format2Duration(start, end); Assert.assertEquals("1d 10s", duration); - d1 = DateUtils.stringToDate("2020-01-20 11:10:00"); - d2 = DateUtils.stringToDate("2020-01-20 16:10:10"); - duration = DateUtils.format2Duration(d2, d1); + start = DateUtils.stringToDate("2020-01-20 11:10:00"); + end = DateUtils.stringToDate("2020-01-20 16:10:10"); + duration = DateUtils.format2Duration(start, end); Assert.assertEquals("5h 10s", duration); - } + // startTime = endTime, default 1s + start = DateUtils.stringToDate("2020-01-20 11:10:00"); + end = DateUtils.stringToDate("2020-01-20 11:10:00"); + duration = DateUtils.format2Duration(start, end); + Assert.assertEquals("1s", duration); - @Test - public void testNullDuration() { - // days hours minutes seconds - Date d1 = DateUtils.stringToDate("2020-01-20 11:00:00"); - Date d2 = null; - Assert.assertNull(DateUtils.format2Duration(d1, d2)); + // endTime is null, use current time + start = DateUtils.stringToDate("2020-01-20 11:10:00"); + duration = DateUtils.format2Duration(start, null); + Assert.assertNotNull(duration); } @Test diff --git a/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/entity/TaskDefinition.java b/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/entity/TaskDefinition.java index 61bbb99f9b..2db49f9e07 100644 --- a/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/entity/TaskDefinition.java +++ b/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/entity/TaskDefinition.java @@ -20,6 +20,7 @@ package org.apache.dolphinscheduler.dao.entity; import org.apache.dolphinscheduler.common.Constants; import org.apache.dolphinscheduler.common.enums.Flag; import org.apache.dolphinscheduler.common.enums.Priority; +import org.apache.dolphinscheduler.common.enums.TaskExecuteType; import org.apache.dolphinscheduler.plugin.task.api.enums.TaskTimeoutStrategy; import org.apache.dolphinscheduler.common.enums.TimeoutFlag; import org.apache.dolphinscheduler.common.utils.JSONUtils; @@ -211,6 +212,11 @@ public class TaskDefinition { */ private Integer memoryMax; + /** + * task execute type + */ + private TaskExecuteType taskExecuteType; + public TaskDefinition() { } @@ -483,6 +489,14 @@ public class TaskDefinition { this.memoryMax = memoryMax; } + public TaskExecuteType getTaskExecuteType() { + return taskExecuteType; + } + + public void setTaskExecuteType(TaskExecuteType taskExecuteType) { + this.taskExecuteType = taskExecuteType; + } + @Override public boolean equals(Object o) { if (o == null) { @@ -509,7 +523,8 @@ public class TaskDefinition { && taskGroupId == that.taskGroupId && taskGroupPriority == that.taskGroupPriority && Objects.equals(cpuQuota, that.cpuQuota) - && Objects.equals(memoryMax, that.memoryMax); + && Objects.equals(memoryMax, that.memoryMax) + && Objects.equals(taskExecuteType, that.taskExecuteType); } @Override @@ -543,6 +558,7 @@ public class TaskDefinition { + ", resourceIds='" + resourceIds + '\'' + ", cpuQuota=" + cpuQuota + ", memoryMax=" + memoryMax + + ", taskExecuteType=" + taskExecuteType + ", createTime=" + createTime + ", updateTime=" + updateTime + '}'; diff --git a/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/entity/TaskInstance.java b/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/entity/TaskInstance.java index b52528b62b..91d82df089 100644 --- a/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/entity/TaskInstance.java +++ b/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/entity/TaskInstance.java @@ -27,6 +27,7 @@ import static org.apache.dolphinscheduler.plugin.task.api.TaskConstants.TASK_TYP import org.apache.dolphinscheduler.common.Constants; import org.apache.dolphinscheduler.common.enums.Flag; import org.apache.dolphinscheduler.common.enums.Priority; +import org.apache.dolphinscheduler.common.enums.TaskExecuteType; import org.apache.dolphinscheduler.common.utils.DateUtils; import org.apache.dolphinscheduler.common.utils.JSONUtils; import org.apache.dolphinscheduler.plugin.task.api.enums.TaskExecutionStatus; @@ -89,6 +90,12 @@ public class TaskInstance implements Serializable { @TableField(exist = false) private String processInstanceName; + /** + * process definition name + */ + @TableField(exist = false) + private String processDefinitionName; + /** * process instance name */ @@ -288,6 +295,11 @@ public class TaskInstance implements Serializable { */ private Integer memoryMax; + /** + * task execute type + */ + private TaskExecuteType taskExecuteType; + public void init(String host, Date startTime, String executePath) { this.host = host; this.startTime = startTime; @@ -355,6 +367,11 @@ public class TaskInstance implements Serializable { return TASK_TYPE_BLOCKING.equalsIgnoreCase(this.taskType); } + public boolean isFirstRun() { + return endTime == null; + } + + /** * determine if a task instance can retry * if subProcess, @@ -388,9 +405,4 @@ public class TaskInstance implements Serializable { // task retry does not over time, return false return getRetryInterval() * SEC_2_MINUTES_TIME_UNIT < failedTimeInterval; } - - public boolean isFirstRun() { - return endTime == null; - } - } diff --git a/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/mapper/TaskDefinitionMapper.java b/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/mapper/TaskDefinitionMapper.java index c426da6b04..4940b4e1d2 100644 --- a/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/mapper/TaskDefinitionMapper.java +++ b/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/mapper/TaskDefinitionMapper.java @@ -17,6 +17,7 @@ package org.apache.dolphinscheduler.dao.mapper; +import org.apache.dolphinscheduler.common.enums.TaskExecuteType; import org.apache.dolphinscheduler.dao.entity.DefinitionGroupByUser; import org.apache.dolphinscheduler.dao.entity.TaskDefinition; import org.apache.dolphinscheduler.dao.entity.TaskDefinitionLog; @@ -113,13 +114,15 @@ public interface TaskDefinitionMapper extends BaseMapper { * @param searchWorkflowName searchWorkflowName * @param searchTaskName searchTaskName * @param taskType taskType + * @param taskExecuteType taskExecuteType * @return task main info IPage */ IPage queryDefineListPaging(IPage page, @Param("projectCode") long projectCode, @Param("searchWorkflowName") String searchWorkflowName, @Param("searchTaskName") String searchTaskName, - @Param("taskType") String taskType); + @Param("taskType") String taskType, + @Param("taskExecuteType")TaskExecuteType taskExecuteType); /** * query task definition by code list diff --git a/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/mapper/TaskInstanceMapper.java b/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/mapper/TaskInstanceMapper.java index 17acff69d9..ca68dc57e5 100644 --- a/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/mapper/TaskInstanceMapper.java +++ b/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/mapper/TaskInstanceMapper.java @@ -20,6 +20,7 @@ package org.apache.dolphinscheduler.dao.mapper; import com.baomidou.mybatisplus.core.mapper.BaseMapper; import com.baomidou.mybatisplus.core.metadata.IPage; import org.apache.dolphinscheduler.common.enums.Flag; +import org.apache.dolphinscheduler.common.enums.TaskExecuteType; import org.apache.dolphinscheduler.dao.entity.ExecuteStatusCount; import org.apache.dolphinscheduler.dao.entity.TaskInstance; import org.apache.dolphinscheduler.plugin.task.api.enums.TaskExecutionStatus; @@ -93,6 +94,20 @@ public interface TaskInstanceMapper extends BaseMapper { @Param("executorId") int executorId, @Param("states") int[] statusArray, @Param("host") String host, + @Param("taskExecuteType") TaskExecuteType taskExecuteType, + @Param("startTime") Date startTime, + @Param("endTime") Date endTime + ); + + IPage queryStreamTaskInstanceListPaging(IPage page, + @Param("projectCode") Long projectCode, + @Param("processDefinitionName") String processDefinitionName, + @Param("searchVal") String searchVal, + @Param("taskName") String taskName, + @Param("executorId") int executorId, + @Param("states") int[] statusArray, + @Param("host") String host, + @Param("taskExecuteType") TaskExecuteType taskExecuteType, @Param("startTime") Date startTime, @Param("endTime") Date endTime); diff --git a/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/utils/TaskInstanceUtils.java b/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/utils/TaskInstanceUtils.java index fa3bfec0ca..498abcb3a4 100644 --- a/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/utils/TaskInstanceUtils.java +++ b/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/utils/TaskInstanceUtils.java @@ -74,6 +74,7 @@ public class TaskInstanceUtils { target.setTaskGroupId(source.getTaskGroupId()); target.setCpuQuota(source.getCpuQuota()); target.setMemoryMax(source.getMemoryMax()); + target.setTaskExecuteType(source.getTaskExecuteType()); } } diff --git a/dolphinscheduler-dao/src/main/resources/org/apache/dolphinscheduler/dao/mapper/TaskDefinitionLogMapper.xml b/dolphinscheduler-dao/src/main/resources/org/apache/dolphinscheduler/dao/mapper/TaskDefinitionLogMapper.xml index 16dc698d26..d914f9db08 100644 --- a/dolphinscheduler-dao/src/main/resources/org/apache/dolphinscheduler/dao/mapper/TaskDefinitionLogMapper.xml +++ b/dolphinscheduler-dao/src/main/resources/org/apache/dolphinscheduler/dao/mapper/TaskDefinitionLogMapper.xml @@ -21,7 +21,7 @@ id, code, name, version, description, project_code, user_id, task_type, task_params, flag, task_priority, worker_group, environment_code, fail_retry_times, fail_retry_interval, timeout_flag, timeout_notify_strategy, timeout, delay_time, - resource_ids, operator, operate_time, create_time, update_time, task_group_id, task_group_priority, cpu_quota, memory_max + resource_ids, operator, operate_time, create_time, update_time, task_group_id, task_group_priority, cpu_quota, memory_max, task_execute_type select @@ -89,14 +89,15 @@ insert into t_ds_task_definition (code, name, version, description, project_code, user_id, task_type, task_params, flag, task_priority, worker_group, environment_code, fail_retry_times, fail_retry_interval, - timeout_flag, timeout_notify_strategy, timeout, delay_time, resource_ids, create_time, update_time,task_group_id) + timeout_flag, timeout_notify_strategy, timeout, delay_time, resource_ids, create_time, update_time,task_group_id, task_execute_type) values (#{taskDefinition.code},#{taskDefinition.name},#{taskDefinition.version},#{taskDefinition.description}, #{taskDefinition.projectCode},#{taskDefinition.userId},#{taskDefinition.taskType},#{taskDefinition.taskParams},#{taskDefinition.flag}, #{taskDefinition.taskPriority},#{taskDefinition.workerGroup},#{taskDefinition.environmentCode},#{taskDefinition.failRetryTimes}, #{taskDefinition.failRetryInterval},#{taskDefinition.timeoutFlag},#{taskDefinition.timeoutNotifyStrategy},#{taskDefinition.timeout}, - #{taskDefinition.delayTime},#{taskDefinition.resourceIds},#{taskDefinition.createTime},#{taskDefinition.updateTime}, #{taskDefinition.taskGroupId}) + #{taskDefinition.delayTime},#{taskDefinition.resourceIds},#{taskDefinition.createTime},#{taskDefinition.updateTime}, #{taskDefinition.taskGroupId}, + #{taskDefinition.taskExecuteType}) +