[agent]-[executor] optimize the command log collect algrothnm

agent/src/main/java/io/wdd/agent/config/beans/executor/CommandLog.java

@@ -18,6 +18,6 @@ public class CommandLog {
 //    @JsonFormat(pattern = "yyyy-MM-dd HH:mm:ss")
     private String lineTime;
 
-    private String lineContend;
+    private Object lineContend;
 
 }

agent/src/main/java/io/wdd/agent/config/beans/executor/StreamSenderEntity.java

@@ -0,0 +1,25 @@
+package io.wdd.agent.config.beans.executor;
+
+
+import lombok.AllArgsConstructor;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+import lombok.experimental.SuperBuilder;
+
+import java.util.ArrayList;
+
+@Data
+@AllArgsConstructor
+@NoArgsConstructor
+@SuperBuilder(toBuilder = true)
+public class StreamSenderEntity {
+
+    private String streamKey;
+
+    private ArrayList<String> cachedCommandLog;
+
+    private boolean waitToSendLog;
+
+    private int startIndex;
+
+}

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

@@ -2,9 +2,9 @@ package io.wdd.agent.executor;
 
 import com.google.common.io.ByteStreams;
 import io.wdd.agent.executor.redis.StreamSender;
-import io.wdd.agent.executor.thread.DaemonLogThread;
-import io.wdd.agent.executor.thread.LogToStreamSender;
+import io.wdd.agent.executor.thread.LogToArrayListCache;
 import io.wdd.common.beans.executor.ExecutionMessage;
+import lombok.SneakyThrows;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.context.annotation.Configuration;
 
@@ -12,6 +12,7 @@ import javax.annotation.Resource;
 import java.io.*;
 import java.nio.ByteBuffer;
 import java.util.List;
+import java.util.concurrent.TimeUnit;
 
 
 @Configuration
