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

byit-myth-admin/src/main/java/com/byit/api/ApiNodeController.java

+package com.byit.api;
+
+import com.byit.service.ApiNodeService;
+import io.swagger.annotations.Api;
+import org.springframework.web.bind.annotation.PostMapping;
+import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.RestController;
+
+import javax.annotation.Resource;
+
+@Api(tags  = "任务节点api")
+@RestController
+@RequestMapping("api/node")
+public class ApiNodeController {
+
+    @Resource
+    private ApiNodeService apiNodeService;
+
+    @PostMapping("runNode")
+    public String runNode(String param){
+        String monitorKey = apiNodeService.runNode(param);
+        return monitorKey;
+    }
+
+}

byit-myth-admin/src/main/java/com/byit/service/ApiNodeService.java

+package com.byit.service;
+
+/**
+ * 任务节点service
+ */
+public interface ApiNodeService {
+    /**
+     * 单独运行节点
+     * @param param
+     * @return
+     */
+    String runNode(String param);
+}

byit-myth-admin/src/main/java/com/byit/service/impl/ApiNodeServiceImpl.java

+package com.byit.service.impl;
+
+import com.alibaba.fastjson.JSONObject;
+import com.byit.model.JobTaskSchedule;
+import com.byit.service.ApiNodeService;
+import com.byit.utils.ValidationUtil;
+import org.springframework.stereotype.Service;
+import org.springframework.transaction.annotation.Transactional;
+
+@Service
+@Transactional(rollbackFor = Exception.class)
+public class ApiNodeServiceImpl implements ApiNodeService {
+
+
+    @Override
+    public String runNode(String param) {
+        ValidationUtil.dataNotBank(param, "请求参数不允许为空！");
+        JSONObject jsonObject = JSONObject.parseObject(param);
+        String nodeId = jsonObject.getString("nodeId");
+        ValidationUtil.dataNotBank(nodeId, "节点id不允许为空！");
+        String nodeName = jsonObject.getString("nodeName");
+        ValidationUtil.dataNotBank(nodeName, "节点名称不允许为空！");
+        String runCmd = jsonObject.getString("runCmd");
+        ValidationUtil.dataNotBank(runCmd, "运行命令不允许为空！");
+        String runParam = jsonObject.getString("runParam");
+        String scriptUrl = jsonObject.getString("scriptUrl");
+        String runSource = jsonObject.getString("runSource");
+        String jobType = jsonObject.getString("jobType");
+        ValidationUtil.dataNotBank(jobType, "节点类型不允许为空！");
+        JobTaskSchedule schedule = new JobTaskSchedule();
+        return null;
+    }
+}

byit-myth-core/myth-executor-core/pom.xml

@@ -47,10 +47,10 @@
             <artifactId>guava</artifactId>
         </dependency>
 
-        <!--<dependency>
-            <groupId>net.oschina.zcx7878</groupId>
-            <artifactId>fastdfs-client-java</artifactId>
-        </dependency>-->
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-data-redis</artifactId>
+        </dependency>
 
         <dependency>
             <groupId>junit</groupId>

byit-myth-core/myth-executor-core/src/main/java/com/byit/executor/config/RedisConfig.java

+package com.byit.executor.config;
+
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.data.redis.connection.RedisConnectionFactory;
+import org.springframework.data.redis.core.StringRedisTemplate;
+import org.springframework.data.redis.serializer.StringRedisSerializer;
+
+@Configuration
+public class RedisConfig {
+
+    @Bean
+    StringRedisTemplate stringRedisTemplate(RedisConnectionFactory connectionFactory){
+        StringRedisTemplate stringRedisTemplate = new StringRedisTemplate();
+        stringRedisTemplate.setConnectionFactory(connectionFactory);
+        stringRedisTemplate.setDefaultSerializer(new StringRedisSerializer());
+        return stringRedisTemplate;
+    }
+}

byit-myth-core/myth-executor-core/src/main/java/com/byit/executor/jobExecutor/process/MythJobProcess.java

