游离节点日志追加

byit-myth-admin/src/main/java/com/byit/api/ApiCallbackController.java

@@ -5,6 +5,7 @@ import com.byit.dto.api.JavaCallbackLogDto;
 import com.byit.dto.executor.JobRunResultDto;
 import com.byit.dto.web.ReturnResult;
 import com.byit.packet.response.PluginRpcResponsePacket;
+import com.byit.thread.JavaFlowFreeTaskCallbackThread;
 import com.byit.thread.JavaTaskCallbackThread;
 import com.byit.thread.LogCallbackThread;
 import io.swagger.annotations.Api;
@@ -27,6 +28,11 @@ public class ApiCallbackController {
         MythJobAutoConfigure.LOG_CALLBACK.execute(new JavaTaskCallbackThread(pluginRpcResponsePacket));
     }
 
+    @PostMapping(value = "flowFreeNodeCallbackRes")
+    public void flowFreeNodeCallbackRes(@RequestBody PluginRpcResponsePacket pluginRpcResponsePacket) {
+        MythJobAutoConfigure.LOG_CALLBACK.execute(new JavaFlowFreeTaskCallbackThread(pluginRpcResponsePacket));
+    }
+
     /**
      * 供ddmp平台回调
      * @param javaCallbackLogDto

byit-myth-core/myth-admin-core/src/main/java/com/byit/mapper/RunRecordingMapper.java

@@ -25,6 +25,8 @@ public interface RunRecordingMapper {
      * @return 一个时间段的数据
      */
     List<RunRecording> findThisDayRunRecording(StatisticsConditionDto statisticsConditionDto);
+
+    RunRecording runIngRunRecording(String flowName);
     /**
      * 查询全部的数据
      * @return

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

@@ -215,7 +215,7 @@ public class JobTaskSchedule implements Serializable {
     /**
      * 跑批类型  1 正常跑批   2 重跑   3  补批  4  插件端立即运行  5  类quartz任务执行
      */
