Merge pull request #166 from FlowCI/fix/agent/164

yang.guo · web-flow · commit e7e32a3e6da0 · 2017-11-15T04:00:15.000-06:00
Fix/agent/164
diff --git a/platform-agent/src/main/java/com/flow/platform/agent/CmdManager.java b/platform-agent/src/main/java/com/flow/platform/agent/CmdManager.java
@@ -173,27 +173,23 @@ public void run() {
                     ProcEventHandler procEventHandler =
                         new ProcEventHandler(getCmd(), extraProcEventListeners, running, finished);
 
-                    CmdExecutor executor;
                     try {
-                        executor = new CmdExecutor(
+                        CmdExecutor executor = new CmdExecutor(
                             procEventHandler,
                             logListener,
                             cmd.getInputs(),
                             cmd.getWorkingDir(),
                             cmd.getOutputEnvFilter(),
                             cmd.getTimeout(),
                             Lists.newArrayList(getCmd().getCmd()));
+
+                        executor.run();
                     } catch (Throwable e) {
                         LOGGER.errorMarker("execute", "Cannot init CmdExecutor for cmd " + cmd, e);
-
                         CmdResult result = new CmdResult();
                         result.getExceptions().add(e);
                         procEventHandler.onException(result);
-
-                        return;
                     }
-
-                    executor.run();
                 }
             });
 
@@ -270,12 +266,9 @@ public synchronized void kill() {
         }
 
         try {
-            if (!cmdExecutor.awaitTermination(10, TimeUnit.SECONDS)) {
-                cmdExecutor.shutdownNow();
-                LOGGER.warn("Force to terminate CmdExecutor since been waiting 10 seconds");
-            }
-        } catch (Throwable e) {
-            LOGGER.error("Exception while waiting for all cmd thread finish", e);
+            cmdExecutor.shutdownNow();
+        } catch (Throwable ignore) {
+
         } finally {
             cmdExecutor = createExecutor(); // reset cmd executor
             LOGGER.trace("Cmd thread terminated");
diff --git a/platform-agent/src/test/java/com/flow/platform/agent/test/AgentManagerTest.java b/platform-agent/src/test/java/com/flow/platform/agent/test/AgentManagerTest.java
@@ -23,10 +23,13 @@
 import com.flow.platform.util.zk.ZKClient;
 import com.flow.platform.util.zk.ZkException;
 import java.util.concurrent.ArrayBlockingQueue;
+import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.ThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
 import org.apache.curator.test.TestingServer;
 import org.apache.curator.utils.ZKPaths;
+import org.apache.zookeeper.ZKUtil;
 import org.junit.After;
 import org.junit.AfterClass;
 import org.junit.Assert;
@@ -82,27 +85,6 @@ public void should_agent_registered() throws Throwable {
         agent.stop();
     }
 
-    @Test
-    public void should_zookeeper_create_node_atom() {
-        ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(5, 5, 1, TimeUnit.SECONDS,
-            new ArrayBlockingQueue<>(5));
-        final int[] size = {0};
-        String agentNodePath = ZKPaths.makePath(ZK_ROOT, "flow");
-        zkClient.delete(agentNodePath, false);
-        for (int i = 0; i < 5; i++) {
-            threadPoolExecutor.execute(() -> {
-                try {
-                    zkClient.createEphemeral(agentNodePath);
-                    size[0] = size[0] + 1;
-                } catch (ZkException e) {
-                    System.out.println(e);
-                }
-            });
-        }
-
-        Assert.assertEquals(1, size[0]);
-    }
-
     @Test
     public void should_receive_command() throws Throwable {
         AgentManager agent = new AgentManager(server.getConnectString(), 20000, ZONE, MACHINE);
@@ -123,6 +105,7 @@ public void should_receive_command() throws Throwable {
 
     @After
     public void after() throws Throwable {
+        zkClient.delete(ZKPaths.makePath(ZK_ROOT, ZONE, MACHINE), true);
         zkClient.close();
     }
 
diff --git a/platform-cmd-runner/src/main/java/com/flow/platform/cmd/CmdExecutor.java b/platform-cmd-runner/src/main/java/com/flow/platform/cmd/CmdExecutor.java
@@ -18,6 +18,7 @@
 
 import com.flow.platform.domain.CmdResult;
 import com.flow.platform.util.DateUtil;
+import com.flow.platform.util.Logger;
 import com.flow.platform.util.SystemUtil;
 import com.google.common.base.Strings;
 import com.google.common.collect.Lists;
@@ -82,6 +83,8 @@ public void onFinish() {
         }
     }
 
+    private final static Logger LOGGER = new Logger(CmdExecutor.class);
+
     private final static File DEFAULT_WORKING_DIR = new File(
         System.getProperty("user.home", System.getProperty("user.dir")));
 
@@ -226,7 +229,7 @@ public CmdResult run() {
 
             outputResult.setExecutedTime(DateUtil.now());
             procListener.onExecuted(outputResult);
-            System.out.println(String.format("====== 1. Process executed : %s ======", outputResult.getExitValue()));
+            LOGGER.trace("====== 1. Process executed : %s ======", outputResult.getExitValue());
 
             // wait for log thread with max 30 seconds to continue upload log
             logThreadCountDown.await(DEFAULT_LOGGING_WAITTING_SECONDS, TimeUnit.SECONDS);
@@ -239,16 +242,17 @@ public CmdResult run() {
 
             outputResult.setFinishTime(DateUtil.now());
             procListener.onLogged(outputResult);
+            LOGGER.trace("====== 2. Logging executed ======");
 
-            System.out.println(String.format("====== 2. Logging executed ======"));
-
+        } catch (InterruptedException ignore) {
+            LOGGER.warn(ignore.getMessage());
         } catch (Throwable e) {
             outputResult.getExceptions().add(e);
             outputResult.setFinishTime(DateUtil.now());
             procListener.onException(outputResult);
-            e.printStackTrace();
+            LOGGER.warn(e.getMessage());
         } finally {
-            System.out.println("====== 3. Process Done ======");
+            LOGGER.trace("====== 3. Process Done ======");
         }
 
         return outputResult;
@@ -290,7 +294,7 @@ private Runnable createCmdListExec(final OutputStream outputStream, final List<S
                 }
 
             } catch (IOException e) {
-                System.out.println("Exception on write cmd: " + e.getMessage());
+                LOGGER.warn("Exception on write cmd: " + e.getMessage());
             }
         };
     }
