Merge pull request #115 from FlowCI/feature/cc/agent_status_change

Feature/cc/agent status change

Author: yang.guo

platform-api/src/main/java/com/flow/platform/api/controller/AgentController.java

@@ -22,7 +22,6 @@
 import com.flow.platform.api.events.AgentStatusChangeEvent;
 import com.flow.platform.api.service.AgentService;
 import com.flow.platform.core.exception.IllegalParameterException;
-import com.flow.platform.core.service.ApplicationEventService;
 import com.flow.platform.domain.Agent;
 import com.flow.platform.domain.AgentPath;
 import com.flow.platform.domain.AgentSettings;
@@ -42,7 +41,7 @@
  */
 @RestController
 @RequestMapping(path = "/agents")
-public class AgentController extends ApplicationEventService {
+public class AgentController {
 
     @Autowired
     private AgentService agentService;
@@ -193,7 +192,7 @@ public BooleanValue shutDown(@RequestParam String zone,
      */
     @PostMapping(path = "/callback")
     public void callback(@RequestBody Agent agent) {
-        this.dispatchEvent(new AgentStatusChangeEvent(this, agent));
+        agentService.onAgentStatusChange(agent);
     }
 
     /**

platform-api/src/main/java/com/flow/platform/api/dao/job/JobDao.java

@@ -69,6 +69,11 @@ public interface JobDao extends BaseDao<BigInteger, Job> {
      */
     Job get(String path, Integer number);
 
+    /**
+     * get job by session id
+     */
+    Job get(String sessionId);
+
     /**
      * get max build number for node path
      */

platform-api/src/main/java/com/flow/platform/api/dao/job/JobDaoImpl.java

@@ -87,6 +87,8 @@ public Job get(BigInteger key) {
         });
     }
 
+
+
     @Override
     public List<Job> latestByPath(List<String> paths) {
         return execute((Session session) -> {
@@ -163,6 +165,13 @@ public Job get(String path, Integer number) {
         });
     }
 
+    @Override
+    public Job get(String sessionId) {
+        return execute(session -> session.createQuery("from Job where sessionId = :sessionId", Job.class)
+            .setParameter("sessionId", sessionId)
+            .uniqueResult());
+    }
+
     @Override
     public Integer maxBuildNumber(String path) {
         return execute((Session session) -> {

platform-api/src/main/java/com/flow/platform/api/service/AgentService.java

@@ -58,4 +58,9 @@ public interface AgentService {
      * @param agentPath required
      */
     void sendSysCmd(AgentPath agentPath);
+
+    /**
+     * Handle agent status call back from cc
+     */
+    void onAgentStatusChange(Agent agent);
 }

platform-api/src/main/java/com/flow/platform/api/service/AgentServiceImpl.java

@@ -19,11 +19,17 @@
 import com.flow.platform.api.dao.job.JobDao;
 import com.flow.platform.api.domain.AgentWithFlow;
 import com.flow.platform.api.domain.job.Job;
+import com.flow.platform.api.domain.job.JobStatus;
+import com.flow.platform.api.domain.job.NodeResult;
 import com.flow.platform.api.domain.job.NodeStatus;
+import com.flow.platform.api.events.AgentStatusChangeEvent;
 import com.flow.platform.api.service.job.CmdService;
+import com.flow.platform.api.service.job.JobService;
+import com.flow.platform.api.service.job.NodeResultService;
 import com.flow.platform.api.util.PlatformURL;
 import com.flow.platform.core.exception.HttpException;
 import com.flow.platform.core.exception.IllegalStatusException;
+import com.flow.platform.core.service.ApplicationEventService;
 import com.flow.platform.domain.Agent;
 import com.flow.platform.domain.AgentPath;
 import com.flow.platform.domain.AgentPathWithWebhook;
@@ -53,7 +59,7 @@
  */
 
 @Service
-public class AgentServiceImpl implements AgentService {
+public class AgentServiceImpl extends ApplicationEventService implements AgentService {
 
     private final Logger LOGGER = new Logger(AgentService.class);
 
@@ -71,6 +77,12 @@ public class AgentServiceImpl implements AgentService {
     @Autowired
     private CmdService cmdService;
 
+    @Autowired
+    private JobService jobService;
+
+    @Autowired
+    private NodeResultService nodeResultService;
+
     @Value(value = "${domain}")
     private String domain;
 
@@ -228,4 +240,30 @@ private Agent findAgent(AgentPath agentPath) {
         return agent;
     }
 
+    @Override
+    public void onAgentStatusChange(Agent agent) {
+        this.dispatchEvent(new AgentStatusChangeEvent(this, agent));
+
+        // do not check related job if agent status not offline
+        if (agent.getStatus() != AgentStatus.OFFLINE) {
+            return;
+        }
+
+        // find related job and set job to failure
+        String sessionId = agent.getSessionId();
+        if (Strings.isNullOrEmpty(sessionId)) {
+            return;
+        }
+
+        // find agent related job by session id
+        Job job = jobService.find(sessionId);
+        if (job == null) {
+            return;
+        }
+
+        if (Job.RUNNING_STATUS.contains(job.getStatus())) {
+            job.setFailureMessage(String.format("Agent %s is offline when job running", agent.getPath()));
+            jobService.updateJobStatusAndSave(job, JobStatus.FAILURE);
+        }
+    }
 }

platform-api/src/main/java/com/flow/platform/api/service/job/JobService.java

@@ -45,6 +45,11 @@ public interface JobService {
      */
     Job find(BigInteger jobId);
 
+    /**
+     * Find by agent session id
+     */
+    Job find(String sessionId);
+
     /**
      * Get job yml content
      */

platform-api/src/main/java/com/flow/platform/api/service/job/JobServiceImpl.java

@@ -146,6 +146,11 @@ public Job find(BigInteger jobId) {
         return find(job);
     }
 
+    @Override
+    public Job find(String sessionId) {
+        return jobDao.get(sessionId);
+    }
+
     @Override
     public String findYml(String path, Integer number) {
         Job job = find(path, number);

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;
+            }
         }
     }
 }