Skip to content

B
byit-myth-job
Commit 5b68486c by huangfusuper
Options

升级fastjson 增加异步日志接口节点日志回写

parent ba3f6ea8
Showing with 48 additions and 41 deletions
byit-myth-admin/src/main/java/com/byit/api/ApiCallbackController.java
......@@ -43,6 +43,6 @@ public class ApiCallbackController {
pluginRpcResponsePacket.setExtension(javaCallbackLogDto.getLogId());
pluginRpcResponsePacket.setResult(javaCallbackLogDto.getReturnResult());
pluginRpcResponsePacket.setExtension1(javaCallbackLogDto.getLogUrl());
MythJobAutoConfigure.LOG_CALLBACK.execute(new JavaTaskCallbackThread(pluginRpcResponsePacket));
MythJobAutoConfigure.LOG_CALLBACK.execute(new JavaFlowFreeTaskCallbackThread(pluginRpcResponsePacket));
}
}
byit-myth-core/myth-admin-core/src/main/java/com/byit/thread/JavaFlowFreeTaskCallbackThread.java
......@@ -7,17 +7,17 @@ import com.byit.enums.JobResultEnum;
import com.byit.enums.task.RunResultEnum;
import com.byit.filesystem.FileSystem;
import com.byit.job.utils.DateUtil;
import com.byit.job.utils.MythLogUtils;
import com.byit.model.JobTaskRunLogWithBLOBs;
import com.byit.model.RunRecording;
import com.byit.model.Workspace;
import com.byit.model.vo.RunLogVo;
import com.byit.packet.response.PluginRpcResponsePacket;
import com.byit.service.RunRecordingService;
import com.byit.service.WorkspaceService;
import com.byit.service.impl.JobTaskRunLogServiceImpl;
import com.byit.util.SpringUtil;
import com.byit.utils.ValidationUtil;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import java.io.IOException;
......@@ -32,6 +32,7 @@ import java.util.Map;
*
* @author huangfu
*/
@Slf4j
public class JavaFlowFreeTaskCallbackThread implements Runnable {
public static final String CODE = "code";
public static final String MSG = "msg";
......@@ -48,29 +49,29 @@ public class JavaFlowFreeTaskCallbackThread implements Runnable {
String logIdStr = pluginRpcResponsePacket.getExtension();
int logId = Integer.parseInt(logIdStr);
JobTaskRunLogServiceImpl mythJobTaskRunLogService = SpringUtil.getBean(JobTaskRunLogServiceImpl.class);
Map<String,String> result = new HashMap<>(2);
Map<String, String> result = new HashMap<>(2);
Object returnReultObject = pluginRpcResponsePacket.getResult();
if(returnReultObject instanceof ReturnResult) {
if (returnReultObject instanceof ReturnResult) {
ReturnResult<String> returnResult = (ReturnResult) returnReultObject;
result.put(CODE,returnResult.getCode());
result.put(MSG,returnResult.getMsg());
result.put(CONTENT,returnResult.getContent());
}else{
result = (Map<String,String>)returnReultObject;
result.put(CODE, returnResult.getCode());
result.put(MSG, returnResult.getMsg());
result.put(CONTENT, returnResult.getContent());
} else {
result = (Map<String, String>) returnReultObject;
}
JobTaskRunLogWithBLOBs jobTaskRunLog = new JobTaskRunLogWithBLOBs();
String code= result.get(CODE);
if(JobResultEnum.SUCCESS.getCode().equals(code)){
String code = result.get(CODE);
if (JobResultEnum.SUCCESS.getCode().equals(code)) {
code = RunResultEnum.RUN_SUCCESS.getCode();
}else if(JobResultEnum.DISPATCH_SUCCESS.getCode().equals(code)){
} else if (JobResultEnum.DISPATCH_SUCCESS.getCode().equals(code)) {
code = RunResultEnum.JOB_RUN_ING.getCode();
}else {
} else {
code = RunResultEnum.RUN_ERROR.getCode();
}
String logUrl = pluginRpcResponsePacket.getExtension1();
if(StringUtils.isNoneBlank(logUrl)){
if (StringUtils.isNoneBlank(logUrl)) {
jobTaskRunLog.setLogRemotelyPath(logUrl);
}
jobTaskRunLog.setRunMsg(result.get(MSG));
......@@ -78,30 +79,34 @@ public class JavaFlowFreeTaskCallbackThread implements Runnable {
jobTaskRunLog.setLogId(logId);
jobTaskRunLog.setEndTime(new Date());
String content = result.get(CONTENT);
if(StringUtils.isNoneBlank(content)){
SpecialJavaNode specialJavaNode = JSON.parseObject(content, SpecialJavaNode.class);
if(specialJavaNode != null) {
if (StringUtils.isNoneBlank(content)) {
SpecialJavaNode specialJavaNode = null;
try {
specialJavaNode = JSON.parseObject(content, SpecialJavaNode.class);
} catch (Exception e) {
log.warn("--------{}该结果集类型不属于追加型节点，不追加-----------", MythLogUtils.getMessage(e));
}
if (specialJavaNode != null) {
String flowName = specialJavaNode.getFlowName();
String nodeName = specialJavaNode.getNodeName();
String thisLogUrlPath = specialJavaNode.getThisLogUrlPath();
String logTemple = "该节点为工作流 【%s】 下的游离节点【%s】，属于二次运行：\n 本次调度信息:%s;\n 上次运行信息日志存放路径为：【%s】";
if(StringUtils.isBlank(flowName) || StringUtils.isBlank(nodeName) || StringUtils.isBlank(thisLogUrlPath)){
if (StringUtils.isBlank(flowName) || StringUtils.isBlank(nodeName) || StringUtils.isBlank(thisLogUrlPath)) {
String errorLogTemple = "该节点为工作流 【%s】 下的游离节点【%s】，属于二次运行：\n 本次调度信息:%s;\n 上次运行信息日志存放路径为：【%s】 " +
"\n\n 但是日志回写过程中发现回写参数中未携带【工作流名称】或者【节点名称】或者【日志路径信息】，本次日志追加失败！";
String logFormat = String.format(errorLogTemple, flowName, nodeName,jobTaskRunLog.getRunMsg(), thisLogUrlPath);
String logFormat = String.format(errorLogTemple, flowName, nodeName, jobTaskRunLog.getRunMsg(), thisLogUrlPath);
jobTaskRunLog.setRunMsg(logFormat);
}else{
String logFormat = String.format(logTemple, flowName, nodeName, jobTaskRunLog.getRunMsg(),thisLogUrlPath);
} else {
String logFormat = String.format(logTemple, flowName, nodeName, jobTaskRunLog.getRunMsg(), thisLogUrlPath);
jobTaskRunLog.setRunMsg(logFormat);
JobTaskRunLogWithBLOBs jobTaskRunLogWithBLOBs = buildJobTaskRunLogWithBLOBs(specialJavaNode);
if(jobTaskRunLogWithBLOBs != null) {
if (jobTaskRunLogWithBLOBs != null) {
mythJobTaskRunLogService.updateJobTaskRunLogWithBLOBs(jobTaskRunLogWithBLOBs);
}else {
} else {
logTemple = "%s\n\n 日志追加失败，失败原因是没有找到正确的节点或者没有正在运行中的实例！";
jobTaskRunLog.setRunMsg(String.format(logTemple,jobTaskRunLog.getRunMsg()));
jobTaskRunLog.setRunMsg(String.format(logTemple, jobTaskRunLog.getRunMsg()));
}
}
......@@ -117,27 +122,27 @@ public class JavaFlowFreeTaskCallbackThread implements Runnable {
String workspaceName = specialJavaNode.getWorkspaceName();
WorkspaceService workspaceService = SpringUtil.getBean(WorkspaceService.class);
Workspace byName = workspaceService.getByName(workspaceName);
if(byName == null) {
if (byName == null) {
return null;
}
RunRecording runRecording = bean.runIngRunRecording(flowName,byName.getWorkspaceId());
if(runRecording == null) {
RunRecording runRecording = bean.runIngRunRecording(flowName, byName.getWorkspaceId());
if (runRecording == null) {
return null;
}
List<JobTaskRunLogWithBLOBs> jobTaskRunLogWithBLOBsByFlowIdAndRunId = mythJobTaskRunLogService.findJobTaskRunLogWithBLOBsByFlowIdAndRunId(runRecording.getFlowId(), runRecording.getRunId());
JobTaskRunLogWithBLOBs jobTaskRunLogWithBLOBs = jobTaskRunLogWithBLOBsByFlowIdAndRunId.stream().filter(log -> log.getNodeName().equals(specialJavaNode.getNodeName())).findFirst().orElse(null);
if(jobTaskRunLogWithBLOBs != null) {
if (jobTaskRunLogWithBLOBs != null) {
String logRemotelyPath = jobTaskRunLogWithBLOBs.getLogRemotelyPath();
//平台传过来的
String thisLogUrlPath = specialJavaNode.getThisLogUrlPath();
if(StringUtils.isBlank(logRemotelyPath)) {
if (StringUtils.isBlank(logRemotelyPath)) {
FileSystem fileSystem = SpringUtil.getBean(FileSystem.class);
byte[] sourceLogBytes = fileSystem.downloaderFile(thisLogUrlPath);
//间隔日志
String interval = "\n\n\n 游离节点日志,时间："+ DateUtil.dateFormat(new Date(),"yyyy-MM-dd HH:mm:ss")+"\n";
String interval = "\n\n\n 游离节点日志,时间：" + DateUtil.dateFormat(new Date(), "yyyy-MM-dd HH:mm:ss") + "\n";
byte[] intervalByte = stringToByteArray(interval);
byte[] intervalByteMergeFile = mergeFile(intervalByte,sourceLogBytes);
byte[] intervalByteMergeFile = mergeFile(intervalByte, sourceLogBytes);
//间隔日志
String runMsg = jobTaskRunLogWithBLOBs.getRunMsg();
......@@ -146,7 +151,7 @@ public class JavaFlowFreeTaskCallbackThread implements Runnable {
Map<String, String> fileMate = fileSystem.getFileMate(thisLogUrlPath);
String logPath = fileSystem.uploadFile(mergeFile, "log", fileMate);
jobTaskRunLogWithBLOBs.setLogRemotelyPath(logPath);
}else{
} else {
String logPath = mergeLog(logRemotelyPath, thisLogUrlPath);
jobTaskRunLogWithBLOBs.setLogRemotelyPath(logPath);
}
......@@ -156,10 +161,10 @@ public class JavaFlowFreeTaskCallbackThread implements Runnable {
return null;
}
public String mergeLog(String sourceLogPath,String targetLogPath) throws IOException {
public String mergeLog(String sourceLogPath, String targetLogPath) throws IOException {
FileSystem fileSystem = SpringUtil.getBean(FileSystem.class);
byte[] sourceLogBytes = fileSystem.downloaderFile(sourceLogPath);
String interval = "\n\n\n 游离节点日志,时间："+ DateUtil.dateFormat(new Date(),"yyyy-MM-dd HH:mm:ss")+"\n";
String interval = "\n\n\n 游离节点日志,时间：" + DateUtil.dateFormat(new Date(), "yyyy-MM-dd HH:mm:ss") + "\n";
byte[] intervalByte = stringToByteArray(interval);
sourceLogBytes = mergeFile(sourceLogBytes, intervalByte);
byte[] targetLogBytes = fileSystem.downloaderFile(targetLogPath);
......@@ -170,6 +175,7 @@ public class JavaFlowFreeTaskCallbackThread implements Runnable {
/**
* 将字符串转换我数组
*
* @param logData 日志字符串
* @return 转换的字节数组
*/
......@@ -179,14 +185,15 @@ public class JavaFlowFreeTaskCallbackThread implements Runnable {
/**
* 合并两个字节数组
*
* @param sourceByte 字节数组1
* @param targetByte 字节数组2
* @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);
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;
}
......
pom.xml
......@@ -46,7 +46,7 @@
<objenesis.version>2.6</objenesis.version>
<kryo.version>4.0.2</kryo.version>
<jackson.version>2.9.10</jackson.version>
<fastjson.version>1.2.31</fastjson.version>
<fastjson.version>1.2.9</fastjson.version>
<zookeeper.version>3.5.5</zookeeper.version>
......
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