Merge pull request #165 from FlowCI/feature/api/agent_dup_error

Feature/api/agent dup error

Author: yang.guo

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

@@ -19,8 +19,8 @@
 import com.flow.platform.domain.Cmd;
 import com.flow.platform.domain.Jsonable;
 import com.flow.platform.util.Logger;
-import com.flow.platform.util.zk.*;
-
+import com.flow.platform.util.zk.ZKClient;
+import com.flow.platform.util.zk.ZkException;
 import java.io.IOException;
 import java.util.LinkedList;
 import java.util.List;
@@ -48,6 +48,9 @@ public class AgentManager implements Runnable, TreeCacheListener, AutoCloseable
     private int zkTimeout;
     private ZKClient zkClient;
 
+    // node delete or not, default true
+    private Boolean canDeleted = true;
+
     private String zone; // agent running zone
     private String name; // agent name, can be machine name
 
@@ -92,6 +95,10 @@ public void stop() {
     public void run() {
         // init zookeeper
         zkClient.start();
+
+        // if node is exists, exit
+        checkNodePathExistAndExit();
+
         registerZkNodeAndWatch();
 
         synchronized (STATUS_LOCKER) {
@@ -103,6 +110,21 @@ public void run() {
         }
     }
 
+    /**
+     * if node exist , exit
+     */
+    private void checkNodePathExistAndExit() {
+        if (this.zkClient.exist(this.nodePath)) {
+            exit();
+        }
+    }
+
+    private void exit(){
+        this.canDeleted = false;
+        LOGGER.info("One Agent is running in other place. Please first to stop another agent, thx!");
+        Runtime.getRuntime().exit(1);
+    }
+
     @Override
     public void childEvent(CuratorFramework client, TreeCacheEvent event) throws Exception {
         ChildData eventData = event.getData();
@@ -143,7 +165,11 @@ public void childEvent(CuratorFramework client, TreeCacheEvent event) throws Exc
 
     @Override
     public void close() throws IOException {
-        removeZkNode();
+        // only this node can delete
+        if (this.canDeleted) {
+            removeZkNode();
+        }
+
         stop();
     }
 
@@ -194,8 +220,13 @@ private void onDataChanged(String path) {
      * @return path of zookeeper or null if failure
      */
     private String registerZkNodeAndWatch() {
-        String path = zkClient.createEphemeral(nodePath, null);
-        zkClient.watchTree(path, this);
+        String path = null;
+        try {
+            path = zkClient.createEphemeral(nodePath);
+            zkClient.watchTree(path, this);
+        } catch (ZkException e) {
+            exit();
+        }
         return path;
     }
 

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

@@ -21,6 +21,10 @@
 import com.flow.platform.domain.Cmd;
 import com.flow.platform.domain.CmdType;
 import com.flow.platform.util.zk.ZKClient;
+import com.flow.platform.util.zk.ZkException;
+import java.util.concurrent.ArrayBlockingQueue;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
 import org.apache.curator.test.TestingServer;
 import org.apache.curator.utils.ZKPaths;
 import org.junit.After;
@@ -32,10 +36,6 @@
 import org.junit.Test;
 import org.junit.runners.MethodSorters;
 
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertNotNull;
-import static org.junit.Assert.assertNull;
-
 /**
  * @author [email protected]
  */
@@ -82,6 +82,27 @@ 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);

platform-util-zk/src/main/java/com/flow/platform/util/zk/ZKClient.java

@@ -149,6 +149,19 @@ public String createEphemeral(String path, byte[] data) {
             return path;
         }
 
+        return createEphemeralPrivate(path, data);
+    }
+
+    /**
+     * if not create node, will throw exception
+     * @param path
+     * @return
+     */
+    public String createEphemeral(String path) {
+        return createEphemeralPrivate(path, null);
+    }
+
+    private String createEphemeralPrivate(String path, byte[] data) {
         try {
             return client.create()
                 .withMode(CreateMode.EPHEMERAL)