Skip to content

B
byit-myth-job
Commit b489a913 by huangfusuper
Options

【修改BUG】运行记录

parent deb0e4e8
Showing with 115 additions and 41 deletions
byit-myth-core/myth-admin-core/src/main/java/com/byit/model/JobTaskRunLog.java
......@@ -76,7 +76,7 @@ public class JobTaskRunLog implements Serializable {
/**
* 运行结果
*/
@ApiModelProperty("运行结果")
@ApiModelProperty("1 成功 2 失败 3 补批成功 4 补批失败")
private String runCode;
/**
......@@ -154,6 +154,11 @@ public class JobTaskRunLog implements Serializable {
@ApiModelProperty("运行标识")
private String runId;
/**
* 补批运行标识
*/
@ApiModelProperty("补批运行标识")
private String reRunId;
/**
*/
private static final long serialVersionUID = 1L;
}
\ No newline at end of file
byit-myth-core/myth-admin-core/src/main/java/com/byit/model/RunRecording.java
......@@ -126,6 +126,12 @@ public class RunRecording implements Serializable {
*/
@ApiModelProperty("是否是内嵌工作流 0 否， 1 是")
private String isInner;
/**
* 是否执行快速失败 0 否， 1 是
*/
@ApiModelProperty("是否执行快速失败 0 否， 1 是")
private String failFast;
/**
*/
private static final long serialVersionUID = 1L;
......
byit-myth-core/myth-admin-core/src/main/java/com/byit/thread/JobScheduleHelper.java
......@@ -329,8 +329,10 @@ public class JobScheduleHelper{
private Integer saveLog(JobTaskSchedule mythJobTaskSchedule){
JobTaskRunLogWithBLOBs jobTaskRunLog = new JobTaskRunLogWithBLOBs();
jobTaskRunLog.setRunId(mythJobTaskSchedule.getRunId());
jobTaskRunLog.setFlowName(mythJobTaskSchedule.getVersionName());
//jobTaskRunLog.setRunId(mythJobTaskSchedule.getRunId());
jobTaskRunLog.setFlowId(mythJobTaskSchedule.getFlowId());
jobTaskRunLog.setFlowName("1");
jobTaskRunLog.setRunId("qwer-tyui-opas-dfgh");
jobTaskRunLog.setNodeName(mythJobTaskSchedule.getNodeName());
jobTaskRunLog.setRunParams(mythJobTaskSchedule.getRunParam());
jobTaskRunLog.setFailedRemainingCount(mythJobTaskSchedule.getFailedRetryCount());
......
byit-myth-core/myth-admin-core/src/main/java/com/byit/thread/LogScanHelper.java
......@@ -42,8 +42,10 @@ public class LogScanHelper {
}
private final String FAILURE_CODE = "100500";
private final String SUCCESS_CODE = "100200";
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";
/**
* 扫描虚节点的线程是否停止
*/
......@@ -89,7 +91,7 @@ public class LogScanHelper {
notEndVirtualNodes.forEach(notEndVirtualNode ->{
log.debug("------------【开始查询虚拟节点的执行情况】----------------");
//根据运行标识和工作流id查询运行日志
RunRecording runRecordingByFlowIdAndRunId = runRecordingService.findRunRecordingByFlowIdAndRunId(notEndVirtualNode.getFlowId( ), notEndVirtualNode.getRunId( ));
RunRecording runRecordingByFlowIdAndRunId = runRecordingService.findRunRecordingByFlowIdAndRunId(notEndVirtualNode.getFlowId( ), notEndVirtualNode.getRunId());
//判断当前的工作流是否已经完结
if(runRecordingByFlowIdAndRunId!=null && "4".equals(runRecordingByFlowIdAndRunId.getFlowStatus())){
log.debug("------------【查询到有已经完成的虚拟节点修改日志】----------------");
......@@ -107,7 +109,7 @@ public class LogScanHelper {
e.printStackTrace();
}finally {
//释放资源
freedResource(conn,preparedStatement,connAutoCommit);
freedResource(conn,preparedStatement,connAutoCommit,virtualNodeScanIsStop);
}
}
......@@ -133,25 +135,47 @@ public class LogScanHelper {
conn.setAutoCommit(false);
preparedStatement = conn.prepareStatement("SELECT * FROM JOB_LOCK WHERE LOCK_NAME = 'log_node_callback_lock' FOR UPDATE ");
preparedStatement.execute();
System.out.println("--------------火球锁成功------------------" );
//查询的是失败的或者是已经结束的节点（完成的）
List<JobTaskRunLog> jobTaskRunLogEndOrFailureNode = jobTaskRunLogService.findJobTaskRunLogEndOrFailureNode( );
if(CollectionUtil.isNotEmpty(jobTaskRunLogEndOrFailureNode)){
log.debug("-------------【查询到有完结而且未告警的节点】-------------");
Map<String, List<JobTaskRunLog>> jobLogMap = jobTaskRunLogEndOrFailureNode.stream( ).collect(Collectors.groupingBy(JobTaskRunLog::getRunCode));
//假设成功的code码是000200 失败的是000500
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("1").flowStatus("4").build());
successNode.setAlertEnd("1");
jobTaskRunLogService.updateJobTaskRunLog(successNode);
});
}else{
dateAligned(1000);
}
/*if(CollectionUtil.isNotEmpty(failureNodes)){
failureNodes.forEach(failureNode ->{
String runId = failureNode.getRunId( );
Integer flowId = failureNode.getFlowId( );
});
}*/
}else{
dateAligned(20000);
}
......@@ -159,7 +183,7 @@ public class LogScanHelper {
}catch (Exception e){
e.printStackTrace();
}finally {
freedResource(conn,preparedStatement,connAutoCommit);
freedResource(conn,preparedStatement,connAutoCommit,notAlarmedNodeScanIsStop);
}
}
});
......@@ -174,12 +198,12 @@ public class LogScanHelper {
* @param preparedStatement 执行器
* @param connAutoCommit 原来的提交状态
*/
private void freedResource(Connection conn,PreparedStatement preparedStatement,boolean connAutoCommit){
private void freedResource(Connection conn,PreparedStatement preparedStatement,boolean connAutoCommit,boolean isStop){
if(conn != null){
try {
conn.commit();
} catch (SQLException e) {
if(!virtualNodeScanIsStop){
if(!isStop){
log.error("--------------------【提交行锁出错】---------------------");
}
}
......@@ -189,28 +213,33 @@ public class LogScanHelper {
assert conn != null;
conn.setAutoCommit(connAutoCommit);
} catch (SQLException e) {
if(!virtualNodeScanIsStop){
if(!isStop){
log.error("--------------------【恢复自动提交出错】---------------------");
}
}
try {
conn.close();
} catch (SQLException e) {
if(!virtualNodeScanIsStop){
log.error("--------------------【关闭链接出错】---------------------");
}
}
if(preparedStatement != null){
try {
preparedStatement.close();
} catch (SQLException e) {
if(!virtualNodeScanIsStop){
if(!isStop){
log.error("--------------------【关闭执行器出错】---------------------");
}
}
}
try {
conn.close();
} catch (SQLException e) {
if(!isStop){
log.error("--------------------【关闭链接出错】---------------------");
}
}
}
public void stopVirtualNodeScanThread(){
this.virtualNodeScanIsStop = true;
}
/**
......
byit-myth-core/myth-admin-core/src/main/resources/mapper/JobTaskRunLogMapper.xml
......@@ -25,6 +25,7 @@
<result column="alert_end" jdbcType="CHAR" property="alertEnd" />
<result column="job_type" jdbcType="CHAR" property="jobType" />
<result column="run_id" jdbcType="VARCHAR" property="runId" />
<result column="re_run_id" jdbcType="VARCHAR" property="reRunId" />
</resultMap>
<resultMap extends="BaseResultMap" id="ResultMapWithBLOBs" type="com.byit.model.JobTaskRunLogWithBLOBs">
<result column="run_msg" jdbcType="LONGVARCHAR" property="runMsg" />
......@@ -33,7 +34,7 @@
<sql id="Base_Column_List">
log_id, failed_remaining_count, version_name, flow_id, flow_name, job_group_id, handler_name,
node_name, is_virtual, run_code, run_params, start_time, run_type, trigger_code,
trigger_time, job_group_ip, map_flow_id, run_command, end_time, node_id,job_type,alert_end,run_id
trigger_time, job_group_ip, map_flow_id, run_command, end_time, node_id,job_type,alert_end,run_id,re_run_id
</sql>
<sql id="Blob_Column_List">
run_msg, trigger_msg
......@@ -42,7 +43,7 @@
<select id="findJobTaskRunLogEndOrFailureNode" resultMap="BaseResultMap">
select <include refid="Base_Column_List" /> from job_task_run_log
where (node_name='END' or run_code = '000500') and alert_end = '0'
where (node_name='END' or run_code = '2' or run_code = '4') and alert_end = '0'
</select>
<select id="findNotEndVirtualNode" resultMap="BaseResultMap">
......@@ -142,6 +143,9 @@
<if test="runId != null">
run_id,
</if>
<if test="reRunId != null">
re_run_id,
</if>
</trim>
<trim prefix="values (" suffix=")" suffixOverrides=",">
<if test="logId != null">
......@@ -217,7 +221,10 @@
#{triggerMsg,jdbcType=LONGVARCHAR},
</if>
<if test="runId != null">
run_id = #{runId,jdbcType=VARCHAR},
#{runId,jdbcType=VARCHAR},
</if>
<if test="reRunId != null">
#{reRunId,jdbcType=VARCHAR},
</if>
</trim>
</insert>
......@@ -296,6 +303,9 @@
<if test="runId != null">
run_id = #{runId,jdbcType=VARCHAR},
</if>
<if test="reRunId != null">
re_run_id = #{reRunId,jdbcType=VARCHAR},
</if>
</set>
where log_id = #{logId,jdbcType=INTEGER}
</update>
......@@ -368,6 +378,9 @@
<if test="runId != null">
run_id = #{runId,jdbcType=VARCHAR},
</if>
<if test="reRunId != null">
re_run_id = #{reRunId,jdbcType=VARCHAR},
</if>
</set>
where log_id = #{logId,jdbcType=INTEGER}
</update>
......
byit-myth-core/myth-admin-core/src/main/resources/mapper/RunRecordingMapper.xml
......@@ -20,11 +20,12 @@
<result column="end_time" jdbcType="TIMESTAMP" property="endTime" />
<result column="is_alarm" jdbcType="CHAR" property="isAlarm" />
<result column="is_inner" jdbcType="CHAR" property="isInner" />
<result column="fail_fast" jdbcType="CHAR" property="failFast" />
</resultMap>
<sql id="Base_Column_List">
recording_id, run_id, alarm_email, dispatch_ip, flow_name, flow_run_result, flow_status,
flow_timeout, flow_version_name, alarml_action, priority, trigger_time, principal,
flow_id, start_time, end_time, is_alarm,is_inner
flow_id, start_time, end_time, is_alarm,is_inner,fail_fast
</sql>
<select id="findRunRecordingByFlowIdAndRunId" resultMap="BaseResultMap" >
......@@ -103,6 +104,9 @@
<if test="isInner != null">
is_inner,
</if>
<if test="failFast != null">
fail_fast,
</if>
</trim>
<trim prefix="values (" suffix=")" suffixOverrides=",">
<if test="recordingId != null">
......@@ -159,10 +163,12 @@
<if test="isInner != null">
#{isInner,jdbcType=CHAR},
</if>
<if test="failFast != null">
#{fail_fast,jdbcType=CHAR},
</if>
</trim>
</insert>
<update id="updateRunRecordingById" parameterType="com.byit.model.RunRecording">
<!-- generated @mbg.generated date: 2019-12-25 -->
update run_recording
<set>
<if test="runId != null">
......@@ -216,6 +222,9 @@
<if test="isInner != null">
is_inner = #{isInner,jdbcType=CHAR},
</if>
<if test="failFast != null">
fail_fast = #{failFast,jdbcType=CHAR},
</if>
</set>
where recording_id = #{recordingId,jdbcType=INTEGER}
</update>
......@@ -274,6 +283,9 @@
<if test="isInner != null">
is_inner = #{isInner,jdbcType=CHAR},
</if>
<if test="failFast != null">
fail_fast = #{failFast,jdbcType=CHAR},
</if>
</set>
where flow_id = #{flowId,jdbcType=INTEGER} and run_id = #{runId,jdbcType=VARCHAR}
</update>
......
byit-myth-core/myth-core-common/src/main/java/com/byit/job/enums/JobResultEnum.java
......@@ -5,20 +5,22 @@ package com.byit.job.enums;
* @author huangfu
*/
public enum JobResultEnum implements IEnum {
SUCCESS("100200","任务执行成功"),
FAIL("100500","任务执行失败"),
FAIL_TIMEOUT("100502","超时错误"),
DISPATCH_SUCCESS("200200","调度成功"),
DISPATCH_FAIL("200500","调度失败");
SUCCESS("100200","任务执行成功","1"),
FAIL("100500","任务执行失败","2"),
FAIL_TIMEOUT("100502","超时错误","2"),
DISPATCH_SUCCESS("200200","调度成功","1"),
DISPATCH_FAIL("200500","调度失败","2");
private String code;
private String msg;
private String res;
JobResultEnum() {
}
JobResultEnum(String code, String msg) {
JobResultEnum(String code, String msg,String res) {
this.code = code;
this.msg = msg;
this.res = res;
}
......@@ -31,4 +33,7 @@ public enum JobResultEnum implements IEnum {
public String getMsg() {
return this.msg;
}
public String getRes() {
return this.res;
}
}
\ No newline at end of file
byit-myth-core/myth-core-common/src/main/java/com/byit/job/vo/ReturnResult.java
......@@ -18,9 +18,9 @@ import java.io.Serializable;
@NoArgsConstructor
public class ReturnResult<T> implements Serializable {
public static final long serialVersionUID = 1573630876693L;
public static final ReturnResult SUCCESS = new ReturnResult(null);
public static final ReturnResult FAIL = new ReturnResult(JobResultEnum.FAIL.getCode(), JobResultEnum.FAIL.getMsg());
public static final ReturnResult FAIL_TIMEOUT = new ReturnResult(JobResultEnum.FAIL_TIMEOUT.getCode(),JobResultEnum.FAIL_TIMEOUT.getMsg());
public static final ReturnResult<String> SUCCESS = new ReturnResult<String>(null);
public static final ReturnResult FAIL = new ReturnResult(JobResultEnum.FAIL.getRes(), JobResultEnum.FAIL.getMsg());
public static final ReturnResult FAIL_TIMEOUT = new ReturnResult(JobResultEnum.FAIL_TIMEOUT.getRes(),JobResultEnum.FAIL_TIMEOUT.getMsg());
private String code;
private String msg;
private T content;
......@@ -39,8 +39,8 @@ public class ReturnResult<T> implements Serializable {
* 默认就是成功
* @param content
*/
public ReturnResult(T content) {
this.code = JobResultEnum.SUCCESS.getCode();
private ReturnResult(T content) {
this.code = JobResultEnum.SUCCESS.getRes();
this.msg = JobResultEnum.SUCCESS.getMsg();
this.content = content;
}
......
byit-myth-executor/myth-exector-plugin/src/main/java/com/byit/rpc/RunJobThread.java
......@@ -20,7 +20,7 @@ import java.util.Date;
public class RunJobThread implements Runnable {
private AdminSenPluginDto adminSenPluginDto;
public RunJobThread(AdminSenPluginDto adminSenPluginDto) {
RunJobThread(AdminSenPluginDto adminSenPluginDto) {
this.adminSenPluginDto = adminSenPluginDto;
}
......
demo-client/byit-demo-client/src/main/java/com/byit/job/Mains.java
......@@ -16,13 +16,13 @@ import java.io.IOException;
public class Mains {
public static void main(String[] args) throws IOException {
new JobRunServerLauncher(8888);
String plServerUrl = "http://10.0.55.237:8888";
String plServerUrl = "http://10.0.55.238:8888";
String mythCron = "时间";
String routingStrategy = LoadBalance.ROUND.name();
String blockingStrategy = "阻塞策略";
String callbackToken = "dsasadsad";
String gatewayToken="asdsadsa";
String requestIP = "10.0.55.237";
String requestIP = "10.0.55.238";
String requestPort="8080";
String param="不延迟任务";
String name="addJob";
......
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