Merge pull request #376 from FlowCI/develop

Develop

Author: gy2006

.run/Application - 1.run.xml

@@ -14,6 +14,7 @@
       <env name="FLOWCI_SERVER_PORT" value="8080" />
       <env name="FLOWCI_TEMPLATES" value="https://raw.githubusercontent.com/FlowCI/templates/master/templates.json" />
       <env name="FLOWCI_AGENT_IMAGE" value="flowci/agent:dev" />
+      <env name="FLOWCI_SERVER_URL" value="http://192.168.0.4:8080" />
     </envs>
     <method v="2">
       <option name="Make" enabled="true" />

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

@@ -1,10 +1,8 @@
 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;
@@ -37,14 +35,13 @@ public enum ShellType {
 
     private List<DockerOption> dockers;
 
-    @JsonIgnore
-    private String condition;
-
     private List<String> bash;
 
     private List<String> pwsh;
 
-    private int timeout = 1800;
+    private int timeout = 1800; // from StepNode.timeout
+
+    private int retry; // from StepNode.retry
 
     private Vars<String> inputs;
 
@@ -75,9 +72,4 @@ 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/agent/manager/AgentEventManager.java

@@ -54,20 +54,19 @@ public <T> void writeMessage(String token, ResponseMessage<T> msg) {
 
         try {
             byte[] bytes = objectMapper.writeValueAsBytes(msg);
-            session.sendMessage(new BinaryMessage(bytes));
+            writeMessage(session, bytes);
         } catch (IOException e) {
             log.warn("Unable to write response message for agent {}: {}", token, e.getMessage());
         }
     }
 
-    public void writeMessage(String token, byte[] bytes) throws IOException {
+    public void writeMessage(String token, byte[] bytes) {
         WebSocketSession session = agentSessionStore.get(token);
         if (session == null) {
             log.warn("Agent {} not connected", token);
             return;
         }
-
-        session.sendMessage(new BinaryMessage(bytes));
+        writeMessage(session, bytes);
     }
 
     @Override
@@ -123,12 +122,22 @@ public void afterConnectionClosed(WebSocketSession session, CloseStatus status)
     private <T> void writeMessage(WebSocketSession session, ResponseMessage<T> msg) {
         try {
             byte[] bytes = objectMapper.writeValueAsBytes(msg);
-            session.sendMessage(new BinaryMessage(bytes));
+            writeMessage(session, bytes);
         } catch (IOException e) {
             log.warn(e);
         }
     }
 
+    private void writeMessage(WebSocketSession session, byte[] body) {
+        synchronized (session) {
+            try {
+                session.sendMessage(new BinaryMessage(body));
+            } catch (IOException e) {
+                log.warn(e);
+            }
+        }
+    }
+
     private void onConnected(WebSocketSession session, String token, byte[] body) {
         try {
             AgentInit init = objectMapper.readValue(body, AgentInit.class);

core/src/main/java/com/flowci/core/common/config/AppConfig.java

@@ -34,6 +34,7 @@
 import org.springframework.core.ResolvableType;
 import org.springframework.core.task.TaskExecutor;
 import org.springframework.scheduling.annotation.EnableScheduling;
+import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
 
 import javax.annotation.PostConstruct;
 import java.io.IOException;
@@ -80,8 +81,10 @@ public ObjectMapper objectMapper() {
     }
 
     @Bean("appTaskExecutor")
-    public TaskExecutor getAppTaskExecutor() {
-        return ThreadHelper.createTaskExecutor(100, 100, 50, "app-task-");
+    public ThreadPoolTaskExecutor getAppTaskExecutor() {
+        int corePoolSize = appProperties.getCorePoolSize();
+        int maxPoolSize = appProperties.getMaxPoolSize();
+        return ThreadHelper.createTaskExecutor(maxPoolSize, corePoolSize, 10, "app-task-");
     }
 
     @Bean(name = "applicationEventMulticaster")

core/src/main/java/com/flowci/core/common/config/AppProperties.java

@@ -57,6 +57,10 @@ public class AppProperties {
 
     private boolean socketContainer;
 
+    private int corePoolSize;
+
+    private int maxPoolSize;
+
     @Bean("zkProperties")
     @ConfigurationProperties(prefix = "app.zookeeper")
     public Zookeeper zk() {

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

@@ -0,0 +1,16 @@
+package com.flowci.core.common.manager;
+
+import com.flowci.domain.Vars;
+import groovy.util.ScriptException;
+
+import javax.annotation.Nullable;
+
+public interface ConditionManager {
+
+    /**
+     * Verify the input condition is groovy script with boolean return or not
+     */
+    void verify(@Nullable String condition) throws ScriptException;
+
+    boolean run(@Nullable String groovyScript, @Nullable Vars<String> envs) throws ScriptException;
+}

core/src/main/java/com/flowci/core/job/manager/ConditionManagerImpl.java renamed to core/src/main/java/com/flowci/core/common/manager/ConditionManagerImpl.java

@@ -1,7 +1,7 @@
-package com.flowci.core.job.manager;
+package com.flowci.core.common.manager;
 
-import com.flowci.core.agent.domain.CmdIn;
-import com.flowci.core.agent.domain.ShellIn;
+import com.flowci.domain.Vars;
+import com.google.common.base.Strings;
 import groovy.lang.Binding;
 import groovy.lang.GroovyRuntimeException;
 import groovy.lang.GroovyShell;
@@ -11,7 +11,11 @@
 import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
 import org.springframework.stereotype.Component;
 
-import java.util.concurrent.*;
+import javax.annotation.Nullable;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
 
 @Log4j2
 @Component
@@ -23,23 +27,29 @@ public class ConditionManagerImpl implements ConditionManager {
     private ThreadPoolTaskExecutor jobConditionExecutor;
 
     @Override
-    public boolean run(CmdIn in) throws ScriptException {
-        if (!(in instanceof ShellIn)) {
-            return true;
+    public void verify(@Nullable String condition) throws ScriptException {
+        try {
+            new GroovyShell().parse(condition);
+        } catch (Exception e) {
+            throw new ScriptException("Invalid groovy condition: " + e.getMessage());
         }
+    }
 
-        ShellIn shell = (ShellIn) in;
-        if (!shell.hasCondition()) {
+    @Override
+    public boolean run(String groovyScript, Vars<String> envs) throws ScriptException {
+        if (Strings.isNullOrEmpty(groovyScript)) {
             return true;
         }
 
         Future<Boolean> submit = jobConditionExecutor.submit(() -> {
             Binding binding = new Binding();
-            shell.getInputs().forEach(binding::setVariable);
+            if (envs != null) {
+                envs.forEach(binding::setVariable);
+            }
 
             try {
                 GroovyShell groovy = new GroovyShell(binding);
-                Object value = groovy.evaluate(shell.getCondition());
+                Object value = groovy.evaluate(groovyScript);
                 if (value instanceof Boolean) {
                     return (Boolean) value;
                 }

core/src/main/java/com/flowci/core/common/service/SettingServiceImpl.java

@@ -35,18 +35,21 @@ public class SettingServiceImpl implements SettingService {
     public void setDefaultValue() {
         taskManager.run("init-default-settings", () -> {
             Optional<Settings> optional = settingsDao.findById(Settings.DefaultId);
-            if (!optional.isPresent()) {
 
-                String address = serverProperties.getAddress().toString().replace("/", "");
-                String serverUrl = environment.getProperty(
-                        Variables.App.ServerUrl,
-                        String.format("http://%s:%s", address, serverProperties.getPort())
-                );
+            String address = serverProperties.getAddress().toString().replace("/", "");
+            String serverUrl = environment.getProperty(
+                    Variables.App.ServerUrl,
+                    String.format("http://%s:%s", address, serverProperties.getPort())
+            );
 
-                Settings s = new Settings();
-                s.setServerUrl(serverUrl);
-                settingsDao.save(s);
+            Settings s = new Settings();
+
+            if (optional.isPresent()) {
+                s = optional.get();
             }
+
+            s.setServerUrl(serverUrl);
+            settingsDao.save(s);
         });
     }
 

core/src/main/java/com/flowci/core/flow/service/FlowServiceImpl.java

@@ -17,29 +17,41 @@
 package com.flowci.core.flow.service;
 
 import com.flowci.core.common.domain.Variables;
+import com.flowci.core.common.manager.ConditionManager;
 import com.flowci.core.common.manager.SessionManager;
 import com.flowci.core.common.manager.SpringEventManager;
 import com.flowci.core.flow.dao.FlowDao;
 import com.flowci.core.flow.dao.FlowUserDao;
+import com.flowci.core.flow.dao.YmlDao;
 import com.flowci.core.flow.domain.ConfirmOption;
 import com.flowci.core.flow.domain.Flow;
 import com.flowci.core.flow.domain.Flow.Status;
+import com.flowci.core.flow.domain.WebhookStatus;
+import com.flowci.core.flow.domain.Yml;
 import com.flowci.core.flow.event.FlowCreatedEvent;
 import com.flowci.core.flow.event.FlowDeletedEvent;
 import com.flowci.core.flow.event.FlowInitEvent;
+import com.flowci.core.job.domain.Job;
+import com.flowci.core.job.event.CreateNewJobEvent;
 import com.flowci.core.secret.domain.Secret;
 import com.flowci.core.secret.event.CreateAuthEvent;
 import com.flowci.core.secret.event.CreateRsaEvent;
 import com.flowci.core.secret.event.GetSecretEvent;
+import com.flowci.core.trigger.domain.GitPingTrigger;
+import com.flowci.core.trigger.domain.GitTrigger;
+import com.flowci.core.trigger.event.GitHookEvent;
 import com.flowci.core.user.event.UserDeletedEvent;
 import com.flowci.domain.*;
 import com.flowci.exception.ArgumentException;
 import com.flowci.exception.DuplicateException;
 import com.flowci.exception.NotFoundException;
 import com.flowci.exception.StatusException;
 import com.flowci.store.FileManager;
+import com.flowci.tree.FlowNode;
+import com.flowci.tree.YmlParser;
 import com.flowci.util.ObjectsHelper;
 import com.google.common.collect.Sets;
+import groovy.util.ScriptException;
 import lombok.extern.log4j.Log4j2;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.context.event.ContextRefreshedEvent;
@@ -60,6 +72,9 @@ public class FlowServiceImpl implements FlowService {
     @Autowired
     private FlowDao flowDao;
 
+    @Autowired
+    private YmlDao ymlDao;
+
     @Autowired
     private FlowUserDao flowUserDao;
 
@@ -72,6 +87,9 @@ public class FlowServiceImpl implements FlowService {
     @Autowired
     private FileManager fileManager;
 
+    @Autowired
+    private ConditionManager conditionManager;
+
     @Autowired
     private YmlService ymlService;
 
@@ -293,10 +311,64 @@ public void deleteUserFromFlow(UserDeletedEvent event) {
         // TODO:
     }
 
+    @EventListener
+    public void onGitHookEvent(GitHookEvent event) {
+        GitTrigger trigger = event.getTrigger();
+        Flow flow = flowDao.findByName(event.getFlow());
+
+        if (event.isPingEvent()) {
+            GitPingTrigger ping = (GitPingTrigger) trigger;
+
+            WebhookStatus ws = new WebhookStatus();
+            ws.setAdded(true);
+            ws.setCreatedAt(ping.getCreatedAt());
+            ws.setEvents(ping.getEvents());
+
+            flow.setWebhookStatus(ws);
+            flowDao.save(flow);
+            return;
+        }
+
+        if (trigger.isSkip()) {
+            log.info("Ignore git trigger {} since skip message", trigger);
+            return;
+        }
+
+        Optional<Yml> optional = ymlDao.findById(flow.getId());
+        if (!optional.isPresent()) {
+            log.warn("No available yml for flow {}", flow.getName());
+            return;
+        }
+
+        Yml yml = optional.get();
+        FlowNode root = YmlParser.load(flow.getName(), yml.getRaw());
+        if (!canStartJob(root, trigger)) {
+            log.debug("Cannot start job, condition not match: {}", root.getCondition());
+            return;
+        }
+
+        StringVars gitInput = trigger.toVariableMap();
+        Job.Trigger jobTrigger = trigger.toJobTrigger();
+
+        eventManager.publish(new CreateNewJobEvent(this, flow, yml.getRaw(), jobTrigger, gitInput));
+    }
+
+
     // ====================================================================
     // %% Utils
     // ====================================================================
 
+    private boolean canStartJob(FlowNode root, GitTrigger trigger) {
+        try {
+            String groovy = root.getCondition();
+            Vars<String> envs = trigger.toVariableMap();
+            return conditionManager.run(groovy, envs);
+        } catch (ScriptException e) {
+            log.warn("Illegal groovy script at condition section", e);
+            return false;
+        }
+    }
+
     private void setupDefaultVars(Flow flow) {
         Vars<VarValue> localVars = flow.getLocally();
         localVars.put(Variables.Flow.Name, VarValue.of(flow.getName(), VarType.STRING, false));