@@ -48,9 +48,11 @@ public class MythJobProcess implements Callable<String>{
     private String effectiveUser = null;
     //本次任务对应的日志id
     private Integer logId;
+    //redis的发布key
+    private String runId;
 
     public MythJobProcess(final List<String> cmd, final Map<String, String> env,
-                          final String workingDir, final Integer logId) {
+                          final String workingDir, final Integer logId, final String runId) {
         this.cmd = cmd;
         this.env = env;
         this.workingDir = workingDir;
@@ -58,12 +60,13 @@ public class MythJobProcess implements Callable<String>{
         this.startupLatch = new CountDownLatch(1);
         this.completeLatch = new CountDownLatch(1);
         this.logId = logId;
+        this.runId = runId;
     }
 
     public MythJobProcess(final List<String> cmd, final Map<String, String> env,
                           final String workingDir, final String executeAsUserBinary,
-                          final String effectiveUser, Integer logId) {
-        this(cmd, env, workingDir, logId);
+                          final String effectiveUser, Integer logId, final String runId) {
+        this(cmd, env, workingDir, logId, runId);
         this.isExecuteAsUser = true;
         this.executeAsUserBinary = executeAsUserBinary;
         this.effectiveUser = effectiveUser;
@@ -107,8 +110,8 @@ public class MythJobProcess implements Callable<String>{
                 log.info("进程的id " + this.processId);
             }
 
-            outputGobbler = new LogGobbler(new InputStreamReader(this.process.getInputStream(), "utf-8"), Boolean.FALSE, 30);
-            errorGobbler = new LogGobbler(new InputStreamReader(this.process.getErrorStream(), "utf-8"), Boolean.TRUE, 30);
+            outputGobbler = new LogGobbler(new InputStreamReader(this.process.getInputStream(), "utf-8"), Boolean.FALSE, 30, runId);
+            errorGobbler = new LogGobbler(new InputStreamReader(this.process.getErrorStream(), "utf-8"), Boolean.TRUE, 30, runId);
             //开始获取进程的运行日志
             outputGobbler.start();
             errorGobbler.start();
@@ -124,7 +127,7 @@ public class MythJobProcess implements Callable<String>{
             outputGobbler.awaitCompletion(5000);
             errorGobbler.awaitCompletion(5000);
 
-            System.out.println(exitCode);
+            log.info("exit code 【{}】", exitCode);
             if (exitCode != 0) {
                 throw new ProcessFailureException(exitCode);
             }

byit-myth-core/myth-executor-core/src/main/java/com/byit/executor/util/LogGobbler.java

 package com.byit.executor.util;
 
 import lombok.extern.slf4j.Slf4j;
+import org.springframework.data.redis.core.StringRedisTemplate;
 
+import javax.annotation.Resource;
 import java.io.BufferedReader;
 import java.io.IOException;
 import java.io.Reader;
@@ -18,11 +20,16 @@ public class LogGobbler extends Thread {
     private final BufferedReader inputReader;
     private final boolean isError;
     private final StringBuilder buffer;
+    private final String runId;
 
-    public LogGobbler(final Reader inputReader, final boolean isError, final int bufferLines) {
+    @Resource
+    private StringRedisTemplate stringRedisTemplate;
+
+    public LogGobbler(final Reader inputReader, final boolean isError, final int bufferLines, final String runId) {
         this.inputReader = new BufferedReader(inputReader);
         this.isError = isError;
         this.buffer = new StringBuilder(bufferLines);
+        this.runId = runId;
     }
 
     @Override
@@ -34,6 +41,7 @@ public class LogGobbler extends Thread {
                     return;
                 }
                 this.buffer.append(line + "\n");
+                stringRedisTemplate.convertAndSend(runId, line);
                 printLog(line);
             }
         } catch (final IOException e) {

byit-myth-executor/myth-executor-server/src/main/java/com/byit/service/ScriptExecutorServiceImpl.java

@@ -81,7 +81,7 @@ public class ScriptExecutorServiceImpl implements ScriptExecutorService {
                     command = PlaceholderUtils.commandReplace(command,scriptParamAndPlaceholderDto.getParam());
                 }
                 List<String> cmdList = Arrays.asList(command.split(" "));
-                MythJobProcess mythJobProcess = new MythJobProcess(cmdList, null, null, scriptDto.getLogId());
+                MythJobProcess mythJobProcess = new MythJobProcess(cmdList, null, null, scriptDto.getLogId(), scriptDto.getRunId());
                 //保存日志
                 String logData = mythJobProcess.call();
                 byte[] logDataByte = stringToByteArray(logData);