Commit be7f205a by guo_minglei@163.com

补批代码修改

parent 7593c888
......@@ -134,12 +134,6 @@ public class ApiFlowController {
return result;
}
@PostMapping("getLogUrl")
@ApiOperation("获取日志文件的url地址")
public String getLogUrl(String param){
String logUrl = apiFlowService.getLogUrl(param);
return logUrl;
}
/**
* 补批节点
......
......@@ -73,12 +73,6 @@ public interface ApiFlowService {
*/
void repairFlow(String param);
/**
* 获取运行日志的存放地址
* @param param
* @return
*/
String getLogUrl(String param);
/**
* 加载运行日志
......
......@@ -735,9 +735,28 @@ public class ApiFlowServiceImpl implements ApiFlowService {
//获取工作流名称
String flowName = jsonObject.getString("flowName");
ValidationUtil.dataNotBank(flowName, "工作流名称不允许为空!");
String nodeNames = jsonObject.getString("nodeNames");
ValidationUtil.dataNotBank(nodeNames, "补批节点不允许为空!");
List<String> nodeNameList = Arrays.asList(nodeNames.split(","));
//获取补批的日期
String repairTime = jsonObject.getString("repairTime");
ValidationUtil.dataNotBank(repairTime, "补批日期不允许为空!");
String repairTimes = jsonObject.getString("repairTimes");
ValidationUtil.dataNotBank(repairTimes, "补批日期不允许为空!");
List<String> repairTimeList = Arrays.asList(repairTimes.split(","));
//开始校验
Workspace workspace = workspaceMapper.getByName(workspaceName);
ValidationUtil.dataNotNull(workspace, workspaceName + "工作空间不存在");
Flow flow = flowMapper.getByWorkSpaceAndName(workspace.getWorkspaceId(), flowName);
ValidationUtil.dataNotNull(flow, flowName + "工作流不存在");
List<Node> nodeList = nodeMapper.findByFlowIdAndName(flow.getFlowId(), nodeNameList);
//校验是否存在未运行结束的工作流实例
List<RunRecording> unFinishList = runRecordingMapper.findUnFinishByFlowId(flow.getFlowId());
//校验时间
for (String repairTime : repairTimeList){
try {
Date repairDate = sdf.parse(repairTime);
ValidationUtil.isTrueValidation(!repairDate.before(new Date()), "只能补过去时间的批次!");
......@@ -745,17 +764,17 @@ public class ApiFlowServiceImpl implements ApiFlowService {
log.error("补批日期不符合规范,例:20200101");
ValidationUtil.isTrueValidation(true, "补批日期不符合规范,例:20200101");
}
//TODO 替换参数
//开始校验
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) {
}
private List<WaitingTask> buildTask(List<Node> nodeList, String repairTimeList){
return null;
}
......
......@@ -55,7 +55,8 @@
type="XMLMAPPER">
<property name="enableSubPackages" value="false"/>
</javaClientGenerator>
<table tableName="job_flow_current" domainObjectName="JobFlowCurrent" />
<table tableName="waiting_record" domainObjectName="WaitingRecord" />
<table tableName="waiting_task" domainObjectName="WaitingTask" />
<!--<table tableName="t_publish_result" domainObjectName="PublishResult" />
<table tableName="t_publish_approve" domainObjectName="PublishApprove" />-->
</context>
......
......@@ -85,4 +85,12 @@ public interface NodeMapper {
* @return
*/
Node getByMapFlowId(Integer mapFlowId);
/**
* 根据工作流id和任务名称查找任务集合
* @param flowId
* @param nodeNameList
* @return
*/
List<Node> findByFlowIdAndName(@Param("flowId") Integer flowId, @Param("nodeNameList")List<String> nodeNameList);
}
\ No newline at end of file
......@@ -132,4 +132,11 @@ public interface RunRecordingMapper {
* @return
*/
RunRecording findByRunIdAndFlowName(@Param("runId")String runId, @Param("flowName")String flowName);
/**
* 查询未完成的工作流实例
* @param flowId
* @return
*/
List<RunRecording> findUnFinishByFlowId(Integer flowId);
}
\ No newline at end of file
package com.byit.mapper;
import com.byit.model.WaitingRecord;
public interface WaitingRecordMapper {
int deleteById(Integer waitId);
int insert(WaitingRecord record);
int insertSelective(WaitingRecord record);
WaitingRecord getById(Integer waitId);
int updateByIdSelective(WaitingRecord record);
int updateById(WaitingRecord record);
}
\ No newline at end of file
package com.byit.mapper;
import com.byit.model.WaitingTask;
public interface WaitingTaskMapper {
int deleteById(Integer id);
int insert(WaitingTask record);
int insertSelective(WaitingTask record);
WaitingTask getById(Integer id);
int updateByIdSelective(WaitingTask record);
int updateByIdWithBLOBs(WaitingTask record);
int updateById(WaitingTask record);
}
\ No newline at end of file
package com.byit.model;
import io.swagger.annotations.ApiModel;
import io.swagger.annotations.ApiModelProperty;
import java.io.Serializable;
import lombok.Data;
/**
*
*/
@ApiModel
@Data
public class WaitingRecord implements Serializable {
/**
* 运行标识
*/
@ApiModelProperty("运行标识")
private Integer waitId;
/**
* 工作流Id
*/
@ApiModelProperty("工作流Id")
private Integer flowId;
/**
* 工作流的名字
*/
@ApiModelProperty("工作流的名字")
private String flowName;
/**
* 工作流的版本名称
*/
@ApiModelProperty("工作流的版本名称")
private String flowVersionName;
/**
* 工作流的超时时间
*/
@ApiModelProperty("工作流的超时时间")
private Long flowTimeout;
/**
* 当前工作流版本的报警邮箱
*/
@ApiModelProperty("当前工作流版本的报警邮箱")
private String alarmEmail;
/**
* 当前工作流版本的告警的时机(0 不告警, 1 完成时告警, 2 失败时告警, 3 成功时告警)
*/
@ApiModelProperty("当前工作流版本的告警的时机(0 不告警, 1 完成时告警, 2 失败时告警, 3 成功时告警)")
private String alarmlAction;
/**
* 设置任务的优先级,1最低 2最高
*/
@ApiModelProperty("设置任务的优先级,1最低 2最高")
private String priority;
/**
* 本次任务的执行时间
*/
@ApiModelProperty("本次任务的执行时间")
private Long triggerTime;
/**
* 责任人
*/
@ApiModelProperty("责任人")
private String principal;
/**
* 是否是内嵌工作流 1 否, 0 是
*/
@ApiModelProperty("是否是内嵌工作流 1 否, 0 是")
private String isInner;
/**
* 工作流节点数量
*/
@ApiModelProperty("工作流节点数量")
private Integer flowNodeCount;
/**
* 跑批类型 1 正常跑批 2 重跑 3 补批
*/
@ApiModelProperty("跑批类型 1 正常跑批 2 重跑 3 补批")
private Integer scheduleType;
/**
* 重跑和补批的操作人
*/
@ApiModelProperty("重跑和补批的操作人")
private String operator;
/**
* 针对工作流设置的排队的从小到大编号
*/
@ApiModelProperty("针对工作流设置的排队的从小到大编号")
private Integer waitOrder;
/**
* 运行标识
*/
@ApiModelProperty("运行标识")
private String runId;
/**
*/
private static final long serialVersionUID = 1L;
}
\ No newline at end of file
package com.byit.model;
import io.swagger.annotations.ApiModel;
import io.swagger.annotations.ApiModelProperty;
import java.io.Serializable;
import lombok.Data;
/**
*
*/
@ApiModel
@Data
public class WaitingTask implements Serializable {
/**
*/
@ApiModelProperty("")
private Integer id;
/**
* 运行标识
*/
@ApiModelProperty("运行标识")
private String runId;
/**
* 当前版本节点主键
*/
@ApiModelProperty("当前版本节点主键")
private Integer nodeId;
/**
* 当前节点的名称
*/
@ApiModelProperty("当前节点的名称")
private String nodeName;
/**
* 节点的说明
*/
@ApiModelProperty("节点的说明")
private String nodeDesc;
/**
* 当前任务的类型 java python shell sql script
*/
@ApiModelProperty("当前任务的类型 java python shell sql script")
private String jobType;
/**
* 本地节点(插件方) 的节点的名字
*/
@ApiModelProperty("本地节点(插件方) 的节点的名字")
private String handlerName;
/**
* 节点的超时时间 -1不超时
*/
@ApiModelProperty("节点的超时时间 -1不超时")
private Long nodeTimeout;
/**
* 工作流ID
*/
@ApiModelProperty("工作流ID")
private Integer flowId;
/**
* 工作流版本名称
*/
@ApiModelProperty("工作流版本名称")
private String versionName;
/**
* 插件的网络令牌
*/
@ApiModelProperty("插件的网络令牌")
private String pluginToken;
/**
* 插件端请求调度中心的令牌
*/
@ApiModelProperty("插件端请求调度中心的令牌")
private String gatewayToken;
/**
* 映射的工作流id(当节点为虚节点,有意义)
*/
@ApiModelProperty("映射的工作流id(当节点为虚节点,有意义)")
private Integer mapFlowId;
/**
* 是否是虚节点(0 是, 1 否),虚节点代表是内嵌工作流
*/
@ApiModelProperty("是否是虚节点(0 是, 1 否),虚节点代表是内嵌工作流")
private String isVirtual;
/**
* 插件端的url集合
*/
@ApiModelProperty("插件端的url集合")
private String pluginUrls;
/**
* 设置任务的优先级,1最低 2最高
*/
@ApiModelProperty("设置任务的优先级,1最低 2最高")
private String priority;
/**
* 当前节点的失败重试次数
*/
@ApiModelProperty("当前节点的失败重试次数")
private Integer failedRetryCount;
/**
* 失败重试的间隔(单位毫秒)
*/
@ApiModelProperty("失败重试的间隔(单位毫秒)")
private Long failedRetryInterval;
/**
* 阻塞策略
*/
@ApiModelProperty("阻塞策略")
private String blockStrategy;
/**
* 路由策略
*/
@ApiModelProperty("路由策略")
private String routingStrategy;
/**
* 节点的参数
*/
@ApiModelProperty("节点的参数")
private String runParam;
/**
* 源码备注
*/
@ApiModelProperty("源码备注")
private String runSourceDesc;
/**
* 运行命令
*/
@ApiModelProperty("运行命令")
private String runCommand;
/**
* 脚本的文件服务器路径集
*/
@ApiModelProperty("脚本的文件服务器路径集")
private String scriptUrls;
/**
* 源码负责人
*/
@ApiModelProperty("源码负责人")
private String sourcePrincipal;
/**
* 节点的本次执行时间
*/
@ApiModelProperty("节点的本次执行时间")
private Long triggerTime;
/**
* 工作流的名称
*/
@ApiModelProperty("工作流的名称")
private String flowName;
/**
* 是否上级运行成功时运行(0 是 1 否)
*/
@ApiModelProperty("是否上级运行成功时运行(0 是 1 否)")
private String superSuccessRun;
/**
* 重跑的运行标识
*/
@ApiModelProperty("重跑的运行标识")
private String reRunId;
/**
* 节点的依赖节点id集合
*/
@ApiModelProperty("节点的依赖节点id集合")
private String nodeDepend;
/**
* 重跑和补批的操作人
*/
@ApiModelProperty("重跑和补批的操作人")
private String operator;
/**
* 跑批类型 1 正常跑批 2 重跑 3 补批
*/
@ApiModelProperty("跑批类型 1 正常跑批 2 重跑 3 补批")
private Integer scheduleType;
/**
* 源码
*/
@ApiModelProperty("源码")
private String runSource;
/**
*/
private static final long serialVersionUID = 1L;
}
\ No newline at end of file
......@@ -55,7 +55,8 @@
type="XMLMAPPER">
<property name="enableSubPackages" value="false"/>
</javaClientGenerator>
<table tableName="node_version_dependency" domainObjectName="NodeVersionDependency" />
<table tableName="waiting_record" domainObjectName="WaitingRecord" />
<table tableName="waiting_task" domainObjectName="WaitingTask" />
</context>
......
......@@ -123,6 +123,20 @@
where map_flow_id = #{mapFlowId,jdbcType=INTEGER}
</select>
<select id="findByFlowIdAndName" resultMap="BaseResultMap">
select
<include refid="Base_Column_List" />
,
<include refid="Blob_Column_List" />
from node
where flow_id = #{flowId,jdbcType=INTEGER}
and node_name in (
<foreach collection="nodeNameList" item="nodeName" separator=",">
#{nodeName}
</foreach>
)
</select>
<delete id="deleteVirtualNode">
delete from node
where flow_id = #{flowId,jdbcType=INTEGER}
......
......@@ -113,6 +113,13 @@
from run_recording
where run_id = #{runId,jdbcType=VARCHAR} and flow_name = #{flowName,jdbcType=VARCHAR}
</select>
<select id="findUnFinishByFlowId" resultMap="BaseResultMap">
select
<include refid="Base_Column_List" />
from run_recording
where flow_id = #{flowId}
and flow_status != '4'
</select>
<delete id="deleteById" parameterType="java.lang.Integer">
<!-- generated @mbg.generated date: 2019-12-25 -->
......
<?xml version="1.0" encoding="UTF-8"?>
<!DOCTYPE mapper PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN" "http://mybatis.org/dtd/mybatis-3-mapper.dtd">
<mapper namespace="com.byit.mapper.WaitingRecordMapper">
<resultMap id="BaseResultMap" type="com.byit.model.WaitingRecord">
<!-- generated @mbg.generated date: 2020-03-11 -->
<id column="wait_id" jdbcType="INTEGER" property="waitId" />
<result column="flow_id" jdbcType="INTEGER" property="flowId" />
<result column="flow_name" jdbcType="VARCHAR" property="flowName" />
<result column="flow_version_name" jdbcType="VARCHAR" property="flowVersionName" />
<result column="flow_timeout" jdbcType="BIGINT" property="flowTimeout" />
<result column="alarm_email" jdbcType="VARCHAR" property="alarmEmail" />
<result column="alarml_action" jdbcType="CHAR" property="alarmlAction" />
<result column="priority" jdbcType="CHAR" property="priority" />
<result column="trigger_time" jdbcType="BIGINT" property="triggerTime" />
<result column="principal" jdbcType="VARCHAR" property="principal" />
<result column="is_inner" jdbcType="CHAR" property="isInner" />
<result column="flow_node_count" jdbcType="INTEGER" property="flowNodeCount" />
<result column="schedule_type" jdbcType="INTEGER" property="scheduleType" />
<result column="operator" jdbcType="VARCHAR" property="operator" />
<result column="wait_order" jdbcType="INTEGER" property="waitOrder" />
<result column="run_id" jdbcType="VARCHAR" property="runId" />
</resultMap>
<sql id="Base_Column_List">
<!-- generated @mbg.generated date: 2020-03-11 -->
wait_id, flow_id, flow_name, flow_version_name, flow_timeout, alarm_email, alarml_action,
priority, trigger_time, principal, is_inner, flow_node_count, schedule_type, `operator`,
wait_order, run_id
</sql>
<select id="getById" parameterType="java.lang.Integer" resultMap="BaseResultMap">
<!-- generated @mbg.generated date: 2020-03-11 -->
select
<include refid="Base_Column_List" />
from waiting_record
where wait_id = #{waitId,jdbcType=INTEGER}
</select>
<delete id="deleteById" parameterType="java.lang.Integer">
<!-- generated @mbg.generated date: 2020-03-11 -->
delete from waiting_record
where wait_id = #{waitId,jdbcType=INTEGER}
</delete>
<insert id="insert" parameterType="com.byit.model.WaitingRecord">
<!-- generated @mbg.generated date: 2020-03-11 -->
insert into waiting_record (wait_id, flow_id, flow_name,
flow_version_name, flow_timeout, alarm_email,
alarml_action, priority, trigger_time,
principal, is_inner, flow_node_count,
schedule_type, `operator`, wait_order,
run_id)
values (#{waitId,jdbcType=INTEGER}, #{flowId,jdbcType=INTEGER}, #{flowName,jdbcType=VARCHAR},
#{flowVersionName,jdbcType=VARCHAR}, #{flowTimeout,jdbcType=BIGINT}, #{alarmEmail,jdbcType=VARCHAR},
#{alarmlAction,jdbcType=CHAR}, #{priority,jdbcType=CHAR}, #{triggerTime,jdbcType=BIGINT},
#{principal,jdbcType=VARCHAR}, #{isInner,jdbcType=CHAR}, #{flowNodeCount,jdbcType=INTEGER},
#{scheduleType,jdbcType=INTEGER}, #{operator,jdbcType=VARCHAR}, #{waitOrder,jdbcType=INTEGER},
#{runId,jdbcType=VARCHAR})
</insert>
<insert id="insertSelective" parameterType="com.byit.model.WaitingRecord">
<!-- generated @mbg.generated date: 2020-03-11 -->
insert into waiting_record
<trim prefix="(" suffix=")" suffixOverrides=",">
<if test="waitId != null">
wait_id,
</if>
<if test="flowId != null">
flow_id,
</if>
<if test="flowName != null">
flow_name,
</if>
<if test="flowVersionName != null">
flow_version_name,
</if>
<if test="flowTimeout != null">
flow_timeout,
</if>
<if test="alarmEmail != null">
alarm_email,
</if>
<if test="alarmlAction != null">
alarml_action,
</if>
<if test="priority != null">
priority,
</if>
<if test="triggerTime != null">
trigger_time,
</if>
<if test="principal != null">
principal,
</if>
<if test="isInner != null">
is_inner,
</if>
<if test="flowNodeCount != null">
flow_node_count,
</if>
<if test="scheduleType != null">
schedule_type,
</if>
<if test="operator != null">
`operator`,
</if>
<if test="waitOrder != null">
wait_order,
</if>
<if test="runId != null">
run_id,
</if>
</trim>
<trim prefix="values (" suffix=")" suffixOverrides=",">
<if test="waitId != null">
#{waitId,jdbcType=INTEGER},
</if>
<if test="flowId != null">
#{flowId,jdbcType=INTEGER},
</if>
<if test="flowName != null">
#{flowName,jdbcType=VARCHAR},
</if>
<if test="flowVersionName != null">
#{flowVersionName,jdbcType=VARCHAR},
</if>
<if test="flowTimeout != null">
#{flowTimeout,jdbcType=BIGINT},
</if>
<if test="alarmEmail != null">
#{alarmEmail,jdbcType=VARCHAR},
</if>
<if test="alarmlAction != null">
#{alarmlAction,jdbcType=CHAR},
</if>
<if test="priority != null">
#{priority,jdbcType=CHAR},
</if>
<if test="triggerTime != null">
#{triggerTime,jdbcType=BIGINT},
</if>
<if test="principal != null">
#{principal,jdbcType=VARCHAR},
</if>
<if test="isInner != null">
#{isInner,jdbcType=CHAR},
</if>
<if test="flowNodeCount != null">
#{flowNodeCount,jdbcType=INTEGER},
</if>
<if test="scheduleType != null">
#{scheduleType,jdbcType=INTEGER},
</if>
<if test="operator != null">
#{operator,jdbcType=VARCHAR},
</if>
<if test="waitOrder != null">
#{waitOrder,jdbcType=INTEGER},
</if>
<if test="runId != null">
#{runId,jdbcType=VARCHAR},
</if>
</trim>
</insert>
<update id="updateByIdSelective" parameterType="com.byit.model.WaitingRecord">
<!-- generated @mbg.generated date: 2020-03-11 -->
update waiting_record
<set>
<if test="flowId != null">
flow_id = #{flowId,jdbcType=INTEGER},
</if>
<if test="flowName != null">
flow_name = #{flowName,jdbcType=VARCHAR},
</if>
<if test="flowVersionName != null">
flow_version_name = #{flowVersionName,jdbcType=VARCHAR},
</if>
<if test="flowTimeout != null">
flow_timeout = #{flowTimeout,jdbcType=BIGINT},
</if>
<if test="alarmEmail != null">
alarm_email = #{alarmEmail,jdbcType=VARCHAR},
</if>
<if test="alarmlAction != null">
alarml_action = #{alarmlAction,jdbcType=CHAR},
</if>
<if test="priority != null">
priority = #{priority,jdbcType=CHAR},
</if>
<if test="triggerTime != null">
trigger_time = #{triggerTime,jdbcType=BIGINT},
</if>
<if test="principal != null">
principal = #{principal,jdbcType=VARCHAR},
</if>
<if test="isInner != null">
is_inner = #{isInner,jdbcType=CHAR},
</if>
<if test="flowNodeCount != null">
flow_node_count = #{flowNodeCount,jdbcType=INTEGER},
</if>
<if test="scheduleType != null">
schedule_type = #{scheduleType,jdbcType=INTEGER},
</if>
<if test="operator != null">
`operator` = #{operator,jdbcType=VARCHAR},
</if>
<if test="waitOrder != null">
wait_order = #{waitOrder,jdbcType=INTEGER},
</if>
<if test="runId != null">
run_id = #{runId,jdbcType=VARCHAR},
</if>
</set>
where wait_id = #{waitId,jdbcType=INTEGER}
</update>
<update id="updateById" parameterType="com.byit.model.WaitingRecord">
<!-- generated @mbg.generated date: 2020-03-11 -->
update waiting_record
set flow_id = #{flowId,jdbcType=INTEGER},
flow_name = #{flowName,jdbcType=VARCHAR},
flow_version_name = #{flowVersionName,jdbcType=VARCHAR},
flow_timeout = #{flowTimeout,jdbcType=BIGINT},
alarm_email = #{alarmEmail,jdbcType=VARCHAR},
alarml_action = #{alarmlAction,jdbcType=CHAR},
priority = #{priority,jdbcType=CHAR},
trigger_time = #{triggerTime,jdbcType=BIGINT},
principal = #{principal,jdbcType=VARCHAR},
is_inner = #{isInner,jdbcType=CHAR},
flow_node_count = #{flowNodeCount,jdbcType=INTEGER},
schedule_type = #{scheduleType,jdbcType=INTEGER},
`operator` = #{operator,jdbcType=VARCHAR},
wait_order = #{waitOrder,jdbcType=INTEGER},
run_id = #{runId,jdbcType=VARCHAR}
where wait_id = #{waitId,jdbcType=INTEGER}
</update>
</mapper>
\ No newline at end of file
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