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

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

 package com.byit.api;
 
+import com.byit.model.vo.RunRecordingVo;
 import com.byit.service.ApiFlowService;
 import com.byit.service.FlowService;
 import io.swagger.annotations.Api;
@@ -10,6 +11,7 @@ import org.springframework.web.bind.annotation.RestController;
 
 import javax.annotation.Resource;
 import java.text.ParseException;
+import java.util.List;
 
 /**
  * @description: 工作流操作的API接口
@@ -55,6 +57,7 @@ public class ApiFlowController {
         return "SUCCESS";
     }
 
+
     @PostMapping("killJob")
     @ApiOperation("杀死节点")
     public Boolean killJob(String param) throws InterruptedException {
@@ -103,4 +106,50 @@ public class ApiFlowController {
         return "SUCCESS";
     }
 
+    /**
+     *
+     * @param param  startTime
+     *               endTime
+     *               flowName
+     *               workspaceName
+     * @return
+     */
+    @PostMapping("loadScheduleResult")
+    @ApiOperation("加载运行记录")
+    public List<RunRecordingVo> loadScheduleResult(String param){
+        List<RunRecordingVo> result = apiFlowService.loadScheduleResult(param);
+        return result;
+    }
+
+    @PostMapping("getLogUrl")
+    @ApiOperation("获取日志文件的url地址")
+    public String getLogUrl(String param){
+        String logUrl = apiFlowService.getLogUrl(param);
+        return logUrl;
+    }
+
+    /**
+     * 补批节点
+     * @param param
+     * @return
+     */
+    @PostMapping("/repairJob")
+    @ApiOperation("补批")
+    public String repairJob(String param){
+        apiFlowService.repairJob(param);
+        return "SUCCESS";
+    }
+
+    /**
+     * 补批工作流
+     * @param param
+     * @return
+     */
+    @PostMapping("/repairFlow")
+    @ApiOperation("补批工作流")
+    public String repairFlow(String param){
+        apiFlowService.repairFlow(param);
+        return "SUCCESS";
+    }
+
 }

byit-myth-admin/src/main/java/com/byit/service/ApiFlowService.java

 package com.byit.service;
 
 import com.byit.job.dto.plugin.PluginFlow;
+import com.byit.model.vo.RunRecordingVo;
 
 import java.text.ParseException;
+import java.util.List;
 
 /**
  * @description: 工作流的api请求业务处理接口
@@ -50,4 +52,30 @@ public interface ApiFlowService {
      * @param param
      */
     void reRunFlow(String param);
+
+    /**
+     * 加载运行记录
+     * @param param
+     * @return
+     */
+    List<RunRecordingVo> loadScheduleResult(String param);
+
+    /**
+     * 补批
+     * @param param
+     */
+    void repairJob(String param);
+
+    /**
+     * 补批工作流
+     * @param param
+     */
+    void repairFlow(String param);
+
+    /**
+     * 获取运行日志的存放地址
+     * @param param
+     * @return
+     */
+    String getLogUrl(String param);
 }

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

@@ -13,6 +13,7 @@ import com.byit.job.enums.plugin.PluginNodeTypeEnum;
 import com.byit.job.utils.CronExpression;
 import com.byit.mapper.*;
 import com.byit.model.*;
+import com.byit.model.vo.RunRecordingVo;
 import com.byit.service.ApiFlowService;
 import com.byit.util.ApiFlowDagCheck;
 import com.byit.utils.ValidationUtil;
@@ -24,6 +25,7 @@ import org.springframework.transaction.annotation.Transactional;
 
 import javax.annotation.Resource;
 import java.text.ParseException;
