Merge pull request #175 from FlowCI/feature/171

yang.guo · web-flow · commit 5e4cf40426cc · 2017-11-18T19:15:37.000+08:00
Feature/171
diff --git a/platform-api/src/main/java/com/flow/platform/api/consumer/CmdCallbackQueueConsumer.java b/platform-api/src/main/java/com/flow/platform/api/consumer/CmdCallbackQueueConsumer.java
@@ -18,10 +18,12 @@
 
 import com.flow.platform.api.domain.CmdCallbackQueueItem;
 import com.flow.platform.api.service.job.JobService;
+import com.flow.platform.core.exception.FlowException;
 import com.flow.platform.core.exception.NotFoundException;
 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.util.Logger;
 import javax.annotation.PostConstruct;
 import org.springframework.beans.factory.annotation.Autowired;
@@ -35,6 +37,9 @@ public class CmdCallbackQueueConsumer implements QueueListener<PriorityMessage>
 
     private final static Logger LOGGER = new Logger(CmdCallbackQueueConsumer.class);
 
+    // requeue 1 s
+    private final static int REQUEUE_DELAY_TIME = 1000;
+
     @Autowired
     private PlatformQueue<PriorityMessage> cmdCallbackQueue;
 
@@ -58,20 +63,34 @@ public void onQueueItem(PriorityMessage message) {
             jobService.callback(item);
         } catch (NotFoundException notFoundException) {
 
+            // detect retry times is reach the limit or not
+            detectRetryTimes(item);
+
             // re-enqueue cmd callback if job not found since transaction problem
-            reEnqueueJobCallback(item, 1000);
+            reEnqueueJobCallback(item, REQUEUE_DELAY_TIME, message.getMessageProperties().getPriority());
 
         } catch (Throwable throwable) {
             LOGGER.traceMarker("onQueueItem", String.format("exception - %s", throwable));
         }
     }
 
-    private void reEnqueueJobCallback(CmdCallbackQueueItem item, long wait) {
-        try {
-            Thread.sleep(wait);
-        } catch (Throwable ignore) {
+    private void detectRetryTimes(CmdCallbackQueueItem item) {
+        if (item.getRetryTimes() <= 0) {
+            throw new FlowException("retry times has reach the limit");
         }
+    }
+
+    private void reEnqueueJobCallback(CmdCallbackQueueItem item, long wait, int priority) {
+
+        // sleep seconds
+        ThreadUtil.sleep(wait);
+
+        // set retry times
+        item.setRetryTimes(item.getRetryTimes() - 1);
+
+        //priority inc 1
+        priority = priority + 1;
 
-        jobService.enterQueue(item);
+        jobService.enterQueue(item, priority);
     }
 }
diff --git a/platform-api/src/main/java/com/flow/platform/api/controller/CmdWebhookController.java b/platform-api/src/main/java/com/flow/platform/api/controller/CmdWebhookController.java
@@ -64,7 +64,7 @@ public void execute(@RequestBody Cmd cmd, @RequestParam String identifier) {
                 cmd.getStatus(),
                 cmd.getId(),
                 cmd.getCmdResult());
-            jobService.enterQueue(new CmdCallbackQueueItem(jobId, cmd));
+            jobService.enterQueue(new CmdCallbackQueueItem(jobId, cmd), 1);
         } catch (NumberFormatException warn) {
             LOGGER.warn("Invalid job id format");
         }
diff --git a/platform-api/src/main/java/com/flow/platform/api/dao/job/NodeResultDaoImpl.java b/platform-api/src/main/java/com/flow/platform/api/dao/job/NodeResultDaoImpl.java
@@ -28,7 +28,6 @@
 import javax.persistence.criteria.Predicate;
 import javax.persistence.criteria.Root;
 import org.hibernate.Session;
