Merge pull request #114 from FlowCI/feature/cc/cmd_queue_pause_bug

Feature/cc/cmd queue pause bug

Author: yang.guo

platform-api/src/main/java/com/flow/platform/api/service/node/YmlServiceImpl.java

@@ -167,6 +167,10 @@ public Node loadYmlContent(final Node root, final Consumer<Yml> onSuccess, final
         } catch (ExecutionException | TaskRejectedException e) {
             LOGGER.warn("Fail to get task executor for node: " + root.getPath());
             nodeService.updateYmlState(root, YmlStatusValue.ERROR, e.getMessage());
+
+            if (onError != null) {
+                onError.accept(e);
+            }
         }
 
         return root;

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

@@ -30,7 +30,9 @@
  */
 public interface AgentService extends WebhookService {
 
-    int AGENT_SESSION_TIMEOUT_TASK_PERIOD = 60 * 1000; // millisecond
+    int SESSION_TIMEOUT_TASK_HEARTBEAT = 60 * 1000; // millisecond
+
+    int IDLE_AGENT_TASK_HEARTBEAT = 30 * 1000; // millisecond
 
     /**
      * Async update agent offline and online list by zone, will send to agent report queue
@@ -77,11 +79,6 @@ public interface AgentService extends WebhookService {
      */
     boolean isSessionTimeout(Agent agent, ZonedDateTime compareDate, long timeoutInSeconds);
 
-    /**
-     * To check agent session timeout
-     */
-    void sessionTimeoutTask();
-
     /**
      * Create agent and return token
      */
@@ -105,4 +102,14 @@ public interface AgentService extends WebhookService {
      * delete agent
      */
     void delete(Agent agent);
+
+    /**
+     * To check agent session timeout
+     */
+    void sessionTimeoutTask();
+
+    /**
+     * Task to check num of idle agent for zone
+     */
+    void idleAgentTask();
 }

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

@@ -165,31 +165,6 @@ public boolean isSessionTimeout(Agent agent, ZonedDateTime compareDate, long tim
         return sessionAlive >= timeoutInSeconds;
     }
 
-    @Override
-    @Transactional(propagation = Propagation.NEVER)
-    @Scheduled(initialDelay = 10 * 1000, fixedDelay = AGENT_SESSION_TIMEOUT_TASK_PERIOD)
-    public void sessionTimeoutTask() {
-        if (!taskConfig.isEnableAgentSessionTimeoutTask()) {
-            return;
-        }
-
-        LOGGER.traceMarker("sessionTimeoutTask", "start");
-        ZonedDateTime now = DateUtil.utcNow();
-
-        for (Zone zone : zoneService.getZones()) {
-            Collection<Agent> agents = listForOnline(zone.getName());
-            for (Agent agent : agents) {
-                if (agent.getSessionId() != null && isSessionTimeout(agent, now, zone.getAgentSessionTimeout())) {
-                    Cmd delSessionCmd = cmdService.create(new CmdInfo(agent.getPath(), CmdType.DELETE_SESSION, null));
-                    cmdDispatchService.dispatch(delSessionCmd.getId(), false);
-                    LOGGER.traceMarker("sessionTimeoutTask", "Send DELETE_SESSION to agent %s", agent);
-                }
-            }
-        }
-
-        LOGGER.traceMarker("sessionTimeoutTask", "end");
-    }
-
     @Override
     public Agent create(AgentPath agentPath, String webhook) {
         Agent agent = agentDao.get(agentPath);
@@ -246,4 +221,41 @@ public void delete(Agent agent){
         }
 
     }
+
+    @Override
+    @Transactional(propagation = Propagation.NEVER)
+    @Scheduled(initialDelay = 10 * 1000, fixedDelay = SESSION_TIMEOUT_TASK_HEARTBEAT)
+    public void sessionTimeoutTask() {
+        if (!taskConfig.isEnableAgentSessionTimeoutTask()) {
+            return;
+        }
+
+        LOGGER.traceMarker("sessionTimeoutTask", "start");
+        ZonedDateTime now = DateUtil.utcNow();
+
+        for (Zone zone : zoneService.getZones()) {
+            Collection<Agent> agents = listForOnline(zone.getName());
+            for (Agent agent : agents) {
+                if (agent.getSessionId() != null && isSessionTimeout(agent, now, zone.getAgentSessionTimeout())) {
+                    Cmd delSessionCmd = cmdService.create(new CmdInfo(agent.getPath(), CmdType.DELETE_SESSION, null));
+                    cmdDispatchService.dispatch(delSessionCmd.getId(), false);
+                    LOGGER.traceMarker("sessionTimeoutTask", "Send DELETE_SESSION to agent %s", agent);
+                }
+            }
+        }
+
+        LOGGER.traceMarker("sessionTimeoutTask", "end");
+    }
+
+    @Override
+    @Transactional(propagation = Propagation.NEVER)
+    @Scheduled(initialDelay = 10 * 1000, fixedDelay = IDLE_AGENT_TASK_HEARTBEAT)
+    public void idleAgentTask() {
+        for (Zone zone : zoneService.getZones()) {
+            List<Agent> availableList = findAvailable(zone.getName());
+            if (availableList.size() > 0) {
+                this.dispatchEvent(new AgentResourceEvent(this, zone.getName(), Category.RELEASED));
+            }
+        }
+    }
 }