From 6120eeecdee8707a53c5df93046c6b4faffde307 Mon Sep 17 00:00:00 2001
From: IceDerce <zeaslity@qq.com>
Date: Wed, 28 Dec 2022 15:22:38 +0800
Subject: [PATCH] [agent]-[executor] optimize execution code -2

---
 .../message/handler/OMHandlerExecutor.java    |  2 +-
 .../wdd/agent/executor/CommandExecutor.java   | 46 +++++++++++++++----
 .../wdd/agent/executor/FunctionExecutor.java  |  2 +-
 .../agent/executor/redis/StreamSender.java    |  2 +-
 .../executor/thread/LogToArrayListCache.java  |  2 +-
 agent/src/main/resources/bootstrap.yml        |  6 ++-
 index.html                                    |  7 +++
 7 files changed, 54 insertions(+), 13 deletions(-)
 create mode 100644 index.html

diff --git a/agent/src/main/java/io/wdd/agent/config/message/handler/OMHandlerExecutor.java b/agent/src/main/java/io/wdd/agent/config/message/handler/OMHandlerExecutor.java
index bcd963a..10f6b68 100644
--- a/agent/src/main/java/io/wdd/agent/config/message/handler/OMHandlerExecutor.java
+++ b/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();
 
diff --git a/agent/src/main/java/io/wdd/agent/executor/CommandExecutor.java b/agent/src/main/java/io/wdd/agent/executor/CommandExecutor.java
index 3ce6f9b..3a1f3b3 100644
--- a/agent/src/main/java/io/wdd/agent/executor/CommandExecutor.java
+++ b/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);
diff --git a/agent/src/main/java/io/wdd/agent/executor/FunctionExecutor.java b/agent/src/main/java/io/wdd/agent/executor/FunctionExecutor.java
index eb4fefc..995a605 100644
--- a/agent/src/main/java/io/wdd/agent/executor/FunctionExecutor.java
+++ b/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();
 
diff --git a/agent/src/main/java/io/wdd/agent/executor/redis/StreamSender.java b/agent/src/main/java/io/wdd/agent/executor/redis/StreamSender.java
index 4c102ce..27d59fa 100644
--- a/agent/src/main/java/io/wdd/agent/executor/redis/StreamSender.java
+++ b/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);
         }
diff --git a/agent/src/main/java/io/wdd/agent/executor/thread/LogToArrayListCache.java b/agent/src/main/java/io/wdd/agent/executor/thread/LogToArrayListCache.java
index c28f479..1e4e9a0 100644
--- a/agent/src/main/java/io/wdd/agent/executor/thread/LogToArrayListCache.java
+++ b/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) {
diff --git a/agent/src/main/resources/bootstrap.yml b/agent/src/main/resources/bootstrap.yml
index 013d9e3..4b3844a 100644
--- a/agent/src/main/resources/bootstrap.yml
+++ b/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
diff --git a/index.html b/index.html
new file mode 100644
index 0000000..b3cbb6d
--- /dev/null
+++ b/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>