Merge pull request #56 from st-tech/patch/makeCommandExecutorPluggable

patch make command executor pluggable

digdag-core/src/test/java/io/digdag/core/workflow/MockCommandExecutorFactory.java

@@ -0,0 +1,25 @@
+package io.digdag.core.workflow;
+
+import com.google.inject.Inject;
+import io.digdag.spi.CommandExecutor;
+import io.digdag.spi.CommandExecutorFactory;
+
+public class MockCommandExecutorFactory
+        implements CommandExecutorFactory
+{
+    @Inject
+    public MockCommandExecutorFactory()
+    { }
+
+    @Override
+    public String getType()
+    {
+        return "mock";
+    }
+
+    @Override
+    public CommandExecutor newCommandExecutor()
+    {
+        return new MockCommandExecutor();
+    }
+}

digdag-docs/src/command_reference.rst

@@ -403,6 +403,8 @@ In the config file, following parameters are available
 * api.max_archive_total_size_limit (integer. The maximum size of an archived project. i.e. ``digdag push`` size. default: 2MB(2\*1024\*1024))
 * eval.js-engine-type (type of ConfigEvalEngine. "nashorn" or "graal". "nashorn" is default on Java8 and "graal" is default on Java11)
 * eval.extended-syntax (boolean, default: true. Enable or disable extended syntax in graal. If true, nested ``{..}`` is allowed)
+* agent.max-task-threads (integer. The maximum number of task execution threads)
+* agent.command_executor.type (type of command executor, "ecs", "docker", "simple" or "kubernetes". See also CommandExecutor Plugins section bellow. default: "ecs")
 
 Authenticator Plugins
 *********************
@@ -419,6 +421,21 @@ Configuration:
 * server.authenticator.basic.password (string. Required if username is set)
 * server.authenticator.basic.admin (boolean. default `false`)
 
+CommandExecutor Plugins
+*********************
+
+CommandExecutor implementation is to be provided by a system plugin (See `System plugins section in Internal architecture <internal.html#system-plugins>`). Interface is ``io.digdag.spi.CommandExecutorFactory``.
+
+You can specify the command executor to use by `agent.command_executor.type`. (digdag version v0.10.4 and up)
+
+The following executors can be set by default.
+
+* ecs (default)
+* docker
+* simple
+* kubernetes (EXPERIMENTAL)
+
+See also [Command Executor](command_executor.html) for details.
 
 Secret Encryption Key
 *********************
@@ -993,4 +1010,3 @@ Common options
 
 :command:`-X KEY=VALUE`
   Add a performance system configuration. This option is for experimental use.
-

digdag-spi/src/main/java/io/digdag/spi/CommandExecutorFactory.java

@@ -0,0 +1,8 @@
+package io.digdag.spi;
+
+public interface CommandExecutorFactory
+{
+    String getType();
+
+    CommandExecutor newCommandExecutor();
+}

digdag-standards/src/main/java/io/digdag/standards/command/CommandExecutorModule.java

@@ -3,21 +3,30 @@
 import com.google.inject.Binder;
 import com.google.inject.Module;
 import com.google.inject.Scopes;
+import com.google.inject.multibindings.Multibinder;
 import io.digdag.spi.CommandExecutor;
+import io.digdag.spi.CommandExecutorFactory;
 import io.digdag.standards.command.ecs.DefaultEcsClientFactory;
 import io.digdag.standards.command.ecs.EcsClientFactory;
+import io.digdag.standards.command.kubernetes.DefaultKubernetesClientFactory;
+import io.digdag.standards.command.kubernetes.KubernetesClientFactory;
 
 public class CommandExecutorModule
     implements Module
 {
     @Override
     public void configure(Binder binder)
     {
-        binder.bind(CommandExecutor.class).to(EcsCommandExecutor.class).in(Scopes.SINGLETON);
         binder.bind(EcsClientFactory.class).to(DefaultEcsClientFactory.class).in(Scopes.SINGLETON);
-        //binder.bind(CommandExecutor.class).to(KubernetesCommandExecutor.class).in(Scopes.SINGLETON);
-        //binder.bind(KubernetesClientFactory.class).to(DefaultKubernetesClientFactory.class).in(Scopes.SINGLETON);
-        binder.bind(SimpleCommandExecutor.class).in(Scopes.SINGLETON);
-        binder.bind(DockerCommandExecutor.class).in(Scopes.SINGLETON);
+        binder.bind(KubernetesClientFactory.class).to(DefaultKubernetesClientFactory.class).in(Scopes.SINGLETON);
+
+        // CommandExecutor
+        Multibinder<CommandExecutorFactory> commandExecutorBinder = Multibinder.newSetBinder(binder, CommandExecutorFactory.class);
+        commandExecutorBinder.addBinding().to(SimpleCommandExecutorFactory.class).in(Scopes.SINGLETON);
+        commandExecutorBinder.addBinding().to(DockerCommandExecutorFactory.class).in(Scopes.SINGLETON);
+        commandExecutorBinder.addBinding().to(EcsCommandExecutorFactory.class).in(Scopes.SINGLETON);
+        commandExecutorBinder.addBinding().to(KubernetesCommandExecutorFactory.class).in(Scopes.SINGLETON);
+
+        binder.bind(CommandExecutor.class).toProvider(CommandExecutorProvider.class).in(Scopes.SINGLETON);
     }
 }