+import java.text.SimpleDateFormat;
 import java.util.*;
 
 /**
@@ -35,7 +37,7 @@ import java.util.*;
 @Service
 @Transactional(rollbackFor = Exception.class)
 public class ApiFlowServiceImpl implements ApiFlowService {
-
+    private final static SimpleDateFormat sdf = new SimpleDateFormat("yyyyMMdd");
     @Resource
     private FlowMapper flowMapper;
 
@@ -590,6 +592,135 @@ public class ApiFlowServiceImpl implements ApiFlowService {
         jobTaskRunLogMapper.updateJobTaskRunLog(jobTaskRunLog);
     }
 
+    @Override
+    public List<RunRecordingVo> loadScheduleResult(String param){
+        ValidationUtil.dataNotBank(param, "请求参数不允许为空！");
+        JSONObject jsonObject = JSON.parseObject(param);
+        String startTime = jsonObject.getString("startTime");
+        ValidationUtil.dataNotNull(startTime, "开始时间不允许为空！");
+
+        String endTime = jsonObject.getString("endTime");
+        ValidationUtil.dataNotNull(endTime, "结束时间不允许为空！");
+
+        Date startDate = null;
+        Date endDate = null;
+        try {
+            startDate = sdf.parse(startTime);
+            endDate = sdf.parse(endTime);
+            //TODO 最长不能超过60天
+            ValidationUtil.isTrueValidation(endDate.getTime() - startDate.getTime() > 60 * 24 * 60 * 60 * 1000, "查询时间最长为60天!");
+        } catch (ParseException e) {
+            log.error("时间格式不正确, startTime {}  endTime {}", startTime, endTime);
+            ValidationUtil.isTrueValidation(true, "时间格式不符合标准! 例：20200101");
+        }
+
+        //获取工作空间名称
+        String workspaceName = jsonObject.getString("workspaceName");
+        String flowName = jsonObject.getString("flowName");
+        List<Integer> flowIds = new ArrayList<>();
+        if (StringUtils.isNotBlank(workspaceName)){
+            Workspace workspace = workspaceMapper.getByName(workspaceName);
+            ValidationUtil.dataNotNull(workspace, workspaceName + "工作空间不存在");
+            if (StringUtils.isNotBlank(flowName)){
+                Flow flow = flowMapper.getByWorkSpaceAndName(workspace.getWorkspaceId(), flowName);
+                ValidationUtil.dataNotNull(flow, flowName + "工作流不存在");
+                flowIds.add(flow.getFlowId());
+            }else {
+                List<Flow> flowList = flowMapper.findByWorkspace(workspace.getWorkspaceId());
+                if (flowList != null && flowList.size() > 0){
+                    flowList.forEach(flow -> flowIds.add(flow.getFlowId()));
+                }
+            }
+        }
+
+        List<RunRecording> runRecordList = runRecordingMapper.findByStartAndEndTime(startDate.getTime(), endDate.getTime(), flowIds);
+        if (runRecordList != null && runRecordList.size() > 0){
+            List<RunRecordingVo> runRecordingVoList = new ArrayList<>();
+            runRecordList.forEach(runRecording -> {
+                RunRecordingVo runRecordingVo = new RunRecordingVo();
+                BeanUtils.copyProperties(runRecording, runRecordingVo);
+                List<JobTaskRunLog> jobTaskRunLogList = jobTaskRunLogMapper.findByRunIdAndFlowName(runRecording.getRunId(), runRecording.getFlowName());
+                runRecordingVo.setJobTaskRunLogList(jobTaskRunLogList);
+
+                runRecordingVoList.add(runRecordingVo);
+            });
+
+            return runRecordingVoList;
+        }
+        return null;
+    }
+
+    /**
+     * runState  补批机制   1 补批当前节点   2   补批当前节点及以下节点
+     * @param param
+     */
+    @Override
+    public void repairJob(String param) {
+        ValidationUtil.dataNotBank(param, "请求参数不允许为空！");
+        JSONObject jsonObject = JSON.parseObject(param);
+        //获取工作空间名称
+        String workspaceName = jsonObject.getString("workspaceName");
+        ValidationUtil.dataNotBank(workspaceName, "工作空间名称不允许为空！");
+        //获取工作流名称
+        String flowName = jsonObject.getString("flowName");
+        ValidationUtil.dataNotBank(flowName, "工作流名称不允许为空！");
+        //获取节点名称
+        String nodeName = jsonObject.getString("nodeName");
+        ValidationUtil.dataNotBank(nodeName, "节点名称不允许为空！");
+        //获取重跑机制（运行当前节点，或运行当前节点及以下节点）
+        String runState = jsonObject.getString("runState");
+        ValidationUtil.dataNotBank(runState, "补批机制不允许为空！");
+        //获取运行参数
+        String runParam = jsonObject.getString("runParam");
+
+        //开始校验
+        Workspace workspace = workspaceMapper.getByName(workspaceName);
+        ValidationUtil.dataNotNull(workspace, workspaceName + "工作空间不存在");
+        Flow flow = flowMapper.getByWorkSpaceAndName(workspace.getWorkspaceId(), flowName);
+        ValidationUtil.dataNotNull(flow, flowName + "工作流不存在");
+        Node node = nodeMapper.getByNameAndFlow(nodeName, flow.getFlowId());
+        ValidationUtil.dataNotNull(node, nodeName + "节点不存在");
+
+        if (StringUtils.isNotEmpty(node.getRunParam())){
+            ValidationUtil.dataNotBank(runParam, "运行参数不允许为空！");
+        }
+    }
+
+    @Override
+    public void repairFlow(String param) {
+        ValidationUtil.dataNotBank(param, "请求参数不允许为空！");
+        JSONObject jsonObject = JSON.parseObject(param);
+        //获取工作空间名称
+        String workspaceName = jsonObject.getString("workspaceName");
+        ValidationUtil.dataNotBank(workspaceName, "工作空间名称不允许为空！");
+        //获取工作流名称
+        String flowName = jsonObject.getString("flowName");
+        ValidationUtil.dataNotBank(flowName, "工作流名称不允许为空！");
+        //获取补批的日期
+        String repairTime = jsonObject.getString("repairTime");
+        ValidationUtil.dataNotBank(repairTime, "补批日期不允许为空！");
+        try {
+            Date repairDate = sdf.parse(repairTime);
+            ValidationUtil.isTrueValidation(!repairDate.before(new Date()), "只能补过去时间的批次!");
+        } catch (ParseException e) {
+            log.error("补批日期不符合规范，例：20200101");
+            ValidationUtil.isTrueValidation(true, "补批日期不符合规范，例：20200101");
+        }
+
+        //开始校验
+        Workspace workspace = workspaceMapper.getByName(workspaceName);
+        ValidationUtil.dataNotNull(workspace, workspaceName + "工作空间不存在");
+        Flow flow = flowMapper.getByWorkSpaceAndName(workspace.getWorkspaceId(), flowName);
+        ValidationUtil.dataNotNull(flow, flowName + "工作流不存在");
+
+    }
+
+    @Override
+    public String getLogUrl(String param) {
+        return null;
+    }
+
+
     /**
      * 工作流生成新版本
      * @param flow

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

@@ -35,4 +35,11 @@ public interface FlowMapper {
      * @return
      */
     Flow getByWorkSpaceAndName(@Param("workspaceId") Integer workspaceId, @Param("flowName") String flowName);
