海豚调度调优 | 正在运行的工作流(DAG)如何重新拉起失败的任务(Task)
💡 本系列文章是DolphinScheduler由浅入深的教程，涵盖搭建、二开迭代、核心原理解读、运维和管理等一系列内容。适用于想对 DolphinScheduler了解或想要加深理解的读者。
**祝开卷有益。 **
本系列教程基于 DolphinScheduler 2.0.5 做的优化。（稳定版推荐使用3.1.9）
先抛出问题
1.场景描述
工作流 A 正在运行，里面有很多节点，依赖关系比较复杂。凌晨用户接到报警，a 节点失败了，此时其他分支的任务还在运行。此时工作流是不会是失败的，需要等待其他分支的任务运行结束，整个工作流才会失败。
失败的任务：是失败状态且重试次数用完了。
2.目前的处理流程
目前的做法是，先 kill 工作流，等待工作流变成失败状态，然后再点击恢复失败按钮，即可把失败的节点重新拉起来。
画外音：kill 工作流我也做了优化，后面会有文章介绍，kill之后，工作流会变成失败状态，这样做是为了可以恢复失败。
3.困惑
这种做法会影响正在运行的任务，强制把正在跑的任务 kill 掉，对一些运行时间比较久的任务来说，会降低执行效率。跑的好好的，被干掉了，恢复失败，又得重新跑。
这非常不划算。
优化建议：如何在不停止工作流的情况下，单独把失败的节点重新拉起来呢？
解决方案
后端优化：
分析了工作流启动、停止、恢复失败等操作类型 Master 和 Worker 的原理，打算新增一个操作类型、命令类型枚举值：RUN_FAILED_ONLY。
优化后的大致流程如下：
-
用户在页面上点击按钮，提交 executeType = RUN_FAILED_ONLY、processInstanceId=xxxx的请求。
-
API服务收到请求，判断是 RUN_FAILED_ONLY 操作，就封装一个 StateEventChangeCommand 命令，进行RPC请求。
-
Master服务的 StateEventProcessor 监听到命令，提交给 StateEventResponseService ，它负责找到对应的工作流 WorkflowExecuteThread ，然后把这个stateEvent 给这个 WorkflowExecuteThread.
-
WorkflowExecuteThread处理 stateEvent，判断这个 stateEvent 的 StateEventType 是 RUN_FAILED_ONLY_EVENT ，进行下面的处理：
找到改工作流失败且重试次数用完的任务列表，然后依次处理它们的执行记录(标记为失效，从失败列表移除，添加到待提交队列），最后提交到等待队列。
后端流程结束。
前端页面优化：
比较简单：新增一个按钮，文案是【重新失败节点】，在工作流列表上展示，用户可以点击。
源码
其中，新增了三个枚举类的值：
org.apache.dolphinscheduler.api.enums.ExecuteType
RUN_FAILED_ONLY
org.apache.dolphinscheduler.common.enums.CommandType
RUN_FAILED_ONLY(44, "run failed only");
org.apache.dolphinscheduler.common.enums.StateEventType
RUN_FAILED_ONLY_EVENT(4, "run failed only event");
几个关键步骤的代码，为了方便查看，上下文的代码、涉及改动的方法也会贴出来。
提示：序号对应上面的流程。
② org.apache.dolphinscheduler.api.service.impl.ExecutorServiceImpl#execute
switch (executeType) {
case REPEAT_RUNNING:
result = insertCommand(loginUser, processInstanceId, processDefinition.getCode(), processDefinition.getVersion(), CommandType.REPEAT_RUNNING, startParams);
break;
case RECOVER_SUSPENDED_PROCESS:
result = insertCommand(loginUser, processInstanceId, processDefinition.getCode(), processDefinition.getVersion(), CommandType.RECOVER_SUSPENDED_PROCESS, startParams);
break;
// 新增 9-11 行代码
case RUN_FAILED_ONLY:
result = sendRunFailedOnlyMsg(processInstance, CommandType.RUN_FAILED_ONLY);
break;
case START_FAILURE_TASK_PROCESS:
result = insertCommand(loginUser, processInstanceId, processDefinition.getCode(), processDefinition.getVersion(), CommandType.START_FAILURE_TASK_PROCESS, startParams);
break;
case STOP:
if (processInstance.getState() == ExecutionStatus.READY_STOP) {
putMsg(result, Status.PROCESS_INSTANCE_ALREADY_CHANGED, processInstance.getName(), processInstance.getState());
} else {
result = updateProcessInstancePrepare(processInstance, CommandType.STOP, ExecutionStatus.READY_STOP);
}
break;
case PAUSE:
if (processInstance.getState() == ExecutionStatus.READY_PAUSE) {
putMsg(result, Status.PROCESS_INSTANCE_ALREADY_CHANGED, processInstance.getName(), processInstance.getState());
} else {
result = updateProcessInstancePrepare(processInstance, CommandType.PAUSE, ExecutionStatus.READY_PAUSE);
}
break;
default:
logger.error("unknown execute type : {}", executeType);
putMsg(result, Status.REQUEST_PARAMS_NOT_VALID_ERROR, "unknown execute type");
break;
}
sendRunFailedOnlyMsg 方法的逻辑，封装 stateEventChangeCommand，提交 RPC 请求。
/**
* send msg to master, run failed only
*
* @param processInstance process instance
* @param commandType command type
* @return update result
*/
private Map<String, Object> sendRunFailedOnlyMsg(ProcessInstance processInstance, CommandType commandType) {
Map<String, Object> result = new HashMap<>();
String host = processInstance.getHost();
String address = host.split(":")[0];
int port = Integer.parseInt(host.split(":")[1]);
StateEventChangeCommand stateEventChangeCommand = new StateEventChangeCommand(
processInstance.getId(), 0, processInstance.getState(), processInstance.getId(), 0,StateEventType.RUN_FAILED_ONLY_EVENT
);
stateEventCallbackService.sendResult(address, port, stateEventChangeCommand.convert2Command());
putMsg(result, Status.SUCCESS);
return result;
}
这里也给 StateEventChangeCommand 家了一个状态类型的字段，对应枚举类：StateEventType
方便下游判断状态。
查看下面 ③ 处的代码，用这个状态赋值给 stateEvent 的 type。
③ org.apache.dolphinscheduler.server.master.processor.StateEventProcessor#process 用上面 ② 处的 StateEventType，赋值给 stateEvent 的 type，然后提交给 stateEventResponseService 的 BlockingQueueeventQueue 队列。
@Override
public void process(Channel channel, Command command) {
Preconditions.checkArgument(CommandType.STATE_EVENT_REQUEST == command.getType(), String.format("invalid command type: %s", command.getType()));
StateEventChangeCommand stateEventChangeCommand = JSONUtils.parseObject(command.getBody(), StateEventChangeCommand.class);
StateEvent stateEvent = new StateEvent();
stateEvent.setKey(stateEventChangeCommand.getKey());
if (stateEventChangeCommand.getSourceProcessInstanceId() != stateEventChangeCommand.getDestProcessInstanceId()) {
stateEvent.setExecutionStatus(ExecutionStatus.RUNNING_EXECUTION);
} else {
stateEvent.setExecutionStatus(stateEventChangeCommand.getSourceStatus());
}
stateEvent.setProcessInstanceId(stateEventChangeCommand.getDestProcessInstanceId());
stateEvent.setTaskInstanceId(stateEventChangeCommand.getDestTaskInstanceId());
// TODO 修改
StateEventType stateEventType = stateEventChangeCommand.getStateEventType();
if (stateEventType != null){
stateEvent.setType(stateEventType);
}else {
StateEventType type = stateEvent.getTaskInstanceId() == 0 ? StateEventType.PROCESS_STATE_CHANGE : StateEventType.TASK_STATE_CHANGE;
stateEvent.setType(type);
}
logger.info("received command : {}", stateEvent);
stateEventResponseService.addResponse(stateEvent);
}
StateEventResponseWorker 线程一直扫描这个队列，拿到 stateEvent，找到要处理的工作流对应的 WorkflowExecuteThread 线程，把这个事件提交给 WorkflowExecuteThread 线程。
/**
* task worker thread
*/
class StateEventResponseWorker extends Thread {
@Override
public void run() {
while (Stopper.isRunning()) {
try {
// if not task , blocking here
StateEvent stateEvent = eventQueue.take();
persist(stateEvent);
} catch (InterruptedException e) {
logger.warn("persist task error", e);
Thread.currentThread().interrupt();
break;
}
}
logger.info("StateEventResponseWorker stopped");
}
}
private void persist(StateEvent stateEvent) {
try {
if (!this.processInstanceMapper.containsKey(stateEvent.getProcessInstanceId())) {
writeResponse(stateEvent, ExecutionStatus.FAILURE);
return;
}
WorkflowExecuteThread workflowExecuteThread = this.processInstanceMapper.get(stateEvent.getProcessInstanceId());
workflowExecuteThread.addStateEvent(stateEvent);
writeResponse(stateEvent, ExecutionStatus.SUCCESS);
} catch (Exception e) {
logger.error("persist event queue error, event: {}", stateEvent, e);
}
}
④WorkflowExecuteThread 内部循环扫描事件列表。
private void handleEvents() {
while (!this.stateEvents.isEmpty()) {
try {
StateEvent stateEvent = this.stateEvents.peek();
if (stateEventHandler(stateEvent)) {
this.stateEvents.remove(stateEvent);
}
} catch (Exception e) {
logger.error("state handle error:", e);
}
}
}
stateEventHandler 处理 RUN_FAILED_ONLY_EVENT 类型的事件，处理方法是：runFailedHandler
private boolean stateEventHandler(StateEvent stateEvent) {
logger.info("process event: {}", stateEvent.toString());
if (!checkStateEvent(stateEvent)) {
return false;
}
boolean result = false;
switch (stateEvent.getType()) {
case RUN_FAILED_ONLY_EVENT:
result = runFailedHandler(stateEvent);
break;
case PROCESS_STATE_CHANGE:
result = processStateChangeHandler(stateEvent);
break;
case TASK_STATE_CHANGE:
result = taskStateChangeHandler(stateEvent);
break;
case PROCESS_TIMEOUT:
result = processTimeout();
break;
case TASK_TIMEOUT:
result = taskTimeout(stateEvent);
break;
default:
break;
}
if (result) {
this.stateEvents.remove(stateEvent);
}
return result;
}
runFailedHandler 的内部逻辑如下：找到改工作流失败且重试次数用完的任务列表，然后依次处理它们的执行记录(标记为失效，从失败列表移除，添加到待提交队列），最后提交到等待队列。
private boolean runFailedHandler(StateEvent stateEvent) {
try {
logger.info("process:{} will do {}", processInstance.getId(), stateEvent.getExecutionStatus());
// find failed tasks with max retry times and init these tasks
List<Integer> failedList = processService.queryTaskByProcessIdAndStateWithMaxRetry(processInstance.getId(), ExecutionStatus.FAILURE);
logger.info("run failed task size is : {}", failedList.size());
for (Integer taskId : failedList) {
logger.info("run failed task id is : {}", taskId);
TaskInstance taskInstance = processService.findTaskInstanceById(taskId);
taskInstance.setFlag(Flag.NO);
// remove it from errorTaskList
errorTaskList.remove(Long.toString(taskInstance.getTaskCode()));
processService.updateTaskInstance(taskInstance);
// submit current task nodes
if (readyToSubmitTaskQueue.contains(taskInstance)) {
continue;
}
logger.info("run failed task ，submit current task nodes : {}", taskInstance.toString());
addTaskToStandByList(taskInstance);
}
submitStandByTask();
// updateProcessInstanceState();
} catch (Exception e) {
logger.error("process only run failed task error:", e);
}
return true;
}
最终效果
再次回到文章开头的场景：
工作流 A 正在运行，里面有很多节点，依赖关系比较复杂。凌晨用户接到报警，a 节点失败了，此时其他分支的任务还在运行。
此时用户可以直接点击【重新拉起失败任务】按钮，失败的任务就会重新进入等待队列，后续流程就像任务正常运行一样，也会继续拉起下游任务。
画外音：本次优化简化了失败任务运维的复杂度，提高了效率。
作者从1.x开始使用海豚调度，那是还叫做 Easy Scheduler，是一个忠实用户，我们基于 2.x版本做了很多内部的改造，后续会分享出来，同样社区也推荐大家使用3.1.9版本，这是相对比较稳定的版本。
本文由 白鲸开源 提供发布支持！