digdag-standards/src/main/java/io/digdag/standards/command/CommandExecutorProvider.java

@@ -0,0 +1,43 @@
+package io.digdag.standards.command;
+
+import com.google.inject.Inject;
+import com.google.inject.Provider;
+import io.digdag.client.config.Config;
+import io.digdag.client.config.ConfigException;
+import io.digdag.core.plugin.PluginSet;
+import io.digdag.spi.CommandExecutor;
+import io.digdag.spi.CommandExecutorFactory;
+
+import java.util.Set;
+import java.util.stream.Stream;
+
+public class CommandExecutorProvider
+        implements Provider<CommandExecutor>
+{
+    private final CommandExecutor commandExecutor;
+
+    @Inject
+    public CommandExecutorProvider(Set<CommandExecutorFactory> injectedFactories, PluginSet.WithInjector pluginSet, Config systemConfig)
+    {
+        // Set ECS as default command executor type
+        String executorName = systemConfig.get("agent.command_executor.type", String.class, "ecs");
+        Stream<CommandExecutorFactory> candidates = Stream.concat(
+                // Search from PluginSet first
+                pluginSet.getServiceProviders(CommandExecutorFactory.class).stream(),
+                // Then fallback to statically-injected commandExecutor
+                injectedFactories.stream());
+
+        CommandExecutorFactory factory = candidates
+                .filter(candidate -> candidate.getType().equals(executorName))
+                .findFirst()
+                .orElseThrow(() -> new ConfigException("Configured commandExecutor name is not found: " + executorName));
+
+        this.commandExecutor = factory.newCommandExecutor();
+    }
+
+    @Override
+    public CommandExecutor get()
+    {
+        return commandExecutor;
+    }
+}

digdag-standards/src/main/java/io/digdag/standards/command/DockerCommandExecutor.java

@@ -38,7 +38,6 @@ public class DockerCommandExecutor
     private final CommandLogger clog;
     private final SimpleCommandExecutor simple;
 
