[Fix-17906] Fix can't get pod's log in k8s task (#18075)

Author: SbloodyS

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.common.utils.JSONUtils;
 import org.apache.dolphinscheduler.plugin.task.api.K8sTaskExecutionContext;
 import org.apache.dolphinscheduler.plugin.task.api.enums.ResourceType;
 import org.apache.dolphinscheduler.plugin.task.api.model.Label;
@@ -31,8 +32,10 @@
 import java.util.List;
 
 import lombok.Data;
+import lombok.EqualsAndHashCode;
 import lombok.extern.slf4j.Slf4j;
 
+@EqualsAndHashCode(callSuper = true)
 @Data
 @Slf4j
 public class K8sTaskParameters extends AbstractParameters {
@@ -79,7 +82,18 @@ public K8sTaskExecutionContext generateK8sTaskExecutionContext(
             connectionParams = dataSourceParameters.getConnectionParams();
         }
 
+        String configYaml = null;
+        String namespace = null;
+        if (StringUtils.isNotEmpty(connectionParams) && JSONUtils.checkJsonValid(connectionParams, false)) {
+            K8sTaskParameters connectionTaskParameters =
+                    JSONUtils.parseObject(connectionParams, K8sTaskParameters.class);
+            configYaml = connectionTaskParameters.getKubeConfig();
+            namespace = connectionTaskParameters.getNamespace();
+        }
+
         return K8sTaskExecutionContext.builder()
+                .configYaml(configYaml)
+                .namespace(namespace)
                 .connectionParams(connectionParams)
                 .build();
     }

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

@@ -19,38 +19,20 @@
 
 import org.apache.dolphinscheduler.spi.enums.DbType;
 
+import lombok.Data;
+import lombok.EqualsAndHashCode;
+
 import com.fasterxml.jackson.annotation.JsonProperty;
 
+@EqualsAndHashCode(callSuper = true)
+@Data
 public class DataSourceParameters extends AbstractResourceParameters {
 
     private DbType type;
 
-    public String getResourceType() {
-        return resourceType;
-    }
-
-    public void setResourceType(String resourceType) {
-        this.resourceType = resourceType;
-    }
-
     @JsonProperty(value = "DATASOURCE")
     private String resourceType;
 
     private String connectionParams;
 
-    public DbType getType() {
-        return type;
-    }
-
-    public void setType(DbType type) {
-        this.type = type;
-    }
-
-    public String getConnectionParams() {
-        return connectionParams;
-    }
-
-    public void setConnectionParams(String connectionParams) {
-        this.connectionParams = connectionParams;
-    }
 }

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

@@ -79,6 +79,7 @@ public void init() {
         k8sTaskParameters.setNamespace(k8sConnectionParam.getNamespace());
         k8sTaskParameters.setKubeConfig(kubeConfig);
         k8sTaskExecutionContext.setConfigYaml(kubeConfig);
+        k8sTaskExecutionContext.setNamespace(k8sConnectionParam.getNamespace());
         taskRequest.setK8sTaskExecutionContext(k8sTaskExecutionContext);
         log.info("Initialize k8s task params:{}", JSONUtils.toPrettyJsonString(k8sTaskParameters));
     }

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

@@ -17,9 +17,13 @@
 
 package org.apache.dolphinscheduler.plugin.task.k8s;
 
+import org.apache.dolphinscheduler.plugin.task.api.K8sTaskExecutionContext;
+import org.apache.dolphinscheduler.plugin.task.api.enums.ResourceType;
 import org.apache.dolphinscheduler.plugin.task.api.model.Label;
 import org.apache.dolphinscheduler.plugin.task.api.model.NodeSelectorExpression;
 import org.apache.dolphinscheduler.plugin.task.api.parameters.K8sTaskParameters;
+import org.apache.dolphinscheduler.plugin.task.api.parameters.resource.DataSourceParameters;
+import org.apache.dolphinscheduler.plugin.task.api.parameters.resource.ResourceParametersHelper;
 
 import java.util.Arrays;
 import java.util.List;
@@ -80,4 +84,21 @@ public void testK8sParameters() {
         Assertions.assertEquals(nodeSelectorExpressions, k8sTaskParameters.getNodeSelectors());
     }
 
+    @Test
+    public void testGenerateK8sTaskExecutionContextShouldContainDatasourceNamespace() {
+        String connectionParams = "{\"kubeConfig\":\"{}\",\"namespace\":\"namespace\"}";
+        DataSourceParameters dataSourceParameters = new DataSourceParameters();
+        dataSourceParameters.setConnectionParams(connectionParams);
+
+        ResourceParametersHelper resourceParametersHelper = new ResourceParametersHelper();
+        resourceParametersHelper.put(ResourceType.DATASOURCE, 1, dataSourceParameters);
+
+        K8sTaskExecutionContext executionContext =
+                k8sTaskParameters.generateK8sTaskExecutionContext(resourceParametersHelper, 1);
+
+        Assertions.assertEquals(connectionParams, executionContext.getConnectionParams());
+        Assertions.assertEquals("{}", executionContext.getConfigYaml());
+        Assertions.assertEquals(namespace, executionContext.getNamespace());
+    }
+
 }