@@ -21,6 +22,8 @@ public class CommandExecutor {
     @Resource
     StreamSender streamSender;
 
+    @Resource
+    LogToArrayListCache logToArrayListCache;
 
     /**
      * handle command from octopus server
@@ -52,28 +55,26 @@ public class CommandExecutor {
 
         processBuilder.redirectErrorStream(true);
 //        processBuilder.inheritIO();
-//        processBuilder.directory(new File(System.getProperty("user.home")));
-
+        processBuilder.directory(new File(System.getProperty("user.home")));
         int processResult = 233;
 
         try {
 
             Process process = processBuilder.start();
 
+            // cache log lines
+            logToArrayListCache.cacheLog(streamKey, process.getInputStream());
 
+            streamSender.startToWaitLog(streamKey);
 
-            LogToStreamSender toStreamSender = new LogToStreamSender(streamKey, process.getInputStream(), streamSender::send);
-            DaemonLogThread.start(toStreamSender);
-
-            log.warn("---------------------------------------------");
-            new BufferedReader(new InputStreamReader(process.getInputStream())).lines()
-                    .map(
-                            String::valueOf
-                    ).forEach(System.out::println);
-            log.warn("---------------------------------------------");
+//            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
             processResult = process.waitFor();
+            streamSender.endWaitLog(streamKey);
 
             log.info("current shell command {} result is {}", processBuilder.command(), processResult);
 
@@ -105,4 +106,17 @@ public class CommandExecutor {
     }
 
 
+    @SneakyThrows
+    public void clearCommandCache(String streamKey) {
+
+        // wait
+        TimeUnit.SECONDS.sleep(1);
+
+        // clear the log Cache Thread scope
+        logToArrayListCache.getCommandCachedLog(streamKey).clear();
+
+        // clear the stream sender
+        streamSender.clearLocalCache(streamKey);
+
+    }
 }

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

@@ -57,15 +57,15 @@ public class FunctionExecutor {
         Iterator<List<String>> iterator = commandList.iterator();
 
         while (iterator.hasNext()) {
-
             int execute = commandExecutor.execute(streamKey, iterator.next());
 
             if (execute != 0) {
                 log.error("command list execute failed !");
                 break;
             }
-
         }
+
+        commandExecutor.clearCommandCache(streamKey);
     }
 
 

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

@@ -2,6 +2,8 @@ package io.wdd.agent.executor.redis;
 
 
 import io.wdd.agent.config.beans.executor.CommandLog;
+import io.wdd.agent.config.beans.executor.StreamSenderEntity;
+import io.wdd.agent.executor.thread.LogToArrayListCache;
 import lombok.SneakyThrows;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.beanutils.BeanUtils;
@@ -11,16 +13,17 @@ import org.springframework.context.annotation.Configuration;
 import org.springframework.data.redis.connection.stream.MapRecord;
 import org.springframework.data.redis.connection.stream.RecordId;
 import org.springframework.data.redis.connection.stream.StreamRecords;
-import org.springframework.data.redis.connection.stream.StringRecord;
 import org.springframework.data.redis.core.RedisTemplate;
 
 import javax.annotation.Resource;
-import java.lang.reflect.InvocationTargetException;
 import java.nio.ByteBuffer;
 import java.nio.charset.StandardCharsets;
 import java.time.LocalDateTime;
 import java.time.ZoneId;
 import java.time.format.DateTimeFormatter;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
 import java.util.Map;
 import java.util.concurrent.TimeUnit;
 
@@ -31,6 +34,82 @@ public class StreamSender {
     @Resource
     RedisTemplate redisTemplate;
 
+    @Resource
+    LogToArrayListCache logToArrayListCache;
+
+
+    private HashMap<String, StreamSenderEntity> AllNeededStreamSender = new HashMap<>(16);
+
+
+    private ArrayList<String> cacheLogList = new ArrayList<>(256);
+
+    public void startToWaitLog(String streamKey) throws InterruptedException {
+
+        if (!AllNeededStreamSender.containsKey(streamKey)) {
+
+            StreamSenderEntity streamSender = StreamSenderEntity.builder()
+                    .cachedCommandLog(logToArrayListCache.getCommandCachedLog(streamKey))
+                    .waitToSendLog(true)
+                    .startIndex(0)
+                    .streamKey(streamKey).build();
+
+            AllNeededStreamSender.put(streamKey, streamSender);
+
+        }
+
+        TimeUnit.SECONDS.sleep(2);
+        if (AllNeededStreamSender.get(streamKey).isWaitToSendLog()) {
+            log.info("stream sender wait 1 s to send message");
+            AllNeededStreamSender.get(streamKey).setWaitToSendLog(false);
+            batchSendLog(streamKey);
+        }
+    }
+
+    public void endWaitLog(String streamKey){
+
+        StreamSenderEntity streamSenderEntity = AllNeededStreamSender.get(streamKey);
+        streamSenderEntity.setWaitToSendLog(false);
+
+        batchSendLog(streamKey);
+
+    }
+
+    public void batchSendLog(String streamKey) {
+        StreamSenderEntity streamSenderEntity = AllNeededStreamSender.get(streamKey);
+
+        log.info("batch send log == {}", streamSenderEntity);
+
+        ArrayList<String> cachedCommandLog = streamSenderEntity.getCachedCommandLog();
+
+//        System.out.println("cachedCommandLog = " + cachedCommandLog);
+
+        int startIndex = streamSenderEntity.getStartIndex();
+        int endIndex = cachedCommandLog.size();
+
+        List<String> content = cachedCommandLog.subList(startIndex, endIndex);
+
+//        System.out.println("content = " + content);
+
+        this.send(streamKey, content);
+        // for next time
+        startIndex = endIndex;
+    }
+
+    public boolean send(String streamKey, String content){
+
+        return this.send(streamKey, List.of(content));
+
+    }
+
+    private boolean send(String streamKey, List<String> content) {
+
+        HashMap<String, List<String>> map = new HashMap<>(16);
+
+        map.put(currentTimeString(), content);
+
+        return doSendLogToStream(streamKey, map);
+    }
+
     private static ByteBuffer currentTimeByteBuffer(){
 
         byte[] timeBytes = LocalDateTime.now(ZoneId.of("UTC+8")).format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss")).getBytes(StandardCharsets.UTF_8);
@@ -43,33 +122,22 @@ public class StreamSender {
         return LocalDateTime.now(ZoneId.of("UTC+8")).format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));
     }
 
+    private boolean doSendLogToStream(String streamKey, HashMap map) {
 
-    public static String TEST_STREAM_JAVA = "test-stream-java";
+            log.info("redis stream sender message is {}", map);
 
+            MapRecord<String, String, List<String>> stringRecord = StreamRecords.mapBacked(map).withStreamKey(streamKey);
 
-    public boolean send(String streamKey, String content){
+            RecordId recordId = redisTemplate.opsForStream().add(stringRecord);
 
-        CommandLog commandLog = new CommandLog(currentTimeString(), content);
-        Map<String, String> map = null;
-        try {
-            map = BeanUtils.describe(commandLog);
-        } catch (IllegalAccessException | InvocationTargetException | NoSuchMethodException e) {
-            throw new RuntimeException(e);
-        }
+//            log.info("redis send recordId is {}",recordId);
 
-        log.info("redis stream sender message is {}", map);
-
-        StringRecord stringRecord = StreamRecords.string(map).withStreamKey(streamKey);
-
-        RecordId recordId = redisTemplate.opsForStream().add(stringRecord);
-
-        log.info("redis send recordId is {}",recordId);
-
-        return ObjectUtils.isNotEmpty(recordId);
+            return ObjectUtils.isNotEmpty(recordId);
 
     }
 
 
+    public static String TEST_STREAM_JAVA = "test-stream-java";
 
     @SneakyThrows
     public void test(){
@@ -93,8 +161,6 @@ public class StreamSender {
 
         }
 
-
-
     }
 
     @SneakyThrows
@@ -107,4 +173,7 @@ public class StreamSender {
         return map;
     }
 
+    public void clearLocalCache(String streamKey) {
+        AllNeededStreamSender.remove(streamKey);
+    }
 }

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

@@ -1,30 +1,32 @@
 package io.wdd.agent.executor.thread;
 
 import com.google.common.util.concurrent.ThreadFactoryBuilder;
+import io.wdd.common.beans.response.R;
 
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
-import java.util.concurrent.Future;
-import java.util.concurrent.ThreadFactory;
+import java.util.concurrent.*;
 
-public class DaemonLogThread {
-
-    private static final ExecutorService executorService;
-
-    static {
-
-        ThreadFactory daemonLogThread = new ThreadFactoryBuilder()
-                .setDaemon(true)
-                .setNameFormat("DaemonLogThread")
-                .setPriority(1)
-                .build();
-
-        executorService = Executors.newSingleThreadExecutor(daemonLogThread);
-
-    }
-
-    public static Future<?> start(Runnable logToSenderTask) {
-
-        return executorService.submit(logToSenderTask);
-    }
-}
+//public class DaemonLogThread {
+//
+//    private static final ExecutorService executorService;
+//
+//    static {
+//
+//        ThreadFactory daemonLogThread = new ThreadFactoryBuilder()
+//                .setDaemon(true)
+//                .setNameFormat("BackendToRedisThread")
+//                .setPriority(1)
+//                .build();
+//
+//        executorService = Executors.newSingleThreadExecutor(daemonLogThread);
+//
+//    }
+//
+//    public static Future<?> start(Runnable backendToRedisStream) {
+//
+//        return executorService.submit(backendToRedisStream);
+//    }
+//
+//    public static void stop(Runnable backendToRedisStream) {
+//        executorService.shutdownNow();
+//    }
+//}

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

@@ -0,0 +1,56 @@
+package io.wdd.agent.executor.thread;
+
+
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.stereotype.Component;
+
+import java.io.BufferedReader;
+import java.io.InputStream;
+import java.io.InputStreamReader;
+import java.util.ArrayList;
+import java.util.List;
+
+
+@Component
+@Slf4j
+public class LogToArrayListCache {
+
+    // concurrent command execute logs
+    public static List<ArrayList<String>> CachedCommandLog = List.of(
+            new ArrayList<>(256),
+            new ArrayList<>(256),
+            new ArrayList<>(256),
+            new ArrayList<>(256),
+            new ArrayList<>(256)
+    );
+
+    public void cacheLog(String streamKey, InputStream commandLogStream) {
+
+        ArrayList<String> commandCachedLog = this.getCommandCachedLog(streamKey);
+
+//        log.info(String.valueOf(commandCachedLog));
+
+        new BufferedReader(new InputStreamReader(commandLogStream))
+                .lines()
+                .forEach(
+                        commandCachedLog::add
+                );
+
+        log.info("current streamKey is {} and CacheLog is {}", streamKey, commandCachedLog);
+    }
+
+    public ArrayList<String> getCommandCachedLog(String streamKey) {
+
+        int keyToIndex = this.hashStreamKeyToIndex(streamKey);
+
+        return CachedCommandLog.get(keyToIndex);
+    }
+
+    private int hashStreamKeyToIndex(String streamKey) {
+
+        int size = CachedCommandLog.size();
+
+        return Math.abs(streamKey.hashCode() % size);
+    }
+
+}