-    @ApiModelProperty("跑批类型  1 正常跑批   2 重跑   3  补批  4  插件端立即运行  5  类quartz任务执行")
+    @ApiModelProperty("跑批类型  1 正常跑批   2 重跑   3  补批  4  插件端立即运行  5  类quartz任务执行 6 游离节点")
     private Integer scheduleType;
 
     /**

byit-myth-core/myth-admin-core/src/main/java/com/byit/service/RunRecordingService.java

@@ -12,6 +12,8 @@ import java.util.List;
  */
 public interface RunRecordingService {
 
+    RunRecording runIngRunRecording(String flowName);
+
     /**
      * 查询当天的实例  查询一个时间段的范围
      * @param startTime 开始时间

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

@@ -80,7 +80,7 @@ public class JobTaskRunLogServiceImpl implements JobTaskRunLogService {
 
     @Override
     public List<RunLogVo> findAllRunLogIdByFlowIdAndRunId(Integer flowId, String runId) {
-        List<JobTaskRunLogWithBLOBs> jobTaskRunLogWithBLOBsByFlowIdAndRunId = jobTaskRunLogMapper.findJobTaskRunLogWithBLOBsByFlowIdAndRunId(flowId, runId);
+        List<JobTaskRunLogWithBLOBs> jobTaskRunLogWithBLOBsByFlowIdAndRunId = findJobTaskRunLogWithBLOBsByFlowIdAndRunId(flowId, runId);
         List<RunLogVo> collect = jobTaskRunLogWithBLOBsByFlowIdAndRunId.stream().map(log -> {
             RunLogVo runLogVo = new RunLogVo();
             runLogVo.setHasTrigger(true);

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

@@ -35,6 +35,11 @@ public class RunRecordingServiceImpl implements RunRecordingService  {
         this.runRecordingMapper = runRecordingMapper;
     }
 
+    @Override
+    public RunRecording runIngRunRecording(String flowName) {
+        return runRecordingMapper.runIngRunRecording(flowName);
+    }
+
     /**
      * 查询当天的数据
      * @param startTime 开始时间

byit-myth-core/myth-admin-core/src/main/java/com/byit/task/JavaNodeExecutorTask.java

@@ -31,7 +31,7 @@ import java.util.stream.Collectors;
 import static com.alibaba.fastjson.serializer.SerializerFeature.WriteClassName;
 
 /**
- * java节点执行
+ * java节点执行 关联工作流
  *
  * @author huangfu
  */

byit-myth-core/myth-admin-core/src/main/java/com/byit/task/JavaTaskJobTask.java

@@ -3,6 +3,7 @@ package com.byit.task;
 import com.alibaba.fastjson.JSON;
 import com.byit.conf.MythJobAutoConfigure;
 import com.byit.dto.executor.RunParamWrapped;
+import com.byit.dto.personalise.JavaJobConfDTO;
 import com.byit.dto.plugin.JavaTask;
 import com.byit.enums.ScheduleTypeEnum;
 import com.byit.enums.task.RunResultEnum;
@@ -25,7 +26,7 @@ import java.util.TreeSet;
 import java.util.stream.Collectors;
 
 /**
- * 立即运行  java单任务 java节点的执行器
+ * 立即运行  java单任务 java节点的执行器   不关联任何的工作流任务
  *
  * @author huangfu
  */
@@ -33,6 +34,7 @@ public class JavaTaskJobTask implements TimerTask {
     public static final String JAVA_SYNC = "JAVA_SYNC";
     public static final String HTTP_PRE = "http://";
     public static final String HTTP_SUFFIX = "/myth-job-admin/api/callback/callbackRes";
+    public static final String HTTP_SUFFIX_FLOW = "/myth-job-admin/api/callback/flowFreeNodeCallbackRes";
     private final JavaTask javaTask;
 
     public JavaTaskJobTask(JavaTask javaTask) {
@@ -49,11 +51,20 @@ public class JavaTaskJobTask implements TimerTask {
         RunJavaServiceImpl service = SpringUtil.getBean(RunJavaServiceImpl.class);
         PluginRpcRequestPacket request = new PluginRpcRequestPacket();
         String param = javaTask.getParam();
+        String personalise = javaTask.getPersonalise();
+        JavaJobConfDTO javaJobConfDTO = JSON.parseObject(personalise, JavaJobConfDTO.class);
+        String hasRelatedFlow = javaJobConfDTO.getHasRelatedFlow();
         //保存到日志
         Integer logId = saveLog(javaTask);
         GetRegConfig getRegConfig = SpringUtil.getBean(GetRegConfig.class);
         TreeSet<String> gatewayConf = getRegConfig.getGatewayConf();
-        List<String> callUrlList = gatewayConf.stream().map(callbackIp -> String.format("%s%s%s", HTTP_PRE, callbackIp, HTTP_SUFFIX)).collect(Collectors.toList());
+        List<String> callUrlList = gatewayConf.stream().map(callbackIp -> {
+            if("2".equals(hasRelatedFlow)){
+                return String.format("%s%s%s", HTTP_PRE, callbackIp, HTTP_SUFFIX);
+            }else{
+                return String.format("%s%s%s", HTTP_PRE, callbackIp, HTTP_SUFFIX_FLOW);
+            }
+        }).collect(Collectors.toList());
 
         request.setCallbackUrl(JSON.toJSONString(callUrlList));
         CommunicationParam communicationParam = new CommunicationParam();

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

+package com.byit.thread;
+
+import com.alibaba.fastjson.JSON;
+import com.byit.dto.specials.SpecialJavaNode;
+import com.byit.dto.web.ReturnResult;
+import com.byit.enums.JobResultEnum;
+import com.byit.enums.task.RunResultEnum;
+import com.byit.filesystem.FileSystem;
+import com.byit.model.JobTaskRunLogWithBLOBs;
+import com.byit.model.RunRecording;
+import com.byit.model.vo.RunLogVo;
+import com.byit.packet.response.PluginRpcResponsePacket;
+import com.byit.service.RunRecordingService;
+import com.byit.service.impl.JobTaskRunLogServiceImpl;
+import com.byit.util.SpringUtil;
+import com.byit.utils.ValidationUtil;
+import lombok.SneakyThrows;
+import org.apache.commons.lang3.StringUtils;
+
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.util.Date;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * java任务 游离的工作流节点任务回调线程
+ *
+ * @author huangfu
+ */
+public class JavaFlowFreeTaskCallbackThread implements Runnable {
+    public static final String CODE = "code";
+    public static final String MSG = "msg";
+    public static final String CONTENT = "content";
+    private PluginRpcResponsePacket pluginRpcResponsePacket;
+
+    public JavaFlowFreeTaskCallbackThread(PluginRpcResponsePacket pluginRpcResponsePacket) {
+        this.pluginRpcResponsePacket = pluginRpcResponsePacket;
+    }
+
+    @SneakyThrows
+    @Override
+    public void run() {
+        String logIdStr = pluginRpcResponsePacket.getExtension();
+        int logId = Integer.parseInt(logIdStr);
+        JobTaskRunLogServiceImpl mythJobTaskRunLogService = SpringUtil.getBean(JobTaskRunLogServiceImpl.class);
+        Map<String,String> result = new HashMap<>(2);
+        Object returnReultObject = pluginRpcResponsePacket.getResult();
+        if(returnReultObject instanceof ReturnResult) {
+            ReturnResult<String> returnResult = (ReturnResult) returnReultObject;
+            result.put(CODE,returnResult.getCode());
+            result.put(MSG,returnResult.getMsg());
+            result.put(CONTENT,returnResult.getContent());
+        }else{
+            result = (Map<String,String>)returnReultObject;
+        }
+
+        JobTaskRunLogWithBLOBs jobTaskRunLog = new JobTaskRunLogWithBLOBs();
+        String code= result.get(CODE);
+        if(JobResultEnum.SUCCESS.getCode().equals(code)){
+            code = RunResultEnum.RUN_SUCCESS.getCode();
+        }else if(JobResultEnum.DISPATCH_SUCCESS.getCode().equals(code)){
+            code = RunResultEnum.JOB_RUN_ING.getCode();
+        }else {
+            code = RunResultEnum.RUN_ERROR.getCode();
+        }
+
+        String logUrl = pluginRpcResponsePacket.getExtension1();
+        if(StringUtils.isNoneBlank(logUrl)){
+            jobTaskRunLog.setLogRemotelyPath(logUrl);
+        }
+        jobTaskRunLog.setRunMsg(result.get(MSG));
+        jobTaskRunLog.setRunCode(code);
+        jobTaskRunLog.setLogId(logId);
+        jobTaskRunLog.setEndTime(new Date());
+
+
+        String content = result.get(CONTENT);
+        if(StringUtils.isNoneBlank(content)){
+            SpecialJavaNode specialJavaNode = JSON.parseObject(content, SpecialJavaNode.class);
+            if(specialJavaNode != null) {
+                String flowName = specialJavaNode.getFlowName();
+                String nodeName = specialJavaNode.getNodeName();
+                String thisLogUrlPath = specialJavaNode.getThisLogUrlPath();
+
+                String logTemple = "该节点为工作流 【%s】 下的游离节点【%s】，属于二次运行：\n 本次调度信息:%s;\n 上次运行信息日志存放路径为：【%s】";
+                if(StringUtils.isBlank(flowName) || StringUtils.isBlank(nodeName) || StringUtils.isBlank(thisLogUrlPath)){
+                    logTemple = "该节点为工作流 【%s】 下的游离节点【%s】，属于二次运行：\n 本次调度信息:%s;\n 上次运行信息日志存放路径为：【%s】 " +
+                            "\n\n 但是日志回写过程中发现回写参数中未携带【工作流名称】或者【节点名称】或者【日志路径信息】，本次日志追加失败！";
+                    String logFormat = String.format(logTemple, flowName, nodeName, thisLogUrlPath);
+                    jobTaskRunLog.setRunMsg(logFormat);
+                }else{
+                    String logFormat = String.format(logTemple, flowName, nodeName, thisLogUrlPath);
+                    jobTaskRunLog.setRunMsg(logFormat);
+                    JobTaskRunLogWithBLOBs jobTaskRunLogWithBLOBs = buildJobTaskRunLogWithBLOBs(specialJavaNode);
+                    if(jobTaskRunLogWithBLOBs != null) {
+                        mythJobTaskRunLogService.saveJobTaskRunLog(jobTaskRunLogWithBLOBs);
+                    }else {
+                        logTemple = "%s\n\n 日志追加失败，失败原因是没有找到正确的节点！";
+                        jobTaskRunLog.setRunMsg(String.format(logTemple,jobTaskRunLog.getRunMsg()));
+                    }
+
+                }
+            }
+        }
+        mythJobTaskRunLogService.updateJobTaskRunLogWithBLOBs(jobTaskRunLog);
+    }
+
+    public JobTaskRunLogWithBLOBs buildJobTaskRunLogWithBLOBs(SpecialJavaNode specialJavaNode) throws IOException {
+        RunRecordingService bean = SpringUtil.getBean(RunRecordingService.class);
+        JobTaskRunLogServiceImpl mythJobTaskRunLogService = SpringUtil.getBean(JobTaskRunLogServiceImpl.class);
+        String flowName = specialJavaNode.getFlowName();
+
+        RunRecording runRecording = bean.runIngRunRecording(flowName);
+        List<JobTaskRunLogWithBLOBs> jobTaskRunLogWithBLOBsByFlowIdAndRunId = mythJobTaskRunLogService.findJobTaskRunLogWithBLOBsByFlowIdAndRunId(runRecording.getFlowId(), runRecording.getRunId());
+        JobTaskRunLogWithBLOBs jobTaskRunLogWithBLOBs = jobTaskRunLogWithBLOBsByFlowIdAndRunId.stream().filter(log -> log.getNodeName().equals(specialJavaNode.getNodeName())).findFirst().orElse(null);
+        if(jobTaskRunLogWithBLOBs != null) {
+            String logPath = mergeLog(jobTaskRunLogWithBLOBs.getLogRemotelyPath(), specialJavaNode.getThisLogUrlPath());
+            jobTaskRunLogWithBLOBs.setLogRemotelyPath(logPath);
+            return jobTaskRunLogWithBLOBs;
+        }
+        return null;
+    }
+
+    public String mergeLog(String sourceLogPath,String targetLogPath) throws IOException {
+        FileSystem fileSystem = SpringUtil.getBean(FileSystem.class);
+        byte[] sourceLogBytes = fileSystem.downloaderFile(sourceLogPath);
+        String interval = "\n\n\n 游离节点日志：\n";
+        byte[] intervalByte = stringToByteArray(interval);
+        sourceLogBytes = mergeFile(sourceLogBytes, intervalByte);
+        byte[] targetLogBytes = fileSystem.downloaderFile(targetLogPath);
+        byte[] bytes = mergeFile(sourceLogBytes, targetLogBytes);
+        Map<String, String> fileMate = fileSystem.getFileMate(sourceLogPath);
+        return fileSystem.uploadFile(bytes, "log", fileMate);
+    }
+
+    /**
+     * 将字符串转换我数组
+     * @param logData 日志字符串
+     * @return 转换的字节数组
+     */
+    private byte[] stringToByteArray(String logData) {
+        return logData.getBytes(StandardCharsets.UTF_8);
+    }
+
+    /**
+     * 合并两个字节数组
+     * @param sourceByte 字节数组1
+     * @param targetByte 字节数组2
+     * @return 合并后的字节数组
+     */
+    private byte[] mergeFile(byte[] sourceByte, byte[] targetByte){
+        byte[] result = new byte[sourceByte.length+targetByte.length];
+        System.arraycopy(sourceByte,0,result,0,sourceByte.length);
+        System.arraycopy(targetByte,0,result,sourceByte.length,targetByte.length);
+        return result;
+    }
+}

byit-myth-core/myth-admin-core/src/main/resources/mapper/RunRecordingMapper.xml

@@ -315,7 +315,15 @@
     ) flow
   </select>
 
-  <delete id="deleteById">
+  <!--查询运行中的实例   根据工作流名称-->
+  <select id="runIngRunRecording"  resultMap="BaseResultMap">
+    select
+    <include refid="Base_Column_List" />
+    from run_recording
+    where (flow_status = '2' or flow_status = '3') and schedule_type != 4 and flow_name = #{flowName}
+  </select>
+
+    <delete id="deleteById">
     <!-- generated @mbg.generated date: 2019-12-25 -->
     delete from run_recording
     where recording_id = #{recordingId,jdbcType=INTEGER}

byit-myth-core/myth-dto-core/src/main/java/com/byit/dto/personalise/JavaJobConfDTO.java

@@ -28,4 +28,8 @@ public class JavaJobConfDTO implements Serializable {
      * 任务的时间单位
      */
     private TimeUnit timeUnit;
+    /**
+     * 是否关联工作流  关联1  不关联  2
+     */
+    private String hasRelatedFlow = "2";
 }

byit-myth-executor/myth-executor-server/src/main/java/com/byit/node/impl/LogDataProcessingMachine.java

@@ -47,7 +47,7 @@ public class LogDataProcessingMachine implements ProcessingMachine {
             String logPath;
             byte[] logDataByte = stringToByteArray(scriptDto.getRunLogData());
             Map<String,String> fileMateData = new HashMap<>(2);
-            fileMateData.put(FILE_NAME,scriptDto.getRunId()+scriptDto.getRunId()+ LOG_RETOUCH_SUFFIX);
+            fileMateData.put(FILE_NAME,scriptDto.getRunId()+ LOG_RETOUCH_SUFFIX);
 
             //不为空 则追加
             if(scriptDto.getLogRemotePath() != null){