Skip to content

B
byit-myth-job
Commit b8bfb368 by huangfusuper
Options

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

parents 8e8528d0 6076546f
Showing with 70 additions and 130 deletions
byit-myth-admin/src/main/java/com/byit/service/impl/ApiFlowServiceImpl.java
......@@ -761,64 +761,71 @@ public class ApiFlowServiceImpl implements ApiFlowService {
}
//如果工作空间下面有数据
if (flowIdList.size() > 0){
StatisticData flowStatisticData = runRecordingMapper.findStatusByStartAndEndTime(new Date(startTime), new Date(endTime), flowIdList);
if (null == flowStatisticData){
flowStatisticData = new StatisticData();
flowStatisticData.setUnstart(flowList.size());
}else {
Integer sum = runRecordingMapper.findFlowNum(new Date(startTime), new Date(endTime), flowIdList);
if (flowList.size() > sum){
//如果工作流
flowStatisticData.setUnstart(flowList.size() - sum + flowStatisticData.getUnstart());
}
return buildStatisticDate(startTime, endTime, flowIdList);
}
return null;
}
private CollectData buildStatisticDate(Long startTime, Long endTime, List<Integer> flowIdList){
Date startDate = new Date(startTime);
Date endDate = new Date(endTime);
StatisticData flowStatisticData = runRecordingMapper.findStatusByStartAndEndTime(startDate, endDate, flowIdList);
if (null == flowStatisticData){
flowStatisticData = new StatisticData();
flowStatisticData.setUnstart(flowIdList.size());
}else {
Integer sum = runRecordingMapper.findFlowNum(new Date(startTime), new Date(endTime), flowIdList);
if (flowIdList.size() > sum){
//如果工作流
flowStatisticData.setUnstart(flowIdList.size() - sum + flowStatisticData.getUnstart());
}
}
StatisticData nodeStatisticData = jobTaskRunLogMapper.findStatusByStartAndEndTime(startDate, endDate, flowIdList);
List<Node> nodeList = nodeMapper.findbyFlowIds(flowIdList);
if (null == nodeStatisticData){
nodeStatisticData = new StatisticData();
nodeStatisticData.setUnstart(nodeList.size());
}else {
int sum = nodeStatisticData.getFail() + nodeStatisticData.getStop() + nodeStatisticData.getRunIng() + nodeStatisticData.getSuccess() + nodeStatisticData.getKill();
nodeStatisticData.setUnstart(nodeList.size() > sum ? nodeList.size()- sum : 0 );
}
List<FlowStatusSnapshoot> flowStatusSnapshootList = flowStatusSnapshootMapper.findByTime(flowIdList, startTime, endTime);
StatisticData nodeStatisticData = jobTaskRunLogMapper.findStatusByStartAndEndTime(new Date(startTime), new Date(endTime), flowIdList);
List<Node> nodeList = nodeMapper.findbyFlowIds(flowIdList);
if (null == nodeStatisticData){
nodeStatisticData = new StatisticData();
nodeStatisticData.setUnstart(nodeList.size());
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 {
int sum = nodeStatisticData.getFail() + nodeStatisticData.getStop() + nodeStatisticData.getRunIng() + nodeStatisticData.getSuccess() + nodeStatisticData.getKill();
nodeStatisticData.setUnstart(nodeList.size() > sum ? nodeList.size()- sum : 0 );
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());
}
List<FlowStatusSnapshoot> flowStatusSnapshootList = flowStatusSnapshootMapper.findByTime(flowIdList, 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);
});
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;
CollectData collectData = new CollectData();
collectData.setFlowData(flowStatisticData);
collectData.setNodeData(nodeStatisticData);
collectData.setNodeCollectMap(nodeCollectMap);
collectData.setFlowCollectMap(flowCollectMap);
return collectData;
}
/**
......@@ -875,42 +882,7 @@ public class ApiFlowServiceImpl implements ApiFlowService {
}
}
StatisticData nodeStatisticData = jobTaskRunLogMapper.findStatusByStartAndEndTime(new Date(startTime), new Date(endTime), flowIdList);
List<Node> nodeList = nodeMapper.findbyFlowIds(flowIdList);
if (null == nodeStatisticData){
nodeStatisticData = new StatisticData();
nodeStatisticData.setUnstart(nodeList.size());
}else {
int sum = nodeStatisticData.getFail() + nodeStatisticData.getStop() + nodeStatisticData.getRunIng() + nodeStatisticData.getSuccess() + nodeStatisticData.getKill();
nodeStatisticData.setUnstart(nodeList.size() > sum ? nodeList.size()- sum : 0 );
}
List<FlowStatusSnapshoot> flowStatusSnapshootList = flowStatusSnapshootMapper.findByTime(flowIdList,startTime, endTime);
Map<String, StatisticData> nodeCollectMap = new HashMap<>();
flowStatusSnapshootList.forEach(flowStatusSnapshoot -> {
String key = flowStatusSnapshoot.getDay() + "/" + flowStatusSnapshoot.getHour();
StatisticData nodeHourStatisticData = nodeCollectMap.get(key);
if (null == nodeHourStatisticData){
nodeHourStatisticData = new StatisticData();
nodeHourStatisticData.setUnstart(flowStatusSnapshoot.getUnstartNode());
nodeHourStatisticData.setRunIng(flowStatusSnapshoot.getRuningNode());
nodeHourStatisticData.setSuccess(flowStatusSnapshoot.getSuccessNode());
nodeHourStatisticData.setFail(flowStatusSnapshoot.getFailNode());
nodeHourStatisticData.setKill(flowStatusSnapshoot.getKillNode());
}else {
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());
}
nodeCollectMap.put(key, nodeHourStatisticData);
});
CollectData collectData = new CollectData();
collectData.setNodeData(nodeStatisticData);
collectData.setNodeCollectMap(nodeCollectMap);
return collectData;
return buildStatisticDate(startTime, endTime, flowIdList);
}
@Override
......
byit-myth-core/myth-admin-core/src/main/java/com/byit/service/mapservice/impl/RunRecordingAndJobTaskServiceImpl.java
......@@ -241,6 +241,7 @@ public class RunRecordingAndJobTaskServiceImpl implements RunRecordingAndJobTask
target.setTriggerStatus("1");
target.setRunId(waitingRecord.getRunId());
BeanUtils.copyProperties(waitingTask, target);
target.setReRunId(runRecording.getReRunId());
target.setId(null);
target.setTriggerTime(0L);
return target;
......
byit-myth-core/myth-admin-core/src/main/java/com/byit/thread/helper/StatusSnapshootThreadRunHelper.java
......@@ -66,7 +66,10 @@ public class StatusSnapshootThreadRunHelper extends BaseThreadRunHelper {
List<FlowStatusSnapshoot> flowStatusSnapshootList = new ArrayList<>();
flowList.forEach(flow -> {
List<RunRecording> maxRunRecordingList = runRecordingMapper.findMaxByFlowId(flow.getFlowId());
//获取今天凌晨的时间戳
Long time = LocalDate.now().atStartOfDay(ZoneOffset.ofHours(8)).toInstant().toEpochMilli();
//获取今天的运行实例
List<RunRecording> runRecordingList = runRecordingMapper.findByPreTime(new Date(time), flow.getFlowId());
FlowStatusSnapshoot flowStatusSnapshoot = new FlowStatusSnapshoot();
flowStatusSnapshoot.setDay(date);
flowStatusSnapshoot.setFlowId(flow.getFlowId());
......@@ -77,45 +80,8 @@ public class StatusSnapshootThreadRunHelper extends BaseThreadRunHelper {
//统计运行实例的数据
//如果存在实例
if (null != maxRunRecordingList && maxRunRecordingList.size() > 0){
//获取今天凌晨的时间戳
Long time = LocalDate.now().atStartOfDay(ZoneOffset.ofHours(8)).toInstant().toEpochMilli();
RunRecording maxRunRecording = maxRunRecordingList.get(0);
//从凌晨到现在是否有运行实例
if (time.longValue() > maxRunRecording.getStartTime().getTime()){
flowStatusSnapshoot.setFlowStatus(1);
flowStatusSnapshoot.setUnstartNode(flow.getFlowNodeCount());
flowStatusSnapshootList.add(flowStatusSnapshoot);
} else { //从凌晨到现在存在工作流的运行实例
//获取上一个小时的时间戳
Long preHourTime = hourDateTime.minusHours(1).toInstant(ZoneOffset.of("+8")).toEpochMilli();
//查询过去一个小时的和未完成的工作流实例
List<RunRecording> runRecordingList = runRecordingMapper.findByPreTime(new Date(preHourTime), flow.getFlowId());
//判断是否存在过去一个小时的和未完成的工作流实例
if(null != runRecordingList && runRecordingList.size() > 0){
//处理未完成和过去一个小时的工作流实例
buildUnFinish(runRecordingList, flowStatusSnapshootList, flowStatusSnapshoot, flow);
}else {
//如果不存在实例，获取上一个小时的运行快照保存
LocalDateTime preHourLocalDateTime = localDateTime.minusHours(1);
String preHourDate = preHourLocalDateTime.format(DATE_FORMATTER);
String preHourHour = String.valueOf(preHourLocalDateTime.getHour());
List<FlowStatusSnapshoot> preHourFlowStatusSnapshootList = flowStatusSnapshootMapper.findByDateAndHourAndFlowId(preHourDate, preHourHour, flow.getFlowId());
if (null != preHourFlowStatusSnapshootList && preHourFlowStatusSnapshootList.size() > 0){
preHourFlowStatusSnapshootList.forEach(preHourFlowStatusSnapshoot->{
preHourFlowStatusSnapshoot.setDay(date);
preHourFlowStatusSnapshoot.setHour(hour);
preHourFlowStatusSnapshoot.setSnapshootTime(hourTime);
flowStatusSnapshootList.add(preHourFlowStatusSnapshoot);
});
}else {
//如果上个小时也没有，就设置为未运行
flowStatusSnapshoot.setFlowStatus(1);
flowStatusSnapshoot.setUnstartNode(flow.getFlowNodeCount());
flowStatusSnapshootList.add(flowStatusSnapshoot);
}
}
}
if (null != runRecordingList && runRecordingList.size() > 0){
buildUnFinish(runRecordingList, flowStatusSnapshootList, flowStatusSnapshoot, flow);
}else {
//如果不存在运行实例
flowStatusSnapshoot.setFlowStatus(1);
......@@ -137,7 +103,7 @@ public class StatusSnapshootThreadRunHelper extends BaseThreadRunHelper {
}
/**
* 功能描述 处理过去一小时和未完成的工作流的快照
* 功能描述 当天的和未完成的工作流的快照
* @author gml
* @date 2020-05-08 10:20
* @param runRecordingList
......
byit-myth-core/myth-admin-core/src/main/resources/mapper/RunRecordingMapper.xml
......@@ -234,6 +234,7 @@
where flow_id = #{flowId}
and (
start_time &gt;= #{preHourDate}
or end_time &gt;= #{preHourDate}
or flow_status != '4'
)
</select>
......
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