Skip to content

B
byit-myth-job
Commit ad58eced by huangfusuper
Options

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

parents 0634a7a8 2797f75f
Showing with 329 additions and 77 deletions
byit-myth-admin/src/main/java/com/byit/api/ApiFlowController.java
...@@ -161,9 +161,28 @@ public class ApiFlowController { ...@@ -161,9 +161,28 @@ public class ApiFlowController {
return ResponseResult.ok("SUCCESS"); return ResponseResult.ok("SUCCESS");
} }
/**
* 获取整体运行的统计数据
* @param param
* @return
*/
@PostMapping("/loadStatisticData")
@ApiOperation("获取运行的统计数据")
public ResponseResult loadStatisticData(String param){ public ResponseResult loadStatisticData(String param){
CollectData collectData = apiFlowService.loadStatisticData(param); CollectData collectData = apiFlowService.loadStatisticData(param);
return ResponseResult.ok(collectData); return ResponseResult.ok(collectData);
} }
/**
* 获取节点运行的统计数据
* @param param
* @return
*/
@PostMapping("/loadNodeStatisticData")
@ApiOperation("获取运行的统计数据")
public ResponseResult loadNodeStatisticData(String param){
CollectData collectData = apiFlowService.loadNodeStatisticData(param);
return ResponseResult.ok(collectData);
}
} }
byit-myth-admin/src/main/java/com/byit/service/ApiFlowService.java
...@@ -88,4 +88,11 @@ public interface ApiFlowService { ...@@ -88,4 +88,11 @@ public interface ApiFlowService {
* @return * @return
*/ */
CollectData loadStatisticData(String param); CollectData loadStatisticData(String param);
/**
* 获取节点的汇总数据
* @param param
* @return
*/
CollectData loadNodeStatisticData(String param);
} }
byit-myth-admin/src/main/java/com/byit/service/impl/ApiFlowServiceImpl.java
...@@ -44,6 +44,7 @@ import static com.alibaba.fastjson.serializer.SerializerFeature.WriteClassName; ...@@ -44,6 +44,7 @@ import static com.alibaba.fastjson.serializer.SerializerFeature.WriteClassName;
@Transactional(rollbackFor = Exception.class) @Transactional(rollbackFor = Exception.class)
public class ApiFlowServiceImpl implements ApiFlowService { public class ApiFlowServiceImpl implements ApiFlowService {
private final static SimpleDateFormat sdf = new SimpleDateFormat("yyyyMMdd"); private final static SimpleDateFormat sdf = new SimpleDateFormat("yyyyMMdd");
private final static SimpleDateFormat dateTimeFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
@Resource @Resource
private FlowMapper flowMapper; private FlowMapper flowMapper;
...@@ -658,16 +659,18 @@ public class ApiFlowServiceImpl implements ApiFlowService { ...@@ -658,16 +659,18 @@ public class ApiFlowServiceImpl implements ApiFlowService {
} }
String scheduleStatus = jsonObject.getString("scheduleStatus"); String scheduleStatus = jsonObject.getString("scheduleStatus");
List<String> scheduleStatusList = new ArrayList<>(); List<String> scheduleStatusList = null;
if(StringUtils.isNotEmpty(scheduleStatus)){ if(StringUtils.isNotEmpty(scheduleStatus)){
scheduleStatusList = Arrays.asList(scheduleStatus.split(",")); scheduleStatusList = Arrays.asList(scheduleStatus.split(","));
} }
String executeStatus = jsonObject.getString("executeStatus"); String executeStatus = jsonObject.getString("executeStatus");
List<String> executeStatusList = new ArrayList<>(); List<String> executeStatusList = null;
if(StringUtils.isNotEmpty(executeStatus)){ if(StringUtils.isNotEmpty(executeStatus)){
executeStatusList = Arrays.asList(executeStatus.split(",")); executeStatusList = Arrays.asList(executeStatus.split(","));
} }
List<RunRecording> runRecordList = runRecordingMapper.findByStartAndEndTime(startTime, endTime, flowIds, scheduleStatusList, executeStatusList); String startDate = dateTimeFormat.format(new Date(startTime));
String endDate = dateTimeFormat.format(new Date(endTime));
List<RunRecording> runRecordList = runRecordingMapper.findByStartAndEndTime(startDate, endDate, flowIds, scheduleStatusList, executeStatusList);
return runRecordList; return runRecordList;
} }
...@@ -709,66 +712,151 @@ public class ApiFlowServiceImpl implements ApiFlowService { ...@@ -709,66 +712,151 @@ public class ApiFlowServiceImpl implements ApiFlowService {
ValidationUtil.dataNotNull(endTime, "结束时间不允许为空！"); ValidationUtil.dataNotNull(endTime, "结束时间不允许为空！");
ValidationUtil.isTrueValidation(String.valueOf(startTime).length() < 13 || String.valueOf(endTime).length() < 13 , "时间格式必须是毫秒"); ValidationUtil.isTrueValidation(String.valueOf(startTime).length() < 13 || String.valueOf(endTime).length() < 13 , "时间格式必须是毫秒");
List<Integer> flowIdList = new ArrayList<>();
List<Flow> flowList = flowMapper.findByWorkspace(workspace.getWorkspaceId());
if (flowList != null && flowList.size() > 0){
flowList.forEach(flow -> flowIdList.add(flow.getFlowId()));
}
//如果工作空间下面有数据
if (flowIdList.size() > 0){
StatisticData flowStatisticData = runRecordingMapper.findStatusByStartAndEndTime(startTime, endTime, flowIdList);
StatisticData unstartflowStatisticData = waitingRecordMapper.findStatusByStartAndEndTime(flowIdList);
if (null != unstartflowStatisticData){
flowStatisticData.setUnstart(flowStatisticData.getUnstart() + unstartflowStatisticData.getUnstart());
}
StatisticData nodeStatisticData = jobTaskRunLogMapper.findStatusByStartAndEndTime(startTime, endTime, flowIdList);
StatisticData unstartNodeStatisticData = waitingTaskMapper.findStatusByStartAndEndTime(flowIdList);
if (null != unstartNodeStatisticData){
nodeStatisticData.setUnstart(nodeStatisticData.getUnstart() + unstartNodeStatisticData.getUnstart());
}
List<FlowStatusSnapshoot> flowStatusSnapshootList = flowStatusSnapshootMapper.findByTime(startTime, endTime);
Map<String, StatisticData> flowCollectMap = new HashMap<>();
Map<String, StatisticData> nodeCollectMap = new HashMap<>();
flowStatusSnapshootList.forEach(flowStatusSnapshoot -> {
String key = flowStatusSnapshoot.getDay() + " " + flowStatusSnapshoot.getHour();
StatisticData nodeHourStatisticData = nodeCollectMap.get(key);
StatisticData flowHourStatisticData = flowCollectMap.get(key);
if (null == nodeHourStatisticData){
flowHourStatisticData = new StatisticData();
buildFlowHourStatisticData(flowHourStatisticData, flowStatusSnapshoot);
nodeHourStatisticData = new StatisticData();
nodeHourStatisticData.setUnstart(flowStatusSnapshoot.getUnstartNode());
nodeHourStatisticData.setRunIng(flowStatusSnapshoot.getRuningNode());
nodeHourStatisticData.setSuccess(flowStatusSnapshoot.getSuccessNode());
nodeHourStatisticData.setFail(flowStatusSnapshoot.getFailNode());
nodeHourStatisticData.setKill(flowStatusSnapshoot.getKillNode());
}else {
buildFlowHourStatisticData(flowHourStatisticData, flowStatusSnapshoot);
nodeHourStatisticData.setUnstart(nodeHourStatisticData.getUnstart() + flowStatusSnapshoot.getUnstartNode());
nodeHourStatisticData.setRunIng(nodeHourStatisticData.getRunIng() + flowStatusSnapshoot.getRuningNode());
nodeHourStatisticData.setSuccess(nodeHourStatisticData.getSuccess() + flowStatusSnapshoot.getSuccessNode());
nodeHourStatisticData.setFail(nodeHourStatisticData.getFail() + flowStatusSnapshoot.getFailNode());
nodeHourStatisticData.setKill(nodeHourStatisticData.getKill() + flowStatusSnapshoot.getKillNode());
}
flowCollectMap.put(key, flowHourStatisticData);
nodeCollectMap.put(key, nodeHourStatisticData);
});
CollectData collectData = new CollectData();
collectData.setFlowData(flowStatisticData);
collectData.setNodeData(nodeStatisticData);
collectData.setNodeCollectMap(nodeCollectMap);
collectData.setFlowCollectMap(flowCollectMap);
return collectData;
}
return null;
}
/**
* 构建统计数据
* @param flowHourStatisticData
* @param flowStatusSnapshoot
*/
private void buildFlowHourStatisticData(StatisticData flowHourStatisticData, FlowStatusSnapshoot flowStatusSnapshoot){
switch (flowStatusSnapshoot.getFlowStatus()){
case 1 : flowHourStatisticData.setUnstart(flowHourStatisticData.getUnstart() + 1); break;
case 2 : flowHourStatisticData.setRunIng(flowHourStatisticData.getRunIng() + 1); break;
case 3 : flowHourStatisticData.setStop(flowHourStatisticData.getStop() + 1); break;
case 4 : flowHourStatisticData.setSuccess(flowHourStatisticData.getSuccess() + 1); break;
case 5 : flowHourStatisticData.setFail(flowHourStatisticData.getFail() + 1); break;
case 6 : flowHourStatisticData.setKill(flowHourStatisticData.getFail() + 1); break;
default: break;
}
}
@Override
public CollectData loadNodeStatisticData(String param){
ValidationUtil.dataNotBank(param, "请求参数不允许为空！");
JSONObject jsonObject = JSON.parseObject(param);
//获取工作空间名称
String workspaceName = jsonObject.getString("workspaceName");
ValidationUtil.dataNotBank(workspaceName, "工作空间名称不允许为空！");
//开始校验
Workspace workspace = workspaceMapper.getByName(workspaceName);
ValidationUtil.dataNotNull(workspace, workspaceName + "工作空间不存在");
//获取时间间隔
Long startTime = jsonObject.getLong("startTime");
ValidationUtil.dataNotNull(startTime, "开始时间不允许为空！");
Long endTime = jsonObject.getLong("endTime");
ValidationUtil.dataNotNull(endTime, "结束时间不允许为空！");
//校验时间间隔
ValidationUtil.isTrueValidation(String.valueOf(startTime).length() < 13 || String.valueOf(endTime).length() < 13 , "时间格式必须是毫秒");
//获取工作流名称 //获取工作流名称
String flowName = jsonObject.getString("flowName"); String flowName = jsonObject.getString("flowName");
ValidationUtil.dataNotBank(flowName, "工作流名称不允许为空！");
Flow flow = flowMapper.getByWorkSpaceAndName(workspace.getWorkspaceId(), flowName);
ValidationUtil.dataNotNull(flow, "工作流不存在！");
List<Integer> flowIdList = new ArrayList<>(); List<Integer> flowIdList = new ArrayList<>();
if (StringUtils.isNotEmpty(flowName)){ Queue<Integer> queue = new LinkedList<>();
Flow flow = flowMapper.getByWorkSpaceAndName(workspace.getWorkspaceId(), flowName); flowIdList.add(flow.getFlowId());
ValidationUtil.dataNotNull(flow, "工作流不存在！"); queue.offer(flow.getFlowId());
flowIdList.add(flow.getFlowId()); while(!queue.isEmpty()){
if (FlowPropertyEnum.IS_INNER.getCode().equals(flow.getIsInner())){ List<Node> innerFlowList = nodeMapper.findVirtualByFlowId(queue.poll());
List<RunRecording> runRecordingList = runRecordingMapper.findByStartAndEndTime(startTime, endTime, flowIdList, null, null); if (null != innerFlowList && innerFlowList.size() > 0){
flowIdList.clear(); innerFlowList.forEach(node -> {
runRecordingList.forEach(runRecording -> flowIdList.add(runRecording.getFlowId())); flowIdList.add(node.getMapFlowId());
} queue.offer(node.getMapFlowId());
}else { });
List<Flow> flowList = flowMapper.findByWorkspace(workspace.getWorkspaceId());
if (flowList != null && flowList.size() > 0){
flowList.forEach(flow -> flowIdList.add(flow.getFlowId()));
} }
} }
StatisticData flowStatisticData = runRecordingMapper.findStatusByStartAndEndTime(startTime, endTime, flowIdList);
StatisticData nodeStatisticData = jobTaskRunLogMapper.findStatusByStartAndEndTime(startTime, endTime, flowIdList); StatisticData nodeStatisticData = jobTaskRunLogMapper.findStatusByStartAndEndTime(startTime, endTime, flowIdList);
StatisticData unstartNodeStatisticData = waitingTaskMapper.findStatusByStartAndEndTime(flowIdList);
if (null != unstartNodeStatisticData){
nodeStatisticData.setUnstart(nodeStatisticData.getUnstart() + unstartNodeStatisticData.getUnstart());
}
List<FlowStatusSnapshoot> flowStatusSnapshootList = flowStatusSnapshootMapper.findByTime(startTime, endTime); List<FlowStatusSnapshoot> flowStatusSnapshootList = flowStatusSnapshootMapper.findByTime(startTime, endTime);
Map<String, StatisticData> collectMap = new HashMap<>(); Map<String, StatisticData> nodeCollectMap = new HashMap<>();
flowStatusSnapshootList.forEach(flowStatusSnapshoot -> { flowStatusSnapshootList.forEach(flowStatusSnapshoot -> {
String key = flowStatusSnapshoot.getDay() + " " + flowStatusSnapshoot.getHour(); String key = flowStatusSnapshoot.getDay() + " " + flowStatusSnapshoot.getHour();
StatisticData statisticData = collectMap.get(key); StatisticData nodeHourStatisticData = nodeCollectMap.get(key);
if (null == statisticData){ if (null == nodeHourStatisticData){
statisticData = new StatisticData(); nodeHourStatisticData = new StatisticData();
statisticData.setUnstart(flowStatusSnapshoot.getUnstartNode()); nodeHourStatisticData.setUnstart(flowStatusSnapshoot.getUnstartNode());
statisticData.setRunIng(flowStatusSnapshoot.getRuningNode()); nodeHourStatisticData.setRunIng(flowStatusSnapshoot.getRuningNode());
statisticData.setSuccess(flowStatusSnapshoot.getSuccessNode()); nodeHourStatisticData.setSuccess(flowStatusSnapshoot.getSuccessNode());
statisticData.setFail(flowStatusSnapshoot.getFailNode()); nodeHourStatisticData.setFail(flowStatusSnapshoot.getFailNode());
statisticData.setKill(flowStatusSnapshoot.getKillNode()); nodeHourStatisticData.setKill(flowStatusSnapshoot.getKillNode());
collectMap.put(key, statisticData);
}else { }else {
statisticData.setUnstart(statisticData.getUnstart() + flowStatusSnapshoot.getUnstartNode()); nodeHourStatisticData.setUnstart(nodeHourStatisticData.getUnstart() + flowStatusSnapshoot.getUnstartNode());
statisticData.setRunIng(statisticData.getRunIng() + flowStatusSnapshoot.getRuningNode()); nodeHourStatisticData.setRunIng(nodeHourStatisticData.getRunIng() + flowStatusSnapshoot.getRuningNode());
statisticData.setSuccess(statisticData.getSuccess() + flowStatusSnapshoot.getSuccessNode()); nodeHourStatisticData.setSuccess(nodeHourStatisticData.getSuccess() + flowStatusSnapshoot.getSuccessNode());
statisticData.setFail(statisticData.getFail() + flowStatusSnapshoot.getFailNode()); nodeHourStatisticData.setFail(nodeHourStatisticData.getFail() + flowStatusSnapshoot.getFailNode());
statisticData.setKill(statisticData.getKill() + flowStatusSnapshoot.getKillNode()); nodeHourStatisticData.setKill(nodeHourStatisticData.getKill() + flowStatusSnapshoot.getKillNode());
collectMap.put(key, statisticData);
} }
nodeCollectMap.put(key, nodeHourStatisticData);
}); });
CollectData collectData = new CollectData(); CollectData collectData = new CollectData();
collectData.setFlowData(flowStatisticData);
collectData.setNodeData(nodeStatisticData); collectData.setNodeData(nodeStatisticData);
collectData.setCollectMap(collectMap); collectData.setNodeCollectMap(nodeCollectMap);
return collectData; return collectData;
} }
public static void main(String[] args) throws ParseException {
String str = "20200401";
SimpleDateFormat ssss = new SimpleDateFormat("yyyy-MM-dd hh");
Date date = new Date();
System.out.println(ssss.format(date));
System.out.println(sdf.parse(str));
}
/** /**
* runState 补批机制 1 补批当前节点 2 补批当前节点及以下节点 * runState 补批机制 1 补批当前节点 2 补批当前节点及以下节点
* @param param * @param param
......
byit-myth-core/myth-admin-core/src/main/java/com/byit/mapper/FlowStatusSnapshootMapper.java
package com.byit.mapper; package com.byit.mapper;
import com.byit.model.FlowStatusSnapshoot; import com.byit.model.FlowStatusSnapshoot;
import org.apache.ibatis.annotations.Param;
import java.util.List; import java.util.List;
...@@ -23,14 +24,14 @@ public interface FlowStatusSnapshootMapper { ...@@ -23,14 +24,14 @@ public interface FlowStatusSnapshootMapper {
* @param hour * @param hour
* @return * @return
*/ */
Integer exist(String date, String hour); Integer exist(@Param("date")String date, @Param("hour")String hour);
/** /**
* 批量保存 * 批量保存
* @param flowStatusSnapshootList * @param flowStatusSnapshootList
* @return * @return
*/ */
int saveList(List<FlowStatusSnapshoot> flowStatusSnapshootList); int saveList(@Param("flowStatusSnapshootList")List<FlowStatusSnapshoot> flowStatusSnapshootList);
/** /**
* 删除太长时间的快照 * 删除太长时间的快照
...@@ -45,5 +46,5 @@ public interface FlowStatusSnapshootMapper { ...@@ -45,5 +46,5 @@ public interface FlowStatusSnapshootMapper {
* @param endTime * @param endTime
* @return * @return
*/ */
List<FlowStatusSnapshoot> findByTime(Long startTime, Long endTime); List<FlowStatusSnapshoot> findByTime(@Param("startTime")Long startTime, @Param("endTime")Long endTime);
} }
\ No newline at end of file
byit-myth-core/myth-admin-core/src/main/java/com/byit/mapper/RunRecordingMapper.java
...@@ -5,6 +5,7 @@ import com.byit.model.RunRecording; ...@@ -5,6 +5,7 @@ import com.byit.model.RunRecording;
import org.apache.ibatis.annotations.Param; import org.apache.ibatis.annotations.Param;
import org.springframework.stereotype.Repository; import org.springframework.stereotype.Repository;
import java.util.Date;
import java.util.List; import java.util.List;
/** /**
...@@ -138,8 +139,8 @@ public interface RunRecordingMapper { ...@@ -138,8 +139,8 @@ public interface RunRecordingMapper {
*/ */
List<RunRecording> findUnFinishByRunId(String runId); List<RunRecording> findUnFinishByRunId(String runId);
List<RunRecording> findByStartAndEndTime(@Param("startDate")long startDate, List<RunRecording> findByStartAndEndTime(@Param("startDate")String startDate,
@Param("endDate")long endDate, @Param("endDate")String endDate,
@Param("flowIds")List<Integer> flowIds, @Param("flowIds")List<Integer> flowIds,
@Param("scheduleStatusList")List<String> scheduleStatusList, @Param("scheduleStatusList")List<String> scheduleStatusList,
@Param("executeStatusList")List<String> executeStatusList); @Param("executeStatusList")List<String> executeStatusList);
......
byit-myth-core/myth-admin-core/src/main/java/com/byit/mapper/WaitingRecordMapper.java
package com.byit.mapper; package com.byit.mapper;
import com.byit.dto.plugin.StatisticData;
import com.byit.model.WaitingRecord; import com.byit.model.WaitingRecord;
import org.apache.ibatis.annotations.Param;
import org.springframework.stereotype.Repository; import org.springframework.stereotype.Repository;
import java.util.List; import java.util.List;
...@@ -31,4 +33,17 @@ public interface WaitingRecordMapper { ...@@ -31,4 +33,17 @@ public interface WaitingRecordMapper {
* @return * @return
*/ */
Integer findOrderByFlowId(Integer flowId); Integer findOrderByFlowId(Integer flowId);
/**
* 根据工作流id获取运行实例
* @param flowId
* @return
*/
List<WaitingRecord> findByFlowId(Integer flowId);
/**
* 查询运行汇总
* @param flowIdList
* @return
*/
StatisticData findStatusByStartAndEndTime(@Param("flowIdList")List<Integer> flowIdList);
} }
\ No newline at end of file
byit-myth-core/myth-admin-core/src/main/java/com/byit/mapper/WaitingTaskMapper.java
package com.byit.mapper; package com.byit.mapper;
import com.byit.dto.plugin.StatisticData;
import com.byit.model.WaitingTask; import com.byit.model.WaitingTask;
import org.apache.ibatis.annotations.Param;
import org.springframework.stereotype.Repository; import org.springframework.stereotype.Repository;
import java.util.List; import java.util.List;
...@@ -22,4 +24,10 @@ public interface WaitingTaskMapper { ...@@ -22,4 +24,10 @@ public interface WaitingTaskMapper {
int updateByIdSelective(WaitingTask record); int updateByIdSelective(WaitingTask record);
/**
* 查询运行汇总
* @param flowIdList
* @return
*/
StatisticData findStatusByStartAndEndTime(@Param("flowIdList")List<Integer> flowIdList);
} }
\ No newline at end of file
byit-myth-core/myth-admin-core/src/main/java/com/byit/thread/helper/StatusSnapshootThreadRunHelper.java
package com.byit.thread.helper; package com.byit.thread.helper;
import com.byit.dto.plugin.StatisticData; import com.byit.dto.plugin.StatisticData;
import com.byit.mapper.FlowMapper; import com.byit.mapper.*;
import com.byit.mapper.FlowStatusSnapshootMapper; import com.byit.model.*;
import com.byit.mapper.JobTaskRunLogMapper;
import com.byit.mapper.RunRecordingMapper;
import com.byit.model.Flow;
import com.byit.model.FlowStatusSnapshoot;
import com.byit.model.RunRecording;
import com.byit.thread.BaseThreadRunHelper; import com.byit.thread.BaseThreadRunHelper;
import lombok.extern.slf4j.Slf4j; import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Value; import org.springframework.beans.factory.annotation.Value;
...@@ -39,6 +34,10 @@ public class StatusSnapshootThreadRunHelper extends BaseThreadRunHelper { ...@@ -39,6 +34,10 @@ public class StatusSnapshootThreadRunHelper extends BaseThreadRunHelper {
@Resource @Resource
private JobTaskRunLogMapper jobTaskRunLogMapper; private JobTaskRunLogMapper jobTaskRunLogMapper;
@Resource @Resource
private WaitingRecordMapper waitingRecordMapper;
@Resource
private WaitingTaskMapper waitingTaskMapper;
@Resource
private DataSource dataSource; private DataSource dataSource;
@Value("${myth-job.snapshoot-date}") @Value("${myth-job.snapshoot-date}")
private Integer snapshootDate; private Integer snapshootDate;
...@@ -66,20 +65,25 @@ public class StatusSnapshootThreadRunHelper extends BaseThreadRunHelper { ...@@ -66,20 +65,25 @@ public class StatusSnapshootThreadRunHelper extends BaseThreadRunHelper {
List<FlowStatusSnapshoot> flowStatusSnapshootList = new ArrayList<>(); List<FlowStatusSnapshoot> flowStatusSnapshootList = new ArrayList<>();
flowList.forEach(flow -> { flowList.forEach(flow -> {
RunRecording runRecording = runRecordingMapper.findMaxByFlowId(flow.getFlowId()); RunRecording runRecording = runRecordingMapper.findMaxByFlowId(flow.getFlowId());
FlowStatusSnapshoot flowStatusSnapshoot = new FlowStatusSnapshoot();
flowStatusSnapshoot.setDay(date);
flowStatusSnapshoot.setFlowId(flow.getFlowId());
flowStatusSnapshoot.setHour(hour);
flowStatusSnapshoot.setFlowName(flow.getFlowName());
flowStatusSnapshoot.setWorkspaceId(flow.getWorkspaceId());
flowStatusSnapshoot.setSnapshootTime(hourTime);
//统计运行实例的数据
//如果存在实例
if (runRecording != null){ if (runRecording != null){
//获取凌晨的时间戳 //获取凌晨的时间戳
Long time = LocalDate.now().atStartOfDay(ZoneOffset.ofHours(8)).toInstant().toEpochMilli(); Long time = LocalDate.now().atStartOfDay(ZoneOffset.ofHours(8)).toInstant().toEpochMilli();
FlowStatusSnapshoot flowStatusSnapshoot = new FlowStatusSnapshoot(); //从凌晨到现在是否有运行实例
flowStatusSnapshoot.setDay(date); if (time.longValue() > runRecording.getTriggerTime().longValue()){//今天不存在工作流的运行实例
flowStatusSnapshoot.setFlowId(flow.getFlowId());
flowStatusSnapshoot.setHour(hour);
flowStatusSnapshoot.setFlowName(flow.getFlowName());
flowStatusSnapshoot.setWorkspaceId(flow.getWorkspaceId());
flowStatusSnapshoot.setSnapshootTime(hourTime);
if (time.longValue() > runRecording.getTriggerTime().longValue()){
flowStatusSnapshoot.setFlowStatus(1); flowStatusSnapshoot.setFlowStatus(1);
flowStatusSnapshoot.setUnstartNode(flow.getFlowNodeCount()); flowStatusSnapshoot.setUnstartNode(flow.getFlowNodeCount());
}else { }else { //存在工作流的运行实例
//判断运行状态
if (runRecording.getFlowStatus().equals("1") || runRecording.getFlowStatus().equals("2") || runRecording.getFlowStatus().equals("3")){ if (runRecording.getFlowStatus().equals("1") || runRecording.getFlowStatus().equals("2") || runRecording.getFlowStatus().equals("3")){
flowStatusSnapshoot.setFlowStatus(Integer.valueOf(runRecording.getFlowStatus())); flowStatusSnapshoot.setFlowStatus(Integer.valueOf(runRecording.getFlowStatus()));
}else { }else {
...@@ -94,6 +98,8 @@ public class StatusSnapshootThreadRunHelper extends BaseThreadRunHelper { ...@@ -94,6 +98,8 @@ public class StatusSnapshootThreadRunHelper extends BaseThreadRunHelper {
} }
//获取运行实例下各类状态的节点数目 //获取运行实例下各类状态的节点数目
StatisticData nodeStatistic = jobTaskRunLogMapper.findStatisticDataByRunIdAndFlowId(runRecording.getRunId(), flow.getFlowId()); StatisticData nodeStatistic = jobTaskRunLogMapper.findStatisticDataByRunIdAndFlowId(runRecording.getRunId(), flow.getFlowId());
//判断是否存在节点的运行日志
//存在节点运行日志
if (nodeStatistic != null){ if (nodeStatistic != null){
int sum = nodeStatistic.getFail() + nodeStatistic.getSuccess() + nodeStatistic.getRunIng() + nodeStatistic.getKill(); int sum = nodeStatistic.getFail() + nodeStatistic.getSuccess() + nodeStatistic.getRunIng() + nodeStatistic.getKill();
flowStatusSnapshoot.setUnstartNode(flow.getFlowNodeCount() - sum); flowStatusSnapshoot.setUnstartNode(flow.getFlowNodeCount() - sum);
...@@ -102,12 +108,38 @@ public class StatusSnapshootThreadRunHelper extends BaseThreadRunHelper { ...@@ -102,12 +108,38 @@ public class StatusSnapshootThreadRunHelper extends BaseThreadRunHelper {
flowStatusSnapshoot.setFailNode(nodeStatistic.getFail()); flowStatusSnapshoot.setFailNode(nodeStatistic.getFail());
flowStatusSnapshoot.setKillNode(nodeStatistic.getKill()); flowStatusSnapshoot.setKillNode(nodeStatistic.getKill());
}else { }else {
//不存在节点的运行日志
nodeStatistic.setUnstart(flow.getFlowNodeCount()); nodeStatistic.setUnstart(flow.getFlowNodeCount());
} }
} }
flowStatusSnapshootList.add(flowStatusSnapshoot); }else {
//如果不存在运行实例
flowStatusSnapshoot.setUnstartNode(flow.getFlowNodeCount());
} }
flowStatusSnapshootList.add(flowStatusSnapshoot);
//获取排队的运行实例
List<WaitingRecord> waitingRecordList = waitingRecordMapper.findByFlowId(flow.getFlowId());
if (waitingRecordList != null && waitingRecordList.size() > 0){
//如果存在正在排队的实例
waitingRecordList.forEach(waitingRecord -> {
//创建工作流快照
FlowStatusSnapshoot waitingStatusSnapshoot = new FlowStatusSnapshoot();
waitingStatusSnapshoot.setDay(date);
waitingStatusSnapshoot.setFlowId(flow.getFlowId());
waitingStatusSnapshoot.setHour(hour);
waitingStatusSnapshoot.setFlowName(flow.getFlowName());
waitingStatusSnapshoot.setWorkspaceId(flow.getWorkspaceId());
waitingStatusSnapshoot.setSnapshootTime(hourTime);
waitingStatusSnapshoot.setFlowStatus(1);
//获取排队的节点数目
List<WaitingTask> waitingTaskList = waitingTaskMapper.findAllByWaitId(waitingRecord.getWaitId());
if (waitingTaskList != null && waitingTaskList.size() > 0){
waitingStatusSnapshoot.setUnstartNode(waitingTaskList.size());
}
flowStatusSnapshootList.add(waitingStatusSnapshoot);
});
}
}); });
if (flowList != null && flowList.size() > 0){ if (flowList != null && flowList.size() > 0){
flowStatusSnapshootMapper.saveList(flowStatusSnapshootList); flowStatusSnapshootMapper.saveList(flowStatusSnapshootList);
......
byit-myth-core/myth-admin-core/src/main/resources/mapper/FlowStatusSnapshootMapper.xml
...@@ -198,7 +198,7 @@ ...@@ -198,7 +198,7 @@
from flow_status_snapshoot from flow_status_snapshoot
where 'day' = #{date} and `hour` = #{hour} where 'day' = #{date} and `hour` = #{hour}
</select> </select>
<select id="findByTime" resultType="com.byit.model.FlowStatusSnapshoot"> <select id="findByTime" resultMap="BaseResultMap">
select select
<include refid="Base_Column_List" /> <include refid="Base_Column_List" />
from flow_status_snapshoot from flow_status_snapshoot
...@@ -233,7 +233,7 @@ ...@@ -233,7 +233,7 @@
#{flowStatusSnapshoot.runingNode,jdbcType=INTEGER}, #{flowStatusSnapshoot.runingNode,jdbcType=INTEGER},
#{flowStatusSnapshoot.successNode,jdbcType=INTEGER}, #{flowStatusSnapshoot.successNode,jdbcType=INTEGER},
#{flowStatusSnapshoot.failNode,jdbcType=INTEGER}, #{flowStatusSnapshoot.failNode,jdbcType=INTEGER},
#{flowStatusSnapshoot.killNode,jdbcType=INTEGER}, #{flowStatusSnapshoot.killNode,jdbcType=INTEGER}
) )
</foreach> </foreach>
</insert> </insert>
......
byit-myth-core/myth-admin-core/src/main/resources/mapper/JobTaskRunLogMapper.xml
...@@ -129,9 +129,9 @@ ...@@ -129,9 +129,9 @@
from job_task_run_log from job_task_run_log
where start_time &gt;= ${startDate} where start_time &gt;= ${startDate}
and end_time &lt;= ${endDate} and end_time &lt;= ${endDate}
<if test="flowIds != null"> <if test="flowIdList != null">
and flow_id in ( and flow_id in (
<foreach collection="flowIds" item="flowId" separator=","> <foreach collection="flowIdList" item="flowId" separator=",">
#{flowId} #{flowId}
</foreach> </foreach>
) )
......
byit-myth-core/myth-admin-core/src/main/resources/mapper/RunRecordingMapper.xml
...@@ -158,15 +158,15 @@ ...@@ -158,15 +158,15 @@
from run_recording from run_recording
where start_time &gt;= ${startDate} where start_time &gt;= ${startDate}
and end_time &lt;= ${endDate} and end_time &lt;= ${endDate}
<if test="flowIds != null"> <if test="flowIdList != null">
and flow_id in ( and flow_id in (
<foreach collection="flowIds" item="flowId" separator=","> <foreach collection="flowIdList" item="flowId" separator=",">
#{flowId} #{flowId}
</foreach> </foreach>
) )
</if> </if>
</select> </select>
<select id="findMaxByFlowId" resultType="com.byit.model.RunRecording"> <select id="findMaxByFlowId" resultMap="BaseResultMap">
select <include refid="Base_Column_List" /> select <include refid="Base_Column_List" />
from run_recording from run_recording
where flow_id = #{flowId} where flow_id = #{flowId}
......
byit-myth-core/myth-admin-core/src/main/resources/mapper/WaitingRecordMapper.xml
...@@ -48,6 +48,24 @@ ...@@ -48,6 +48,24 @@
from waiting_record from waiting_record
where flow_id = #{flowId} where flow_id = #{flowId}
</select> </select>
<select id="findByFlowId" resultMap="BaseResultMap">
select
<include refid="Base_Column_List" />
from waiting_record
where flow_id = #{flowId}
</select>
<select id="findStatusByStartAndEndTime" resultType="com.byit.dto.plugin.StatisticData">
select count(1) unstart
from waiting_record
where 1 = 1
<if test="flowIdList != null">
and flow_id in (
<foreach collection="flowIdList" item="flowId" separator=",">
#{flowId}
</foreach>
)
</if>
</select>
<delete id="deleteById" parameterType="java.lang.Integer"> <delete id="deleteById" parameterType="java.lang.Integer">
<!-- generated @mbg.generated date: 2020-03-12 --> <!-- generated @mbg.generated date: 2020-03-12 -->
......
byit-myth-core/myth-admin-core/src/main/resources/mapper/WaitingTaskMapper.xml
...@@ -70,8 +70,20 @@ ...@@ -70,8 +70,20 @@
from waiting_task from waiting_task
where wait_id=#{waitId,jdbcType=INTEGER} where wait_id=#{waitId,jdbcType=INTEGER}
</select> </select>
<select id="findStatusByStartAndEndTime" resultType="com.byit.dto.plugin.StatisticData">
select count(1) unstart
from waiting_task
where 1 = 1
<if test="flowIdList != null">
and flow_id in (
<foreach collection="flowIdList" item="flowId" separator=",">
#{flowId}
</foreach>
)
</if>
</select>
<delete id="deleteById" parameterType="java.lang.Integer"> <delete id="deleteById" parameterType="java.lang.Integer">
<!-- generated @mbg.generated date: 2020-03-12 --> <!-- generated @mbg.generated date: 2020-03-12 -->
delete from waiting_task delete from waiting_task
where id = #{id,jdbcType=INTEGER} where id = #{id,jdbcType=INTEGER}
......
byit-myth-core/myth-dto-core/src/main/java/com/byit/dto/plugin/CollectData.java
...@@ -22,8 +22,13 @@ public class CollectData { ...@@ -22,8 +22,13 @@ public class CollectData {
private StatisticData nodeData; private StatisticData nodeData;
/** /**
* 按时间段的汇总 * 节点按时间段的汇总
*/ */
private Map<String, StatisticData> collectMap; private Map<String, StatisticData> nodeCollectMap;
/**
* 工作流按时间段的汇总
*/
private Map<String, StatisticData> flowCollectMap;
} }
byit-myth-executor/myth-executor-plugin/src/main/java/com/byit/utils/JobUtils.java
...@@ -91,7 +91,14 @@ public class JobUtils { ...@@ -91,7 +91,14 @@ public class JobUtils {
* 获取运行实例日志 * 获取运行实例日志
*/ */
public static final String REQUEST_LOADSCHEDULELOG = "/api/flow/loadScheduleLog"; public static final String REQUEST_LOADSCHEDULELOG = "/api/flow/loadScheduleLog";
/**
* 获取运行的统计数据
*/
public static final String REQUEST_LOADSTATISTICDATA = "/api/flow/loadStatisticData";
/**
* 获取节点运行的统计数据
*/
public static final String REQUEST_LOADNODESTATISTICDATA = "/api/flow/loadNodeStatisticData";
/** /**
* 补批工作流 * 补批工作流
*/ */
...@@ -344,6 +351,42 @@ public class JobUtils { ...@@ -344,6 +351,42 @@ public class JobUtils {
} }
/** /**
* 获取运行的统计数据接口
* @param startTime 开始时间 毫秒级别时间
* @param endTime 开始时间 毫秒级别时间
* @param workspaceName 工作空间名称
* @return
*/
public static ResponseResult loadStatisticData(Long startTime, Long endTime, String workspaceName){
Map<String, Object> param = new HashMap<>();
param.put("startTime", startTime);
param.put("endTime", endTime);
param.put("workspaceName", workspaceName);
String response = createHttpRequest(REQUEST_LOADSTATISTICDATA, "param=" + JSON.toJSONString(param));
log.info("--------------------获取运行的统计数据接口调用成功，结果为：{}------------------------",response);
return JSON.parseObject(response, ResponseResult.class);
}
/**
* 获取节点运行的统计数据
* @param startTime 开始时间 毫秒级别时间
* @param endTime 结束时间 毫秒级别时间
* @param workspaceName 工作空间名称
* @param flowName 工作流名称
* @return
*/
public static ResponseResult loadNodeStatisticData(Long startTime, Long endTime, String workspaceName, String flowName){
Map<String, Object> param = new HashMap<>();
param.put("startTime", startTime);
param.put("endTime", endTime);
param.put("workspaceName", workspaceName);
param.put("flowName", flowName);
String response = createHttpRequest(REQUEST_LOADNODESTATISTICDATA, "param=" + JSON.toJSONString(param));
log.info("--------------------获取节点运行的统计数据接口调用成功，结果为：{}------------------------",response);
return JSON.parseObject(response, ResponseResult.class);
}
/**
* 补批工作流 * 补批工作流
* @param workspaceName 工作空间名称 * @param workspaceName 工作空间名称
* @param flowName 工作流名称 * @param flowName 工作流名称
......
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