Skip to content

B
byit-myth-job
Commit 904bd700 by guo_minglei@163.com
Options

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

parents 038c20ef f4f25c72
Showing with 868 additions and 58 deletions
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 {
......@@ -82,10 +85,17 @@ public class ApiFlowController {
return "SUCCESS";
}
@PostMapping("reRun")
@ApiOperation("重跑")
public String reRun(String param){
apiFlowService.reRun(param);
@PostMapping("reRunJob")
@ApiOperation("重跑节点")
public String reRunJob(String param){
apiFlowService.reRunJob(param);
return "SUCCESS";
}
@PostMapping("reRunFlow")
@ApiOperation("重跑工作流")
public String reRunFlow(String param){
apiFlowService.reRunFlow(param);
return "SUCCESS";
}
......@@ -96,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请求业务处理接口
......@@ -37,11 +39,43 @@ public interface ApiFlowService {
* 重跑任务
* @param param
*/
void reRun(String param);
void reRunJob(String param);
/**
* 手动置为成功
* @param param
*/
void madeSuccess(String param);
/**
* 重跑工作流
* @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;
......@@ -416,7 +418,7 @@ public class ApiFlowServiceImpl implements ApiFlowService {
* @param param
*/
@Override
public void reRun(String param) {
public void reRunJob(String param) {
ValidationUtil.dataNotBank(param, "请求参数不允许为空！");
JSONObject jsonObject = JSON.parseObject(param);
//获取工作空间名称
......@@ -471,7 +473,7 @@ public class ApiFlowServiceImpl implements ApiFlowService {
//校验通过，开始设置重跑
//判断重跑机制（单节点重跑，节点及下游重跑
if (!"1".equals(runState)) {//如果是只重跑当前节点
if (!"1".equals(runState)) {//如果不是只重跑当前节点
//查询依赖本节点的节点，并添加到集合中
List<Integer> subNodeIdList = nodeDependencyMapper.findSubNodeList(node.getNodeId());
if (subNodeIdList != null && subNodeIdList.size() > 0){
......@@ -483,6 +485,58 @@ public class ApiFlowServiceImpl implements ApiFlowService {
}
@Override
public void reRunFlow(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, "工作流名称不允许为空！");
//获取要重跑的运行记录id
String runId = jsonObject.getString("runId");
ValidationUtil.dataNotBank(runId, "运行实例id不允许为空！");
//开始校验
Workspace workspace = workspaceMapper.getByName(workspaceName);
ValidationUtil.dataNotNull(workspace, workspaceName + "工作空间不存在");
Flow flow = flowMapper.getByWorkSpaceAndName(workspace.getWorkspaceId(), flowName);
ValidationUtil.dataNotNull(flow, flowName + "工作流不存在");
//获取运行日志实例
RunRecording runRecording = runRecordingMapper.findRunRecordingByFlowIdAndRunId(flow.getFlowId(), runId);
ValidationUtil.dataNotNull(runRecording, "没有找到对应的运行记录");
//判断是否是内嵌工作流
if (FlowPropertyEnum.IS_INNER.getCode().equals(flow.getIsInner())){//是内嵌工作流
Node node = nodeMapper.getByMapFlowId(flow.getFlowId());
Map<String, Object> requestMap = new HashMap<>(10);
requestMap.put("workspaceName", workspaceName);
requestMap.put("flowName", flowName);
requestMap.put("runId", runId);
requestMap.put("runState", "1");
requestMap.put("nodeName", node.getNodeName());
reRunJob(JSON.toJSONString(requestMap));
}else {
//不是内嵌工作流
Long triggerTime = System.currentTimeMillis();
String reRunId = UUID.randomUUID().toString().replace("-","");
List<JobTask> jobTaskList = new ArrayList<>();
List<Node> nodeList = nodeMapper.findOnforkByFlowId(flow.getFlowId());
ValidationUtil.dataNotNull(nodeList, "该工作流没有在调度上的任务");
nodeList.forEach(node -> {
JobTask jobTask = new JobTask();
BeanUtils.copyProperties(node, jobTask);
jobTask.setTriggerTime(triggerTime);
jobTask.setTriggerStatus("1");
jobTask.setRunId(reRunId);
jobTask.setReRunId(runId);
jobTaskList.add(jobTask);
});
jobTaskMapper.saveJobTasks(jobTaskList);
}
}
private void addDependNode(String reRunId, String runId, Long triggerTime, List<JobTask> jobTaskList, List<Integer> subNodeIdList) {
subNodeIdList.forEach(childNodeId -> {
Node subNode = nodeMapper.getById(childNodeId);
......@@ -538,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-admin/src/main/resources/application-dev.yml
......@@ -6,13 +6,13 @@ spring:
password: 123456
mail:
host: smtp.163.com #我自己的SMTP服务器地址
username: huangfusuper@163.com #登录用户名
password: huangfu0110 #授权密码
host: smtp.163.com #���Լ���SMTP��������ַ
username: huangfusuper@163.com #��¼�û���
password: huangfu0110 #��Ȩ����
default-encoding: UTF-8
protocol: smtp #协议
protocol: smtp #��
properties:
from: huangfusuper@163.com #真实邮箱
from: huangfusuper@163.com #��ʵ����
mybatis:
mapper-locations: /mapper/*.xml
......@@ -26,3 +26,11 @@ myth-rpc:
logging:
path: /data/mythjob
file: myth_log_file
file:
system:
ip: 10.0.120.2
port: 88
myth-job:
filestystem: FASTDFS
\ No newline at end of file
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/NodeMapper.java
......@@ -78,4 +78,11 @@ public interface NodeMapper {
* @param flowId
*/
int deleteByFlowId(Integer flowId);
/**
* 根据内嵌工作流的id来查找结点
* @param mapFlowId
* @return
*/
Node getByMapFlowId(Integer mapFlowId);
}
\ 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 0 → 100644
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/java/com/byit/service/impl/JobTaskRunLogAndJobTaskServiceImpl.java
......@@ -35,6 +35,7 @@ public class JobTaskRunLogAndJobTaskServiceImpl implements JobTaskRunLogAndJobTa
log.debug("-------------【{}节点失败重试------------------】",jobTaskRunLog);
jobTaskRunLog.setFailedRemainingCount(jobTaskRunLog.getFailedRemainingCount()-1);
jobTaskRunLog.setRunCount(jobTaskRunLog.getRunCount()+1);
jobTaskRunLog.setRunCode("0");
jobTaskRunLogService.updateJobTaskRunLog(jobTaskRunLog);
//查询节点
......
byit-myth-core/myth-admin-core/src/main/java/com/byit/task/ScriptExecutorJobTask.java
......@@ -8,6 +8,7 @@ import com.byit.job.enums.JobResultEnum;
import com.byit.model.JobTaskRunLogWithBLOBs;
import com.byit.model.JobTaskSchedule;
import com.byit.model.RunRecording;
import com.byit.rpc.util.RpcException;
import com.byit.service.RunScriptService;
import com.byit.service.impl.JobTaskRunLogServiceImpl;
import com.byit.service.impl.RunRecordingServiceImpl;
......@@ -77,6 +78,10 @@ public class ScriptExecutorJobTask implements TimerTask {
*/
private void runJob(JobTaskSchedule mythJobTaskSchedule) {
RunScriptService runScriptService = SpringUtil.getBean(RunScriptService.class);
JobTaskRunLogServiceImpl jobTaskRunLogService = SpringUtil.getBean(JobTaskRunLogServiceImpl.class);
JobTaskRunLogWithBLOBs jobTaskRunLogById = jobTaskRunLogService.findJobTaskRunLogById(mythJobTaskSchedule.getLogId());
ScriptDto scriptDto = new ScriptDto();
scriptDto.setLogId(mythJobTaskSchedule.getLogId());
scriptDto.setCommand(mythJobTaskSchedule.getRunCommand());
......@@ -84,7 +89,17 @@ public class ScriptExecutorJobTask implements TimerTask {
scriptDto.setRunId(mythJobTaskSchedule.getRunId());
scriptDto.setRemotePath(mythJobTaskSchedule.getScriptUrls());
scriptDto.setCallbackUrl("http://127.0.0.1:8998/job/callbackRes");
DispatchResponseDto dispatchResponseDto = runScriptService.runScript(scriptDto);
//二次执行的情况下 会有这个信息
scriptDto.setLogRemotePath(jobTaskRunLogById.getLogRemotelyPath());
DispatchResponseDto dispatchResponseDto = new DispatchResponseDto();
try{
dispatchResponseDto = runScriptService.runScript(scriptDto);
}catch (RpcException rpcException){
dispatchResponseDto.setMsg(rpcException.getMessage());
dispatchResponseDto.setCode(JobResultEnum.DISPATCH_FAIL.getRes());
log.error("--------------执行机不存在----------------");
}
saveLog(mythJobTaskSchedule,dispatchResponseDto);
}
......@@ -121,7 +136,7 @@ public class ScriptExecutorJobTask implements TimerTask {
if (jobTaskRunLogById.getRunCount()>1) {
//上一次的执行日志
String runMsg = jobTaskRunLogById.getRunMsg();
jobTaskRunLog.setRunMsg(runMsg+"|"+"调度失败");
jobTaskRunLog.setRunMsg(runMsg+"|"+dispatchResponseDto.getMsg());
}
}
//获取执行机地址
......
byit-myth-core/myth-admin-core/src/main/java/com/byit/thread/RunRecordingScanHelper.java
......@@ -3,6 +3,7 @@ package com.byit.thread;
import cn.hutool.core.collection.CollectionUtil;
import cn.hutool.core.date.DateUtil;
import com.byit.enums.RunRecordingEnum;
import com.byit.filesystem.FileSystem;
import com.byit.model.EmailAlarm;
import com.byit.model.JobTaskRunLogWithBLOBs;
import com.byit.model.RunRecording;
......@@ -11,14 +12,17 @@ import com.byit.service.JobTaskRunLogService;
import com.byit.service.RunRecordingService;
import com.byit.util.TimeFormatUtil;
import lombok.extern.slf4j.Slf4j;
import org.csource.common.MyException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;
import javax.sql.DataSource;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.TimeUnit;
......@@ -45,20 +49,36 @@ public class RunRecordingScanHelper {
*/
public static final String SUCCESS_DONE = "3";
/**
* 文件服务器ip
*/
@Value("${file.system.ip}")
private String fileSystemIp;
/**
* 文件服务器端口号
*/
@Value("${file.system.port}")
private String fileSystemPort;
private static final String FILE_SYSTEM_PRE="http://";
private DataSource dataSource;
private final EmailAlarmService emailAlarmService;
private final JobTaskRunLogService jobTaskRunLogService;
private final RunRecordingService runRecordingService;
private final FileSystem fileSystem;
private volatile boolean runRecordingThreadStop = false;
private Thread runRecordingThread;
@Autowired
public RunRecordingScanHelper(EmailAlarmService emailAlarmService, JobTaskRunLogService jobTaskRunLogService, RunRecordingService runRecordingService) {
public RunRecordingScanHelper(EmailAlarmService emailAlarmService, JobTaskRunLogService jobTaskRunLogService, RunRecordingService runRecordingService, FileSystem fileSystem) {
this.emailAlarmService = emailAlarmService;
this.jobTaskRunLogService = jobTaskRunLogService;
this.runRecordingService = runRecordingService;
this.fileSystem = fileSystem;
}
public void start() {
......@@ -238,7 +258,19 @@ public class RunRecordingScanHelper {
if (CollectionUtil.isNotEmpty(jobTaskRunLogs)) {
jobTaskRunLogs.forEach(jobTaskRunLog -> {
long timeConsuming = jobTaskRunLog.getEndTime().getTime() - jobTaskRunLog.getStartTime().getTime();
log.info("######################{}#####################3",timeConsuming);
/*
* String iframeHtml = "<iframe src='%s'></iframe>";
* String logFilePath = FILE_SYSTEM_PRE+fileSystemIp+":"+fileSystemPort+"/"+jobTaskRunLog.getLogRemotelyPath();
* String logFileUrl = String.format(iframeHtml,logFilePath);
*/
String logStr = null;
try {
if(null != jobTaskRunLog.getLogRemotelyPath()){
logStr = new String(fileSystem.downloaderFile(jobTaskRunLog.getLogRemotelyPath()), StandardCharsets.UTF_8);
}
} catch (IOException | MyException e) {
e.printStackTrace();
}
stringBuilder.append("<tr align='center'>")
.append(String.format("<td>%s</td>", jobTaskRunLog.getNodeName()))
.append(String.format("<td>%s</td>", DateUtil.format(jobTaskRunLog.getStartTime(), DATE_FORMAT)))
......@@ -249,7 +281,7 @@ public class RunRecordingScanHelper {
: "4".equals(jobTaskRunLog.getRunCode()) ? "补批失败" : "失败"))
.append("<td>")
.append("<div style='display:inline-block;width:100%;word-break:break-all;height: auto;overflow: auto;text-align: left;'>")
.append(String.format("%s", jobTaskRunLog.getRunMsg()))
.append(String.format("%s", logStr==null?jobTaskRunLog.getRunMsg():logStr))
.append("</div>")
.append("</td></tr>");
......
byit-myth-core/myth-admin-core/src/main/java/com/byit/util/TimeFormatUtil.java
package com.byit.util;
import java.util.concurrent.TimeUnit;
/**
* 对于时间的格式化
* @author huangfu
......@@ -13,23 +11,35 @@ public class TimeFormatUtil {
* @return
*/
public static String timeFormat(long time) {
//秒
long seconds = TimeUnit.MILLISECONDS.toSeconds(time);
long minutes = TimeUnit.MILLISECONDS.toMinutes(time);
long hours = TimeUnit.MILLISECONDS.toHours(time);
//小时
if(hours > 0){
return hours + "小时";
//分钟
} if (minutes > 0) {
return minutes + "分钟";
//秒
} else if (seconds > 0) {
return seconds + "秒";
//毫秒
} else{
return time+"毫秒";
StringBuffer sb=new StringBuffer();
long millis=1;
long seconds=1000*millis;
long minutes=60*seconds;
long hours=60*minutes;
long days=24*hours;
if(time/days>=1){
sb.append((int)(time/days)+"天");
}
if(time%days/hours>=1){
sb.append((int)(time%days/hours)+"小时");
}
if(time%days%hours/minutes>=1){
sb.append((int)(time%days%hours/minutes)+"分钟");
}
if(time%days%hours%minutes/seconds>=1){
sb.append((int)(time%days%hours%minutes/seconds)+"秒");
}
if(time%days%hours%minutes%seconds/millis>=1){
sb.append((int)(time%days%hours%minutes%seconds/millis)+"毫秒");
}
return sb.toString();
}
}
byit-myth-core/myth-admin-core/src/main/resources/mapper/FlowMapper.xml
......@@ -67,6 +67,13 @@
where flow_id = #{id,jdbcType=INTEGER}
</select>
<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
......
byit-myth-core/myth-admin-core/src/main/resources/mapper/JobTaskRunLogMapper.xml
......@@ -61,7 +61,7 @@
<select id="findErrorNode" resultMap="BaseResultMap">
select <include refid="Base_Column_List" />
from job_task_run_log
where run_code != '6' and (run_code = '2' || run_code = '4') and failed_remaining_count > 0
where (run_code != '6' || run_code != '5') and (run_code = '2' || run_code = '4') and failed_remaining_count > 0
and is_virtual = '1'
</select>
......
byit-myth-core/myth-admin-core/src/main/resources/mapper/NodeMapper.xml
......@@ -114,6 +114,15 @@
where flow_id = #{flowId,jdbcType=INTEGER}
</select>
<select id="getByMapFlowId" parameterType="integer" resultMap="BaseResultMap">
select
<include refid="Base_Column_List" />
,
<include refid="Blob_Column_List" />
from node
where map_flow_id = #{mapFlowId,jdbcType=INTEGER}
</select>
<delete id="deleteVirtualNode">
delete from node
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-core/myth-core-common/pom.xml
......@@ -13,6 +13,10 @@
<artifactId>myth-core-common</artifactId>
<dependencies>
<dependency>
<groupId>net.oschina.zcx7878</groupId>
<artifactId>fastdfs-client-java</artifactId>
</dependency>
<!-- slf4j -->
<dependency>
<groupId>org.slf4j</groupId>
......
byit-myth-core/myth-executor-core/src/main/java/com/byit/executor/filesystem/FastDfsFileSystem.java → byit-myth-core/myth-core-common/src/main/java/com/byit/filesystem/FastDfsFileSystem.java
package com.byit.executor.filesystem;
package com.byit.filesystem;
import cn.hutool.core.collection.CollectionUtil;
import org.csource.common.MyException;
import org.csource.common.NameValuePair;
import org.csource.fastdfs.*;
import org.csource.fastdfs.ClientGlobal;
import org.csource.fastdfs.StorageClient1;
import org.csource.fastdfs.TrackerClient;
import org.csource.fastdfs.TrackerServer;
import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression;
import org.springframework.stereotype.Component;
import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
......
byit-myth-core/myth-executor-core/src/main/java/com/byit/executor/filesystem/FileSystem.java → byit-myth-core/myth-core-common/src/main/java/com/byit/filesystem/FileSystem.java
package com.byit.executor.filesystem;
package com.byit.filesystem;
import org.csource.common.MyException;
......@@ -19,7 +19,7 @@ public interface FileSystem {
* @throws IOException
* @throws MyException
*/
String uploadFile(byte[] fileBuffer, String fileExtName, Map<String,String> mateDaTA) throws IOException, MyException;
String uploadFile(byte[] fileBuffer, String fileExtName, Map<String, String> mateDaTA) throws IOException, MyException;
/**
* 文件下载接口
......
byit-myth-core/myth-core-common/src/main/java/com/byit/job/dto/ScriptDto.java
......@@ -34,4 +34,8 @@ public class ScriptDto implements Serializable {
* 回调URL
*/
private String callbackUrl;
/**
* 已有的日志路径，不存在就为null
*/
private String logRemotePath;
}
byit-myth-core/myth-core-common/src/main/java/com/byit/job/dto/ScriptParamAndPlaceholderDto.java 0 → 100644
package com.byit.job.dto;
import com.sun.xml.internal.ws.developer.Serialization;
import lombok.*;
import java.io.Serializable;
import java.util.List;
import java.util.Map;
/**
* 参数和占位符的处理
* @author huangfu
*/
@Builder
@Data
@ToString
@NoArgsConstructor
@AllArgsConstructor
public class ScriptParamAndPlaceholderDto implements Serializable {
/**
* 参数的处理
*/
private Map<String,String> param;
/**
* 占位符的处理
*/
private Map<String,String> Placeholder;
}
byit-myth-core/myth-core-common/src/main/java/com/byit/job/dto/plugin/PluginNode.java
package com.byit.job.dto.plugin;
import com.alibaba.fastjson.JSON;
import com.byit.job.dto.ScriptParamAndPlaceholderDto;
import com.sun.istack.internal.NotNull;
import lombok.Data;
import lombok.NonNull;
import java.io.Serializable;
......@@ -56,5 +60,13 @@ public class PluginNode extends PluginBaseNode implements Serializable {
*/
private PluginNodeConfig config;
/**
* 脚本运行参数信息
* @param scriptParam 脚本
*/
public void setScriptParam(@NotNull ScriptParamAndPlaceholderDto scriptParam){
this.runParam = JSON.toJSONString(scriptParam);
}
private static final long serialVersionUID = 2L;
}
byit-myth-core/myth-core-common/src/main/java/com/byit/job/utils/PlaceholderUtils.java 0 → 100644
package com.byit.job.utils;
import lombok.extern.java.Log;
import lombok.extern.slf4j.Slf4j;
import java.io.BufferedReader;
import java.io.File;
import java.io.FileReader;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.charset.Charset;
import java.nio.charset.StandardCharsets;
import java.util.HashMap;
import java.util.Map;
/**
* 占位符替换工具类
* @author huangfu
*/
@Slf4j
public class PlaceholderUtils {
/**
* 占位符前缀
*/
private static final String PLACEHOLDER_PREFIX = "${";
/**
* 占位符后缀
*/
private static final String PLACEHOLDER_SUFFIX = "}";
/**
* 占位符数据替换
* @param scriptData
* @param parameter
* @return
*/
public static byte[] resolvePlaceholders(byte[] scriptData, Map<String,String> parameter){
//替换为null 不做操作
if(parameter == null || parameter.isEmpty()){
log.warn("------------占位符数据为空--------");
return scriptData;
}
//读取脚本文件
String scriptContent = new String(scriptData, StandardCharsets.UTF_8);
StringBuilder sbt = new StringBuilder(scriptContent);
//找到第一个需要替换的位置
int startIndex = sbt.indexOf(PLACEHOLDER_PREFIX);
//开始循环遍历
while (startIndex != -1){
//寻找这个替换符 结束的位置
int endIndex = sbt.indexOf(PLACEHOLDER_SUFFIX, startIndex + PLACEHOLDER_PREFIX.length());
//确定是否存在后缀
if(endIndex != -1){
//截取变量值
String placeholder = sbt.substring(startIndex + PLACEHOLDER_PREFIX.length(), endIndex);
//确定下一次查询的位置
int nextIndex = endIndex+PLACEHOLDER_SUFFIX.length();
//开始从 map中解析
if(parameter.containsKey(placeholder)){
String placeholderValue = parameter.get(placeholder);
//替换这个位置
sbt.replace(startIndex,endIndex+PLACEHOLDER_SUFFIX.length(),placeholderValue);
//再次计算替换后的查询开始位置
nextIndex = startIndex+placeholderValue.length();
}else{
throw new RuntimeException("站位解析异常");
}
startIndex = sbt.indexOf(PLACEHOLDER_PREFIX,nextIndex);
}
}
return sbt.toString().getBytes(StandardCharsets.UTF_8);
}
public static void main(String[] args) {
String text = "123${name},1232321${sex}123123";
Map<String,String> map = new HashMap<String,String>();
map.put("name","皇甫科星");
map.put("sex","'男'");
byte[] bytes = resolvePlaceholders(text.getBytes(StandardCharsets.UTF_8), map);
System.out.println(new String(bytes, StandardCharsets.UTF_8));
}
}
byit-myth-core/myth-executor-core/pom.xml
......@@ -47,10 +47,10 @@
<artifactId>guava</artifactId>
</dependency>
<dependency>
<!--<dependency>
<groupId>net.oschina.zcx7878</groupId>
<artifactId>fastdfs-client-java</artifactId>
</dependency>
</dependency>-->
<dependency>
<groupId>junit</groupId>
......
byit-myth-core/myth-executor-core/src/test/java/com/test/Test1.java
package com.test;
import com.byit.executor.filesystem.FastDfsFileSystem;
import com.byit.filesystem.FastDfsFileSystem;
import org.apache.commons.io.FileUtils;
import org.csource.common.MyException;
......@@ -12,6 +13,7 @@ import java.util.Map;
public class Test1 {
public static void main(String[] args) throws IOException, MyException {
FastDfsFileSystem fds = new FastDfsFileSystem();
fds.fileRemove("ddmp/M00/00/00/CgB4Al5XoBSAZg1qAAAAW4eTPeg3226.py");
byte[] bytes = FileUtils.readFileToByteArray(new File("D:\\2020project\\byit-myth-job\\byit-myth-core\\myth-executor-core\\src\\test\\java\\com\\test\\test-ex.py"));
Map<String,String> map = new HashMap<>();
map.put("filename","test-ex.py");
......
byit-myth-core/myth-executor-core/src/test/java/com/test/test-ex.py
import time
i = 0
while i<100:
print("--------%d" %i)
time.sleep(1)
print("-----${name}---%d" %i)
i+=1
\ No newline at end of file
byit-myth-executor/myth-executor-plugin/src/main/java/com/byit/utils/JobUtils.java
......@@ -37,10 +37,45 @@ public class JobUtils {
*/
private static final String REQUEST_WORKSPACE_ADD = "/api/workspace/add";
/**
* 开始工作流
*/
private static final String REQUEST_FLOW_START = "/api/flow/start";
/**
* 暂停工作流
*/
private static final String REQUEST_FLOW_STOP = "/api/flow/stopSchedule";
/**
* 删除工作流
*/
private static final String REQUEST_FLOW_DELETE = "/api/flow/deleteFlow";
/**
* 撤销工作流调度
*/
private static final String REQUEST_FLOW_REPEAL = "/api/flow/repealSchedule";
/**
* 重新开始撤销的工作流的调度
*/
private static final String REQUEST_FLOW_REREPEAL = "/api/flow/reStartSchedule";
/**
* 杀死任务
*/
private static final String REQUEST_FLOW_KILL_JOB = "/api/flow/killJob";
/**
* 杀死任务
*/
private static final String REQUEST_FLOW_KILL_FLOW = "/api/flow/killFlow";
/**
* 重跑节点
*/
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";
private static final String SERVER_PORT = "8998";
/**
* 当前项目运行环境 jar file
......@@ -96,6 +131,24 @@ public class JobUtils {
* @param workspaceName
* @return
*/
public static String startFlow(String flowName, String workspaceName){
//请求的路径
String requestUrl = REQUEST_PREFIX + "127.0.0.1" + ":" + SERVER_PORT + REQUEST_FLOW_START;
Map<String,String> map = new HashMap<>(5);
map.put("flowName",flowName);
map.put("workspaceName",workspaceName);
//发送请求 添加任务
String addRequestResult = HttpUtil.post(requestUrl,"param="+JSON.toJSONString(map,WriteClassName));
log.info("--------------------开始接口调用成功，结果为：{}------------------------",addRequestResult);
return addRequestResult;
}
/**
* 暂停工作流
* @param flowName
* @param workspaceName
* @return
*/
public static String stopFlow(String flowName, String workspaceName){
//请求的路径
String requestUrl = REQUEST_PREFIX + "127.0.0.1" + ":" + SERVER_PORT + REQUEST_FLOW_STOP;
......@@ -105,7 +158,161 @@ public class JobUtils {
//发送请求 添加任务
String addRequestResult = HttpUtil.post(requestUrl,"param="+JSON.toJSONString(map,WriteClassName));
//String addRequestResult = HttpUtil.post(requestUrl, JSON.toJSONString(pluginPackage, WriteClassName))
log.info("--------------------暂停成功，结果为：{}------------------------",addRequestResult);
log.info("--------------------暂停接口调用成功，结果为：{}------------------------",addRequestResult);
return addRequestResult;
}
/**
* 删除工作流
* @param flowName
* @param workspaceName
* @return
*/
public static String deleteFlow(String flowName, String workspaceName){
//请求的路径
String requestUrl = REQUEST_PREFIX + "127.0.0.1" + ":" + SERVER_PORT + REQUEST_FLOW_DELETE;
Map<String,String> map = new HashMap<>(5);
map.put("flowName",flowName);
map.put("workspaceName",workspaceName);
//发送请求 添加任务
String addRequestResult = HttpUtil.post(requestUrl,"param="+JSON.toJSONString(map,WriteClassName));
log.info("--------------------删除接口调用成功，结果为：{}------------------------",addRequestResult);
return addRequestResult;
}
/**
* 删除工作流
* @param flowName
* @param workspaceName
* @return
*/
public static String repealSchedule(String flowName, String workspaceName){
//请求的路径
String requestUrl = REQUEST_PREFIX + "127.0.0.1" + ":" + SERVER_PORT + REQUEST_FLOW_REPEAL;
Map<String,String> map = new HashMap<>(5);
map.put("flowName",flowName);
map.put("workspaceName",workspaceName);
//发送请求 添加任务
String addRequestResult = HttpUtil.post(requestUrl,"param="+JSON.toJSONString(map,WriteClassName));
log.info("--------------------删除接口调用成功，结果为：{}------------------------",addRequestResult);
return addRequestResult;
}
/**
* 重新开始某次调度
* @param runId
* @return
*/
public static String startSchedule(String runId){
//请求的路径
String requestUrl = REQUEST_PREFIX + "127.0.0.1" + ":" + SERVER_PORT + REQUEST_FLOW_REREPEAL;
Map<String,Object> map = new HashMap<>(5);
map.put("runId", runId);
//发送请求 添加任务
String addRequestResult = HttpUtil.post(requestUrl, map);
log.info("--------------------重新开始调度接口调用成功，结果为：{}------------------------",addRequestResult);
return addRequestResult;
}
/**
* 杀死任务
* @param runId
* @param flowName
* @param nodeName
* @return
*/
public static String killJob(String runId, String flowName, String nodeName){
//请求的路径
String requestUrl = REQUEST_PREFIX + "127.0.0.1" + ":" + SERVER_PORT + REQUEST_FLOW_KILL_JOB;
Map<String,String> map = new HashMap<>(5);
map.put("runId", runId);
map.put("flowName", flowName);
map.put("nodeName", nodeName);
//发送请求 添加任务
String addRequestResult = HttpUtil.post(requestUrl,"param="+JSON.toJSONString(map,WriteClassName));
log.info("--------------------杀死任务接口调用成功，结果为：{}------------------------",addRequestResult);
return addRequestResult;
}
/**
* 杀死工作流
* @param runId
* @return
*/
public static String killFlow(String runId){
//请求的路径
String requestUrl = REQUEST_PREFIX + "127.0.0.1" + ":" + SERVER_PORT + REQUEST_FLOW_KILL_FLOW;
Map<String, Object> map = new HashMap<>(2);
map.put("runId", runId);
//发送请求 添加任务
String addRequestResult = HttpUtil.post(requestUrl, map);
log.info("--------------------杀死工作流接口调用成功，结果为：{}------------------------",addRequestResult);
return addRequestResult;
}
/**
* 重跑节点
* @param runId
* @param runState
* @param workspaceName
* @param flowName
* @param nodeName
* @return
*/
public static String reRunJob(String runId, String runState, String workspaceName, String flowName, String nodeName){
//请求的路径
String requestUrl = REQUEST_PREFIX + "127.0.0.1" + ":" + SERVER_PORT + REQUEST_FLOW_RERUNJOB;
Map<String,String> map = new HashMap<>(5);
map.put("runId",runId);
map.put("runState",runState);
map.put("workspaceName",workspaceName);
map.put("flowName",flowName);
map.put("nodeName",nodeName);
//发送请求 添加任务
String addRequestResult = HttpUtil.post(requestUrl,"param="+JSON.toJSONString(map,WriteClassName));
log.info("--------------------重跑节点接口调用成功，结果为：{}------------------------",addRequestResult);
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
* @param workspaceName
* @param flowName
* @param nodeName
* @return
*/
public static String makeSuccess(String runId, String workspaceName, String flowName, String nodeName){
//请求的路径
String requestUrl = REQUEST_PREFIX + "127.0.0.1" + ":" + SERVER_PORT + REQUEST_FLOW_MAKESUCCESS;
Map<String,String> map = new HashMap<>(5);
map.put("runId",runId);
map.put("workspaceName",workspaceName);
map.put("flowName",flowName);
map.put("nodeName",nodeName);
//发送请求 添加任务
String addRequestResult = HttpUtil.post(requestUrl,"param="+JSON.toJSONString(map,WriteClassName));
log.info("--------------------手动置为成功接口调用成功，结果为：{}------------------------",addRequestResult);
return addRequestResult;
}
......@@ -121,7 +328,7 @@ public class JobUtils {
String requestUrl = REQUEST_PREFIX + "127.0.0.1" + ":" + SERVER_PORT + REQUEST_WORKSPACE_ADD;
//发送请求 添加任务
String addRequestResult = HttpUtil.post(requestUrl, "workspaceName="+ workspaceName);
log.info("--------------------添加任务完成，添加结果为：{}------------------------",addRequestResult);
log.info("--------------------创建工作空间接口调用成功，结果为：{}------------------------",addRequestResult);
return addRequestResult;
}
......
byit-myth-executor/myth-executor-server/src/main/java/com/byit/service/ScriptExecutorServiceImpl.java
......@@ -2,13 +2,15 @@ package com.byit.service;
import com.alibaba.fastjson.JSON;
import com.byit.executor.api.ScriptExecutorService;
import com.byit.executor.filesystem.FileSystem;
import com.byit.executor.jobExecutor.process.MythJobProcess;
import com.byit.executor.jobExecutor.process.ProcessFailureException;
import com.byit.filesystem.FileSystem;
import com.byit.job.dto.DispatchResponseDto;
import com.byit.job.dto.JobRunResultDto;
import com.byit.job.dto.ScriptDto;
import com.byit.job.dto.ScriptParamAndPlaceholderDto;
import com.byit.job.enums.JobResultEnum;
import com.byit.job.utils.PlaceholderUtils;
import com.byit.job.vo.ReturnResult;
import com.byit.pool.RunThreadPool;
import com.byit.rpc.remoting.provider.annotation.RpcService;
......@@ -62,19 +64,39 @@ public class ScriptExecutorServiceImpl implements ScriptExecutorService {
String remotePath = scriptDto.getRemotePath();
String command = scriptDto.getCommand();
String scriptPath = byteArrayToFile(remotePath);
String param = scriptDto.getParam();
ScriptParamAndPlaceholderDto scriptParamAndPlaceholderDto = null;
if(null != param){
scriptParamAndPlaceholderDto = JSON.parseObject(param, ScriptParamAndPlaceholderDto.class);
}
String scriptPath = byteArrayToFile(remotePath,scriptParamAndPlaceholderDto);
//开始执行脚本
List<String> cmdList = Arrays.asList(command,scriptPath);
MythJobProcess mythJobProcess = new MythJobProcess(cmdList, null, null, scriptDto.getLogId());
//保存日志
String logData = mythJobProcess.call();
byte[] logDataByte = stringToByteArray(logData);
Map<String,String> fileMateData = new HashMap<String,String>(2);
fileMateData.put("filename",scriptDto.getRunId()+scriptDto.getRunId()+".log");
String logPath = "";
try {
//不为空 则追加
if(scriptDto.getLogRemotePath() != null){
byte[] sourceLogByte = fileSystem.downloaderFile(scriptDto.getLogRemotePath());
byte[] resultLogByte = mergeFile(sourceLogByte, logDataByte);
//上传日志文件
logPath = fileSystem.uploadFile(resultLogByte,"log",fileMateData);
//删除原有的日志文件
fileSystem.fileRemove(scriptDto.getLogRemotePath());
}else{
//上传日志文件
logPath = fileSystem.uploadFile(logDataByte,"log",fileMateData);
jobRunResultDto.setReturnResult(ReturnResult.SUCCESS);
}
} catch (ProcessFailureException ignored){
jobRunResultDto.setReturnResult(ReturnResult.FAIL);
} catch (IOException | MyException e) {
......@@ -106,11 +128,25 @@ public class ScriptExecutorServiceImpl implements ScriptExecutorService {
return dispatchResponseDto;
}
/**
* 合并两个字节数组
* @param sourceByte
* @param targetByte
* @return
*/
private byte[] mergeFile(byte[] sourceByte, byte[] targetByte){
byte[] result = new byte[sourceByte.length+targetByte.length];
System.arraycopy(sourceByte,0,result,0,sourceByte.length);
System.arraycopy(targetByte,0,result,sourceByte.length,targetByte.length);
return result;
}
/**
* 将脚本字节转换成文件
* @return 生成文件的本地路径
*/
private String byteArrayToFile(String remotePath){
private String byteArrayToFile(String remotePath,ScriptParamAndPlaceholderDto scriptParamAndPlaceholderDto){
//创建目录
File rootPathMkdir = new File(rootScriptPath,formatDate());
if(!rootPathMkdir.exists()){
......@@ -123,6 +159,12 @@ public class ScriptExecutorServiceImpl implements ScriptExecutorService {
try {
//下载脚本文件
byte[] scriptByteArray = fileSystem.downloaderFile(remotePath);
/**
* 替换脚本占位符
*/
if(scriptParamAndPlaceholderDto != null){
scriptByteArray = PlaceholderUtils.resolvePlaceholders(scriptByteArray,scriptParamAndPlaceholderDto.getPlaceholder());
}
//获取文件元信息
Map<String, String> fileMate = fileSystem.getFileMate(remotePath);
String fileName = fileMate.get("filename");
......
demo-client/byit-demo-client/src/main/java/com/byit/job/ScriptAddFlow.java
package com.byit.job;
import com.alibaba.fastjson.JSON;
import com.byit.job.dto.ScriptParamAndPlaceholderDto;
import com.byit.job.dto.plugin.*;
import com.byit.rpc.remoting.invoker.route.LoadBalance;
import com.byit.utils.JobUtils;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;
import java.util.*;
import java.util.concurrent.TimeUnit;
public class ScriptAddFlow {
......@@ -70,9 +70,15 @@ public class ScriptAddFlow {
pluginNode2.setType("node");
pluginNode2.setAuthor("皇甫");
pluginNode2.setJobType("SCRIPT");
pluginNode2.setScriptUrls("ddmp/M00/00/00/CgB4Al5T7xuACemOAAAAOeUjOqs3779.py");
pluginNode2.setScriptUrls("ddmp/M00/00/00/CgB4Al5d3xmAdhrwAAAAYsb__Ck6673.py");
ScriptParamAndPlaceholderDto scriptParamAndPlaceholderDto = new ScriptParamAndPlaceholderDto();
Map<String,String> map = new HashMap<>();
map.put("name","皇甫科星");
scriptParamAndPlaceholderDto.setPlaceholder(map);
pluginNode2.setScriptParam(scriptParamAndPlaceholderDto);
pluginNode2.setRunCommand("python");
pluginNode2.setRunParam("test1");
pluginNodeConfig2.setFailedRetryCount(2);
pluginNodeConfig2.setFailedRetryInterval(TimeUnit.MINUTES.toSeconds(2));
pluginNodeConfig2.setNodeCron("0 0/7 * * * ? *");
......
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