Merge pull request #360 from FlowCI/feature/1469

Feature/1469

Author: gy2006

core/pom.xml

@@ -64,6 +64,12 @@
       <artifactId>minio</artifactId>
     </dependency>
 
+    <dependency>
+      <groupId>org.codehaus.groovy</groupId>
+      <artifactId>groovy-all</artifactId>
+      <type>pom</type>
+    </dependency>
+
     <dependency>
       <groupId>com.github.ben-manes.caffeine</groupId>
       <artifactId>caffeine</artifactId>

core/src/main/java/com/flowci/core/agent/domain/ShellIn.java

@@ -1,8 +1,10 @@
 package com.flowci.core.agent.domain;
 
+import com.fasterxml.jackson.annotation.JsonIgnore;
 import com.flowci.domain.DockerOption;
 import com.flowci.domain.StringVars;
 import com.flowci.domain.Vars;
+import com.flowci.util.StringHelper;
 import com.google.common.base.Strings;
 import lombok.Getter;
 import lombok.Setter;
@@ -31,6 +33,9 @@ public final class ShellIn extends CmdIn {
 
     private List<DockerOption> dockers;
 
+    @JsonIgnore
+    private String condition;
+
     private List<String> scripts = new LinkedList<>();
 
     private int timeout = 1800;
@@ -57,4 +62,9 @@ public void addEnvFilters(Set<String> exports) {
     public void addInputs(StringVars vars) {
         inputs.putAll(vars);
     }
+
+    @JsonIgnore
+    public boolean hasCondition() {
+        return StringHelper.hasValue(condition);
+    }
 }

core/src/main/java/com/flowci/core/job/config/JobConfig.java

@@ -18,6 +18,7 @@
 
 import com.flowci.core.common.config.AppProperties;
 import com.flowci.core.common.helper.CacheHelper;
+import com.flowci.core.common.helper.ThreadHelper;
 import com.flowci.core.job.domain.Step;
 import com.flowci.tree.NodeTree;
 import com.flowci.util.FileHelper;
@@ -26,11 +27,13 @@
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
+import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
 
 import java.io.IOException;
 import java.nio.file.Path;
 import java.nio.file.Paths;
 import java.util.List;
+import java.util.concurrent.ThreadPoolExecutor;
 
 /**
  * @author yang
@@ -58,4 +61,9 @@ public Path pluginDir() throws IOException {
         Path pluginDir = Paths.get(workspace, "repos");
         return FileHelper.createDirectory(pluginDir);
     }
+
+    @Bean("jobConditionExecutor")
+    public ThreadPoolTaskExecutor jobConditionExecutor() {
+        return ThreadHelper.createTaskExecutor(20, 20, 100, "job-cond-");
+    }
 }

core/src/main/java/com/flowci/core/job/domain/Step.java

@@ -48,6 +48,8 @@
 )
 public class Step implements Executed {
 
+    public static final String MessageSkippedOnCondition = "Skipped due to condition";
+
     @Id
     private String id;
 

core/src/main/java/com/flowci/core/job/manager/CmdManagerImpl.java

@@ -56,6 +56,7 @@ public CmdIn createShellCmd(Job job, Step step, NodeTree tree) {
                 .setId(step.getId())
                 .setFlowId(job.getFlowId())
                 .setJobId(job.getId())
+                .setCondition(node.getCondition())
                 .setAllowFailure(node.isAllowFailure())
                 .setDockers(node.getDockers())
                 .setTimeout(job.getTimeout());

core/src/main/java/com/flowci/core/job/manager/ConditionManager.java

@@ -0,0 +1,9 @@
+package com.flowci.core.job.manager;
+
+import com.flowci.core.agent.domain.CmdIn;
+import groovy.util.ScriptException;
+
+public interface ConditionManager {
+
+    boolean run(CmdIn in) throws ScriptException;
+}

core/src/main/java/com/flowci/core/job/manager/ConditionManagerImpl.java

@@ -0,0 +1,63 @@
+package com.flowci.core.job.manager;
+
+import com.flowci.core.agent.domain.CmdIn;
+import com.flowci.core.agent.domain.ShellIn;
+import groovy.lang.Binding;
+import groovy.lang.GroovyRuntimeException;
+import groovy.lang.GroovyShell;
+import groovy.util.ScriptException;
+import lombok.extern.log4j.Log4j2;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
+import org.springframework.stereotype.Component;
+
+import java.util.concurrent.*;
+
+@Log4j2
+@Component
+public class ConditionManagerImpl implements ConditionManager {
+
+    private final int DefaultTimeout = 2; // seconds
+
+    @Autowired
+    private ThreadPoolTaskExecutor jobConditionExecutor;
+
+    @Override
+    public boolean run(CmdIn in) throws ScriptException {
+        if (!(in instanceof ShellIn)) {
+            return true;
+        }
+
+        ShellIn shell = (ShellIn) in;
+        if (!shell.hasCondition()) {
+            return true;
+        }
+
+        Future<Boolean> submit = jobConditionExecutor.submit(() -> {
+            Binding binding = new Binding();
+            shell.getInputs().forEach(binding::setVariable);
+
+            try {
+                GroovyShell groovy = new GroovyShell(binding);
+                Object value = groovy.evaluate(shell.getCondition());
+                if (value instanceof Boolean) {
+                    return (Boolean) value;
+                }
+                throw new Exception("The return type is not boolean");
+            } catch (GroovyRuntimeException e) {
+                throw new Exception(e.getMessage());
+            }
+        });
+
+        try {
+            return submit.get(DefaultTimeout, TimeUnit.SECONDS);
+        } catch (InterruptedException e) {
+            throw new ScriptException("Condition script interrupted");
+        } catch (ExecutionException e) {
+            throw new ScriptException(e.getMessage());
+        } catch (TimeoutException e) {
+            submit.cancel(true);
+            throw new ScriptException("Condition script timeout");
+        }
+    }
+}

core/src/main/java/com/flowci/core/job/manager/JobActionManagerImpl.java

@@ -7,6 +7,7 @@
 import com.flowci.core.common.manager.SpringEventManager;
 import com.flowci.core.common.rabbit.RabbitOperations;
 import com.flowci.core.job.dao.JobDao;
+import com.flowci.core.job.domain.Executed;
 import com.flowci.core.job.domain.Job;
 import com.flowci.core.job.domain.Step;
 import com.flowci.core.job.event.JobReceivedEvent;
@@ -29,6 +30,7 @@
 import com.flowci.zookeeper.InterLock;
 import com.flowci.zookeeper.ZookeeperClient;
 import com.google.common.base.Strings;
+import groovy.util.ScriptException;
 import lombok.Getter;
 import lombok.Setter;
 import lombok.experimental.Accessors;
@@ -42,10 +44,7 @@
 import java.nio.file.Files;
 import java.nio.file.Path;
 import java.nio.file.Paths;
-import java.util.Base64;
-import java.util.List;
-import java.util.Objects;
-import java.util.Optional;
+import java.util.*;
 import java.util.function.Consumer;
 import java.util.function.Function;
 
@@ -120,6 +119,9 @@ public class JobActionManagerImpl implements JobActionManager {
     @Autowired
     private RabbitOperations jobsQueueManager;
 
+    @Autowired
+    private ConditionManager conditionManager;
+
     @Autowired
     private LocalTaskService localTaskService;
 
@@ -328,7 +330,7 @@ public void accept(JobSmContext context) {
 
         Sm.add(QueuedToRunning, new Action<JobSmContext>() {
             @Override
-            public void accept(JobSmContext context) {
+            public void accept(JobSmContext context) throws Exception {
                 Job job = context.job;
                 eventManager.publish(new JobReceivedEvent(this, job));
 
@@ -387,7 +389,7 @@ public boolean canRun(JobSmContext context) {
             }
 
             @Override
-            public void accept(JobSmContext context) {
+            public void accept(JobSmContext context) throws Exception {
                 Job job = context.job;
                 log.debug("Job {} is locked", job.getId());
 
@@ -618,7 +620,7 @@ private SimpleSecret getSimpleSecret(String credentialName) {
     /**
      * Dispatch job to agent when Queue to Running
      */
-    private void dispatch(Job job, Agent agent) {
+    private void dispatch(Job job, Agent agent) throws ScriptException {
         NodeTree tree = ymlManager.getTree(job);
         StepNode next = tree.next(NodePath.create(job.getCurrentPath()));
 
@@ -636,13 +638,30 @@ private void dispatch(Job job, Agent agent) {
         job.setAgentId(agent.getId());
         job.setAgentSnapshot(agent);
 
-        // set executed cmd step to running
-        Step nextStep = stepService.toStatus(job.getId(), nextPath, Step.Status.RUNNING, null);
-
         // dispatch job to agent queue
+        Step nextStep = stepService.get(job.getId(), nextPath);
         CmdIn cmd = cmdManager.createShellCmd(job, nextStep, tree);
-        setJobStatusAndSave(job, Job.Status.RUNNING, null);
+        boolean canExecute = conditionManager.run(cmd);
+
+        if (!canExecute) {
+            nextStep.setStartAt(new Date());
+            nextStep.setFinishAt(new Date());
+            updateJobTime(job, nextStep, tree, next);
+
+            setJobStatusAndSave(job, Job.Status.RUNNING, null);
+            stepService.toStatus(nextStep, Executed.Status.SKIPPED, Step.MessageSkippedOnCondition);
+
+            JobSmContext context = new JobSmContext();
+            context.job = job;
+            context.step = nextStep;
+            context.agentId = agent.getId();
+
+            this.toNextStep(context);
+            return;
+        }
 
+        setJobStatusAndSave(job, Job.Status.RUNNING, null);
+        stepService.toStatus(nextStep, Executed.Status.RUNNING, null);
         agentService.dispatch(cmd, agent);
         logInfo(job, "send to agent: step={}, agent={}", next.getName(), agent.getName());
     }
@@ -652,9 +671,9 @@ private void dispatch(Job job, Agent agent) {
      *
      * @return true if next step dispatched, false if no more steps or failure
      */
-    private boolean toNextStep(JobSmContext context) {
+    private boolean toNextStep(JobSmContext context) throws ScriptException {
         Job job = context.job;
-        Step step = context.step;
+        Step step = context.step; // current step
 
         // save executed cmd
         stepService.resultUpdate(step);
@@ -677,15 +696,29 @@ private boolean toNextStep(JobSmContext context) {
         Optional<StepNode> next = findNext(tree, node, step.isSuccess());
         if (next.isPresent()) {
             String nextPath = next.get().getPathAsString();
-            job.setCurrentPath(nextPath);
-
-            Step nextStep = stepService.toStatus(job.getId(), nextPath, Step.Status.RUNNING, null);
+            Step nextStep = stepService.get(job.getId(), nextPath);
             context.setStep(nextStep);
 
+            job.setCurrentPath(nextPath);
+            setJobStatusAndSave(job, Job.Status.RUNNING, null);
+
             Agent agent = agentService.get(job.getAgentId());
             CmdIn cmd = cmdManager.createShellCmd(job, nextStep, tree);
-            setJobStatusAndSave(job, Job.Status.RUNNING, null);
+            boolean canExecute = conditionManager.run(cmd);
+
+            if (!canExecute) {
+                nextStep.setStartAt(new Date());
+                nextStep.setFinishAt(new Date());
+                updateJobTime(job, nextStep, tree, next.get());
 
+                setJobStatusAndSave(job, Job.Status.RUNNING, null);
+                stepService.toStatus(nextStep, Step.Status.SKIPPED, Step.MessageSkippedOnCondition);
+                return toNextStep(context);
+            }
+
+            // TODO: run condition
+            setJobStatusAndSave(job, Job.Status.RUNNING, null);
+            stepService.toStatus(nextStep, Step.Status.RUNNING, null);
             agentService.dispatch(cmd, agent);
             logInfo(job, "send to agent: step={}, agent={}", next.get().getName(), agent.getName());
             return true;
@@ -722,11 +755,11 @@ private synchronized void setJobStatusAndSave(Job job, Job.Status newStatus, Str
         logInfo(job, "status = {}", job.getStatus());
     }
 
-    private void updateJobTime(Job job, Step execCmd, NodeTree tree, StepNode node) {
+    private void updateJobTime(Job job, Step step, NodeTree tree, StepNode node) {
         if (tree.isFirst(node.getPath())) {
-            job.setStartAt(execCmd.getStartAt());
+            job.setStartAt(step.getStartAt());
         }
-        job.setFinishAt(execCmd.getFinishAt());
+        job.setFinishAt(step.getFinishAt());
     }
 
     private void updateJobContextAndLatestStatus(Job job, StepNode node, Step cmd) {

core/src/test/java/com/flowci/core/test/job/ConditionManagerTest.java

@@ -0,0 +1,54 @@
+package com.flowci.core.test.job;
+
+import com.flowci.core.agent.domain.ShellIn;
+import com.flowci.core.job.manager.ConditionManager;
+import com.flowci.core.test.SpringScenario;
+import groovy.util.ScriptException;
+import org.junit.Test;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.testng.Assert;
+
+public class ConditionManagerTest extends SpringScenario {
+
+    @Autowired
+    private ConditionManager conditionManager;
+
+    @Test
+    public void should_run_groovy_condition() throws ScriptException {
+        ShellIn in = new ShellIn();
+        in.getInputs().put("foo", "helloword");
+        in.setCondition(
+                "println \"$foo\"; "
+                        + "return true;"
+        );
+
+        Assert.assertTrue(conditionManager.run(in));
+    }
+
+    @Test(expected = ScriptException.class)
+    public void should_throw_exception_if_wrong_return_type() throws ScriptException {
+        ShellIn in = new ShellIn();
+        in.getInputs().put("foo", "helloword");
+        in.setCondition(
+                "println \"$foo\"; "
+                        + "hello = \"1234\";"
+                        + "return hello;"
+        );
+
+        conditionManager.run(in);
+    }
+
+    @Test(expected = ScriptException.class)
+    public void should_throw_exception_if_timeout() throws ScriptException {
+        ShellIn in = new ShellIn();
+        in.getInputs().put("foo", "helloword");
+        in.setCondition(
+                "sleep(6000); "
+                        + "println \"$foo\"; "
+                        + "hello = \"1234\";"
+                        + "return true;"
+        );
+
+        conditionManager.run(in);
+    }
+}

…/src/test/resources/flow-with-before.yml …c/test/resources/flow-with-condition.ymlcore/src/test/resources/flow-with-before.yml renamed to core/src/test/resources/flow-with-condition.yml core/src/test/resources/flow-with-before.yml renamed to core/src/test/resources/flow-with-condition.yml

@@ -10,7 +10,7 @@ steps:
 - envs:
     FLOW_WORKSPACE: "echo step"
     FLOW_VERSION: "echo step version"
-  before: |
+  condition: |
     return false
   script: |
     echo hello