-import org.hibernate.query.Query;
 import org.springframework.stereotype.Repository;
 
 /**
@@ -100,11 +99,8 @@ public int update(BigInteger jobId, NodeStatus target) {
 
     @Override
     public void delete(List<BigInteger> jobIds) {
-        execute((Session session) -> {
-            String delete = String.format("delete from NodeResult where job_id in (:list)");
-            Query query = session.createQuery(delete);
-            query.setParameterList("list", jobIds);
-            return true;
-        });
+        execute((Session session) -> session.createQuery("delete from NodeResult where key.jobId in ( :jobIds )")
+            .setParameterList("jobIds", jobIds)
+            .executeUpdate());
     }
 }
diff --git a/platform-api/src/main/java/com/flow/platform/api/domain/CmdCallbackQueueItem.java b/platform-api/src/main/java/com/flow/platform/api/domain/CmdCallbackQueueItem.java
@@ -31,14 +31,19 @@ public class CmdCallbackQueueItem extends Jsonable {
 
     private final Cmd cmd;
 
-    private Integer retryTimes = 0;
+    // default retry times 5
+    private Integer retryTimes = 5;
 
     public CmdCallbackQueueItem(BigInteger jobId, Cmd cmd) {
         this.jobId = jobId;
         this.cmd = cmd;
         this.path = cmd.getExtra();
     }
 
+    public void setRetryTimes(Integer retryTimes) {
+        this.retryTimes = retryTimes;
+    }
+
     public BigInteger getJobId() {
         return jobId;
     }
diff --git a/platform-api/src/main/java/com/flow/platform/api/service/job/JobService.java b/platform-api/src/main/java/com/flow/platform/api/service/job/JobService.java
@@ -110,7 +110,7 @@ void createWithYmlLoad(String path,
     /**
      * Enqueue cmd callback item
      */
-    void enterQueue(CmdCallbackQueueItem cmdQueueItem);
+    void enterQueue(CmdCallbackQueueItem cmdQueueItem, int priority);
 
     /**
      * stop job
diff --git a/platform-api/src/main/java/com/flow/platform/api/service/job/JobServiceImpl.java b/platform-api/src/main/java/com/flow/platform/api/service/job/JobServiceImpl.java
@@ -462,8 +462,8 @@ private void onRunShellCallback(String path, Cmd cmd, Job job) {
     }
 
     @Override
-    public void enterQueue(CmdCallbackQueueItem cmdQueueItem) {
-        cmdCallbackQueue.enqueue(PriorityMessage.create(cmdQueueItem.toBytes(), 1));
+    public void enterQueue(CmdCallbackQueueItem cmdQueueItem, int priority) {
+        cmdCallbackQueue.enqueue(PriorityMessage.create(cmdQueueItem.toBytes(), priority));
     }
 
     @Override
diff --git a/platform-api/src/main/java/com/flow/platform/api/service/node/NodeCrontabService.java b/platform-api/src/main/java/com/flow/platform/api/service/node/NodeCrontabService.java
@@ -44,4 +44,9 @@ public interface NodeCrontabService extends ContextEvent {
      * List current triggers
      */
     List<Trigger> triggers();
+
+    /**
+     * Clean triggers
+     */
+    void cleanTriggers();
 }
diff --git a/platform-api/src/main/java/com/flow/platform/api/service/node/NodeCrontabServiceImpl.java b/platform-api/src/main/java/com/flow/platform/api/service/node/NodeCrontabServiceImpl.java
@@ -165,4 +165,13 @@ public List<Trigger> triggers() {
             throw new IllegalStatusException(e.getMessage());
         }
     }
+
+    @Override
+    public void cleanTriggers() {
+        try {
+            quartzScheduler.getTriggersOfJob(nodeCrontabDetail.getKey()).clear();
+        } catch (SchedulerException e) {
+            throw new IllegalStatusException(e.getMessage());
+        }
+    }
 }
