job failure when agent offline

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);