Merge remote-tracking branch 'origin/developer' into developer

byit-myth-admin/src/main/resources/config/application-dev.yml

@@ -56,6 +56,7 @@ myth-job:
 
 fdfs:
   so-timeout: 1500
+  http-port: 88
   connect-timeout: 600
   pool:
     jmx-enabled: false

byit-myth-admin/src/main/resources/config/application-local.yml

@@ -60,6 +60,7 @@ myth-job:
 fdfs:
   so-timeout: 1500
   connect-timeout: 600
+  http-port: 88
   pool:
     jmx-enabled: false
   tracker-list:

byit-myth-admin/src/main/resources/config/application-pro.yml

@@ -56,6 +56,7 @@ myth-job:
 fdfs:
   so-timeout: 1500
   connect-timeout: 600
+  http-port: 8888
   pool:
     jmx-enabled: false
   tracker-list:

byit-myth-admin/src/main/resources/config/application-test.yml

@@ -57,6 +57,7 @@ myth-job:
 fdfs:
   so-timeout: 1500
   connect-timeout: 600
+  http-port: 8888
   pool:
     jmx-enabled: false
   tracker-list:

byit-myth-core/myth-admin-core/src/main/java/com/byit/conf/RedissonAutoConfiguration.java

@@ -4,6 +4,7 @@ import com.byit.conf.properties.RedissonProperties;
 import com.byit.util.lock.DistributedLocker;
 import com.byit.util.lock.RedissLockUtil;
 import com.byit.util.lock.RedissonDistributedLocker;
+import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.lang3.StringUtils;
 import org.redisson.Redisson;
 import org.redisson.api.RedissonClient;
@@ -24,6 +25,7 @@ import org.springframework.context.annotation.Configuration;
 @Configuration
 @ConditionalOnClass(Config.class)
 @EnableConfigurationProperties(RedissonProperties.class)
