get rid of cmd queue item instead of cmd id as queue item

Author: Yang Guo

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;
@@ -76,36 +75,37 @@ 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;
             }
 
-            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);
@@ -115,17 +115,16 @@ public void onQueueItem(PriorityMessage message) {
     /**
      * 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);
         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);
         }
     }
 

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

@@ -29,7 +29,6 @@
 import com.flow.platform.cc.dao.CmdDao;
 import com.flow.platform.cc.dao.CmdLogDao;
 import com.flow.platform.cc.dao.CmdResultDao;
-import com.flow.platform.cc.domain.CmdQueueItem;
 import com.flow.platform.cc.domain.CmdStatusItem;
 import com.flow.platform.cc.exception.AgentErr;
 import com.flow.platform.core.exception.IllegalParameterException;
@@ -99,6 +98,9 @@ public class CmdServiceImpl extends WebhookServiceImplBase implements CmdService
     @Autowired
     private AgentDao agentDao;
 
+    /**
+     * The queue item is cmd id as string
+     */
     @Autowired
     private PlatformQueue<PriorityMessage> cmdQueue;
 
@@ -205,9 +207,7 @@ public List<Cmd> listWorkingCmd(AgentPath agentPath) {
     @Transactional(propagation = Propagation.NEVER)
     public Cmd enqueue(CmdInfo cmdInfo, int priority, int retry) {
         Cmd cmd = create(cmdInfo, retry);
-
-        CmdQueueItem item = new CmdQueueItem(cmd.getId(), retry);
-        PriorityMessage message = PriorityMessage.create(item.toBytes(), priority);
+        PriorityMessage message = PriorityMessage.create(cmd.getId().getBytes(), priority);
         cmdQueue.enqueue(message);
 
         return cmd;

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

@@ -189,7 +189,7 @@ public boolean keepIdleAgentMaxSize(final Zone zone, final InstanceManager insta
 
                 // send shutdown cmd
                 Cmd shutdown = cmdService.create(new CmdInfo(idleAgent.getPath(), CmdType.SHUTDOWN, "flow.ci"));
-                cmdDispatchService.dispatch(shutdown.getId(), false);
+                cmdDispatchService.dispatch(shutdown);
                 LOGGER.traceMarker("keepIdleAgentMaxSize", "Send SHUTDOWN to idle agent: %s", idleAgent);
 
                 // add instance to cleanup list

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

@@ -39,15 +39,12 @@
 import com.flow.platform.domain.CmdType;
 import com.flow.platform.domain.Zone;
 import java.util.List;
-import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.TimeUnit;
 import org.junit.After;
 import org.junit.Assert;
 import org.junit.Before;
 import org.junit.FixMethodOrder;
 import org.junit.Test;
 import org.junit.runners.MethodSorters;
-import org.springframework.amqp.core.Message;
 import org.springframework.beans.factory.annotation.Autowired;
 
 /**
@@ -91,7 +88,7 @@ public void toCreateSession() throws Throwable {
         // when: create cmd and dispatch to agent
         Cmd cmd = cmdService.create(new CmdInfo(zoneName, agentName, CmdType.CREATE_SESSION, null));
         Assert.assertNotNull(cmd.getSessionId());
-        cmd = cmdDispatchService.dispatch(cmd.getId(), false);
+        cmd = cmdDispatchService.dispatch(cmd);
 
         // then: check agent is locked by session
         target = agentService.find(cmd.getAgentPath());
@@ -109,7 +106,7 @@ public void should_broadcast_agent_resource_event_if_no_available_agent() throws
         Cmd cmdToCreateSession = cmdService.create(new CmdInfo(agentPath, CmdType.CREATE_SESSION, null));
 
         try {
-            cmdDispatchService.dispatch(cmdToCreateSession.getId(), false);
+            cmdDispatchService.dispatch(cmdToCreateSession);
         } catch (FlowException ignore) {
 
         }
@@ -121,7 +118,7 @@ public void should_broadcast_agent_resource_event_if_no_available_agent() throws
         CmdInfo cmd = new CmdInfo(agentPath, CmdType.DELETE_SESSION, null);
         cmd.setSessionId(target.getSessionId());
         Cmd cmdToDeleteSession = cmdService.create(cmd);
-        cmdDispatchService.dispatch(cmdToDeleteSession.getId(), false);
+        cmdDispatchService.dispatch(cmdToDeleteSession);
 
         // then: queue should be resumed since agent resource released
         Assert.assertEquals(true, cmdQueue.isRunning());
@@ -133,7 +130,7 @@ public void should_raise_exception_if_create_session_again() throws Throwable {
 
         // should throw agent not available exception
         try {
-            cmdDispatchService.dispatch(cmdToFail.getId(), false);
+            cmdDispatchService.dispatch(cmdToFail);
             fail();
         } catch (Throwable e) {
             Assert.assertEquals(AgentErr.NotAvailableException.class, e.getClass());
@@ -147,7 +144,7 @@ public void should_send_cmd_with_session() throws Throwable {
         cmdWithSession.setSessionId(target.getSessionId());
 
         Cmd cmd = cmdService.create(cmdWithSession);
-        cmd = cmdDispatchService.dispatch(cmd.getId(), false);
+        cmd = cmdDispatchService.dispatch(cmd);
 
         // then: check target is found correctly by session id
         Agent sessionAgent = agentService.find(cmd.getAgentPath());
@@ -169,7 +166,7 @@ public void should_running_cmd_been_killed_when_delete_session() throws Throwabl
         // when: delete session
         CmdInfo cmdToDelSession = new CmdInfo(agentPath.getZone(), null, CmdType.DELETE_SESSION, null);
         cmdToDelSession.setSessionId(target.getSessionId());
-        cmdDispatchService.dispatch(cmdService.create(cmdToDelSession).getId(), false);
+        cmdDispatchService.dispatch(cmdService.create(cmdToDelSession));
 
         // then: new kill cmd should been sent to agent
         Cmd killCmd = Cmd.parse(zkClient.getData(ZKHelper.buildPath(agentPath)), Cmd.class);
@@ -199,7 +196,7 @@ public void should_cmd_status_not_changed_for_finished_when_delete_session() thr
         // when: delete session
         CmdInfo cmdToDelSession = new CmdInfo(agentPath.getZone(), null, CmdType.DELETE_SESSION, null);
         cmdToDelSession.setSessionId(cmd.getSessionId());
-        cmdDispatchService.dispatch(cmdService.create(cmdToDelSession).getId(), false);
+        cmdDispatchService.dispatch(cmdService.create(cmdToDelSession));
 
         // then: cmd in agent not changed
         Cmd notChangeCmd = Cmd.parse(zkClient.getData(ZKHelper.buildPath(agentPath)), Cmd.class);
@@ -223,6 +220,6 @@ private Cmd startRunShell(String zone, String sessionId) {
         CmdInfo cmdWithSession = new CmdInfo(zone, null, CmdType.RUN_SHELL, "echo hello");
         cmdWithSession.setSessionId(sessionId);
         Cmd cmd = cmdService.create(cmdWithSession);
-        return cmdDispatchService.dispatch(cmd.getId(), false);
+        return cmdDispatchService.dispatch(cmd);
     }
 }

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

@@ -478,7 +478,7 @@ public void should_write_cmd_log() throws Throwable {
 
     private Cmd send(CmdInfo info) {
         Cmd cmd = cmdService.create(info);
-        cmdDispatchService.dispatch(cmd.getId(), false);
+        cmdDispatchService.dispatch(cmd);
         return cmdService.find(cmd.getId());
     }
 }