[Improvement-12536][k8s] Support the command for the container in k8s task plugin (#12538)

docs/docs/en/guide/task/kubernetes.md

@@ -22,6 +22,7 @@ K8S task type used to execute a batch task. In this task, the worker submits the
 | Min CPU          | Minimum CPU requirement for running k8s task.                                                                    |
 | Min Memory       | Minimum memory requirement for running k8s task.                                                                 |
 | Image            | The registry url for image.                                                                                      |
+| Command          | The container execution command, for example: /bin/echo hello world                                              |
 | Custom parameter | It is a local user-defined parameter for K8S task, these params will pass to container as environment variables. |
 
 ## Task Example

docs/docs/zh/guide/task/kubernetes.md

@@ -22,6 +22,7 @@ kubernetes任务类型，用于在kubernetes上执行一个短时和批处理的
 | 最小CPU    | 任务在kubernetes上运行所需的最小CPU                                        |
 | 最小内存     | 任务在kubernetes上运行所需的最小内存                                         |
 | 镜像       | 镜像地址                                                            |
+| 容器执行命令   | 容器执行命令，例如：/bin/echo hello world                                 |
 | 自定义参数    | kubernetes任务局部的用户自定义参数，自定义参数最终会通过环境变量形式存在于容器中，提供给kubernetes任务使用 |
 
 ## 任务样例

dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/main/java/org/apache/dolphinscheduler/plugin/task/api/TaskConstants.java

@@ -468,6 +468,7 @@ public class TaskConstants {
     public static final int LOG_LINES = 500;
     public static final String NAMESPACE_NAME = "name";
     public static final String CLUSTER = "cluster";
+    public static final String COMMAND_SPLIT_REGEX = "[^\\s\"'`]+|\"([^\"]+)\"|'([^']+)'|`([^`]+)`";
 
     /**
      * conda config used by jupyter task plugin

dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/main/java/org/apache/dolphinscheduler/plugin/task/api/k8s/K8sTaskMainParameters.java

@@ -19,75 +19,19 @@ package org.apache.dolphinscheduler.plugin.task.api.k8s;
 
 import java.util.Map;
 
+import lombok.Data;
+
 /**
  * k8s task parameters
  */
+@Data
 public class K8sTaskMainParameters {
 
     private String image;
+    private String command;
     private String namespaceName;
     private String clusterName;
     private double minCpuCores;
     private double minMemorySpace;
     private Map<String, String> paramsMap;
-
-    public String getImage() {
-        return image;
-    }
-
-    public void setImage(String image) {
-        this.image = image;
-    }
-
-    public double getMinCpuCores() {
-        return minCpuCores;
-    }
-
-    public void setMinCpuCores(double minCpuCores) {
-        this.minCpuCores = minCpuCores;
-    }
-
-    public double getMinMemorySpace() {
-        return minMemorySpace;
-    }
-
-    public void setMinMemorySpace(double minMemorySpace) {
-        this.minMemorySpace = minMemorySpace;
-    }
-
-    public String getNamespaceName() {
-        return namespaceName;
-    }
-
-    public void setNamespaceName(String namespaceName) {
-        this.namespaceName = namespaceName;
-    }
-
-    public String getClusterName() {
-        return clusterName;
-    }
-
-    public void setClusterName(String clusterName) {
-        this.clusterName = clusterName;
-    }
-
-    public Map<String, String> getParamsMap() {
-        return paramsMap;
-    }
-
-    public void setParamsMap(Map<String, String> paramsMap) {
-        this.paramsMap = paramsMap;
-    }
-
-    @Override
-    public String toString() {
-        return "K8sTaskMainParameters{"
-                + "image='" + image + '\''
-                + ", namespaceName='" + namespaceName + '\''
-                + ", clusterName='" + clusterName + '\''
-                + ", minCpuCores=" + minCpuCores
-                + ", minMemorySpace=" + minMemorySpace
-                + ", paramsMap=" + paramsMap
-                + '}';
-    }
 }

dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/main/java/org/apache/dolphinscheduler/plugin/task/api/k8s/impl/K8sTaskExecutor.java

@@ -18,6 +18,7 @@
 package org.apache.dolphinscheduler.plugin.task.api.k8s.impl;
 
 import static org.apache.dolphinscheduler.plugin.task.api.TaskConstants.API_VERSION;
+import static org.apache.dolphinscheduler.plugin.task.api.TaskConstants.COMMAND_SPLIT_REGEX;
 import static org.apache.dolphinscheduler.plugin.task.api.TaskConstants.CPU;
 import static org.apache.dolphinscheduler.plugin.task.api.TaskConstants.EXIT_CODE_FAILURE;
 import static org.apache.dolphinscheduler.plugin.task.api.TaskConstants.EXIT_CODE_KILL;
@@ -53,6 +54,8 @@ import java.util.Locale;
 import java.util.Map;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
 
 import org.slf4j.Logger;
 
@@ -108,6 +111,17 @@ public class K8sTaskExecutor extends AbstractK8sTaskExecutor {
                 envVars.add(envVar);
             }
         }
+
+        String commandString = k8STaskMainParameters.getCommand();
+        List<String> commands = new ArrayList<>();
+
+        if (commandString != null) {
+            Matcher commandMatcher = Pattern.compile(COMMAND_SPLIT_REGEX).matcher(commandString.trim());
+            while (commandMatcher.find()) {
+                commands.add(commandMatcher.group());
+            }
+        }
+
         return new JobBuilder()
                 .withApiVersion(API_VERSION)
                 .withNewMetadata()
@@ -122,6 +136,7 @@ public class K8sTaskExecutor extends AbstractK8sTaskExecutor {
                 .addNewContainer()
                 .withName(k8sJobName)
                 .withImage(image)
+                .withCommand(commands.size() == 0 ? null : commands)
                 .withImagePullPolicy(IMAGE_PULL_POLICY)
                 .withResources(new ResourceRequirements(limitRes, reqRes))
                 .withEnv(envVars)

dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/main/java/org/apache/dolphinscheduler/plugin/task/api/parameters/K8sTaskParameters.java

@@ -24,48 +24,20 @@ import org.apache.commons.lang3.StringUtils;
 import java.util.ArrayList;
 import java.util.List;
 
+import lombok.Data;
+
 /**
  * k8s task parameters
  */
+@Data
 public class K8sTaskParameters extends AbstractParameters {
 
     private String image;
     private String namespace;
+    private String command;
     private double minCpuCores;
     private double minMemorySpace;
 
-    public String getImage() {
-        return image;
-    }
-
-    public void setImage(String image) {
-        this.image = image;
-    }
-
-    public String getNamespace() {
-        return namespace;
-    }
-
-    public void setNamespace(String namespace) {
-        this.namespace = namespace;
-    }
-
-    public double getMinCpuCores() {
-        return minCpuCores;
-    }
-
-    public void setMinCpuCores(double minCpuCores) {
-        this.minCpuCores = minCpuCores;
-    }
-
-    public double getMinMemorySpace() {
-        return minMemorySpace;
-    }
-
-    public void setMinMemorySpace(double minMemorySpace) {
-        this.minMemorySpace = minMemorySpace;
-    }
-
     @Override
     public boolean checkParameters() {
         return StringUtils.isNotEmpty(image) && StringUtils.isNotEmpty(namespace);
@@ -75,14 +47,4 @@ public class K8sTaskParameters extends AbstractParameters {
     public List<ResourceInfo> getResourceFilesList() {
         return new ArrayList<>();
     }
-
-    @Override
-    public String toString() {
-        return "K8sTaskParameters{"
-                + "image='" + image + '\''
-                + ", namespace='" + namespace + '\''
-                + ", minCpuCores=" + minCpuCores
-                + ", minMemorySpace=" + minMemorySpace
-                + '}';
-    }
 }

dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/test/java/org/apache/dolphinscheduler/plugin/task/api/k8s/K8sTaskExecutorTest.java

@@ -62,6 +62,7 @@ public class K8sTaskExecutorTest {
         k8sTaskMainParameters.setClusterName(clusterName);
         k8sTaskMainParameters.setMinCpuCores(minCpuCores);
         k8sTaskMainParameters.setMinMemorySpace(minMemorySpace);
+        k8sTaskMainParameters.setCommand("echo 'hello world'");
         job = k8sTaskExecutor.buildK8sJob(k8sTaskMainParameters);
     }
     @Test

dolphinscheduler-task-plugin/dolphinscheduler-task-k8s/src/main/java/org/apache/dolphinscheduler/plugin/task/k8s/K8sTask.java

@@ -81,6 +81,7 @@ public class K8sTask extends AbstractK8sTask {
         k8sTaskMainParameters.setMinCpuCores(k8sTaskParameters.getMinCpuCores());
         k8sTaskMainParameters.setMinMemorySpace(k8sTaskParameters.getMinMemorySpace());
         k8sTaskMainParameters.setParamsMap(ParamUtils.convert(paramsMap));
+        k8sTaskMainParameters.setCommand(k8sTaskParameters.getCommand());
         return JSONUtils.toJsonString(k8sTaskMainParameters);
     }
 

dolphinscheduler-task-plugin/dolphinscheduler-task-k8s/src/test/java/org/apache/dolphinscheduler/plugin/task/k8s/K8sParametersTest.java

@@ -30,6 +30,7 @@ public class K8sParametersTest {
     private final String namespace = "{\"name\":\"default\",\"cluster\":\"lab\"}";
     private final double minCpuCores = 2;
     private final double minMemorySpace = 10;
+    private final String command = "echo 'hello world'";
 
     @BeforeEach
     public void before() {
@@ -38,6 +39,7 @@ public class K8sParametersTest {
         k8sTaskParameters.setNamespace(namespace);
         k8sTaskParameters.setMinCpuCores(minCpuCores);
         k8sTaskParameters.setMinMemorySpace(minMemorySpace);
+        k8sTaskParameters.setCommand(command);
     }
 
     @Test
@@ -57,6 +59,7 @@ public class K8sParametersTest {
         Assertions.assertEquals(namespace, k8sTaskParameters.getNamespace());
         Assertions.assertEquals(0, Double.compare(minCpuCores, k8sTaskParameters.getMinCpuCores()));
         Assertions.assertEquals(0, Double.compare(minMemorySpace, k8sTaskParameters.getMinMemorySpace()));
+        Assertions.assertEquals(command, k8sTaskParameters.getCommand());
     }
 
 }

dolphinscheduler-task-plugin/dolphinscheduler-task-k8s/src/test/java/org/apache/dolphinscheduler/plugin/task/k8s/K8sTaskTest.java

@@ -48,6 +48,7 @@ public class K8sTaskTest {
 
     private final String DAY = "day";
     private final String date = "20220507";
+    private final String command = "echo 'hello world'";
     @BeforeEach
     public void before() {
         k8sTaskParameters = new K8sTaskParameters();
@@ -55,6 +56,7 @@ public class K8sTaskTest {
         k8sTaskParameters.setNamespace(namespace);
         k8sTaskParameters.setMinCpuCores(minCpuCores);
         k8sTaskParameters.setMinMemorySpace(minMemorySpace);
+        k8sTaskParameters.setCommand(command);
         TaskExecutionContext taskRequest = new TaskExecutionContext();
         taskRequest.setTaskInstanceId(taskInstanceId);
         taskRequest.setTaskName(taskName);
@@ -80,7 +82,7 @@ public class K8sTaskTest {
     @Test
     public void testBuildCommandNormal() {
         String expectedStr =
-                "{\"image\":\"ds-dev\",\"namespaceName\":\"default\",\"clusterName\":\"lab\",\"minCpuCores\":2.0,\"minMemorySpace\":10.0,\"paramsMap\":{\"day\":\"20220507\"}}";
+                "{\"image\":\"ds-dev\",\"command\":\"echo 'hello world'\",\"namespaceName\":\"default\",\"clusterName\":\"lab\",\"minCpuCores\":2.0,\"minMemorySpace\":10.0,\"paramsMap\":{\"day\":\"20220507\"}}";
         String commandStr = k8sTask.buildCommand();
         Assertions.assertEquals(expectedStr, commandStr);
     }
@@ -88,7 +90,7 @@ public class K8sTaskTest {
     @Test
     public void testGetParametersNormal() {
         String expectedStr =
-                "K8sTaskParameters{image='ds-dev', namespace='{\"name\":\"default\",\"cluster\":\"lab\"}', minCpuCores=2.0, minMemorySpace=10.0}";
+                "K8sTaskParameters(image=ds-dev, namespace={\"name\":\"default\",\"cluster\":\"lab\"}, command=echo 'hello world', minCpuCores=2.0, minMemorySpace=10.0)";
         String result = k8sTask.getParameters().toString();
         Assertions.assertEquals(expectedStr, result);
     }

dolphinscheduler-ui/src/locales/en_US/project.ts

@@ -367,6 +367,8 @@ export default {
     mb: 'MB',
     image: 'Image',
     image_tips: 'Please enter image',
+    command: 'Command',
+    command_tips: 'Please enter the container execution command, for example: /bin/echo hello world',
     min_memory_tips: 'Please enter min memory',
     state: 'State',
     branch_flow: 'Branch flow',

dolphinscheduler-ui/src/locales/zh_CN/project.ts

@@ -367,6 +367,8 @@ export default {
     mb: 'MB',
     image: '镜像',
     image_tips: '请输入镜像',
+    command: '容器执行命令',
+    command_tips: '请输入容器执行命令，例如：/bin/echo hello world',
     min_memory_tips: '请输入最小内存',
     state: '状态',
     branch_flow: '分支流转',

dolphinscheduler-ui/src/views/projects/task/components/node/fields/use-k8s.ts

@@ -59,6 +59,14 @@ export function useK8s(model: { [field: string]: any }): IJsonItem[] {
         message: t('project.node.min_memory_tips')
       }
     },
+    {
+      type: 'input',
+      field: 'command',
+      name: t('project.node.command'),
+      props: {
+        placeholder: t('project.node.command_tips')
+      }
+    },
     ...useCustomParams({ model, field: 'localParams', isSimple: true })
   ]
 }

dolphinscheduler-ui/src/views/projects/task/components/node/format-data.ts

@@ -347,6 +347,7 @@ export function formatParams(data: INodeData): {
     taskParams.minCpuCores = data.minCpuCores
     taskParams.minMemorySpace = data.minMemorySpace
     taskParams.image = data.image
+    taskParams.command = data.command
   }
 
   if (data.taskType === 'JUPYTER') {

dolphinscheduler-ui/src/views/projects/task/components/node/types.ts

@@ -337,6 +337,7 @@ interface ITaskParams {
   minCpuCores?: string
   minMemorySpace?: string
   image?: string
+  command?: string
   algorithm?: string
   params?: string
   searchParams?: string