Skip to content

B
byit-myth-job
Commit 06cc2029 by guominglei
Options

提供统计数据接口开发和修改

parent 313d2ce8
Showing with 187 additions and 26 deletions
byit-myth-admin/src/main/java/com/byit/api/ApiFlowController.java
......@@ -161,9 +161,28 @@ public class ApiFlowController {
return ResponseResult.ok("SUCCESS");
}
/**
* 获取整体运行的统计数据
* @param param
* @return
*/
@PostMapping("/loadStatisticData")
@ApiOperation("获取运行的统计数据")
public ResponseResult loadStatisticData(String param){
CollectData collectData = apiFlowService.loadStatisticData(param);
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 {
* @return
*/
CollectData loadStatisticData(String param);
/**
* 获取节点的汇总数据
* @param param
* @return
*/
CollectData loadNodeStatisticData(String param);
}
byit-myth-core/myth-admin-core/src/main/java/com/byit/mapper/WaitingRecordMapper.java
package com.byit.mapper;
import com.byit.dto.plugin.StatisticData;
import com.byit.model.WaitingRecord;
import org.apache.ibatis.annotations.Param;
import org.springframework.stereotype.Repository;
import java.util.List;
......@@ -31,4 +33,17 @@ public interface WaitingRecordMapper {
* @return
*/
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;
import com.byit.dto.plugin.StatisticData;
import com.byit.model.WaitingTask;
import org.apache.ibatis.annotations.Param;
import org.springframework.stereotype.Repository;
import java.util.List;
......@@ -22,4 +24,10 @@ public interface WaitingTaskMapper {
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;
import com.byit.dto.plugin.StatisticData;
import com.byit.mapper.FlowMapper;
import com.byit.mapper.FlowStatusSnapshootMapper;
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.mapper.*;
import com.byit.model.*;
import com.byit.thread.BaseThreadRunHelper;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Value;
......@@ -39,6 +34,10 @@ public class StatusSnapshootThreadRunHelper extends BaseThreadRunHelper {
@Resource
private JobTaskRunLogMapper jobTaskRunLogMapper;
@Resource
private WaitingRecordMapper waitingRecordMapper;
@Resource
private WaitingTaskMapper waitingTaskMapper;
@Resource
private DataSource dataSource;
@Value("${myth-job.snapshoot-date}")
private Integer snapshootDate;
......@@ -66,20 +65,25 @@ public class StatusSnapshootThreadRunHelper extends BaseThreadRunHelper {
List<FlowStatusSnapshoot> flowStatusSnapshootList = new ArrayList<>();
flowList.forEach(flow -> {
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){
//获取凌晨的时间戳
Long time = LocalDate.now().atStartOfDay(ZoneOffset.ofHours(8)).toInstant().toEpochMilli();
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 (time.longValue() > runRecording.getTriggerTime().longValue()){
//从凌晨到现在是否有运行实例
if (time.longValue() > runRecording.getTriggerTime().longValue()){//今天不存在工作流的运行实例
flowStatusSnapshoot.setFlowStatus(1);
flowStatusSnapshoot.setUnstartNode(flow.getFlowNodeCount());
}else {
}else { //存在工作流的运行实例
//判断运行状态
if (runRecording.getFlowStatus().equals("1") || runRecording.getFlowStatus().equals("2") || runRecording.getFlowStatus().equals("3")){
flowStatusSnapshoot.setFlowStatus(Integer.valueOf(runRecording.getFlowStatus()));
}else {
......@@ -94,6 +98,8 @@ public class StatusSnapshootThreadRunHelper extends BaseThreadRunHelper {
}
//获取运行实例下各类状态的节点数目
StatisticData nodeStatistic = jobTaskRunLogMapper.findStatisticDataByRunIdAndFlowId(runRecording.getRunId(), flow.getFlowId());
//判断是否存在节点的运行日志
//存在节点运行日志
if (nodeStatistic != null){
int sum = nodeStatistic.getFail() + nodeStatistic.getSuccess() + nodeStatistic.getRunIng() + nodeStatistic.getKill();
flowStatusSnapshoot.setUnstartNode(flow.getFlowNodeCount() - sum);
......@@ -102,12 +108,38 @@ public class StatusSnapshootThreadRunHelper extends BaseThreadRunHelper {
flowStatusSnapshoot.setFailNode(nodeStatistic.getFail());
flowStatusSnapshoot.setKillNode(nodeStatistic.getKill());
}else {
//不存在节点的运行日志
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){
flowStatusSnapshootMapper.saveList(flowStatusSnapshootList);
......
byit-myth-core/myth-admin-core/src/main/resources/mapper/JobTaskRunLogMapper.xml
......@@ -129,9 +129,9 @@
from job_task_run_log
where start_time &gt;= ${startDate}
and end_time &lt;= ${endDate}
<if test="flowIds != null">
<if test="flowIdList != null">
and flow_id in (
<foreach collection="flowIds" item="flowId" separator=",">
<foreach collection="flowIdList" item="flowId" separator=",">
#{flowId}
</foreach>
)
......
byit-myth-core/myth-admin-core/src/main/resources/mapper/RunRecordingMapper.xml
......@@ -158,9 +158,9 @@
from run_recording
where start_time &gt;= ${startDate}
and end_time &lt;= ${endDate}
<if test="flowIds != null">
<if test="flowIdList != null">
and flow_id in (
<foreach collection="flowIds" item="flowId" separator=",">
<foreach collection="flowIdList" item="flowId" separator=",">
#{flowId}
</foreach>
)
......
byit-myth-core/myth-admin-core/src/main/resources/mapper/WaitingRecordMapper.xml
......@@ -48,6 +48,24 @@
from waiting_record
where flow_id = #{flowId}
</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">
<!-- generated @mbg.generated date: 2020-03-12 -->
......
byit-myth-core/myth-admin-core/src/main/resources/mapper/WaitingTaskMapper.xml
......@@ -70,8 +70,20 @@
from waiting_task
where wait_id=#{waitId,jdbcType=INTEGER}
</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 -->
delete from waiting_task
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 {
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 {
* 获取运行实例日志
*/
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 {
}
/**
* 获取运行的统计数据接口
* @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 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