[agent]-[executor] optimize execution code -2

agent/src/main/java/io/wdd/agent/config/message/handler/OMHandlerExecutor.java

@@ -40,7 +40,7 @@ public class OMHandlerExecutor extends AbstractOctopusMessageHandler {
             ExecutionMessage executionMessage = objectMapper.readValue((String) octopusMessage.getContent(), new TypeReference<ExecutionMessage>() {
             });
 
-            System.out.println("executionMessage = " + executionMessage);
+//            System.out.println("executionMessage = " + executionMessage);
 
             String executionType = executionMessage.getType();
 

agent/src/main/java/io/wdd/agent/executor/CommandExecutor.java

@@ -14,7 +14,8 @@ import java.io.IOException;
 import java.io.InputStream;
 import java.nio.ByteBuffer;
 import java.util.List;
-import java.util.concurrent.TimeUnit;
+import java.util.Optional;
+import java.util.concurrent.*;
 
 
 @Configuration
@@ -27,6 +28,10 @@ public class CommandExecutor {
     @Resource
     LogToArrayListCache logToArrayListCache;
 
+    int processMaxWaitSeconds = 60;
+
+    ExecutorService DaemonCommandProcess = Executors.newFixedThreadPool(1);
+
     /**
      * handle command from octopus server
      *
@@ -67,6 +72,13 @@ public class CommandExecutor {
         try {
 
             Process process = processBuilder.start();
+            // start a backend thread to daemon the process
+            // wait for processMaxWaitSeconds and kill the process if it's still alived
+            DaemonCommandProcess.submit(
+                    StopStuckCommandProcess(
+                            process,
+                            processMaxWaitSeconds
+                    ));
 
             // cache log lines
             logToArrayListCache.cacheLog(streamKey, process.getInputStream());
@@ -74,18 +86,13 @@ public class CommandExecutor {
             // start to send the result log
             streamSender.startToWaitLog(streamKey);
 
-//            log.warn("start---------------------------------------------");
-//            new BufferedReader(new InputStreamReader(process.getInputStream())).lines()
-//                   .forEach(System.out::println);
-//            log.warn("end ---------------------------------------------");
-
-            // a command shell don't understand how long it actually take
+            // get the command result
             processResult = process.waitFor();
 
             // end send logs
             streamSender.endWaitLog(streamKey);
 
-            log.info("current shell command {} result is {}", processBuilder.command(), processResult);
+            log.debug("current shell command {} result is {}", processBuilder.command(), processResult);
 
 
         } catch (IOException | InterruptedException e) {
@@ -95,6 +102,29 @@ public class CommandExecutor {
         return processResult;
     }
 
+    private Runnable StopStuckCommandProcess(Process process, int processMaxWaitSeconds)  {
+        return () -> {
+            try {
+
+
+                log.debug("daemon thread start to wait for {} s for the result", processMaxWaitSeconds);
+
+                TimeUnit.SECONDS.sleep(processMaxWaitSeconds);
+
+                if (process.isAlive()) {
+
+                    log.warn("Command [ {} ] stuck for {} s, destroy the command process !", process.info().commandLine().get(), processMaxWaitSeconds);
+
+                    // shutdown the process
+                    process.destroyForcibly();
+                }
+
+            } catch (InterruptedException e) {
+                throw new RuntimeException(e);
+            }
+        };
+    }
+
     private ByteBuffer cvToByteBuffer(InputStream inputStream) throws IOException {
 
         byte[] toByteArray = ByteStreams.toByteArray(inputStream);

agent/src/main/java/io/wdd/agent/executor/FunctionExecutor.java

@@ -52,7 +52,7 @@ public class FunctionExecutor {
 
 //        List<List<String>> commandList = functionReader.ReadFileToCommandList(functionFileName);
 
-        log.info("all commands are {}", commandList);
+        log.info("[ Function Executor ] all commands are ==> {}", commandList);
 
         Iterator<List<String>> iterator = commandList.iterator();
 

agent/src/main/java/io/wdd/agent/executor/redis/StreamSender.java

@@ -84,7 +84,7 @@ public class StreamSender {
 
         TimeUnit.SECONDS.sleep(2);
         if (AllNeededStreamSender.get(streamKey).isWaitToSendLog()) {
-            log.info("stream sender wait 2 s to send message");
+            log.debug("stream sender wait 2 s to send message");
             AllNeededStreamSender.get(streamKey).setWaitToSendLog(false);
             batchSendLog(streamKey);
         }

agent/src/main/java/io/wdd/agent/executor/thread/LogToArrayListCache.java

@@ -40,7 +40,7 @@ public class LogToArrayListCache {
                         commandCachedLog::add
                 );
 
-        log.info("current streamKey is {} and CacheLog is {}", streamKey, commandCachedLog);
+        log.debug("current streamKey is {} and CacheLog is {}", streamKey, commandCachedLog);
     }
 
     public ArrayList<String> getExecutionCmdCachedLogArrayList(String streamKey) {

agent/src/main/resources/bootstrap.yml

@@ -19,4 +19,8 @@ spring:
           - group: local
             data-id: common-local.yaml
 
-
+debug: true
+logging:
+  level:
+    io.wdd.agent:
+      debug

index.html

@@ -0,0 +1,7 @@
+<html>
+<head><title>Index of /octopus-agent/</title></head>
+<body>
+<h1>Index of /octopus-agent/</h1><hr><pre><a href="../">../</a>
+<a href="octopus-agent-2022-12-26-16-00-00.jar">octopus-agent-2022-12-26-16-00-00.jar</a>              27-Dec-2022 07:01            46309416
+</pre><hr></body>
+</html>