diff --git a/platform-api/src/test/java/com/flow/platform/api/test/service/JobServiceTest.java b/platform-api/src/test/java/com/flow/platform/api/test/service/JobServiceTest.java
@@ -32,6 +32,8 @@
 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.queue.PlatformQueue;
+import com.flow.platform.core.queue.PriorityMessage;
 import com.flow.platform.core.util.ThreadUtil;
 import com.flow.platform.domain.Cmd;
 import com.flow.platform.domain.CmdResult;
@@ -42,10 +44,14 @@
 import java.io.IOException;
 import java.time.ZonedDateTime;
 import java.util.List;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
 import org.junit.Assert;
 import org.junit.Before;
 import org.junit.Test;
 import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
 
 /**
  * @author yh@firim
@@ -55,6 +61,9 @@ public class JobServiceTest extends TestBase {
     @Autowired
     private JobNodeService jobNodeService;
 
+    @Autowired
+    private PlatformQueue<PriorityMessage> cmdCallbackQueue;
+
     @Before
     public void init() {
         stubDemo();
@@ -301,4 +310,28 @@ public void should_get_latest_job_by_node_path() throws IOException {
         Assert.assertEquals(1, jobs.size());
         Assert.assertEquals("2", jobs.get(0).getNumber().toString());
     }
+
+    @Test
+    public void should_cmd_enqueue_limit_times_success() throws InterruptedException {
+        Cmd cmd = new Cmd("default", "test", CmdType.RUN_SHELL, "echo 1");
+        CountDownLatch countDownLatch = new CountDownLatch(5);
+        AtomicInteger atomicInteger = new AtomicInteger(0);
+
+        // register new queue to get item info
+        cmdCallbackQueue.register(message -> {
+            CmdCallbackQueueItem item = CmdCallbackQueueItem.parse(message.getBody(), CmdCallbackQueueItem.class);
+            atomicInteger.set(item.getRetryTimes());
+            countDownLatch.countDown();
+        });
+
+        // when: enter queue one not found job id
+        jobService.enterQueue(new CmdCallbackQueueItem(CommonUtil.randomId(), cmd), 1);
+        countDownLatch.await(6, TimeUnit.SECONDS);
+
+        // then: should try 5 times
+        Assert.assertEquals(1, atomicInteger.get());
+
+        // then: cmdCallbackQueue size should 0
+        Assert.assertEquals(0, cmdCallbackQueue.size());
+    }
 }
diff --git a/platform-api/src/test/java/com/flow/platform/api/test/service/NodeCrontabServiceTest.java b/platform-api/src/test/java/com/flow/platform/api/test/service/NodeCrontabServiceTest.java
@@ -42,6 +42,7 @@ public class NodeCrontabServiceTest extends TestBase {
     @Before
     public void before() {
         stubDemo();
+        flowCrontabService.cleanTriggers();
     }
 
     @Test
diff --git a/platform-api/src/test/java/com/flow/platform/api/test/service/NodeServiceTest.java b/platform-api/src/test/java/com/flow/platform/api/test/service/NodeServiceTest.java
@@ -18,6 +18,8 @@
 import static junit.framework.TestCase.fail;
 
 import com.flow.platform.api.domain.Webhook;
+import com.flow.platform.api.domain.job.Job;
+import com.flow.platform.api.domain.job.NodeResult;
 import com.flow.platform.api.domain.node.Node;
 import com.flow.platform.api.domain.node.Yml;
 import com.flow.platform.api.envs.EnvUtil;
@@ -30,6 +32,7 @@
 import com.flow.platform.api.test.TestBase;
 import com.flow.platform.api.util.NodeUtil;
 import com.flow.platform.core.exception.IllegalParameterException;
+import com.flow.platform.core.exception.NotFoundException;
 import com.flow.platform.util.http.HttpURL;
 import java.nio.file.Files;
 import java.util.HashMap;
@@ -255,8 +258,10 @@ public void should_error_if_node_path_is_not_for_flow() throws Throwable {
         ymlService.get(root);
     }
 
-    @Test
+    @Test(expected = NotFoundException.class)
     public void should_delete_flow() throws Throwable {
+        stubDemo();
+
         Node emptyFlow = nodeService.createEmptyFlow("flow1");
         setFlowToReady(emptyFlow);
 
@@ -266,6 +271,12 @@ public void should_delete_flow() throws Throwable {
         Assert.assertNotNull(nodeService.find(root.getPath()));
         Assert.assertNotNull(ymlService.get(root).getFile());
 
+        setRequiredJobEnvsForFlow(root);
+        Job job = createMockJob(root.getPath());
+
+        Assert.assertNotNull(jobService.find(job.getId()));
+        Assert.assertEquals(2, nodeResultService.list(job, true).size());
+
         // when: delete flow
         nodeService.delete(root.getPath());
 
@@ -275,6 +286,13 @@ public void should_delete_flow() throws Throwable {
 
         // then: should return null if root node is not existed
         Assert.assertNull(ymlService.get(root));
+
+        // then: node result should be null
+        List<NodeResult> nodeResults = nodeResultService.list(job, true);
+        Assert.assertEquals(0, nodeResults.size());
+
+        // then: job should be null
+        jobService.find(job.getId());
     }
 
     @Test

Original file line number	Diff line number	Diff line change
`@@ -64,7 +64,7 @@ public void execute(@RequestBody Cmd cmd, @RequestParam String identifier) {`
`64`	`64`	`cmd.getStatus(),`
`65`	`65`	`cmd.getId(),`
`66`	`66`	`cmd.getCmdResult());`
`67`		`- jobService.enterQueue(new CmdCallbackQueueItem(jobId, cmd));`
	`67`	`+ jobService.enterQueue(new CmdCallbackQueueItem(jobId, cmd), 1);`
`68`	`68`	`} catch (NumberFormatException warn) {`
`69`	`69`	`LOGGER.warn("Invalid job id format");`
`70`	`70`	`}`
Original file line number	Diff line number	Diff line change
`@@ -462,8 +462,8 @@ private void onRunShellCallback(String path, Cmd cmd, Job job) {`
`462`	`462`	`}`
`463`	`463`
`464`	`464`	`@Override`
`465`		`- public void enterQueue(CmdCallbackQueueItem cmdQueueItem) {`
`466`		`- cmdCallbackQueue.enqueue(PriorityMessage.create(cmdQueueItem.toBytes(), 1));`
	`465`	`+ public void enterQueue(CmdCallbackQueueItem cmdQueueItem, int priority) {`
	`466`	`+ cmdCallbackQueue.enqueue(PriorityMessage.create(cmdQueueItem.toBytes(), priority));`
`467`	`467`	`}`
`468`	`468`
`469`	`469`	`@Override`
Original file line number	Diff line number	Diff line change
`@@ -165,4 +165,13 @@ public List<Trigger> triggers() {`
`165`	`165`	`throw new IllegalStatusException(e.getMessage());`
`166`	`166`	`}`
`167`	`167`	`}`
	`168`	`+`
	`169`	`+ @Override`
	`170`	`+ public void cleanTriggers() {`
	`171`	`+ try {`
	`172`	`+ quartzScheduler.getTriggersOfJob(nodeCrontabDetail.getKey()).clear();`
	`173`	`+ } catch (SchedulerException e) {`
	`174`	`+ throw new IllegalStatusException(e.getMessage());`
	`175`	`+ }`
	`176`	`+ }`
`168`	`177`	`}`
Original file line number	Diff line number	Diff line change
`@@ -42,6 +42,7 @@ public class NodeCrontabServiceTest extends TestBase {`
`42`	`42`	`@Before`
`43`	`43`	`public void before() {`
`44`	`44`	`stubDemo();`
	`45`	`+ flowCrontabService.cleanTriggers();`
`45`	`46`	`}`
`46`	`47`
`47`	`48`	`@Test`