@@ -316,7 +320,7 @@ private Runnable createCmdLoggingReader() {
             } finally {
                 logListener.onFinish();
                 logThreadCountDown.countDown();
-                System.out.println(" ===== Logging Reader Thread Finish =====");
+                LOGGER.trace(" ===== Logging Reader Thread Finish =====");
             }
         };
     }
@@ -340,7 +344,7 @@ private Runnable createStdStreamReader(final Log.Type type, final InputStream is
 
             } finally {
                 stdThreadCountDown.countDown();
-                System.out.println(String.format(" ===== %s Stream Reader Thread Finish =====", type));
+                LOGGER.trace(" ===== %s Stream Reader Thread Finish =====", type);
             }
         };
     }
diff --git a/platform-control-center/src/main/java/com/flow/platform/cc/service/AgentServiceImpl.java b/platform-control-center/src/main/java/com/flow/platform/cc/service/AgentServiceImpl.java
@@ -165,6 +165,7 @@ public void saveWithStatus(Agent agent, AgentStatus status) {
 
         agent.setStatus(status);
         agentDao.update(agent);
+        LOGGER.trace("Agent status been updated to '%s'", status);
 
         // send webhook if status changed
         if (statusIsChanged) {
diff --git a/platform-control-center/src/main/java/com/flow/platform/cc/service/CmdDispatchServiceImpl.java b/platform-control-center/src/main/java/com/flow/platform/cc/service/CmdDispatchServiceImpl.java
@@ -323,15 +323,12 @@ public void doExec(Agent target, Cmd cmd) {
                 return;
             }
 
-            List<Cmd> runningCmdForSession = getRunningCmd(target.getSessionId());
-
-            // kill current running cmd
-            for (Cmd runningCmd : runningCmdForSession) {
+            // kill current running cmd and update agent status from cmd callback
+            for (Cmd runningCmd : getRunningCmd(target.getSessionId())) {
                 Cmd killCmd = cmdService.create(new CmdInfo(runningCmd.getAgentPath(), CmdType.KILL, null));
                 handler.get(CmdType.KILL).exec(killCmd);
             }
 
-            // release session from target
             target.setSessionId(null);
             agentService.saveWithStatus(target, AgentStatus.IDLE);
         }
diff --git a/platform-control-center/src/main/java/com/flow/platform/cc/service/CmdServiceImpl.java b/platform-control-center/src/main/java/com/flow/platform/cc/service/CmdServiceImpl.java
@@ -55,6 +55,7 @@
 import java.time.ZonedDateTime;
 import java.util.EnumSet;
 import java.util.HashMap;
+import java.util.LinkedList;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
@@ -229,6 +230,8 @@ public void updateStatus(CmdStatusItem statusItem, boolean inQueue) {
         //TODO: missing unit test
         // set cmd status in sequence
         if (!cmd.addStatus(statusItem.getStatus())) {
+            LOGGER.warn("Cannot add cmd '%s' from '%s' status to '%s'",
+                cmd.getId(), cmd.getStatus(), statusItem.getStatus());
             return;
         }
 
@@ -273,20 +276,19 @@ public void saveLog(String cmdId, MultipartFile file) {
 
     /**
      * Update agent status when report cmd status and result
+     * - DONOT update agent status if cmd with session, since it controlled by session cmd
      * - busy or idle by Cmd.Type.RUN_SHELL while report cmd status
      *
      * @param cmd Cmd object
      */
     private void updateAgentStatusFromCmd(Cmd cmd) {
-        // do not update agent status duration session
-        String sessionId = cmd.getSessionId();
-        if (sessionId != null && agentService.find(sessionId) != null) {
+        if (cmd.hasSession()) {
             return;
         }
 
-        // update agent status by cmd status
         AgentPath agentPath = cmd.getAgentPath();
         boolean isAgentBusy = false;
+
         for (Cmd tmp : listByAgentPath(agentPath)) {
             if (tmp.getType() != CmdType.RUN_SHELL) {
                 continue;
diff --git a/platform-control-center/src/test/java/com/flow/platform/cc/test/service/CmdDispatchServiceTest.java b/platform-control-center/src/test/java/com/flow/platform/cc/test/service/CmdDispatchServiceTest.java
@@ -32,11 +32,14 @@
 import com.flow.platform.domain.AgentStatus;
 import com.flow.platform.domain.Cmd;
 import com.flow.platform.domain.CmdInfo;
+import com.flow.platform.domain.CmdReport;
 import com.flow.platform.domain.CmdResult;
 import com.flow.platform.domain.CmdStatus;
 import com.flow.platform.domain.CmdType;
 import com.flow.platform.domain.Zone;
 import java.util.List;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
 import org.junit.After;
 import org.junit.Assert;
 import org.junit.Before;
@@ -173,6 +176,10 @@ public void should_running_cmd_been_killed_when_delete_session() throws Throwabl
         Assert.assertEquals(CmdType.KILL, killCmd.getType());
         Assert.assertNotEquals(cmd.getId(), killCmd.getId());
 
+        // when: mock the KILLED cmd been reported from agent
+        CmdReport mockKilledReport = new CmdReport(cmd.getId(), CmdStatus.KILLED, new CmdResult(147));
+        cmdService.updateStatus(new CmdStatusItem(mockKilledReport, true, true), false);
+
         // then: verify agent status
         Agent sessionShouldReleased = agentService.find(cmd.getAgentPath());
         Assert.assertNull(sessionShouldReleased.getSessionId());
diff --git a/platform-util-zk/src/test/java/com/flow/platform/util/zk/test/ZkClientTest.java b/platform-util-zk/src/test/java/com/flow/platform/util/zk/test/ZkClientTest.java
@@ -17,6 +17,8 @@
 package com.flow.platform.util.zk.test;
 
 import com.flow.platform.util.zk.ZKClient;
+import com.flow.platform.util.zk.ZkException;
+import java.util.concurrent.ArrayBlockingQueue;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.Executor;
 import java.util.concurrent.Executors;
@@ -76,6 +78,34 @@ public void should_return_false_if_node_not_exist() throws Throwable {
         Assert.assertFalse(zkClient.exist("/hello/not-exit"));
     }
 
+    @Test
+    public void should_create_node_atom() throws Throwable {
+        ThreadPoolExecutor threadPoolExecutor =
+            new ThreadPoolExecutor(5, 5, 1, TimeUnit.SECONDS, new ArrayBlockingQueue<>(5));
+
+        final AtomicInteger size = new AtomicInteger(0);
+        final CountDownLatch latch = new CountDownLatch(5);
+
+        String agentNodePath = ZKPaths.makePath("/flow-agent", "flow-atom");
+        zkClient.delete(agentNodePath, false);
+
+        for (int i = 0; i < 5; i++) {
+            threadPoolExecutor.execute(() -> {
+                try {
+                    zkClient.createEphemeral(agentNodePath);
+                    size.incrementAndGet();
+                } catch (ZkException e) {
+                    System.out.println(e);
+                } finally {
+                    latch.countDown();
+                }
+            });
+        }
+
+        latch.await(30, TimeUnit.SECONDS);
+        Assert.assertEquals(1, size.get());
+    }
+
     @Test
     public void should_create_and_delete_zk_node() throws Throwable {
         // init:

Original file line number	Diff line number	Diff line change
`@@ -323,15 +323,12 @@ public void doExec(Agent target, Cmd cmd) {`
`323`	`323`	`return;`
`324`	`324`	`}`
`325`	`325`
`326`		`- List<Cmd> runningCmdForSession = getRunningCmd(target.getSessionId());`
`327`		`-`
`328`		`- // kill current running cmd`
`329`		`- for (Cmd runningCmd : runningCmdForSession) {`
	`326`	`+ // kill current running cmd and update agent status from cmd callback`
	`327`	`+ for (Cmd runningCmd : getRunningCmd(target.getSessionId())) {`
`330`	`328`	`Cmd killCmd = cmdService.create(new CmdInfo(runningCmd.getAgentPath(), CmdType.KILL, null));`
`331`	`329`	`handler.get(CmdType.KILL).exec(killCmd);`
`332`	`330`	`}`
`333`	`331`
`334`		`- // release session from target`
`335`	`332`	`target.setSessionId(null);`
`336`	`333`	`agentService.saveWithStatus(target, AgentStatus.IDLE);`
`337`	`334`	`}`