+
+    /**
+     * 根据workspace查找所属的工作流
+     * @param workspaceId
+     * @return
+     */
+    List<Flow> findByWorkspace(Integer workspaceId);
 }
\ No newline at end of file

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

@@ -117,4 +117,6 @@ public interface RunRecordingMapper {
      * @return
      */
     List<RunRecording> findUnFinishByRunId(String runId);
+
+    List<RunRecording> findByStartAndEndTime(@Param("startDate")long startDate, @Param("endDate")long endDate, @Param("flowIds")List<Integer> flowIds);
 }
\ No newline at end of file

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

+package com.byit.model.vo;
+
+import com.byit.model.JobTaskRunLog;
+import com.byit.model.RunRecording;
+import io.swagger.annotations.ApiModelProperty;
+import lombok.Data;
+
+import java.util.List;
+
+/**
+ * @description: 运行实例记录VO类
+ * @author: gml
+ * @create: 2020/2/28
+ */
+@Data
+public class RunRecordingVo extends RunRecording {
+
+    @ApiModelProperty("运行实例下的运行任务日志")
+    private List<JobTaskRunLog> jobTaskRunLogList;
+
+}

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

@@ -67,7 +67,14 @@
       where flow_id = #{id,jdbcType=INTEGER}
     </select>
 
-    <delete id="deleteById" parameterType="java.lang.Integer">
+  <select id="findByWorkspace" resultType="com.byit.model.Flow">
+    select
+    <include refid="Base_Column_List" />
+    from flow
+    where false = #{workspaceId,jdbcType=INTEGER}
+  </select>
+
+  <delete id="deleteById" parameterType="java.lang.Integer">
     <!-- generated @mbg.generated date: 2019-12-31 -->
     delete from flow
     where flow_id = #{flowId,jdbcType=INTEGER}

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

@@ -81,6 +81,21 @@
     where run_id = #{runId,jdbcType=VARCHAR}
     and flow_status != '4'
   </select>
+  <select id="findByStartAndEndTime" resultMap="BaseResultMap">
+    select
+    <include refid="Base_Column_List" />
+    from run_recording
+    where start_time &gt;= ${startDate}
+        and end_time &lt;= ${endDate}
+        <if test="flowIds != null">
+         and flow_id in (
+            <foreach collection="flowIds" item="flowId" separator=",">
+              flowId
+            </foreach>
+          )
+        </if>
+
+  </select>
 
   <delete id="deleteById" parameterType="java.lang.Integer">
     <!-- generated @mbg.generated date: 2019-12-25 -->

byit-myth-executor/myth-executor-plugin/src/main/java/com/byit/utils/JobUtils.java

@@ -69,6 +69,10 @@ public class JobUtils {
      */
     private static final String REQUEST_FLOW_RERUNJOB = "/api/flow/reRunJob";
     /**
+     * 重跑工作流
+     */
+    private static final String REQUEST_FLOW_RERUNFLOW = "/api/flow/reRunFlow";
+    /**
      * 手动置为成功
      */
     private static final String REQUEST_FLOW_MAKESUCCESS = "/api/flow/madeSuccess";
@@ -269,6 +273,27 @@ public class JobUtils {
         return addRequestResult;
     }
 
+
+    /**
+     * 重跑工作流
+     * @param runId
+     * @param workspaceName
+     * @param flowName
+     * @return
+     */
+    public static String reRunFlow(String runId, String workspaceName, String flowName){
+        //请求的路径
+        String requestUrl = REQUEST_PREFIX + "127.0.0.1" + ":" + SERVER_PORT + REQUEST_FLOW_RERUNFLOW;
+        Map<String,String> map = new HashMap<>(5);
+        map.put("runId",runId);
+        map.put("workspaceName",workspaceName);
+        map.put("flowName",flowName);
+        //发送请求 添加任务
+        String addRequestResult = HttpUtil.post(requestUrl,"param="+JSON.toJSONString(map,WriteClassName));
+        log.info("--------------------重跑节点接口调用成功，结果为：{}------------------------",addRequestResult);
+        return addRequestResult;
+    }
+
     /**
      * 手动置为成功
      * @param runId