Merge pull request #174 from FlowCI/hotfix/173

Hotfix/173

Author: yang.guo

config/app-api.properties

@@ -29,8 +29,8 @@ system.email = [email protected]
 system.username = admin
 system.password = 123456
 
-task.job.toggle.execution_timeout = false
-## 6s expire job
-task.job.toggle.execution_create_session_duration = 600
-## 1h expire job
+task.job.toggle.execution_timeout = true
+## expired in 1800 seconds for create session
+task.job.toggle.execution_create_session_duration = 1800
+## expired in 3600 seconds for job running
 task.job.toggle.execution_running_duration = 3600

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

@@ -662,14 +662,14 @@ public void checkTimeoutTask() {
 
         LOGGER.trace("job timeout task start");
 
-        // create session job timeout 6s time out
+        // job timeout on create session
         ZonedDateTime finishZoneDateTime = ZonedDateTime.now().minusSeconds(jobExecuteTimeoutCreateSessionDuration);
         List<Job> jobs = jobDao.listForExpired(finishZoneDateTime, JobStatus.SESSION_CREATING);
         for (Job job : jobs) {
             updateJobAndNodeResultTimeout(job);
         }
 
-        // running job timeout 1h time out
+        // job timeout on running
         ZonedDateTime finishRunningZoneDateTime = ZonedDateTime.now().minusSeconds(jobExecuteTimeoutRunningDuration);
         List<Job> runningJobs = jobDao.listForExpired(finishRunningZoneDateTime, JobStatus.RUNNING);
         for (Job job : runningJobs) {

platform-api/src/main/resources/app-default.properties

@@ -29,9 +29,9 @@ system.email = [email protected]
 system.username = admin
 system.password = 123456
 
-task.job.toggle.execution_timeout = false
-## 6s expire job
-task.job.toggle.execution_create_session_duration = 600
-## 1h expire job
+task.job.toggle.execution_timeout = true
+## expired in 1800 seconds for job create session
+task.job.toggle.execution_create_session_duration = 1800
+## expired in 3600 seconds for running job
 task.job.toggle.execution_running_duration = 3600
 

platform-api/src/test/java/com/flow/platform/api/test/service/JobServiceTest.java

@@ -32,6 +32,7 @@
 import com.flow.platform.api.test.TestBase;
 import com.flow.platform.api.util.CommonUtil;
 import com.flow.platform.core.exception.IllegalStatusException;
+import com.flow.platform.core.util.ThreadUtil;
 import com.flow.platform.domain.Cmd;
 import com.flow.platform.domain.CmdResult;
 import com.flow.platform.domain.CmdStatus;
@@ -260,19 +261,16 @@ public void should_stop_running_job_success() throws IOException {
         Assert.assertEquals(NodeStatus.STOPPED, stoppedJob.getRootResult().getStatus());
     }
 
-
     @Test
     public void should_job_time_out_and_reject_callback() throws IOException, InterruptedException {
+        // given: job and mock updated time as expired
         Node rootForFlow = createRootFlow("flow1", "demo_flow2.yaml");
-
         Job job = jobService.createFromFlowYml(rootForFlow.getPath(), JobCategory.TAG, null, mockUser);
-
         Assert.assertNotNull(job.getEnv("FLOW_WORKSPACE"));
         Assert.assertNotNull(job.getEnv("FLOW_VERSION"));
 
-        Thread.sleep(7000);
-
         // when: check job timeout
+        ThreadUtil.sleep(20000);
         jobService.checkTimeoutTask();
 
         // then: job status should be timeout

platform-control-center/src/main/java/com/flow/platform/cc/consumer/CmdQueueConsumer.java

@@ -17,7 +17,6 @@
 package com.flow.platform.cc.consumer;
 
 import com.flow.platform.cc.config.QueueConfig;
-import com.flow.platform.cc.domain.CmdQueueItem;
 import com.flow.platform.cc.exception.AgentErr;
 import com.flow.platform.cc.service.AgentService;
 import com.flow.platform.cc.service.CmdDispatchService;
@@ -27,6 +26,7 @@
 import com.flow.platform.core.queue.PlatformQueue;
 import com.flow.platform.core.queue.PriorityMessage;
 import com.flow.platform.core.queue.QueueListener;
+import com.flow.platform.core.util.ThreadUtil;
 import com.flow.platform.domain.Cmd;
 import com.flow.platform.domain.CmdStatus;
 import com.flow.platform.util.Logger;
@@ -48,6 +48,8 @@ public class CmdQueueConsumer implements QueueListener<PriorityMessage> {
 
     private final static Logger LOGGER = new Logger(CmdQueueConsumer.class);
 
+    private final static long RETRY_WAIT_TIME = 1000; // in millis
+
     @Value("${queue.cmd.idle_agent.period}")
     private Integer idleAgentPeriod; // period for check idle agent in seconds
 
@@ -73,96 +75,56 @@ public void init() {
 
     @Override
     public void onQueueItem(PriorityMessage message) {
-        CmdQueueItem item = CmdQueueItem.parse(message.getBody(), CmdQueueItem.class);
-        String cmdId = item.getCmdId();
-        LOGGER.trace("Receive a cmd queue item: %s", item);
+        String cmdId = new String(message.getBody());
+        LOGGER.trace("Receive a cmd queue item: %s", cmdId);
+
+        Cmd cmd = cmdService.find(cmdId);
 
         try {
-            cmdDispatchService.dispatch(cmdId, false);
+            cmdDispatchService.dispatch(cmd);
         } catch (IllegalParameterException e) {
             LOGGER.warn("Illegal cmd id: %s", e.getMessage());
         } catch (IllegalStatusException e) {
             LOGGER.warn("Illegal cmd status: %s", e.getMessage());
         } catch (AgentErr.NotAvailableException | AgentErr.NotFoundException | ZkException.NotExitException e) {
-            if (item.getRetry() <= 0) {
+            if (cmd.getRetry() <= 0) {
                 return;
             }
 
-            boolean isTimeout = waitForIdleAgent(cmdId, idleAgentPeriod, idleAgentTimeout);
-            if (isTimeout) {
-                LOGGER.trace("wait for idle agent time out %s seconds for cmd %s", idleAgentTimeout, cmdId);
-            }
-
-            Cmd cmd = cmdService.find(cmdId);
+            cmd = cmdService.find(cmdId);
 
             // do not re-enqueue if cmd been stopped or killed
             if (cmd.getStatus() == CmdStatus.STOPPED || cmd.getStatus() == CmdStatus.KILLED) {
                 return;
             }
 
             // reset cmd status to pending, record num of retry
-            int retry = item.getRetry() - 1;
+            int retry = cmd.getRetry() - 1;
             cmd.setStatus(CmdStatus.PENDING);
             cmd.setRetry(retry);
             cmdService.save(cmd);
 
             // re-enqueue
-            resend(item, retry);
+            resend(cmd.getId(), retry);
 
         } catch (Throwable e) {
             LOGGER.error("Unexpected exception", e);
         }
     }
 
-    /**
-     * Block current thread and check idle agent
-     *
-     * @param cmdId cmd id
-     * @param period check idle agent period in seconds
-     * @param timeout timeout in seconds
-     * @return is time out exit or not
-     */
-    private boolean waitForIdleAgent(String cmdId, int period, int timeout) {
-        Instant now = Instant.now();
-
-        while (true) {
-            if (Duration.between(now, Instant.now()).toMillis() >= timeout * 1000) {
-                return true;
-            }
-
-            try {
-                Thread.sleep(period * 1000);
-            } catch (InterruptedException ignore) {
-            }
-
-            String zone = cmdService.find(cmdId).getZoneName();
-            int numOfIdle = agentService.findAvailable(zone).size();
-            if (numOfIdle > 0) {
-                LOGGER.trace("has %s idle agent", numOfIdle);
-                return false;
-            }
-        }
-    }
-
     /**
      * Re-enqueue cmd and return num of retry
      */
-    private void resend(final CmdQueueItem item, final int retry) {
+    private void resend(final String cmdId, final int retry) {
         if (retry <= 0) {
             return;
         }
 
-        item.setRetry(retry);
-
-        try {
-            Thread.sleep(1000); // wait 1 seconds and enqueue again with priority
-        } catch (InterruptedException ignore) {
-            // do nothing
-        }
+        ThreadUtil.sleep(RETRY_WAIT_TIME);
 
         // reset cmd status
-        PriorityMessage message = PriorityMessage.create(item.toBytes(), QueueConfig.MAX_PRIORITY);
+        PriorityMessage message = PriorityMessage.create(cmdId.getBytes(), QueueConfig.MAX_PRIORITY);
         cmdQueue.enqueue(message);
-        LOGGER.trace("Re-enqueue item %s", item);
+        LOGGER.trace("Re-enqueue item %s", cmdId);
     }
 }

platform-control-center/src/main/java/com/flow/platform/cc/controller/CmdController.java

@@ -69,7 +69,7 @@ public CmdType[] getCmdTypes() {
     @PostMapping(path = "/send")
     public Cmd sendCommand(@RequestBody CmdInfo cmd) {
         Cmd cmdToExec = cmdService.create(cmd);
-        return cmdDispatchService.dispatch(cmdToExec.getId(), false);
+        return cmdDispatchService.dispatch(cmdToExec);
     }
 
     @PostMapping(path = "/queue/send")

platform-control-center/src/main/java/com/flow/platform/cc/domain/CmdQueueItem.java

@@ -256,7 +256,7 @@ public void sessionTimeoutTask() {
             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);
+                    cmdDispatchService.dispatch(delSessionCmd);
                     LOGGER.traceMarker("sessionTimeoutTask", "Send DELETE_SESSION to agent %s", agent);
                 }
             }

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

@@ -26,11 +26,9 @@ public interface CmdDispatchService {
     /**
      * Dispatch cmd to agent
      *
-     * @param cmdId cmd id
-     * @param reset is reset cmd status to pending
      * @return cmd
      */
-    Cmd dispatch(String cmdId, boolean reset);
+    Cmd dispatch(Cmd cmd);
 
     /**
      * Task to check cmd is executing timeout

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

@@ -100,15 +100,9 @@ public void init() {
 
     @Override
     @Transactional(noRollbackFor = {Throwable.class})
-    public Cmd dispatch(String cmdId, boolean reset) {
-        Cmd cmd = cmdService.find(cmdId);
+    public Cmd dispatch(Cmd cmd) {
         if (cmd == null) {
-            throw new IllegalParameterException(String.format("Cmd '%s' does not exist", cmdId));
-        }
-
-        // reset cmd status to pending
-        if (reset) {
-            cmd.setStatus(CmdStatus.PENDING);
+            throw new IllegalParameterException("Cmd is null while dispatching");
         }
 
         // do not run cmd if not in working status
@@ -165,7 +159,7 @@ public void checkTimeoutTask() {
         for (Cmd cmd : workingCmdList) {
             if (cmd.isCmdTimeout()) {
                 Cmd killCmd = cmdService.create(new CmdInfo(cmd.getAgentPath(), CmdType.KILL, null));
-                dispatch(killCmd.getId(), false);
+                dispatch(killCmd);
                 LOGGER.traceMarker("checkTimeoutTask", "Send KILL for timeout cmd %s", cmd);
 
                 // update cmd status via queue
@@ -201,11 +195,11 @@ private Cmd createDeleteSessionCmd(Agent target) {
     private void cleanCurrentCmd(Cmd current) {
         if (Strings.isNullOrEmpty(current.getSessionId())) {
             Cmd cmdToKill = cmdService.create(new CmdInfo(current.getAgentPath(), CmdType.KILL, null));
-            dispatch(cmdToKill.getId(), false);
+            dispatch(cmdToKill);
         } else {
             Agent agent = agentService.find(current.getAgentPath());
             Cmd cmdToDelSession = createDeleteSessionCmd(agent);
-            dispatch(cmdToDelSession.getId(), false);
+            dispatch(cmdToDelSession);
         }
     }