delete & create agent

Author: stargwq

.gitignore

@@ -1,6 +1,7 @@
 .idea/
 *.iml
 .DS_Store
+deploy.sh
 
 node_modules/
 apidoc/

docker/app-cc.properties

@@ -25,10 +25,12 @@ zone.default.default_cmd_timeout = 600
 
 ### rabbitmq config ###
 mq.host = amqp://localhost:5672
-mq.queue.cmd.name = flow-cmd-queue-default
 mq.management.host = http://localhost:15672
 
 #### cmd queue settings ###
+queue.cmd.retry.enable = false
+queue.cmd.rabbit.enable = false
+queue.cmd.rabbit.name = flow-cmd-queue-default
 queue.cmd.idle_agent.timeout = 30
 queue.cmd.idle_agent.period = 5
 

platform-agent/pom.xml

@@ -97,6 +97,21 @@
 
   <build>
     <finalName>flow-agent-${client.version}</finalName>
+
+    <resources>
+      <resource>
+        <directory>src/main/resources</directory>
+        <filtering>true</filtering>
+      </resource>
+    </resources>
+
+    <testResources>
+      <testResource>
+        <directory>src/test/resources</directory>
+        <filtering>true</filtering>
+      </testResource>
+    </testResources>
+
     <plugins>
       <plugin>
         <groupId>org.apache.maven.plugins</groupId>

platform-agent/src/main/java/com/flow/platform/agent/CmdManager.java

@@ -17,21 +17,28 @@
 package com.flow.platform.agent;
 
 import com.flow.platform.cmd.CmdExecutor;
+import com.flow.platform.cmd.Log;
+import com.flow.platform.cmd.Log.Type;
 import com.flow.platform.cmd.ProcListener;
 import com.flow.platform.domain.Cmd;
 import com.flow.platform.domain.CmdResult;
 import com.flow.platform.domain.CmdStatus;
 import com.flow.platform.domain.CmdType;
+import com.flow.platform.domain.Jsonable;
 import com.flow.platform.util.Logger;
 import com.google.common.collect.Lists;
 import com.google.common.collect.Maps;
-
 import java.time.ZonedDateTime;
 import java.util.ArrayList;
-import java.util.Date;
+import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
-import java.util.concurrent.*;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.ThreadFactory;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
 
 /**
  * Singleton class to handle command
@@ -166,7 +173,6 @@ public void run() {
                     ProcEventHandler procEventHandler =
                         new ProcEventHandler(getCmd(), extraProcEventListeners, running, finished);
 
-
                     CmdExecutor executor;
                     try {
                         executor = new CmdExecutor(
@@ -194,6 +200,15 @@ public void run() {
             return;
         }
 
+        if (cmd.getType() == CmdType.SYSTEM_INFO) {
+            LogEventHandler logListener = new LogEventHandler(cmd);
+            Log log = new Log(Type.STDERR, collectionAgentInfo());
+            logListener.onLog(log);
+            logListener.onFinish();
+            cmd.setStatus(CmdStatus.EXECUTED);
+            return;
+        }
+
         // kill current running proc
         if (cmd.getType() == CmdType.KILL) {
             defaultExecutor.execute(this::kill);
@@ -214,6 +229,29 @@ public void run() {
         }
     }
 
+    /**
+     * collect agent info
+     * @return
+     */
+    private String collectionAgentInfo() {
+        String javaVersion = System.getProperty("java.version");
+        String osName = System.getProperty("os.name");
+        Runtime runtime = Runtime.getRuntime();
+        int kb = 1024;
+        long total = runtime.totalMemory();
+        long free = runtime.freeMemory();
+        long use = total - free;
+        Map<String, String> dic = new HashMap<>(7);
+        dic.put("javaVersion", javaVersion);
+        dic.put("osName", osName);
+        dic.put("totalMemory", total / kb + "MB");
+        dic.put("useMemory", use / kb + "MB");
+        dic.put("zone", Config.zone());
+        dic.put("name", Config.name());
+        dic.put("agentVersion", Config.getProperty("version"));
+        return Jsonable.GSON_CONFIG.toJson(dic);
+    }
+
     /**
      * Kill all current running process
      */

