[Improvement-13868][K8S Task] Support the custom labels for k8s task (#14067)

* [Improvement-13868] Support the custom labels for k8s task

* fix front-end type mismatched

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

@@ -24,6 +24,7 @@ K8S task type used to execute a batch task. In this task, the worker submits the
 | Image            | The registry url for image.                                                                                      |
 | Command          | The container execution command (yaml-style array), for example: ["printenv"]                                    |
 | Args             | The args of execution command (yaml-style array), for example: ["HOSTNAME", "KUBERNETES_PORT"]                   |
+| Custom label     | The customized labels for k8s Job.                                                                               |
 | 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

@@ -24,6 +24,7 @@ kubernetes任务类型，用于在kubernetes上执行一个短时和批处理的
 | 镜像       | 镜像地址                                                            |
 | 容器执行命令   | 容器执行命令（yaml格式数组），例如：["printenv"]                                |
 | 执行命令参数   | 执行命令参数（yaml格式数组），例如：["HOSTNAME", "KUBERNETES_PORT"]             |
+| 自定义标签    | 作业自定义标签                                                         |
 | 自定义参数    | kubernetes任务局部的用户自定义参数，自定义参数最终会通过环境变量形式存在于容器中，提供给kubernetes任务使用 |
 
 ## 任务样例

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

@@ -35,4 +35,5 @@ public class K8sTaskMainParameters {
     private double minCpuCores;
     private double minMemorySpace;
     private Map<String, String> paramsMap;
+    private Map<String, String> labelMap;
 }

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

@@ -94,7 +94,7 @@ public class K8sTaskExecutor extends AbstractK8sTaskExecutor {
         Map<String, Quantity> limitRes = new HashMap<>();
         limitRes.put(MEMORY, new Quantity(String.format("%s%s", limitPodMem, MI)));
         limitRes.put(CPU, new Quantity(String.valueOf(limitPodCpu)));
-        Map<String, String> labelMap = new HashMap<>();
+        Map<String, String> labelMap = k8STaskMainParameters.getLabelMap();
         labelMap.put(LAYER_LABEL, LAYER_LABEL_VALUE);
         labelMap.put(NAME_LABEL, k8sJobName);
         EnvVar taskInstanceIdVar = new EnvVar(TASK_INSTANCE_ID, taskInstanceId, null);

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

@@ -0,0 +1,40 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.dolphinscheduler.plugin.task.api.model;
+
+import java.io.Serializable;
+
+import lombok.AllArgsConstructor;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+@Data
+@NoArgsConstructor
+@AllArgsConstructor
+public class Label implements Serializable {
+
+    /**
+     * label name
+     */
+    private String label;
+
+    /**
+     * label value
+     */
+    private String value;
+}

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

@@ -17,6 +17,7 @@
 
 package org.apache.dolphinscheduler.plugin.task.api.parameters;
 
+import org.apache.dolphinscheduler.plugin.task.api.model.Label;
 import org.apache.dolphinscheduler.plugin.task.api.model.ResourceInfo;
 
 import org.apache.commons.lang3.StringUtils;
@@ -35,6 +36,7 @@ public class K8sTaskParameters extends AbstractParameters {
     private String image;
     private String namespace;
     private String command;
+    private List<Label> customizedLabels;
     private String args;
     private double minCpuCores;
     private double minMemorySpace;

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

@@ -27,6 +27,7 @@ import org.apache.dolphinscheduler.plugin.task.api.TaskExecutionContext;
 import org.apache.dolphinscheduler.plugin.task.api.k8s.impl.K8sTaskExecutor;
 import org.apache.dolphinscheduler.plugin.task.api.model.TaskResponse;
 
+import java.util.HashMap;
 import java.util.Map;
 
 import org.junit.jupiter.api.Assertions;
@@ -55,6 +56,8 @@ public class K8sTaskExecutorTest {
         Map<String, String> namespace = JSONUtils.toMap(this.namespace);
         String namespaceName = namespace.get(NAMESPACE_NAME);
         String clusterName = namespace.get(CLUSTER);
+        Map<String, String> labelMap = new HashMap<>();
+        labelMap.put("test", "1234");
         k8sTaskExecutor = new K8sTaskExecutor(null, taskRequest);
         k8sTaskMainParameters = new K8sTaskMainParameters();
         k8sTaskMainParameters.setImage(image);
@@ -63,6 +66,7 @@ public class K8sTaskExecutorTest {
         k8sTaskMainParameters.setMinCpuCores(minCpuCores);
         k8sTaskMainParameters.setMinMemorySpace(minMemorySpace);
         k8sTaskMainParameters.setCommand("[\"perl\" ,\"-Mbignum=bpi\", \"-wle\", \"print bpi(2000)\"]");
+        k8sTaskMainParameters.setLabelMap(labelMap);
         job = k8sTaskExecutor.buildK8sJob(k8sTaskMainParameters);
     }
     @Test

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

@@ -25,12 +25,16 @@ import org.apache.dolphinscheduler.plugin.task.api.TaskException;
 import org.apache.dolphinscheduler.plugin.task.api.TaskExecutionContext;
 import org.apache.dolphinscheduler.plugin.task.api.k8s.AbstractK8sTask;
 import org.apache.dolphinscheduler.plugin.task.api.k8s.K8sTaskMainParameters;
+import org.apache.dolphinscheduler.plugin.task.api.model.Label;
 import org.apache.dolphinscheduler.plugin.task.api.model.Property;
 import org.apache.dolphinscheduler.plugin.task.api.parameters.AbstractParameters;
 import org.apache.dolphinscheduler.plugin.task.api.parameters.K8sTaskParameters;
 import org.apache.dolphinscheduler.plugin.task.api.parser.ParamUtils;
 
+import org.apache.commons.collections4.CollectionUtils;
+
 import java.util.Collections;
+import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 
@@ -82,9 +86,22 @@ public class K8sTask extends AbstractK8sTask {
         k8sTaskMainParameters.setMinCpuCores(k8sTaskParameters.getMinCpuCores());
         k8sTaskMainParameters.setMinMemorySpace(k8sTaskParameters.getMinMemorySpace());
         k8sTaskMainParameters.setParamsMap(ParamUtils.convert(paramsMap));
+        k8sTaskMainParameters.setLabelMap(convertToLabelMap(k8sTaskParameters.getCustomizedLabels()));
         k8sTaskMainParameters.setCommand(k8sTaskParameters.getCommand());
         k8sTaskMainParameters.setArgs(k8sTaskParameters.getArgs());
         return JSONUtils.toJsonString(k8sTaskMainParameters);
     }
 
+    public Map<String, String> convertToLabelMap(List<Label> customizedLabels) {
+        if (CollectionUtils.isEmpty(customizedLabels)) {
+            return Collections.emptyMap();
+        }
+
+        HashMap<String, String> labelMap = new HashMap<>();
+        customizedLabels.forEach(label -> {
+            labelMap.put(label.getLabel(), label.getValue());
+        });
+        return labelMap;
+    }
+
 }

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

@@ -17,8 +17,12 @@
 
 package org.apache.dolphinscheduler.plugin.task.k8s;
 
+import org.apache.dolphinscheduler.plugin.task.api.model.Label;
 import org.apache.dolphinscheduler.plugin.task.api.parameters.K8sTaskParameters;
 
+import java.util.Arrays;
+import java.util.List;
+
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
@@ -32,6 +36,7 @@ public class K8sParametersTest {
     private final double minMemorySpace = 10;
     private final String command = "[\"/bin/bash\", \"-c\"]";
     private final String args = "[\"echo hello world\"]";
+    private final List<Label> labels = Arrays.asList(new Label("test", "1234"));
 
     @BeforeEach
     public void before() {
@@ -42,6 +47,7 @@ public class K8sParametersTest {
         k8sTaskParameters.setMinMemorySpace(minMemorySpace);
         k8sTaskParameters.setCommand(command);
         k8sTaskParameters.setArgs(args);
+        k8sTaskParameters.setCustomizedLabels(labels);
     }
 
     @Test
@@ -63,6 +69,7 @@ public class K8sParametersTest {
         Assertions.assertEquals(0, Double.compare(minMemorySpace, k8sTaskParameters.getMinMemorySpace()));
         Assertions.assertEquals(command, k8sTaskParameters.getCommand());
         Assertions.assertEquals(args, k8sTaskParameters.getArgs());
+        Assertions.assertEquals(labels, k8sTaskParameters.getCustomizedLabels());
     }
 
 }

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

@@ -21,10 +21,13 @@ import org.apache.dolphinscheduler.common.utils.JSONUtils;
 import org.apache.dolphinscheduler.plugin.task.api.TaskExecutionContext;
 import org.apache.dolphinscheduler.plugin.task.api.enums.DataType;
 import org.apache.dolphinscheduler.plugin.task.api.enums.Direct;
+import org.apache.dolphinscheduler.plugin.task.api.model.Label;
 import org.apache.dolphinscheduler.plugin.task.api.model.Property;
 import org.apache.dolphinscheduler.plugin.task.api.parameters.K8sTaskParameters;
 
+import java.util.Arrays;
 import java.util.HashMap;
+import java.util.List;
 import java.util.Map;
 
 import org.junit.jupiter.api.Assertions;
@@ -50,6 +53,8 @@ public class K8sTaskTest {
     private final String date = "20220507";
     private final String command = "[\"/bin/bash\", \"-c\"]";
     private final String args = "[\"echo hello world\"]";
+    private final List<Label> labels = Arrays.asList(new Label("test", "1234"));
+
     @BeforeEach
     public void before() {
         k8sTaskParameters = new K8sTaskParameters();
@@ -59,6 +64,7 @@ public class K8sTaskTest {
         k8sTaskParameters.setMinMemorySpace(minMemorySpace);
         k8sTaskParameters.setCommand(command);
         k8sTaskParameters.setArgs(args);
+        k8sTaskParameters.setCustomizedLabels(labels);
         TaskExecutionContext taskRequest = new TaskExecutionContext();
         taskRequest.setTaskInstanceId(taskInstanceId);
         taskRequest.setTaskName(taskName);
@@ -84,7 +90,7 @@ public class K8sTaskTest {
     @Test
     public void testBuildCommandNormal() {
         String expectedStr =
-                "{\"image\":\"ds-dev\",\"command\":\"[\\\"/bin/bash\\\", \\\"-c\\\"]\",\"args\":\"[\\\"echo hello world\\\"]\",\"namespaceName\":\"default\",\"clusterName\":\"lab\",\"minCpuCores\":2.0,\"minMemorySpace\":10.0,\"paramsMap\":{\"day\":\"20220507\"}}";
+                "{\"image\":\"ds-dev\",\"command\":\"[\\\"/bin/bash\\\", \\\"-c\\\"]\",\"args\":\"[\\\"echo hello world\\\"]\",\"namespaceName\":\"default\",\"clusterName\":\"lab\",\"minCpuCores\":2.0,\"minMemorySpace\":10.0,\"paramsMap\":{\"day\":\"20220507\"},\"labelMap\":{\"test\":\"1234\"}}";
         String commandStr = k8sTask.buildCommand();
         Assertions.assertEquals(expectedStr, commandStr);
     }
@@ -92,7 +98,7 @@ public class K8sTaskTest {
     @Test
     public void testGetParametersNormal() {
         String expectedStr =
-                "K8sTaskParameters(image=ds-dev, namespace={\"name\":\"default\",\"cluster\":\"lab\"}, command=[\"/bin/bash\", \"-c\"], args=[\"echo hello world\"], minCpuCores=2.0, minMemorySpace=10.0)";
+                "K8sTaskParameters(image=ds-dev, namespace={\"name\":\"default\",\"cluster\":\"lab\"}, command=[\"/bin/bash\", \"-c\"], customizedLabels=[Label(label=test, value=1234)], args=[\"echo hello world\"], minCpuCores=2.0, minMemorySpace=10.0)";
         String result = k8sTask.getParameters().toString();
         Assertions.assertEquals(expectedStr, result);
     }

dolphinscheduler-ui/src/components/form/fields/get-field.ts

@@ -23,6 +23,7 @@ const TYPES = [
   'radio',
   'editor',
   'custom-parameters',
+  'custom-labels',
   'switch',
   'input-number',
   'select',
@@ -45,7 +46,7 @@ const getField = (
     return widget || null
   }
   // TODO Support other widgets later
-  if (type === 'custom-parameters') {
+  if (type === 'custom-parameters' || type === 'custom-labels') {
     let fieldRules: { [key: string]: IFormItemRule }[] = []
     if (rules && !rules[field]) fieldRules = rules[field] = []
     // @ts-ignore

dolphinscheduler-ui/src/components/form/types.ts

@@ -22,6 +22,7 @@ type IType =
   | 'radio'
   | 'editor'
   | 'custom-parameters'
+  | 'custom-labels'
   | 'switch'
   | 'input-number'
   | 'select'

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

@@ -419,6 +419,10 @@ export default {
     prop_repeat: 'prop is repeat',
     value_tips: 'value(optional)',
     value_required_tips: 'value(required)',
+    custom_labels: 'Customized labels',
+    label_repeat: 'repeated label',
+    label_name_tips: 'label name(required)',
+    label_value_tips: 'label value(required)',
     pre_tasks: 'Pre tasks',
     program_type: 'Program Type',
     main_class: 'Main Class',

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

@@ -413,6 +413,10 @@ export default {
     prop_repeat: 'prop中有重复',
     value_tips: 'value(选填)',
     value_required_tips: 'value(必填)',
+    custom_labels: '自定义标签',
+    label_repeat: 'label中有重复',
+    label_name_tips: 'label name(必填)',
+    label_value_tips: 'label value(必填)',
     pre_tasks: '前置任务',
     program_type: '程序类型',
     main_class: '主函数的Class',

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

@@ -36,6 +36,7 @@ export { useDatasource } from './use-datasource'
 export { useSqlType } from './use-sql-type'
 export { useProcedure } from './use-procedure'
 export { useCustomParams } from './use-custom-params'
+export { useCustomLabels } from './use-custom-labels'
 export { useSourceType } from './use-sqoop-source-type'
 export { useTargetType } from './use-sqoop-target-type'
 export { useRelationCustomParams } from './use-relation-custom-params'

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

@@ -0,0 +1,82 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+import { Ref } from 'vue'
+import { useI18n } from 'vue-i18n'
+import type { IJsonItem } from '../types'
+
+export function useCustomLabels({
+                                    model,
+                                    field,
+                                    name = 'custom_labels',
+                                    span = 24
+                                }: {
+    model: { [field: string]: any }
+    field: string
+    name?: string
+    span?: Ref | number
+}): IJsonItem[] {
+    const { t } = useI18n()
+
+    return [
+        {
+            type: 'custom-parameters',
+            field: field,
+            name: t(`project.node.${name}`),
+            class: 'btn-custom-parameters',
+            span,
+            children: [
+                {
+                    type: 'input',
+                    field: 'label',
+                    span: 8,
+                    class: 'customized-label-name',
+                    props: {
+                        placeholder: t('project.node.label_name_tips'),
+                        maxLength: 256
+                    },
+                    validate: {
+                        trigger: ['input', 'blur'],
+                        required: true,
+                        validator(validate: any, value: string) {
+                            if (!value) {
+                                return new Error(t('project.node.label_name_tips'))
+                            }
+
+                            const sameItems = model[field].filter(
+                                (item: { label: string }) => item.label === value
+                            )
+
+                            if (sameItems.length > 1) {
+                                return new Error(t('project.node.label_repeat'))
+                            }
+                        }
+                    }
+                },
+                {
+                    type: 'input',
+                    field: 'value',
+                    span: 14,
+                    class: 'customized-label-value',
+                    props: {
+                        placeholder: t('project.node.label_value_tips'),
+                        maxLength: 256
+                    }
+                }
+            ]
+        }
+    ]
+}

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

@@ -14,7 +14,7 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-import { useCustomParams, useNamespace } from '.'
+import { useCustomParams, useNamespace, useCustomLabels } from '.'
 import type { IJsonItem } from '../types'
 import { useI18n } from 'vue-i18n'
 
@@ -75,6 +75,7 @@ export function useK8s(model: { [field: string]: any }): IJsonItem[] {
         placeholder: t('project.node.args_tips')
       }
     },
+    ...useCustomLabels({ model, field: 'customizedLabels' }),
     ...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.image = data.image
     taskParams.command = data.command
     taskParams.args = data.args
+    taskParams.customizedLabels = data.customizedLabels
   }
 
   if (data.taskType === 'JUPYTER') {

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

@@ -38,6 +38,7 @@ export function useK8s({
     description: '',
     timeoutFlag: false,
     localParams: [],
+    customizedLabels: [],
     environmentCode: null,
     failRetryInterval: 1,
     failRetryTimes: 0,

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

@@ -66,6 +66,11 @@ interface ILocalParam {
   value?: string
 }
 
+interface ICustomLabel {
+  label: string
+  value: string
+}
+
 interface IResponseJsonItem extends Omit<IJsonItemParams, 'type'> {
   type: 'input' | 'select' | 'radio' | 'group'
   emit: 'change'[]
@@ -355,6 +360,7 @@ interface ITaskParams {
   image?: string
   command?: string
   args?: string
+  customizedLabels?: ICustomLabel[]
   algorithm?: string
   params?: string
   searchParams?: string