deal with diff zk event for agent stauts

Author: Yang Guo

platform-control-center/src/main/java/com/flow/platform/cc/service/AgentService.java

@@ -35,9 +35,9 @@ public interface AgentService extends WebhookService {
     int IDLE_AGENT_TASK_HEARTBEAT = 30 * 1000; // millisecond
 
     /**
-     * Async update agent offline and online list by zone, will send to agent report queue
+     * Async update agent offline or online status
      */
-    void reportOnline(String zone, Set<String> agents);
+    void report(AgentPath path, AgentStatus status);
 
     /**
      * List agent by zone name

platform-control-center/src/main/java/com/flow/platform/cc/service/AgentServiceImpl.java

@@ -49,6 +49,7 @@
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.dao.DataAccessException;
+import org.springframework.dao.DataIntegrityViolationException;
 import org.springframework.scheduling.annotation.Scheduled;
 import org.springframework.stereotype.Service;
 import org.springframework.transaction.annotation.Isolation;
@@ -89,14 +90,34 @@ public class AgentServiceImpl extends WebhookServiceImplBase implements AgentSer
     private String secretKey;
 
     @Override
-    public void reportOnline(String zone, Set<String> agents) {
-        // set offline agents
-        agentDao.batchUpdateStatus(zone, AgentStatus.OFFLINE, agents, true);
-
-        // send to report queue
-        for (String agent : agents) {
-            AgentPath key = new AgentPath(zone, agent);
-            agentReportQueue.enqueue(key);
+    public void report(AgentPath path, AgentStatus status) {
+        Agent exist = find(path);
+
+        // For agent offline status
+        if (status == AgentStatus.OFFLINE) {
+            saveWithStatus(exist, AgentStatus.OFFLINE);
+            return;
+        }
+
+        // create new agent with idle status
+        if (exist == null) {
+            try {
+                exist = create(path, null);
+                LOGGER.trace("Create agent %s from report", path);
+            } catch (DataIntegrityViolationException ignore) {
+                // agent been created at some other threads
+                return;
+            }
+        }
+
+        // update exist offline agent to idle status
+        if (exist.getStatus() == AgentStatus.OFFLINE) {
+            saveWithStatus(exist, AgentStatus.IDLE);
+        }
+
+        // do not update agent status when its busy
+        if (exist.getStatus() == AgentStatus.BUSY) {
+            // do nothing
         }
     }
 

platform-control-center/src/main/java/com/flow/platform/cc/service/ZoneServiceImpl.java

@@ -22,7 +22,9 @@
 import com.flow.platform.core.context.ContextEvent;
 import com.flow.platform.core.context.SpringContext;
 import com.flow.platform.domain.Agent;
+import com.flow.platform.domain.AgentPath;
 import com.flow.platform.domain.AgentSettings;
+import com.flow.platform.domain.AgentStatus;
 import com.flow.platform.domain.Cmd;
 import com.flow.platform.domain.CmdInfo;
 import com.flow.platform.domain.CmdType;
@@ -41,6 +43,7 @@
 import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent;
 import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent.Type;
 import org.apache.curator.framework.recipes.cache.PathChildrenCacheListener;
+import org.apache.zookeeper.ZKUtil;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.scheduling.annotation.Scheduled;
 import org.springframework.stereotype.Service;
@@ -116,7 +119,9 @@ public String createZone(Zone zone) {
         List<String> agents = zkClient.getChildren(zonePath);
 
         if (!agents.isEmpty()) {
-            agentService.reportOnline(zone.getName(), Sets.newHashSet(agents));
+            for (String agent : agents) {
+                agentService.report(new AgentPath(zone.getName(), agent), AgentStatus.IDLE);
+            }
         }
 
         ZoneEventListener zoneEventWatcher = zoneEventWatchers.computeIfAbsent(zone, ZoneEventListener::new);
@@ -236,14 +241,20 @@ private class ZoneEventListener implements PathChildrenCacheListener {
 
         @Override
         public void childEvent(CuratorFramework client, PathChildrenCacheEvent event) throws Exception {
-            // TODO: should optimize by event type
-            Type type = event.getType();
-            LOGGER.debugMarker("ZoneEventListener", "Receive zookeeper event %s", type);
-
-            taskExecutor.execute(() -> {
-                List<String> agents = zkClient.getChildren(zone.getPath());
-                agentService.reportOnline(zone.getName(), Sets.newHashSet(agents));
-            });
+            final Type eventType = event.getType();
+            final String path = event.getData().getPath();
+            final String name = ZKHelper.getNameFromPath(path);
+            LOGGER.debugMarker("ZoneEventListener", "Receive zookeeper event %s %s", eventType, path);
+
+            if (eventType == Type.CHILD_ADDED || eventType == Type.CHILD_UPDATED) {
+                agentService.report(new AgentPath(zone.getName(), name), AgentStatus.IDLE);
+                return;
+            }
+
+            if (eventType == Type.CHILD_REMOVED) {
+                agentService.report(new AgentPath(zone.getName(), name), AgentStatus.OFFLINE);
+                return;
+            }
         }
     }
 }

platform-control-center/src/main/java/com/flow/platform/cc/util/ZKHelper.java

@@ -52,4 +52,8 @@ public static String buildPath(String zone, String name) {
     public static String buildPath(AgentPath agentPath) {
         return ZKPaths.makePath(ROOT_NODE, agentPath.getZone(), agentPath.getName());
     }
+
+    public static String getNameFromPath(String path) {
+        return ZKPaths.getNodeFromPath(path);
+    }
 }

platform-control-center/src/test/java/com/flow/platform/cc/test/controller/CmdControllerTest.java

@@ -30,6 +30,7 @@
 import com.flow.platform.cc.test.TestBase;
 import com.flow.platform.cc.util.ZKHelper;
 import com.flow.platform.domain.AgentPath;
+import com.flow.platform.domain.AgentStatus;
 import com.flow.platform.domain.Cmd;
 import com.flow.platform.domain.CmdInfo;
 import com.flow.platform.domain.CmdReport;
@@ -94,7 +95,7 @@ public void should_update_cmd_status() throws Throwable {
         String agent = "test-001";
 
         AgentPath path = new AgentPath(zone, agent);
-        agentService.reportOnline(zone, Sets.newHashSet(agent));
+        agentService.report(path, AgentStatus.IDLE);
         Thread.sleep(1000);
 
         CmdInfo base = new CmdInfo(zone, agent, CmdType.STOP, null);

platform-control-center/src/test/java/com/flow/platform/cc/test/service/AgentServiceTest.java

@@ -252,10 +252,10 @@ public void should_delete_agent() throws Exception{
         // when: report status
         AgentPath pathObj = new AgentPath(zoneName, agentName);
         Agent created = agentService.find(pathObj);
+        Assert.assertNotNull(created);
         agentService.saveWithStatus(created, AgentStatus.BUSY);
 
         agentService.delete(created);
         Assert.assertNull(agentService.find(new AgentPath(zoneName, agentName)));
-
     }
 }

platform-control-center/src/test/java/com/flow/platform/cc/test/service/CmdServiceTest.java

@@ -180,7 +180,7 @@ public void should_report_cmd_status() throws Throwable {
         // given:
         String zoneName = defaultZones.get(0).getName();
         AgentPath agentPath = new AgentPath(zoneName, "test-agent-for-report-cmd");
-        agentService.reportOnline(zoneName, Sets.newHashSet("test-agent-for-report-cmd"));
+        agentService.report(agentPath, AgentStatus.IDLE);
         Thread.sleep(5000);
 
         Agent agent = agentService.find(agentPath);