Skip to content

B
byit-myth-job
Commit 20dad9e0 by huangfusuper
Options

快速失败完善

parent 7dcb3768
Showing with 92 additions and 11 deletions
byit-myth-core/myth-admin-core/src/main/java/com/byit/mapper/JobTaskMapper.java
......@@ -29,6 +29,13 @@ public interface JobTaskMapper {
List<JobTask> findAllByRunId(@Param("runId")String runId,@Param("flowId")Integer flowId);
/**
*
* @param runId
* @return
*/
List<JobTask> findByRunId(@Param("runId") String runId);
/**
* 根据id查询
* @param id
* @return
......
byit-myth-core/myth-admin-core/src/main/java/com/byit/service/JobTaskService.java
......@@ -27,6 +27,8 @@ public interface JobTaskService {
*/
List<JobTask> findJobTaskByRunId(String runId,Integer flowId);
List<JobTask> findByRunId(String runId);
/**
* 添加单个任务节点
* @param jobTask
......
byit-myth-core/myth-admin-core/src/main/java/com/byit/service/impl/JobTaskServiceImpl.java
......@@ -43,6 +43,11 @@ public class JobTaskServiceImpl implements JobTaskService {
return jobTaskMapper.findAllByRunId(runId,flowId);
}
@Override
public List<JobTask> findByRunId(String runId) {
return jobTaskMapper.findByRunId(runId);
}
/**
* 添加一个任务
* @param jobTask 任务实体
......
byit-myth-core/myth-admin-core/src/main/java/com/byit/service/mapservice/impl/RunRecordingAndLogServiceImpl.java
package com.byit.service.mapservice.impl;
import com.byit.enums.EmailEnum;
import com.byit.enums.FlowPropertyEnum;
import com.byit.enums.RunRecordingEnum;
import com.byit.enums.task.RunResultEnum;
import com.byit.model.*;
import com.byit.service.JobTaskRunLogService;
import com.byit.service.JobTaskService;
import com.byit.service.NodeVersionService;
import com.byit.service.RunRecordingService;
import com.byit.service.*;
import com.byit.service.mapservice.RunRecordingAndLogService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.BeanUtils;
import org.springframework.stereotype.Service;
import java.net.InetAddress;
import java.net.UnknownHostException;
import java.util.Date;
import java.util.List;
import java.util.stream.Collectors;
......@@ -20,37 +22,42 @@ import java.util.stream.Collectors;
* @author huangfu
*/
@Service
@Slf4j
public class RunRecordingAndLogServiceImpl implements RunRecordingAndLogService {
public static final String TRIGGER_MSG = "任务执行超时，被强制执行快速失败！";
private final JobTaskRunLogService jobTaskRunLogService;
private final RunRecordingService runRecordingService;
private final NodeVersionService nodeVersionService;
private final JobTaskService jobTaskService;
private final FlowService flowService;
public RunRecordingAndLogServiceImpl(JobTaskRunLogService jobTaskRunLogService, RunRecordingService runRecordingService,
NodeVersionService nodeVersionService, JobTaskService jobTaskService) {
NodeVersionService nodeVersionService, JobTaskService jobTaskService, FlowService flowService) {
this.jobTaskRunLogService = jobTaskRunLogService;
this.runRecordingService = runRecordingService;
this.nodeVersionService = nodeVersionService;
this.jobTaskService = jobTaskService;
this.flowService = flowService;
}
@Override
public void logAndRunRecordingFailFast(RunRecording runRecording) {
String runId = runRecording.getRunId();
Integer flowId = runRecording.getFlowId();
//Integer flowId = runRecording.getFlowId();
//查询该实例对应的所由日志节点
List<JobTaskRunLogWithBLOBs> jobTaskRunLog = jobTaskRunLogService.findJobTaskRunLogWithBLOBsByFlowIdAndRunId(flowId, runId);
List<Integer> logNodeId = jobTaskRunLog.stream().map(JobTaskRunLog::getNodeId).collect(Collectors.toList());
//List<JobTaskRunLogWithBLOBs> jobTaskRunLog = jobTaskRunLogService.findJobTaskRunLogWithBLOBsByFlowIdAndRunId(flowId, runId);
//List<Integer> logNodeId = jobTaskRunLog.stream().map(JobTaskRunLog::getNodeId).collect(Collectors.toList());
//List<NodeVersion> allByFlowId = nodeVersionService.findAllByFlowId(flowId);
//筛选没在日志里面的节点
//List<NodeVersion> notLogNode = allByFlowId.stream().filter(nodeVersion -> !(logNodeId.contains(nodeVersion.getNodeId()))).collect(Collectors.toList());
//将这些节点置为失败并将实例也置为失败
List<JobTask> jobTaskByRunId = jobTaskService.findJobTaskByRunId(runId, flowId);
//保存错误日志节点
jobTaskByRunId.forEach(node ->{
//查询task所有的对应节点
List<JobTask> byRunId = jobTaskService.findByRunId(runId);
byRunId.forEach(task ->{
JobTaskRunLogWithBLOBs log = new JobTaskRunLogWithBLOBs();
BeanUtils.copyProperties(node,log);
BeanUtils.copyProperties(task,log);
log.setTriggerCode(RunResultEnum.TRIGGER_ERROR.getCode());
log.setTriggerMsg(TRIGGER_MSG);
log.setRunCode(RunResultEnum.RUN_ERROR.getCode());
......@@ -72,7 +79,67 @@ public class RunRecordingAndLogServiceImpl implements RunRecordingAndLogService
runRecording.setFlowRunResult(RunRecordingEnum.RUN_FLOW_FAILURE.getCode());
runRecording.setFailFast(RunRecordingEnum.FAIL_FAST_YES.getCode());
runRecordingService.updateRunRecordingById(runRecording);
//筛选虚节点
List<JobTask> virtualTasks = byRunId.stream().filter(task -> "0".equals(task.getIsVirtual())).collect(Collectors.toList());
virtualTasks.forEach(virtualTask -> {
Integer mapFlowId = virtualTask.getMapFlowId();
List<NodeVersion> allByFlowId = nodeVersionService.findAllByFlowId(mapFlowId);
allByFlowId.forEach(node ->{
JobTaskRunLogWithBLOBs log = new JobTaskRunLogWithBLOBs();
BeanUtils.copyProperties(node,log);
log.setTriggerCode(RunResultEnum.TRIGGER_ERROR.getCode());
log.setTriggerMsg(TRIGGER_MSG);
log.setRunCode(RunResultEnum.RUN_ERROR.getCode());
log.setTriggerMsg(TRIGGER_MSG);
Date thisTime = new Date();
log.setStartTime(thisTime);
log.setEndTime(thisTime);
log.setRunId(runId);
log.setFlowName(runRecording.getFlowName());
log.setTriggerTime(thisTime);
jobTaskRunLogService.saveJobTaskRunLog(log);
});
//执行实例的快速失败
Flow virFlow = flowService.findFlowById(mapFlowId);
log.info("-----------【虚节点保存运行记录】--------------");
//保存进运行记录表
RunRecording virtualRunRecording = null;
try {
virtualRunRecording = RunRecording.builder()
.runId(virtualTask.getRunId())
.flowId(mapFlowId)
.flowName(virFlow.getFlowName())
.flowVersionName(virFlow.getVersionName())
.flowStatus("4")
.flowRunResult("2")
.flowTimeout(virFlow.getFlowTimeout())
.dispatchIp(InetAddress.getLocalHost().getHostAddress())
.alarmEmail(virFlow.getAlarmEmail())
.alarmlAction(virFlow.getAlarmlAction())
.priority(virFlow.getPriority())
.triggerTime(runRecording.getTriggerTime())
.principal(virFlow.getPrincipal())
.startTime(new Date())
.flowNodeCount(virFlow.getFlowNodeCount())
.isAlarm(EmailEnum.IS_ALARM_NO.getCode())
.isInner(FlowPropertyEnum.IS_INNER.getCode())
.failFast(RunRecordingEnum.FAIL_FAST_YES.getCode())
.workspaceId(virFlow.getWorkspaceId())
.repeatTime(runRecording.getRepeatTime())
.operator(virtualTask.getOperator())
.scheduleType(virtualTask.getScheduleType())
.build();
} catch (UnknownHostException e) {
e.printStackTrace();
}
runRecordingService.saveRunRecording(virtualRunRecording);
});
//删除task里面的数据
jobTaskService.removeByRunId(runId);
}
}
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment