Skip to content

B
byit-myth-job
Commit 2cdc2029 by huangfusuper
Options

【执行记录过滤】执行记录过滤，将对应的记录添加到日志

parent 96cc7186
Showing with 450 additions and 73 deletions
byit-myth-admin/src/main/java/com/byit/controller/JobController.java
......@@ -4,9 +4,11 @@ import com.byit.conf.MythJobAutoConfigure;
import com.byit.job.dto.JobRunResultDto;
import com.byit.job.dto.PluginBeanJobInfo;
import com.byit.model.JobTask;
import com.byit.model.RunRecording;
import com.byit.service.JobTaskService;
import com.byit.thread.LogCallbackThread;
import com.byit.thread.LogScanHelper;
import com.byit.thread.RunRecordingScanHelper;
import com.byit.util.SourceObj2TargetObjUtil;
import com.byit.utils.ValidationUtil;
import org.springframework.beans.factory.annotation.Autowired;
......@@ -51,10 +53,10 @@ public class JobController {
}
@Autowired
private LogScanHelper logScanHelper;
private RunRecordingScanHelper runRecordingScanThread;
@GetMapping("test")
public String test(){
logScanHelper.doStop();
runRecordingScanThread.start();
return "ASDSADSADSADSAD";
}
}
byit-myth-core/myth-admin-core/src/main/java/com/byit/conf/MythJobScheduler.java
......@@ -2,6 +2,7 @@ package com.byit.conf;
import com.byit.thread.JobScheduleHelper;
import com.byit.thread.LogScanHelper;
import com.byit.thread.RunRecordingScanHelper;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;
......@@ -19,11 +20,13 @@ import org.springframework.stereotype.Component;
public class MythJobScheduler implements InitializingBean, DisposableBean {
private final JobScheduleHelper jobScheduleHelper;
private final LogScanHelper logScanHelper;
private final RunRecordingScanHelper runRecordingScanHelper;
@Autowired
public MythJobScheduler(JobScheduleHelper jobScheduleHelper, LogScanHelper logScanHelper) {
public MythJobScheduler(JobScheduleHelper jobScheduleHelper, LogScanHelper logScanHelper, RunRecordingScanHelper runRecordingScanHelper) {
this.jobScheduleHelper = jobScheduleHelper;
this.logScanHelper = logScanHelper;
this.runRecordingScanHelper = runRecordingScanHelper;
}
/**
......@@ -34,6 +37,7 @@ public class MythJobScheduler implements InitializingBean, DisposableBean {
public void destroy() throws Exception {
this.jobScheduleHelper.doStop();
this.logScanHelper.doStop();
runRecordingScanHelper.doStop();
}
/**
......@@ -45,6 +49,7 @@ public class MythJobScheduler implements InitializingBean, DisposableBean {
//启用扫描线程
this.jobScheduleHelper.start();
this.logScanHelper.start();
runRecordingScanHelper.start();
}
}
byit-myth-core/myth-admin-core/src/main/java/com/byit/mapper/EmailAlarmMapper.java
......@@ -2,8 +2,11 @@ package com.byit.mapper;
import com.byit.model.EmailAlarm;
import com.byit.model.vo.EmailAlarmVo;
import org.apache.ibatis.annotations.Param;
import org.springframework.stereotype.Repository;
import java.util.List;
/**
* @author huangfu
*/
......@@ -31,6 +34,13 @@ public interface EmailAlarmMapper {
int saveEmailAlarm(EmailAlarm record);
/**
* 批量保存邮件
* @param emailAlarms
* @return
*/
int saveEmailAlarms(@Param("emailAlarms") List<EmailAlarm> emailAlarms);
/**
* 修改邮件发送情况
* @param record
* @return
......
byit-myth-core/myth-admin-core/src/main/java/com/byit/mapper/JobTaskRunLogMapper.java
......@@ -2,6 +2,7 @@ package com.byit.mapper;
import com.byit.model.JobTaskRunLog;
import com.byit.model.JobTaskRunLogWithBLOBs;
import org.apache.ibatis.annotations.Param;
import org.springframework.stereotype.Repository;
import java.util.List;
......@@ -13,6 +14,14 @@ import java.util.List;
@Repository
public interface JobTaskRunLogMapper {
/**
* 查根据flowId和RunId查询一批节点
* @param flowId
* @param runId
* @return
*/
List<JobTaskRunLogWithBLOBs> findJobTaskRunLogWithBLOBsByFlowIdAndRunId(@Param("flowId") Integer flowId, @Param("runId")String runId);
/**
* 查询已经结束或者失败的节点
* @return
*/
......
byit-myth-core/myth-admin-core/src/main/java/com/byit/mapper/RunRecordingMapper.java
......@@ -4,6 +4,8 @@ import com.byit.model.RunRecording;
import org.apache.ibatis.annotations.Param;
import org.springframework.stereotype.Repository;
import java.util.List;
/**
* 运行记录表
* @author huangfu
......@@ -11,6 +13,11 @@ import org.springframework.stereotype.Repository;
@Repository
public interface RunRecordingMapper {
/**
* 查询已经完结的，并且没有告警的任务流
* @return
*/
List<RunRecording> findRunRecordingByEndAndNotIsAlarm();
/**
* 这个方法是会根据任务流的id和运行标识找到唯一对应的一个任务流，这个任务流就是一个虚拟节点
* @param flowId
* @param runId
......
byit-myth-core/myth-admin-core/src/main/java/com/byit/model/EmailAlarm.java
......@@ -4,13 +4,20 @@ import io.swagger.annotations.ApiModel;
import io.swagger.annotations.ApiModelProperty;
import java.io.Serializable;
import java.util.Date;
import lombok.AllArgsConstructor;
import lombok.Builder;
import lombok.Data;
import lombok.NoArgsConstructor;
/**
*
*/
@ApiModel
@Data
@AllArgsConstructor
@NoArgsConstructor
@Builder
public class EmailAlarm implements Serializable {
/**
* 主键
......@@ -71,6 +78,16 @@ public class EmailAlarm implements Serializable {
*/
@ApiModelProperty("告警内容")
private String alarmContent;
/**
* 告警标题
*/
@ApiModelProperty("告警标题")
private String alarmTitle;
/**
* 告警标题
*/
@ApiModelProperty("1成功 2 失败 3补批成功 4补批失败")
private String flowRes;
/**
*/
......
byit-myth-core/myth-admin-core/src/main/java/com/byit/model/JobTaskRunLog.java
package com.byit.model;
import com.fasterxml.jackson.annotation.JsonFormat;
import io.swagger.annotations.ApiModel;
import io.swagger.annotations.ApiModelProperty;
import java.io.Serializable;
......@@ -78,11 +79,6 @@ public class JobTaskRunLog implements Serializable {
@ApiModelProperty("运行参数")
private String runParams;
/**
* 开始时间
*/
@ApiModelProperty("开始时间")
private Date startTime;
/**
* 运行方式 1执行机运行，2插件运行
......@@ -127,6 +123,12 @@ public class JobTaskRunLog implements Serializable {
private Date endTime;
/**
* 开始时间
*/
@ApiModelProperty("开始时间")
private Date startTime;
/**
* 节点id
*/
@ApiModelProperty("节点id")
......
byit-myth-core/myth-admin-core/src/main/java/com/byit/model/RunRecording.java
......@@ -133,11 +133,6 @@ public class RunRecording implements Serializable {
@ApiModelProperty("快速失败标识 0 不快速失败 1 快速失败")
private String failFast;
/**
* 工作流版本名称
*/
@ApiModelProperty("工作流版本名称")
private String versionName;
/**
*/
......
byit-myth-core/myth-admin-core/src/main/java/com/byit/service/EmailAlarmService.java
......@@ -2,6 +2,9 @@ package com.byit.service;
import com.byit.model.EmailAlarm;
import com.byit.model.vo.EmailAlarmVo;
import org.apache.ibatis.annotations.Param;
import java.util.List;
/**
* @program: byit-myth-job->EmailAlarmService
......@@ -30,6 +33,12 @@ public interface EmailAlarmService {
* @return
*/
int saveEmailAlarm(EmailAlarm record);
/**
* 批量保存邮件
* @param emailAlarms
* @return
*/
int saveEmailAlarms(List<EmailAlarm> emailAlarms);
/**
* 发送邮件
......
byit-myth-core/myth-admin-core/src/main/java/com/byit/service/JobTaskRunLogService.java
......@@ -13,6 +13,13 @@ import java.util.List;
**/
public interface JobTaskRunLogService {
/**
* 查根据flowId和RunId查询一批节点
* @param flowId
* @param runId
* @return
*/
List<JobTaskRunLogWithBLOBs> findJobTaskRunLogWithBLOBsByFlowIdAndRunId(Integer flowId,String runId);
/**
* 查询已经结束或者失败的节点
* @return
*/
......
byit-myth-core/myth-admin-core/src/main/java/com/byit/service/RunRecordingService.java
......@@ -2,12 +2,19 @@ package com.byit.service;
import com.byit.model.RunRecording;
import java.util.List;
/**
* 运行记录表
* @author huangfu
*/
public interface RunRecordingService {
/**
* 查询已经完结的，并且没有告警的任务流
* @return
*/
List<RunRecording> findRunRecordingByEndAndNotIsAlarm();
/**
* 这个方法是会根据任务流的id和运行标识找到唯一对应的一个任务流，这个任务流就是一个虚拟节点
* @param flowId
* @param runId
......
byit-myth-core/myth-admin-core/src/main/java/com/byit/service/impl/EmailAlarmServiceImpl.java
......@@ -18,6 +18,7 @@ import org.springframework.stereotype.Service;
import javax.mail.MessagingException;
import java.util.Date;
import java.util.List;
/**
* @program: byit-myth-job->EmailAlarmServiceImpl
......@@ -56,6 +57,11 @@ public class EmailAlarmServiceImpl implements EmailAlarmService {
}
@Override
public int saveEmailAlarms(List<EmailAlarm> emailAlarms) {
return emailAlarmMapper.saveEmailAlarms(emailAlarms);
}
@Override
@DataValidation
public void sendEmail(@ParamValidation EmailAlarmVo record) {
record.setSendTime(new Date());
......
byit-myth-core/myth-admin-core/src/main/java/com/byit/service/impl/JobTaskRunLogServiceImpl.java
......@@ -29,6 +29,11 @@ public class JobTaskRunLogServiceImpl implements JobTaskRunLogService {
}
@Override
public List<JobTaskRunLogWithBLOBs> findJobTaskRunLogWithBLOBsByFlowIdAndRunId(Integer flowId, String runId) {
return jobTaskRunLogMapper.findJobTaskRunLogWithBLOBsByFlowIdAndRunId(flowId,runId);
}
@Override
public List<JobTaskRunLog> findJobTaskRunLogEndOrFailureNode() {
return jobTaskRunLogMapper.findJobTaskRunLogEndOrFailureNode();
}
......
byit-myth-core/myth-admin-core/src/main/java/com/byit/service/impl/RunRecordingServiceImpl.java
......@@ -6,6 +6,8 @@ import com.byit.service.RunRecordingService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import java.util.List;
/**
* @program: byit-myth-job->RunRecordingServiceImpl
* @description: 对应的任务流运行记录操作
......@@ -22,6 +24,11 @@ public class RunRecordingServiceImpl implements RunRecordingService {
}
@Override
public List<RunRecording> findRunRecordingByEndAndNotIsAlarm() {
return runRecordingMapper.findRunRecordingByEndAndNotIsAlarm();
}
@Override
public RunRecording findRunRecordingByFlowIdAndRunId(Integer flowId, String runId) {
return runRecordingMapper.findRunRecordingByFlowIdAndRunId(flowId,runId);
}
......
byit-myth-core/myth-admin-core/src/main/java/com/byit/task/JavaBeanJobTask.java
......@@ -4,6 +4,7 @@ import cn.hutool.http.HttpUtil;
import com.alibaba.fastjson.JSON;
import com.byit.job.dto.AdminSenPluginDto;
import com.byit.job.dto.DispatchResponseDto;
import com.byit.job.enums.plugin.PluginEnum;
import com.byit.job.exceptions.plugin.PluginException;
import com.byit.job.utils.IpUtil;
import com.byit.model.JobTaskRunLogWithBLOBs;
......@@ -55,7 +56,7 @@ public class JavaBeanJobTask implements TimerTask {
log.debug("---------------{}------------",result);
}catch (PluginException ignored){
DispatchResponseDto dispatchResponseDto = DispatchResponseDto.builder().code(ignored.getIEnum( ).getCode( )).msg(ignored.getIEnum( ).getMsg( )).build( );
DispatchResponseDto dispatchResponseDto = DispatchResponseDto.builder().code(PluginEnum.NO_SERVICE_AVAILABLE.getRes()).msg(ignored.getIEnum( ).getMsg( )).build( );
saveLog(mythJobTaskSchedule, "", JSON.toJSONString(dispatchResponseDto));
log.error("-------------通讯异常：{}，{}",ignored.getIEnum().getCode(),ignored.getIEnum().getMsg());
}catch (Exception e){
......
byit-myth-core/myth-admin-core/src/main/java/com/byit/thread/LogScanHelper.java
......@@ -2,11 +2,14 @@ package com.byit.thread;
import cn.hutool.core.collection.CollectionUtil;
import com.byit.enums.RunRecordingEnum;
import com.byit.model.EmailAlarm;
import com.byit.model.JobTaskRunLog;
import com.byit.model.RunRecording;
import com.byit.service.EmailAlarmService;
import com.byit.service.JobTaskRunLogService;
import com.byit.service.RunRecordingService;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
......@@ -15,9 +18,7 @@ import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;
/**
* @program: byit-myth-job->LogScanHelper
......@@ -31,10 +32,12 @@ public class LogScanHelper {
private DataSource dataSource;
private final JobTaskRunLogService jobTaskRunLogService;
private final RunRecordingService runRecordingService;
private final EmailAlarmService emailAlarmService;
@Autowired
public LogScanHelper(JobTaskRunLogService jobTaskRunLogService, RunRecordingService runRecordingService) {
public LogScanHelper(JobTaskRunLogService jobTaskRunLogService, RunRecordingService runRecordingService, EmailAlarmService emailAlarmService) {
this.jobTaskRunLogService = jobTaskRunLogService;
this.runRecordingService = runRecordingService;
this.emailAlarmService = emailAlarmService;
}
@Autowired
......@@ -43,10 +46,6 @@ public class LogScanHelper {
}
private final String SUCCESS_CODE = "1";
private final String FAILURE_CODE = "2";
private final String RE_SUCCESS_CODE = "3";
private final String RE_FAILURE_CODE = "4";
/**
* 扫描虚节点的线程是否停止
*/
......@@ -176,42 +175,17 @@ public class LogScanHelper {
List<JobTaskRunLog> jobTaskRunLogEndOrFailureNode = jobTaskRunLogService.findJobTaskRunLogEndOrFailureNode( );
if(CollectionUtil.isNotEmpty(jobTaskRunLogEndOrFailureNode)){
log.debug("-------------【查询到有完结而且未告警的节点】-------------");
Map<String, List<JobTaskRunLog>> jobLogMap = jobTaskRunLogEndOrFailureNode.stream( )
.collect(Collectors.groupingBy(JobTaskRunLog::getRunCode));
//失败的
List<JobTaskRunLog> failureNodes = jobLogMap.get(FAILURE_CODE);
List<JobTaskRunLog> reFailure = jobLogMap.get(RE_FAILURE_CODE);
if(CollectionUtil.isNotEmpty(reFailure)){
failureNodes.addAll(reFailure);
}
//成功的
List<JobTaskRunLog> successNodes = jobLogMap.get(SUCCESS_CODE);
List<JobTaskRunLog> reSuccessNodes = jobLogMap.get(RE_SUCCESS_CODE);
if(CollectionUtil.isNotEmpty(reSuccessNodes)){
successNodes.addAll(reSuccessNodes);
}
//遍历成功的节点 修改执行记录表
if(CollectionUtil.isNotEmpty(successNodes)){
successNodes.forEach(successNode ->{
String runId = successNode.getRunId( );
Integer flowId = successNode.getFlowId( );
runRecordingService.updateRunRecordingByFlowIdAndRunId(RunRecording.builder().runId(runId).flowId(flowId).flowRunResult(RunRecordingEnum.RUN_FLOW_SUCCESS.getCode()).flowStatus(RunRecordingEnum.FLOW_STATUS_IS_END.getCode()).build());
successNode.setAlertEnd("1");
jobTaskRunLogService.updateJobTaskRunLog(successNode);
});
}else{
dateAligned(1000);
}
jobTaskRunLogEndOrFailureNode.forEach(endNode ->{
String runId = endNode.getRunId( );
Integer flowId = endNode.getFlowId( );
//如果运行结果为null 那么就是调度都没成功 那么就取调度的值
String code = StringUtils.isNotBlank(endNode.getRunCode())?endNode.getRunCode() : endNode.getTriggerCode();
runRecordingService.updateRunRecordingByFlowIdAndRunId(RunRecording.builder().runId(runId).flowId(flowId).flowRunResult(code).flowStatus(RunRecordingEnum.FLOW_STATUS_IS_END.getCode()).build());
endNode.setAlertEnd("1");
jobTaskRunLogService.updateJobTaskRunLog(endNode);
});
/*if(CollectionUtil.isNotEmpty(failureNodes)){
failureNodes.forEach(failureNode ->{
String runId = failureNode.getRunId( );
Integer flowId = failureNode.getFlowId( );
});
}*/
}else{
dateAligned(20000);
}
......
byit-myth-core/myth-admin-core/src/main/java/com/byit/thread/RunRecordingScanHelper.java 0 → 100644
package com.byit.thread;
import cn.hutool.core.collection.CollectionUtil;
import cn.hutool.core.date.DateUtil;
import com.byit.enums.RunRecordingEnum;
import com.byit.model.EmailAlarm;
import com.byit.model.JobTaskRunLogWithBLOBs;
import com.byit.model.RunRecording;
import com.byit.service.EmailAlarmService;
import com.byit.service.JobTaskRunLogService;
import com.byit.service.RunRecordingService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import javax.sql.DataSource;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.List;
import java.util.Set;
import java.util.concurrent.TimeUnit;
/**
* 运行记录扫描线程
* @author huangfu
*/
@Component
@Slf4j
public class RunRecordingScanHelper {
public static final String DATE_FORMAT = "yyyy-MM-dd HH:mm:ss";
/**
* 完成时告警
*/
public static final String WHEN_DONE = "1";
/**
* 失败时告警
*/
public static final String FAILURE_DONE = "2";
public static final String SUCCESS_DONE = "3";
private DataSource dataSource;
private final EmailAlarmService emailAlarmService;
private final JobTaskRunLogService jobTaskRunLogService;
private final RunRecordingService runRecordingService;
private volatile boolean runRecordingThreadStop = false;
private Thread runRecordingThread;
@Autowired
public RunRecordingScanHelper(EmailAlarmService emailAlarmService, JobTaskRunLogService jobTaskRunLogService, RunRecordingService runRecordingService) {
this.emailAlarmService = emailAlarmService;
this.jobTaskRunLogService = jobTaskRunLogService;
this.runRecordingService = runRecordingService;
}
public void start(){
runRecordingThread = new Thread(() ->{
log.info("--------------------【com.byit.thread.RunRecordingScanThread#start】init success----------------------");
while (!runRecordingThreadStop){
dateAligned(5000);
Connection conn = null;
Boolean connAutoCommit = null;
PreparedStatement preparedStatement = null;
try {
conn = dataSource.getConnection();
connAutoCommit = conn.getAutoCommit();
conn.setAutoCommit(false);
preparedStatement = conn.prepareStatement("SELECT * FROM JOB_LOCK WHERE LOCK_NAME = 'run_recording_lock' FOR UPDATE ");
preparedStatement.execute();
//进行操作
scanRunRec();
}catch (Exception e){
e.printStackTrace();
}finally {
if(conn != null){
try {
conn.commit();
} catch (SQLException e) {
if(!runRecordingThreadStop){
log.error("--------------------【提交行锁出错】---------------------");
}
}
}
try {
if(conn != null){
conn.setAutoCommit(connAutoCommit);
}
} catch (SQLException e) {
if(!runRecordingThreadStop){
log.error("--------------------【恢复自动提交出错】---------------------");
}
}
if(preparedStatement != null){
try {
preparedStatement.close();
} catch (SQLException e) {
if(!runRecordingThreadStop){
log.error("--------------------【关闭执行器出错】---------------------");
}
}
}
try {
conn.close();
} catch (SQLException e) {
if(!runRecordingThreadStop){
log.error("--------------------【关闭链接出错】---------------------");
}
}
}
}
runRecordingThread.setDaemon(true);
runRecordingThread.setName("myth-job#【RunRecordingScanThread】# start");
runRecordingThread.start();
});
}
private void scanRunRec(){
//查询完结且未告警的工作流信息
List<RunRecording> runRecordingByEndAndNotIsAlarm = runRecordingService.findRunRecordingByEndAndNotIsAlarm();
List<EmailAlarm> emailAlarms = new ArrayList<>(15);
if(CollectionUtil.isNotEmpty(runRecordingByEndAndNotIsAlarm)){
runRecordingByEndAndNotIsAlarm.forEach(runRecording -> {
//如果设置为完成时告警
switch (runRecording.getAlarmlAction()){
//设置为完成时告警
case WHEN_DONE:
if(RunRecordingEnum.FLOW_STATUS_IS_END.getCode().equals(runRecording.getFlowStatus())){
saveEmailAlarms(emailAlarms,runRecording);
}
break;
//失败时告警
case FAILURE_DONE:
if(RunRecordingEnum.RUN_FLOW_FAILURE.getCode().equals(runRecording.getFlowRunResult())
|| RunRecordingEnum.RUN_FLOW_RE_FAILURE.getCode().equals(runRecording.getFlowRunResult())){
saveEmailAlarms(emailAlarms,runRecording);
}
break;
//成功时告警
case SUCCESS_DONE:
if(RunRecordingEnum.RUN_FLOW_SUCCESS.getCode().equals(runRecording.getFlowRunResult())
|| RunRecordingEnum.RUN_FLOW_RE_SUCCESS.getCode().equals(runRecording.getFlowRunResult())){
saveEmailAlarms(emailAlarms,runRecording);
}
break;
default:
break;
}
});
if(CollectionUtil.isNotEmpty(emailAlarms)){
int saveCount = emailAlarmService.saveEmailAlarms(emailAlarms);
}else{
dateAligned(20000);
}
}
}
public void doStop(){
this.runRecordingThreadStop = true;
try {
TimeUnit.SECONDS.sleep(1);
} catch (InterruptedException e) {
e.printStackTrace( );
}
if (runRecordingThread.getState() != Thread.State.TERMINATED) {
runRecordingThread.interrupt();
try {
runRecordingThread.join();
} catch (InterruptedException e) {
e.printStackTrace( );
}
}
log.warn("---------------【运行记录扫描日志线程被注销】-----------------------");
}
/**
* 对齐时钟。整秒运行
*/
private void dateAligned(long waitTime){
try {
TimeUnit.MILLISECONDS.sleep(waitTime - System.currentTimeMillis()%1000);
} catch (InterruptedException e) {
e.printStackTrace( );
}
}
private void saveEmailAlarms(List<EmailAlarm> emailAlarms,RunRecording runRecording){
List<JobTaskRunLogWithBLOBs> jobTaskRunLogByFlowIdAndRunId = jobTaskRunLogService.findJobTaskRunLogWithBLOBsByFlowIdAndRunId(runRecording.getFlowId(), runRecording.getRunId());
String flowName = runRecording.getFlowName();
String senContentHtml = runMsgHtml(jobTaskRunLogByFlowIdAndRunId, flowName);
EmailAlarm build = EmailAlarm.builder()
.flowId(runRecording.getFlowId())
.flowName(flowName)
.runId(runRecording.getRunId())
.versionName(runRecording.getFlowVersionName())
.alarmEmail(runRecording.getAlarmEmail())
.alarmAction(runRecording.getAlarmlAction())
.alarmContent(senContentHtml)
.alarmResult("0")
.flowRes(runRecording.getFlowRunResult())
.alarmTitle(flowName)
.build();
emailAlarms.add(build);
}
private String runMsgHtml(List<JobTaskRunLogWithBLOBs> jobTaskRunLogs,String title){
StringBuilder stringBuilder= new StringBuilder();
stringBuilder.append("<table border='1' width='80%' align='center' cellspacing='0' cellpadding='6'>")
.append(String.format("<h2 style='text-align:center;color:red'>%s</h2>",title))
.append("<thead align='center' style='background: blue;color: #fff'>")
.append("<th width = '10%'>节点名称</th>")
.append("<th width = '10%'>开始时间</th>")
.append("<th width = '10%'>结束时间</th>")
.append("<th width = '10%'>耗费时间</th>")
.append("<th width = '10%'>运行结果</th>")
.append("<th width = '50%'>运行日志</th>")
.append("</thead>")
.append("<tbody>");
if(CollectionUtil.isNotEmpty(jobTaskRunLogs)){
jobTaskRunLogs.forEach(jobTaskRunLog -> {
long timeConsuming = TimeUnit.MILLISECONDS.toSeconds(jobTaskRunLog.getEndTime().getTime() - jobTaskRunLog.getStartTime().getTime());
stringBuilder.append("<tr align='center'>")
.append(String.format("<td>%s</td>",jobTaskRunLog.getNodeName()))
.append(String.format("<td>%s</td>", DateUtil.format(jobTaskRunLog.getStartTime(), DATE_FORMAT)))
.append(String.format("<td>%s</td>",DateUtil.format(jobTaskRunLog.getEndTime(),DATE_FORMAT)))
.append(String.format("<td>%s</td>", timeConsuming+"分钟"))
.append(String.format("<td>%s</td>","1".equals(jobTaskRunLog.getRunCode())? "成功"
: "3".equals(jobTaskRunLog.getRunCode())? "补批成功"
: "4".equals(jobTaskRunLog.getRunCode())? "补批失败" : "失败"))
.append("<td>")
.append("<div style='display:inline-block;width:100%;word-break:break-all;height: auto;overflow: auto;text-align: left;'>")
.append(String.format("%s",jobTaskRunLog.getRunMsg()))
.append("</div>")
.append("</td></tr>");
});
}
stringBuilder.append("</tbody>")
.append("</table>");
return stringBuilder.toString();
}
@Autowired
public void setDataSource(DataSource dataSource) {
this.dataSource = dataSource;
}
}
byit-myth-core/myth-admin-core/src/main/resources/mapper/EmailAlarmMapper.xml
......@@ -2,7 +2,6 @@
<!DOCTYPE mapper PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN" "http://mybatis.org/dtd/mybatis-3-mapper.dtd">
<mapper namespace="com.byit.mapper.EmailAlarmMapper">
<resultMap id="BaseResultMap" type="com.byit.model.EmailAlarm">
<!-- generated @mbg.generated date: 2019-12-25 -->
<id column="id" jdbcType="INTEGER" property="id" />
<result column="flow_id" jdbcType="INTEGER" property="flowId" />
<result column="flow_name" jdbcType="VARCHAR" property="flowName" />
......@@ -12,23 +11,21 @@
<result column="alarm_action" jdbcType="CHAR" property="alarmAction" />
<result column="alarm_result" jdbcType="CHAR" property="alarmResult" />
<result column="send_time" jdbcType="TIMESTAMP" property="sendTime" />
<result column="alarm_title" jdbcType="VARCHAR" property="alarmTitle" />
<result column="flow_res" jdbcType="CHAR" property="flowRes" />
</resultMap>
<resultMap extends="BaseResultMap" id="ResultMapWithBLOBs" type="com.byit.model.EmailAlarm">
<!-- generated @mbg.generated date: 2019-12-25 -->
<result column="alarm_content" jdbcType="LONGVARCHAR" property="alarmContent" />
</resultMap>
<sql id="Base_Column_List">
<!-- generated @mbg.generated date: 2019-12-25 -->
id, flow_id, flow_name, run_id, version_name, alarm_email, alarm_action, alarm_result,
send_time
id, flow_id, flow_name, run_id, version_name, alarm_email, alarm_action, alarm_result,
send_time,alarm_title,flow_res
</sql>
<sql id="Blob_Column_List">
<!-- generated @mbg.generated date: 2019-12-25 -->
alarm_content
</sql>
<select id="findEmailAlarmById" parameterType="java.lang.Integer" resultMap="ResultMapWithBLOBs">
<!-- generated @mbg.generated date: 2019-12-25 -->
select
select
<include refid="Base_Column_List" />
,
<include refid="Blob_Column_List" />
......@@ -36,12 +33,10 @@
where id = #{id,jdbcType=INTEGER}
</select>
<delete id="deleteById" parameterType="java.lang.Integer">
<!-- generated @mbg.generated date: 2019-12-25 -->
delete from email_alarm
where id = #{id,jdbcType=INTEGER}
</delete>
<insert id="saveEmailAlarm" parameterType="com.byit.model.EmailAlarm">
<!-- generated @mbg.generated date: 2019-12-25 -->
insert into email_alarm
<trim prefix="(" suffix=")" suffixOverrides=",">
<if test="id != null">
......@@ -74,6 +69,12 @@
<if test="alarmContent != null">
alarm_content,
</if>
<if test="alarmTitle != null">
alarm_title,
</if>
<if test="flowRes != null">
flow_res,
</if>
</trim>
<trim prefix="values (" suffix=")" suffixOverrides=",">
<if test="id != null">
......@@ -106,10 +107,33 @@
<if test="alarmContent != null">
#{alarmContent,jdbcType=LONGVARCHAR},
</if>
<if test="alarmTitle != null">
#{alarmTitle,jdbcType=VARCHAR},
</if>
<if test="flowRes != null">
#{flowRes,jdbcType=CHAR},
</if>
</trim>
</insert>
<insert id="saveEmailAlarms" parameterType="com.byit.model.EmailAlarm">
insert into email_alarm (
id, flow_id, flow_name, run_id, version_name, alarm_email, alarm_action, alarm_result,
send_time,alarm_content,alarm_title,flow_res
) values
<foreach collection="emailAlarms" item="emailAlarm" separator =",">
(
#{emailAlarm.id,jdbcType=INTEGER},#{emailAlarm.flowId,jdbcType=INTEGER},
#{emailAlarm.flowName,jdbcType=VARCHAR},#{emailAlarm.runId,jdbcType=VARCHAR},
#{emailAlarm.versionName,jdbcType=VARCHAR},#{emailAlarm.alarmEmail,jdbcType=VARCHAR},
#{emailAlarm.alarmAction,jdbcType=CHAR},#{emailAlarm.alarmResult,jdbcType=CHAR},
#{emailAlarm.sendTime,jdbcType=TIMESTAMP},#{emailAlarm.alarmContent,jdbcType=LONGVARCHAR},
#{emailAlarm.alarmTitle,jdbcType=VARCHAR},#{emailAlarm.flowRes,jdbcType=CHAR}
)
</foreach>
</insert>
<update id="updateEmailAlarm" parameterType="com.byit.model.EmailAlarm">
<!-- generated @mbg.generated date: 2019-12-25 -->
update email_alarm
<set>
<if test="flowId != null">
......@@ -139,6 +163,12 @@
<if test="alarmContent != null">
alarm_content = #{alarmContent,jdbcType=LONGVARCHAR},
</if>
<if test="alarmTitle != null">
alarm_title = #{alarmTitle,jdbcType=VARCHAR},
</if>
<if test="flowRes != null">
flow_res = #{flowRes,jdbcType=CHAR},
</if>
</set>
where id = #{id,jdbcType=INTEGER}
</update>
......
byit-myth-core/myth-admin-core/src/main/resources/mapper/JobTaskRunLogMapper.xml
......@@ -39,11 +39,19 @@
<sql id="Blob_Column_List">
run_msg, trigger_msg
</sql>
<select id="findJobTaskRunLogWithBLOBsByFlowIdAndRunId" resultMap="ResultMapWithBLOBs">
select
<include refid="Base_Column_List" />
,
<include refid="Blob_Column_List" />
from job_task_run_log
where flow_id = #{flowId,jdbcType=INTEGER} and run_id = #{runId,jdbcType=VARCHAR}
</select>
<select id="findJobTaskRunLogEndOrFailureNode" resultMap="BaseResultMap">
select <include refid="Base_Column_List" /> from job_task_run_log
where (node_name='END' or run_code = '2' or run_code = '4') and alert_end = '0'
where (node_name='END' or run_code = '2' or run_code = '4' or trigger_code = '2') and alert_end = '0'
</select>
<select id="findNotEndVirtualNode" resultMap="BaseResultMap">
......
byit-myth-core/myth-admin-core/src/main/resources/mapper/RunRecordingMapper.xml
......@@ -28,6 +28,14 @@
flow_id, start_time, end_time, is_alarm,is_inner,fail_fast
</sql>
<!--查询已完结 没有告警的-->
<select id="findRunRecordingByEndAndNotIsAlarm" resultMap="BaseResultMap">
select
<include refid="Base_Column_List" />
from run_recording
where flow_status ='4' and is_alarm = '1'
</select>
<select id="findRunRecordingByFlowIdAndRunId" resultMap="BaseResultMap" >
select
<include refid="Base_Column_List" />
......
byit-myth-core/myth-core-common/src/main/java/com/byit/job/enums/plugin/PluginEnum.java
......@@ -8,14 +8,20 @@ import com.byit.job.enums.IEnum;
*/
public enum PluginEnum implements IEnum {
REQUEST_PORT_OR_IP_IS_MISSING("请求ip或者port为null","10000"),
NO_SERVICE_AVAILABLE("无可用的服务","11000");
REQUEST_PORT_OR_IP_IS_MISSING("请求ip或者port为null","10000","2"),
NO_SERVICE_AVAILABLE("无可用的服务","11000","2");
private String res;
private String msg;
private String code;
PluginEnum(String msg, String code) {
PluginEnum(String msg, String code,String res) {
this.msg = msg;
this.code = code;
this.res = res;
}
public String getRes() {
return res;
}
@Override
......
byit-myth-core/myth-core-common/src/main/java/com/byit/job/exceptions/plugin/PluginException.java
......@@ -19,6 +19,8 @@ public class PluginException extends RuntimeException implements IException {
this.iEnum = e;
}
@Override
public IEnum getIEnum() {
return this.iEnum;
......
byit-myth-executor/myth-exector-plugin/src/main/java/com/byit/rpc/RunJobServerHandler.java
......@@ -41,7 +41,7 @@ public class RunJobServerHandler extends SimpleChannelInboundHandler<FullHttpReq
protected void channelRead0(ChannelHandlerContext ctx, FullHttpRequest req) throws Exception {
log.debug("----------------------有请求过来了------------------------");
DispatchResponseDto dispatchResponseDto = new DispatchResponseDto();
dispatchResponseDto.setCode(JobResultEnum.DISPATCH_FAIL.getCode());
dispatchResponseDto.setCode(JobResultEnum.DISPATCH_FAIL.getRes());
dispatchResponseDto.setMsg(JobResultEnum.DISPATCH_FAIL.getMsg());
if(req != null){
//解析 调度中心 的数据对象
......@@ -53,11 +53,11 @@ public class RunJobServerHandler extends SimpleChannelInboundHandler<FullHttpReq
String heartbeat = adminSenPluginDto.getHeartbeat();
if(null == heartbeat){
JOB_TRIGGER_POOL.execute(new RunJobThread(adminSenPluginDto));
dispatchResponseDto.setCode(JobResultEnum.DISPATCH_SUCCESS.getCode());
dispatchResponseDto.setCode(JobResultEnum.DISPATCH_SUCCESS.getRes());
dispatchResponseDto.setMsg(JobResultEnum.DISPATCH_SUCCESS.getMsg());
}else if(PENG.equals(heartbeat)){
log.debug("----------调度平台心跳检测-------------");
dispatchResponseDto.setCode(JobResultEnum.DISPATCH_SUCCESS.getCode());
dispatchResponseDto.setCode(JobResultEnum.DISPATCH_SUCCESS.getRes());
dispatchResponseDto.setMsg(JobResultEnum.DISPATCH_SUCCESS.getMsg());
dispatchResponseDto.setContent(PONG);
}
......
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