+@Slf4j
 public class RedissonAutoConfiguration {
     @Autowired
     private RedissonProperties redssionProperties;
@@ -79,7 +81,12 @@ public class RedissonAutoConfiguration {
     DistributedLocker distributedLocker(RedissonClient redissonSingle) {
         RedissonDistributedLocker locker = new RedissonDistributedLocker();
         locker.setRedissonClient(redissonSingle);
+        log.info("----------redis分布式锁{}，被设置加载---------",locker);
         RedissLockUtil.setLocker(locker);
+        DistributedLocker redissLock = RedissLockUtil.getRedissLock();
+        if(redissLock == null) {
+            throw new RuntimeException("redis分布式锁设置异常，请联系调度中心开发团队！");
+        }
         return locker;
     }
 

byit-myth-core/myth-admin-core/src/main/java/com/byit/flowservice/impl/NodeVerificationImpl.java

@@ -3,6 +3,7 @@ package com.byit.flowservice.impl;
 import cn.hutool.core.collection.CollectionUtil;
 import com.byit.enums.NodePropertyEnum;
 import com.byit.enums.NodeRunStatusPropertyEnum;
+import com.byit.enums.ScheduleTypeEnum;
 import com.byit.exceptions.SuperiorNodeRunException;
 import com.byit.flowservice.NodeVerification;
 import com.byit.job.exceptions.BusinessException;
@@ -54,6 +55,10 @@ public class NodeVerificationImpl implements NodeVerification {
     public boolean superiorNodeStatus(JobTask thisJobTask) {
         //获取该节点的运行标识
         String runId = thisJobTask.getRunId();
+        if(ScheduleTypeEnum.REPEAT.getCode().equals(thisJobTask.getScheduleType())) {
+            runId = thisJobTask.getReRunId();
+        }
+
         //查询该节点的依赖节点
         String nodeDepend = thisJobTask.getNodeDepend();
         if(StringUtils.isBlank(nodeDepend)){

byit-myth-core/myth-admin-core/src/main/java/com/byit/job/WorkRoulette.java

@@ -29,6 +29,7 @@ public class WorkRoulette {
 
     public static void addJob(TimerTask timerTask,long triggerNextTime) {
         log.debug("-----任务{}毫秒后执行-------",triggerNextTime-System.currentTimeMillis());
+        //设置这个的根部原因是因为保证任务的抛出在事务提交动作完成之后抛出
         long time = triggerNextTime-System.currentTimeMillis();
         if(time < INT){
             time = INT;

byit-myth-core/myth-admin-core/src/main/java/com/byit/model/RunRecording.java

@@ -76,7 +76,7 @@ public class RunRecording implements Serializable {
     /**
      * 当前工作流版本的告警的时机(0 不告警， 1 完成时告警， 2 失败时告警， 3 成功时告警)
      */
-    @ApiModelProperty("当前工作流版本的告警的时机(0 不告警， 1 完成时告警， 3 失败时告警， 2 成功时告警)")
+    @ApiModelProperty("当前工作流版本的告警的时机(0 不告警， 1 完成时告警， 2 失败时告警， 3 成功时告警)")
     private String alarmlAction;
 
     /**

byit-myth-core/myth-admin-core/src/main/java/com/byit/service/mapservice/impl/RunRecordingAndEmailServiceImpl.java

@@ -2,7 +2,6 @@ package com.byit.service.mapservice.impl;
 
 import cn.hutool.core.collection.CollectionUtil;
 import cn.hutool.core.date.DateUtil;
-import com.byit.filesystem.FileSystem;
 import com.byit.model.EmailAlarm;
 import com.byit.model.JobTaskRunLogWithBLOBs;
 import com.byit.model.RunRecording;
@@ -11,12 +10,12 @@ import com.byit.service.JobTaskRunLogService;
 import com.byit.service.RunRecordingService;
 import com.byit.service.mapservice.RunRecordingAndEmailService;
 import com.byit.util.TimeFormatUtil;
+import com.github.tobato.fastdfs.conn.TrackerConnectionManager;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.lang3.StringUtils;
+import org.springframework.beans.factory.annotation.Value;
 import org.springframework.stereotype.Component;
 
-import java.io.IOException;
-import java.nio.charset.StandardCharsets;
 import java.util.List;
 
 /**
@@ -35,20 +34,20 @@ public class RunRecordingAndEmailServiceImpl implements RunRecordingAndEmailServ
      */
     private final JobTaskRunLogService jobTaskRunLogService;
     /**
-     * 文件系统
-     */
-    private final FileSystem fileSystem;
-    /**
      * 运行实例操作
      */
     private final RunRecordingService runRecordingService;
 
+    private final TrackerConnectionManager trackerConnectionManager;
+    @Value("${fdfs.http-port}")
+    private String trackerIpsPort;
+
     public RunRecordingAndEmailServiceImpl(EmailAlarmService emailAlarmService, JobTaskRunLogService jobTaskRunLogService,
-                                           FileSystem fileSystem, RunRecordingService runRecordingService) {
+                                           RunRecordingService runRecordingService, TrackerConnectionManager trackerConnectionManager) {
         this.emailAlarmService = emailAlarmService;
         this.jobTaskRunLogService = jobTaskRunLogService;
-        this.fileSystem = fileSystem;
         this.runRecordingService = runRecordingService;
+        this.trackerConnectionManager = trackerConnectionManager;
     }
 
     @Override
@@ -91,7 +90,7 @@ public class RunRecordingAndEmailServiceImpl implements RunRecordingAndEmailServ
                 .append("<th width = '10%'>结束时间</th>")
                 .append("<th width = '10%'>耗费时间</th>")
                 .append("<th width = '10%'>运行结果</th>")
-                .append("<th width = '50%'>运行日志</th>")
+                .append("<th width = '50%'>日志远程路径</th>")
                 .append("</thead>")
                 .append("<tbody>");
 
@@ -103,19 +102,19 @@ public class RunRecordingAndEmailServiceImpl implements RunRecordingAndEmailServ
                  * String logFilePath = FILE_SYSTEM_PRE+fileSystemIp+":"+fileSystemPort+"/"+jobTaskRunLog.getLogRemotelyPath();
                  * String logFileUrl = String.format(iframeHtml,logFilePath);
                  */
-                String logStr = null;
-                try {
+                String logStr = "无远程路径";
                 if(StringUtils.isNotBlank(jobTaskRunLog.getLogRemotelyPath())){
-                        logStr = new String(fileSystem.downloaderFile(jobTaskRunLog.getLogRemotelyPath()), StandardCharsets.UTF_8);
-                    }else if(StringUtils.isNotBlank(jobTaskRunLog.getRunMsg())) {
-                        logStr = jobTaskRunLog.getRunMsg();
-                    } else if (StringUtils.isNotBlank(jobTaskRunLog.getTriggerMsg())){
-                        logStr = jobTaskRunLog.getTriggerMsg();
-                    } else {
-                        logStr = "未知异常，出现此错误属于系统错误，请联系myth-job-admin开发团队！";
+                    logStr = jobTaskRunLog.getLogRemotelyPath();
+                    List<String> trackerList = trackerConnectionManager.getTrackerList();
+                    if (CollectionUtil.isNotEmpty(trackerList)) {
+                        for (String ipPort : trackerList) {
+                            String ip = ipPort.split(":")[0];
+                            String baseUrl = "http://"+ip+":"+trackerIpsPort+"/";
+                            logStr += baseUrl + logStr+"\n\n";
+                            log.debug("--------log远程路径为:{}------",logStr);
                         }
-                } catch (IOException e) {
-                    e.printStackTrace();
+                    }
+
                 }
                 stringBuilder.append("<tr align='center'>")
                         .append(String.format("<td>%s</td>", jobTaskRunLog.getNodeName()))
@@ -127,7 +126,7 @@ public class RunRecordingAndEmailServiceImpl implements RunRecordingAndEmailServ
                                 : "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", logStr==null?jobTaskRunLog.getRunMsg():logStr))
+                        .append(String.format("%s", logStr))
                         .append("</div>")
                         .append("</td></tr>");
 

byit-myth-core/myth-admin-core/src/main/java/com/byit/thread/helper/EndAndNotWarningThreadRunHelper.java

@@ -7,6 +7,7 @@ import com.byit.service.RunRecordingService;
 import com.byit.service.mapservice.RunRecordingAndEmailService;
 import com.byit.thread.BaseThreadRunHelper;
 import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Value;
 import org.springframework.stereotype.Component;
 
 import javax.sql.DataSource;
@@ -32,11 +33,11 @@ public class EndAndNotWarningThreadRunHelper extends BaseThreadRunHelper {
     /**
      * 失败时告警
      */
-    public static final String FAILURE_DONE = "3";
+    public static final String FAILURE_DONE = "2";
     /**
      * 成功时告警
      */
-    public static final String SUCCESS_DONE = "2";
+    public static final String SUCCESS_DONE = "3";
     /**
      * 运行实例操作
      */
@@ -84,11 +85,11 @@ public class EndAndNotWarningThreadRunHelper extends BaseThreadRunHelper {
                         break;
                     default:
                         log.debug("------工作流{},告警类别不告警-----",runRecording);
+                        break;
+                }
                 //修改为已告警
                 runRecording.setIsAlarm("0");
                 runRecordingService.updateRunRecordingById(runRecording);
-                        break;
-                }
             });
         }else{
             return UNIVERSAL_WAIT_TIME;

byit-myth-core/myth-admin-core/src/main/java/com/byit/thread/helper/JobTaskThreadRunHelper.java

@@ -16,6 +16,7 @@ import com.byit.service.mapservice.TaskAndLogServer;
 import com.byit.service.mapservice.TaskAndScheduleService;
 import com.byit.thread.BaseThreadRunHelper;
 import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.lang3.StringUtils;
 import org.springframework.beans.BeanUtils;
 import org.springframework.stereotype.Component;
 
@@ -115,8 +116,12 @@ public class JobTaskThreadRunHelper extends BaseThreadRunHelper {
                     }else {
                         //普通节点
                         try {
+                            //如果是补批节点 而且没有依赖节点的话，按照开始节点处理
+                            if(ScheduleTypeEnum.REPAIR.getCode().equals(jobTask.getScheduleType()) &&
+                                    StringUtils.isBlank(jobTask.getNodeDepend())){
+                                startNodeOperating(jobTask,jobTaskSchedules);
+                            }else if (nodeVerification.superiorNodeStatus(jobTask)) {
                                 //断定上级节点是否执行成功
-                            if (nodeVerification.superiorNodeStatus(jobTask)) {
                                 runJobTask(jobTask,jobTaskSchedules);
                             }
                         }catch (SuperiorNodeRunException se) {

byit-myth-core/myth-admin-core/src/main/java/com/byit/util/lock/RedissLockUtil.java

@@ -10,7 +10,11 @@ public class RedissLockUtil {
     private static DistributedLocker redissLock;
     
     public static void setLocker(DistributedLocker locker) {
-        redissLock = locker;
+        RedissLockUtil.redissLock = locker;
+    }
+
+    public static DistributedLocker getRedissLock(){
+        return RedissLockUtil.redissLock;
     }
     
     public static void lock(String lockKey) {