-    @Inject
     public DockerCommandExecutor(final CommandLogger clog, final SimpleCommandExecutor simple)
     {
         this.clog = clog;

digdag-standards/src/main/java/io/digdag/standards/command/DockerCommandExecutorFactory.java

@@ -0,0 +1,32 @@
+package io.digdag.standards.command;
+
+import com.google.inject.Inject;
+import io.digdag.spi.CommandExecutor;
+import io.digdag.spi.CommandExecutorFactory;
+import io.digdag.spi.CommandLogger;
+
+public class DockerCommandExecutorFactory
+        implements CommandExecutorFactory
+{
+    private final CommandLogger clog;
+    private final SimpleCommandExecutor simple;
+
+    @Inject
+    public DockerCommandExecutorFactory(CommandLogger clog)
+    {
+        this.clog = clog;
+        this.simple = new SimpleCommandExecutor(clog);
+    }
+
+    @Override
+    public String getType()
+    {
+        return "docker";
+    }
+
+    @Override
+    public CommandExecutor newCommandExecutor()
+    {
+        return new DockerCommandExecutor(clog, simple);
+    }
+}

digdag-standards/src/main/java/io/digdag/standards/command/EcsCommandExecutor.java

@@ -86,7 +86,6 @@ public class EcsCommandExecutor
     private final int retryDownloads, retryUploads;
     private final boolean curlFailOptOnUploads; // false by the default
 
-    @Inject
     public EcsCommandExecutor(
             final Config systemConfig,
             final EcsClientFactory ecsClientFactory,

digdag-standards/src/main/java/io/digdag/standards/command/EcsCommandExecutorFactory.java

@@ -0,0 +1,55 @@
+package io.digdag.standards.command;
+
+import com.google.inject.Inject;
+import io.digdag.client.config.Config;
+import io.digdag.core.archive.ProjectArchiveLoader;
+import io.digdag.core.storage.StorageManager;
+import io.digdag.spi.CommandExecutor;
+import io.digdag.spi.CommandExecutorFactory;
+import io.digdag.spi.CommandLogger;
+import io.digdag.standards.command.ecs.EcsClientFactory;
+
+public class EcsCommandExecutorFactory
+        implements CommandExecutorFactory
+{
+    private final Config systemConfig;
+    private final EcsClientFactory ecsClientFactory;
+    private final DockerCommandExecutor docker;
+    private final StorageManager storageManager;
+    private final ProjectArchiveLoader projectArchiveLoader;
+    private final CommandLogger clog;
+
+    @Inject
+    public EcsCommandExecutorFactory(
+            final Config systemConfig,
+            final EcsClientFactory ecsClientFactory,
+            final StorageManager storageManager,
+            final ProjectArchiveLoader projectArchiveLoader,
+            final CommandLogger clog)
+    {
+        this.systemConfig = systemConfig;
+        this.ecsClientFactory = ecsClientFactory;
+        this.docker = new DockerCommandExecutor(clog, new SimpleCommandExecutor(clog));
+        this.storageManager = storageManager;
+        this.projectArchiveLoader = projectArchiveLoader;
+        this.clog = clog;
+    }
+
+    @Override
+    public String getType()
+    {
+        return "ecs";
+    }
+
+    @Override
+    public CommandExecutor newCommandExecutor()
+    {
+        return new EcsCommandExecutor(
+                systemConfig,
+                ecsClientFactory,
+                docker,
+                storageManager,
+                projectArchiveLoader,
+                clog);
+    }
+}

digdag-standards/src/main/java/io/digdag/standards/command/KubernetesCommandExecutorFactory.java

@@ -0,0 +1,55 @@
+package io.digdag.standards.command;
+
+import com.google.inject.Inject;
+import io.digdag.client.config.Config;
+import io.digdag.core.archive.ProjectArchiveLoader;
+import io.digdag.core.storage.StorageManager;
+import io.digdag.spi.CommandExecutor;
+import io.digdag.spi.CommandExecutorFactory;
+import io.digdag.spi.CommandLogger;
+import io.digdag.standards.command.kubernetes.KubernetesClientFactory;
+
+public class KubernetesCommandExecutorFactory
+    implements CommandExecutorFactory
+{
+    private final Config systemConfig;
+    private final KubernetesClientFactory kubernetesClientFactory;
+    private final DockerCommandExecutor docker;
+    private final StorageManager storageManager;
+    private final ProjectArchiveLoader projectArchiveLoader;
+    private final CommandLogger clog;
+
+    @Inject
+    public KubernetesCommandExecutorFactory(
+            final Config systemConfig,
+            final KubernetesClientFactory kubernetesClientFactory,
+            final StorageManager storageManager,
+            final ProjectArchiveLoader projectArchiveLoader,
+            final CommandLogger clog)
+    {
+        this.systemConfig = systemConfig;
+        this.kubernetesClientFactory = kubernetesClientFactory;
+        this.docker = new DockerCommandExecutor(clog, new SimpleCommandExecutor(clog));
+        this.storageManager = storageManager;
+        this.projectArchiveLoader = projectArchiveLoader;
+        this.clog = clog;
+    }
+
+    @Override
+    public String getType()
+    {
+        return "kubernetes";
+    }
+
+    @Override
+    public CommandExecutor newCommandExecutor()
+    {
+        return new KubernetesCommandExecutor(
+                systemConfig,
+                kubernetesClientFactory,
+                docker,
+                storageManager,
+                projectArchiveLoader,
+                clog);
+    }
+}

digdag-standards/src/main/java/io/digdag/standards/command/SimpleCommandExecutor.java

@@ -16,7 +16,6 @@ public class SimpleCommandExecutor
 {
     private final CommandLogger clog;
 
-    @Inject
     public SimpleCommandExecutor(final CommandLogger clog)
     {
         this.clog = clog;

digdag-standards/src/main/java/io/digdag/standards/command/SimpleCommandExecutorFactory.java

@@ -0,0 +1,30 @@
+package io.digdag.standards.command;
+
+import com.google.inject.Inject;
+import io.digdag.spi.CommandExecutor;
+import io.digdag.spi.CommandExecutorFactory;
+import io.digdag.spi.CommandLogger;
+
+public class SimpleCommandExecutorFactory
+        implements CommandExecutorFactory
+{
+    private final CommandLogger clog;
+
+    @Inject
+    public SimpleCommandExecutorFactory(CommandLogger clog)
+    {
+        this.clog = clog;
+    }
+
+    @Override
+    public String getType()
+    {
+        return "simple";
+    }
+
+    @Override
+    public CommandExecutor newCommandExecutor()
+    {
+        return new SimpleCommandExecutor(clog);
+    }
+}