refactor clean up notification and local docker task bean

Author: gy2006

core/src/main/java/com/flowci/core/flow/domain/SaveNotify.java

@@ -23,7 +23,7 @@
 import com.flowci.core.flow.domain.Yml;
 import com.flowci.core.flow.event.FlowDeletedEvent;
 import com.flowci.core.plugin.event.GetPluginEvent;
-import com.flowci.domain.Notification;
+import com.flowci.domain.LocalTask;
 import com.flowci.domain.Vars;
 import com.flowci.exception.ArgumentException;
 import com.flowci.exception.NotFoundException;
@@ -146,8 +146,8 @@ private Optional<RuntimeException> verifyPlugins(FlowNode flowNode) {
             }
         }
 
-        for (Notification n : flowNode.getNotifications()) {
-            plugins.add(n.getPlugin());
+        for (LocalTask t : flowNode.getNotifications()) {
+            plugins.add(t.getPlugin());
         }
 
         for (String plugin : plugins) {

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

@@ -1,7 +1,7 @@
 package com.flowci.core.job.event;
 
 import com.flowci.core.common.event.AbstractSyncEvent;
-import com.flowci.core.job.domain.LocalTask;
+import com.flowci.domain.LocalTask;
 import lombok.Getter;
 
 /**

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

@@ -3,14 +3,17 @@
 import com.flowci.core.api.adviser.ApiAuth;
 import com.flowci.core.common.domain.Variables;
 import com.flowci.core.common.helper.ThreadHelper;
-import com.flowci.core.job.domain.LocalDockerTask;
 import com.flowci.domain.StringVars;
+import com.flowci.domain.Vars;
+import com.flowci.util.StringHelper;
 import com.github.dockerjava.api.DockerClient;
 import com.github.dockerjava.api.async.ResultCallback;
 import com.github.dockerjava.api.command.CreateContainerCmd;
 import com.github.dockerjava.api.command.CreateContainerResponse;
 import com.github.dockerjava.api.command.InspectContainerResponse;
 import com.github.dockerjava.api.model.*;
+import lombok.Getter;
+import lombok.Setter;
 import lombok.extern.log4j.Log4j2;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Component;
@@ -58,20 +61,20 @@ public boolean pullImage(String image) throws InterruptedException {
         return callback.isSuccess;
     }
 
-    public String createAndStartContainer(LocalDockerTask task) {
+    public String createAndStartContainer(Option option) {
         StringVars defaultEnv = new StringVars(4);
         defaultEnv.put(Variables.App.Url, serverUrl);
         defaultEnv.put(Variables.Agent.Token, ApiAuth.LocalTaskToken);
         defaultEnv.put(Variables.Agent.Workspace, "/ws/");
         defaultEnv.put(Variables.Agent.PluginDir, "/ws/.plugins");
 
-        CreateContainerCmd createContainerCmd = client.createContainerCmd(task.getImage())
-                .withEnv(defaultEnv.merge(task.getInputs()).toList())
-                .withCmd("/bin/bash", "-c", task.getScript());
+        CreateContainerCmd createContainerCmd = client.createContainerCmd(option.image)
+                .withEnv(defaultEnv.merge(option.inputs).toList())
+                .withCmd("/bin/bash", "-c", option.script);
 
-        if (task.hasPlugin()) {
+        if (option.hasPlugin()) {
             createContainerCmd.withBinds(
-                    new Bind(task.getPluginDir().toString(), new Volume("/ws/.plugins/" + task.getPlugin())));
+                    new Bind(option.pluginDir, new Volume("/ws/.plugins/" + option.plugin)));
         }
 
         CreateContainerResponse container = createContainerCmd.exec();
@@ -132,6 +135,25 @@ public void removeContainer(String cid) {
         log.debug("Container {} been removed", cid);
     }
 
+    @Getter
+    @Setter
+    public static final class Option {
+
+        private String image;
+
+        private Vars<String> inputs;
+
+        private String script;
+
+        private String plugin;
+
+        private String pluginDir;
+
+        private boolean hasPlugin() {
+            return StringHelper.hasValue(plugin) && StringHelper.hasValue(pluginDir) ;
+        }
+    }
+
     private static abstract class DockerCallback<T> implements ResultCallback<T> {
 
         protected CountDownLatch counter = new CountDownLatch(1);

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

@@ -8,7 +8,6 @@
 import com.flowci.core.job.dao.JobDao;
 import com.flowci.core.job.domain.ExecutedCmd;
 import com.flowci.core.job.domain.Job;
-import com.flowci.core.job.domain.LocalDockerTask;
 import com.flowci.core.job.event.JobReceivedEvent;
 import com.flowci.core.job.event.JobStatusChangeEvent;
 import com.flowci.core.job.manager.CmdManager;
@@ -807,21 +806,8 @@ private Consumer<JobSmContext> notificationConsumer() {
             Job job = context.job;
             NodeTree tree = ymlManager.getTree(job);
 
-            for (Notification n : tree.getRoot().getNotifications()) {
-                if (!n.isEnabled()) {
-                    continue;
-                }
-
-                StringVars input = new StringVars(job.getContext());
-                input.merge(n.getInputs());
-
-                LocalDockerTask task = new LocalDockerTask();
-                task.setName(n.getPlugin()); // plugin name as task name
-                task.setPlugin(n.getPlugin());
-                task.setJobId(job.getId());
-                task.setInputs(input);
-
-                localTaskService.executeAsync(task);
+            for (LocalTask t : tree.getRoot().getNotifications()) {
+                localTaskService.executeAsync(job, t);
             }
         };
     }

core/src/main/java/com/flowci/core/job/event/StartAsyncLocalTaskEvent.java

@@ -2,7 +2,7 @@
 
 import com.flowci.core.job.domain.ExecutedLocalTask;
 import com.flowci.core.job.domain.Job;
-import com.flowci.core.job.domain.LocalDockerTask;
+import com.flowci.domain.LocalTask;
 
 import java.util.List;
 
@@ -12,7 +12,7 @@ public interface LocalTaskService {
 
     List<ExecutedLocalTask> list(Job job);
 
-    void executeAsync(LocalDockerTask task);
+    void executeAsync(Job job, LocalTask task);
 
-    ExecutedLocalTask execute(LocalDockerTask task);
+    ExecutedLocalTask execute(Job job, LocalTask task);
 }

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

@@ -5,13 +5,13 @@
 import com.flowci.core.job.domain.Executed;
 import com.flowci.core.job.domain.ExecutedLocalTask;
 import com.flowci.core.job.domain.Job;
-import com.flowci.core.job.domain.LocalDockerTask;
 import com.flowci.core.job.manager.DockerManager;
 import com.flowci.core.job.manager.YmlManager;
 import com.flowci.core.plugin.domain.Plugin;
 import com.flowci.core.plugin.event.GetPluginAndVerifySetContext;
 import com.flowci.core.plugin.event.GetPluginEvent;
-import com.flowci.domain.Notification;
+import com.flowci.domain.LocalTask;
+import com.flowci.domain.StringVars;
 import com.flowci.exception.NotAvailableException;
 import com.flowci.exception.StatusException;
 import com.flowci.tree.NodeTree;
@@ -30,6 +30,9 @@
 @Service
 public class LocalTaskServiceImpl implements LocalTaskService {
 
+    private final static String DefaultImage = "flowci/plugin-runtime";
+    private final static int DefaultTimeout = 60; // seconds
+
     @Autowired
     private ExecutedLocalTaskDao executedLocalTaskDao;
 
@@ -48,19 +51,15 @@ public class LocalTaskServiceImpl implements LocalTaskService {
     @Override
     public void init(Job job) {
         NodeTree tree = ymlManager.getTree(job);
-        List<Notification> list = tree.getRoot().getNotifications();
+        List<LocalTask> list = tree.getRoot().getNotifications();
         if (list.isEmpty()) {
             return;
         }
 
         List<ExecutedLocalTask> tasks = new ArrayList<>(list.size());
-        for (Notification n : list) {
-            if (!n.isEnabled()) {
-                continue;
-            }
-
+        for (LocalTask n : list) {
             ExecutedLocalTask t = new ExecutedLocalTask();
-            t.setName(n.getPlugin());
+            t.setName(n.getPlugin()); // name is plugin name
             t.setJobId(job.getId());
             tasks.add(t);
         }
@@ -73,14 +72,14 @@ public List<ExecutedLocalTask> list(Job job) {
     }
 
     @Override
-    public void executeAsync(LocalDockerTask task) {
-        localTaskExecutor.execute(() -> execute(task));
+    public void executeAsync(Job job, LocalTask task) {
+        localTaskExecutor.execute(() -> execute(job, task));
     }
 
     @Override
-    public ExecutedLocalTask execute(LocalDockerTask task) {
+    public ExecutedLocalTask execute(Job job, LocalTask task) {
         Optional<ExecutedLocalTask> optional =
-                executedLocalTaskDao.findByJobIdAndAndName(task.getJobId(), task.getName());
+                executedLocalTaskDao.findByJobIdAndAndName(job.getId(), task.getPlugin());
 
         if (!optional.isPresent()) {
             throw new StatusException("Executed local task should be init() before execute");
@@ -91,47 +90,55 @@ public ExecutedLocalTask execute(LocalDockerTask task) {
         exec.setStatus(Executed.Status.RUNNING);
         executedLocalTaskDao.save(exec);
 
+        DockerManager.Option option = new DockerManager.Option();
+        option.setImage(DefaultImage);
+        option.setInputs(new StringVars(job.getContext()).merge(task.getEnvs()));
+
         if (task.hasPlugin()) {
             String name = task.getPlugin();
-            GetPluginEvent event = eventManager.publish(new GetPluginAndVerifySetContext(this, name, task.getInputs()));
+            GetPluginEvent event = eventManager.publish(new GetPluginAndVerifySetContext(this, name, option.getInputs()));
 
             if (event.hasError()) {
-                exec.setError(event.getError().getMessage());
+                String message = event.getError().getMessage();
+                log.warn(message);
+
+                exec.setError(message);
                 exec.setStatus(Executed.Status.EXCEPTION);
                 exec.setFinishAt(new Date());
                 executedLocalTaskDao.save(exec);
                 return exec;
             }
 
             Plugin plugin = event.getFetched();
-            task.setScript(plugin.getScript());
-            task.setPluginDir(event.getDir());
+
+            option.setScript(plugin.getScript());
+            option.setPlugin(plugin.getName());
+            option.setPluginDir(event.getDir().toString());
 
             // apply docker image only from plugin if it's specified
             ObjectsHelper.ifNotNull(plugin.getDocker(), (docker) -> {
-                task.setImage(plugin.getDocker().getImage());
+                option.setImage(plugin.getDocker().getImage());
             });
         }
 
-        log.info("Start local task {} image = {} for job {}", task.getName(), task.getImage(), task.getJobId());
-        executedLocalTaskDao.save(runDockerTask(task, exec));
+        log.info("Start local task {} image = {} for job {}", task.getPlugin(), option.getImage(), job.getId());
+        executedLocalTaskDao.save(runDockerTask(option, exec));
         return exec;
     }
 
-    private ExecutedLocalTask runDockerTask(LocalDockerTask task, ExecutedLocalTask r) {
+    private ExecutedLocalTask runDockerTask(DockerManager.Option option, ExecutedLocalTask r) {
         try {
-            String image = task.getImage();
-
+            String image = option.getImage();
             boolean isSuccess = dockerManager.pullImage(image);
             if (!isSuccess) {
                 throw new NotAvailableException("Docker image {0} not available", image);
             }
 
-            String cid = dockerManager.createAndStartContainer(task);
+            String cid = dockerManager.createAndStartContainer(option);
             r.setContainerId(cid);
             dockerManager.printContainerLog(cid);
 
-            if (!dockerManager.waitContainer(cid, task.getTimeoutInSecond())) {
+            if (!dockerManager.waitContainer(cid, DefaultTimeout)) {
                 dockerManager.killContainer(cid);
             }
 

core/src/main/java/com/flowci/core/job/service/JobActionServiceImpl.java

@@ -225,7 +225,7 @@ public void should_finish_whole_job() throws InterruptedException, IOException {
         Mockito.doAnswer((Answer<Void>) invocation -> {
             localTaskCountDown.countDown();
             return null;
-        }).when(localTaskManager).executeAsync(Mockito.any());
+        }).when(localTaskManager).executeAsync(Mockito.any(), Mockito.any());
 
         addEventListener((ApplicationListener<StartAsyncLocalTaskEvent>) event -> localTaskCountDown.countDown());