platform-agent/src/main/java/com/flow/platform/agent/Config.java

@@ -18,18 +18,26 @@
 
 import com.flow.platform.domain.AgentSettings;
 import com.flow.platform.domain.Jsonable;
+import com.flow.platform.util.ExceptionUtil;
+import com.flow.platform.util.Logger;
 import com.flow.platform.util.http.HttpClient;
 import com.flow.platform.util.http.HttpResponse;
 import com.flow.platform.util.zk.ZKClient;
+import java.io.FileInputStream;
+import java.io.InputStream;
+import java.net.URL;
 import java.nio.file.Path;
 import java.nio.file.Paths;
+import java.util.Properties;
 import org.apache.curator.utils.ZKPaths;
 
 /**
  * @author [email protected]
  */
 public class Config {
 
+    private final static Logger LOGGER = new Logger(Config.class);
+
     public final static String ZK_ROOT = "flow-agents";
 
     /* Config properties by using -Dxxx.xxx = xxx as JVM parameter */
@@ -50,6 +58,8 @@ public class Config {
     public static String ZONE;
     public static String NAME;
 
+    private static Properties properties;
+
     public static boolean isDebug() {
         String boolStr = System.getProperty(PROP_IS_DEBUG, "false");
         return Boolean.parseBoolean(boolStr);
@@ -60,6 +70,29 @@ public static int zkTimeout() {
         return Integer.parseInt(intStr);
     }
 
+    /**
+     * get property from application.properties
+     * @param name
+     * @return
+     */
+    public static String getProperty(String name) {
+        String value;
+        URL resource = Config.class.getClassLoader().getResource("application.properties");
+        if (properties == null) {
+            try (InputStream fileInputStream = new FileInputStream(resource.getFile())) {
+                properties = new Properties();
+                properties.load(fileInputStream);
+            } catch (Throwable e) {
+                LOGGER.warn("get property from application.properties error %s",
+                    ExceptionUtil.findRootCause(e).getMessage());
+            }
+        }
+
+        value = properties.getProperty(name);
+
+        return value;
+    }
+
     /**
      * Is delete cmd log after uploaded
      */
@@ -104,7 +137,7 @@ public static Path logDir() {
     }
 
     public static int concurrentThreadNum() {
-        String intStr = System.getProperty(PROP_CONCURRENT_THREAD, "1");
+        String intStr = System.getProperty(PROP_CONCURRENT_THREAD, "2");
         return Integer.parseInt(intStr);
     }
 

platform-agent/src/main/java/com/flow/platform/agent/LogEventHandler.java

@@ -121,8 +121,10 @@ public void onFinish() {
     }
 
     public String websocketLogFormat(Log log) {
-        return String.format("%s#%s#%s#%s#%s", log.getNumber(), cmd.getZoneName(), cmd.getAgentName(), cmd.getId(),
-            log.getContent());
+        return String
+            .format("%s#%s#%s#%s#%s#%s", cmd.getType(), log.getNumber(), cmd.getZoneName(), cmd.getAgentName(),
+                cmd.getId(),
+                log.getContent());
     }
 
     private void initWebSocketSession(String url, int wsConnectionTimeout) throws Exception {

platform-agent/src/main/resources/application.properties

@@ -0,0 +1 @@
+version=${client.version}

platform-agent/src/test/java/com/flow/platform/agent/test/CmdManagerTest.java

@@ -21,6 +21,7 @@
 import com.flow.platform.cmd.ProcListener;
 import com.flow.platform.domain.Cmd;
 import com.flow.platform.domain.CmdResult;
+import com.flow.platform.domain.CmdStatus;
 import com.flow.platform.domain.CmdType;
 import java.io.IOException;
 import java.util.Map;
@@ -213,4 +214,56 @@ public void onException(CmdResult result) {
         Assert.assertNotNull(result.getExitValue());
         Assert.assertEquals(CmdResult.EXIT_VALUE_FOR_KILL, result.getExitValue());
     }
+
+
+    @Test
+    public void should_success_run_sys_cmd() throws InterruptedException {
+
+        Cmd cmd = new Cmd("zone1", "agent1", CmdType.RUN_SHELL, resourcePath);
+        cmd.setId(UUID.randomUUID().toString());
+        CountDownLatch finishCountDownLatch = new CountDownLatch(1);
+        CountDownLatch startCountDownLatch = new CountDownLatch(1);
+
+        cmdManager.getExtraProcEventListeners().add(new ProcListener() {
+            @Override
+            public void onStarted(CmdResult result) {
+                startCountDownLatch.countDown();
+                try {
+                    Thread.sleep(3000);
+                } catch (Throwable e) {
+                }
+            }
+
+            @Override
+            public void onLogged(CmdResult result) {
+                finishCountDownLatch.countDown();
+            }
+
+            @Override
+            public void onExecuted(CmdResult result) {
+
+            }
+
+            @Override
+            public void onException(CmdResult result) {
+
+            }
+        });
+
+        // when: start and kill task immediately
+        cmdManager.execute(cmd);
+
+        startCountDownLatch.await();
+        Assert.assertEquals(1, cmdManager.getRunning().size());
+
+        Cmd cmdSys = new Cmd("zone1", "agent1", CmdType.SYSTEM_INFO, "");
+        cmdSys.setId(UUID.randomUUID().toString());
+
+        cmdManager.execute(cmdSys);
+        Assert.assertEquals(CmdStatus.EXECUTED, cmdSys.getStatus());
+
+        finishCountDownLatch.await();
+        Assert.assertEquals(1, cmdManager.getFinished().size());
+
+    }
 }

platform-agent/src/test/java/com/flow/platform/agent/test/LogEventHandlerTest.java

@@ -58,7 +58,8 @@ public void should_get_correct_format_websocket() throws Throwable {
 
         // then:
         String expect = String
-            .format("%s#%s#%s#%s#%s", null, cmd.getZoneName(), cmd.getAgentName(), cmd.getId(), mockLogContent);
+            .format("%s#%s#%s#%s#%s#%s", CmdType.RUN_SHELL, null, cmd.getZoneName(), cmd.getAgentName(), cmd.getId(),
+                mockLogContent);
         Assert.assertEquals(expect, socketIoData);
     }
 }

platform-api/src/main/java/com/flow/platform/api/config/AppConfig.java

@@ -21,6 +21,8 @@
 import com.flow.platform.api.util.PlatformURL;
 import com.flow.platform.core.config.AppConfigBase;
 import com.flow.platform.core.config.DatabaseConfig;
+import com.flow.platform.core.queue.InMemoryQueue;
+import com.flow.platform.core.queue.PlatformQueue;
 import com.flow.platform.core.util.ThreadUtil;
 import com.flow.platform.util.Logger;
 import java.io.IOException;
@@ -68,6 +70,9 @@ public class AppConfig extends AppConfigBase {
     public final static String DEFAULT_USER_NAME = "admin";
     public final static String DEFAULT_USER_PASSWORD = "123456";
 
+    private final static ThreadPoolTaskExecutor executor =
+        ThreadUtil.createTaskExecutor(ASYNC_POOL_SIZE, ASYNC_POOL_SIZE / 10, 100, THREAD_NAME_PREFIX);
+
     @Value("${api.workspace}")
     private String workspace;
 
@@ -98,12 +103,15 @@ public ThreadLocal<String> currentNodePath() {
     @Bean
     @Override
     public ThreadPoolTaskExecutor taskExecutor() {
-        return ThreadUtil.createTaskExecutor(ASYNC_POOL_SIZE, ASYNC_POOL_SIZE / 10, 100, THREAD_NAME_PREFIX);
+        return executor;
     }
 
+    /**
+     * Queue to process cmd callback task
+     */
     @Bean
-    public BlockingQueue<CmdCallbackQueueItem> cmdBaseBlockingQueue() {
-        return new LinkedBlockingQueue<>(50);
+    public PlatformQueue<CmdCallbackQueueItem> cmdCallbackQueue() {
+        return new InMemoryQueue<>(executor, 50);
     }
 
     @Override