You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@dolphinscheduler.apache.org by we...@apache.org on 2023/12/14 13:34:49 UTC
(dolphinscheduler) branch dev updated: Fix k8sTaskExecutionContext setting configYaml (#15116)
This is an automated email from the ASF dual-hosted git repository.
weiraowang pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/dolphinscheduler.git
The following commit(s) were added to refs/heads/dev by this push:
new ce11674668 Fix k8sTaskExecutionContext setting configYaml (#15116)
ce11674668 is described below
commit ce11674668cf26747f34c3474763c32e1ba5b6d1
Author: chenrj <10...@users.noreply.github.com>
AuthorDate: Thu Dec 14 21:34:42 2023 +0800
Fix k8sTaskExecutionContext setting configYaml (#15116)
* fixed the issue of obtaining kubeConfig of k8s tasks
* removed excess files
* removed excess files
* formatted the code
---------
Co-authored-by: xiangzihao <46...@qq.com>
Co-authored-by: Aaron Wang <wa...@gmail.com>
---
.../plugin/task/api/am/KubernetesApplicationManager.java | 3 ---
.../plugin/task/api/k8s/impl/K8sTaskExecutor.java | 4 +---
.../plugin/task/api/parameters/AbstractParameters.java | 13 +++++++++++++
.../plugin/task/api/parameters/K8sTaskParameters.java | 13 ++-----------
.../apache/dolphinscheduler/plugin/task/k8s/K8sTask.java | 6 ++++--
5 files changed, 20 insertions(+), 19 deletions(-)
diff --git a/dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/main/java/org/apache/dolphinscheduler/plugin/task/api/am/KubernetesApplicationManager.java b/dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/main/java/org/apache/dolphinscheduler/plugin/task/api/am/KubernetesApplicationManager.java
index a18637a4ff..ac1ce69f76 100644
--- a/dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/main/java/org/apache/dolphinscheduler/plugin/task/api/am/KubernetesApplicationManager.java
+++ b/dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/main/java/org/apache/dolphinscheduler/plugin/task/api/am/KubernetesApplicationManager.java
@@ -22,7 +22,6 @@ import static org.apache.dolphinscheduler.plugin.task.api.TaskConstants.UNIQUE_L
import org.apache.dolphinscheduler.common.enums.ResourceManagerType;
import org.apache.dolphinscheduler.common.thread.ThreadUtils;
-import org.apache.dolphinscheduler.common.utils.JSONUtils;
import org.apache.dolphinscheduler.plugin.task.api.K8sTaskExecutionContext;
import org.apache.dolphinscheduler.plugin.task.api.TaskException;
import org.apache.dolphinscheduler.plugin.task.api.enums.TaskExecutionStatus;
@@ -132,8 +131,6 @@ public class KubernetesApplicationManager implements ApplicationManager {
private KubernetesClient getClient(KubernetesApplicationManagerContext kubernetesApplicationManagerContext) {
K8sTaskExecutionContext k8sTaskExecutionContext =
kubernetesApplicationManagerContext.getK8sTaskExecutionContext();
- k8sTaskExecutionContext
- .setConfigYaml(JSONUtils.getNodeString(k8sTaskExecutionContext.getConnectionParams(), "kubeConfig"));
return cacheClientMap.computeIfAbsent(kubernetesApplicationManagerContext.getLabelValue(),
key -> new KubernetesClientBuilder()
.withConfig(Config.fromKubeconfig(k8sTaskExecutionContext.getConfigYaml())).build());
diff --git a/dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/main/java/org/apache/dolphinscheduler/plugin/task/api/k8s/impl/K8sTaskExecutor.java b/dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/main/java/org/apache/dolphinscheduler/plugin/task/api/k8s/impl/K8sTaskExecutor.java
index 476ed85be9..f5f794859b 100644
--- a/dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/main/java/org/apache/dolphinscheduler/plugin/task/api/k8s/impl/K8sTaskExecutor.java
+++ b/dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/main/java/org/apache/dolphinscheduler/plugin/task/api/k8s/impl/K8sTaskExecutor.java
@@ -293,9 +293,7 @@ public class K8sTaskExecutor extends AbstractK8sTaskExecutor {
return result;
}
K8sTaskExecutionContext k8sTaskExecutionContext = taskRequest.getK8sTaskExecutionContext();
- String connectionParams = k8sTaskExecutionContext.getConnectionParams();
- String kubeConfig = JSONUtils.getNodeString(connectionParams, "kubeConfig");
- String configYaml = kubeConfig;
+ String configYaml = k8sTaskExecutionContext.getConfigYaml();
k8sUtils.buildClient(configYaml);
submitJob2k8s(k8sParameterStr);
parsePodLogOutput();
diff --git a/dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/main/java/org/apache/dolphinscheduler/plugin/task/api/parameters/AbstractParameters.java b/dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/main/java/org/apache/dolphinscheduler/plugin/task/api/parameters/AbstractParameters.java
index a57eececf5..6ca1be7d7a 100644
--- a/dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/main/java/org/apache/dolphinscheduler/plugin/task/api/parameters/AbstractParameters.java
+++ b/dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/main/java/org/apache/dolphinscheduler/plugin/task/api/parameters/AbstractParameters.java
@@ -18,9 +18,12 @@
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.Direct;
+import org.apache.dolphinscheduler.plugin.task.api.enums.ResourceType;
import org.apache.dolphinscheduler.plugin.task.api.model.Property;
import org.apache.dolphinscheduler.plugin.task.api.model.ResourceInfo;
+import org.apache.dolphinscheduler.plugin.task.api.parameters.resource.DataSourceParameters;
import org.apache.dolphinscheduler.plugin.task.api.parameters.resource.ResourceParametersHelper;
import org.apache.commons.collections4.CollectionUtils;
@@ -86,6 +89,16 @@ public abstract class AbstractParameters implements IParameters {
return localParametersMaps;
}
+ public K8sTaskExecutionContext generateK8sTaskExecutionContext(ResourceParametersHelper parametersHelper,
+ int datasource) {
+ DataSourceParameters dataSourceParameters =
+ (DataSourceParameters) parametersHelper.getResourceParameters(ResourceType.DATASOURCE, datasource);
+ K8sTaskExecutionContext k8sTaskExecutionContext = new K8sTaskExecutionContext();
+ k8sTaskExecutionContext.setConnectionParams(
+ Objects.nonNull(dataSourceParameters) ? dataSourceParameters.getConnectionParams() : null);
+ return k8sTaskExecutionContext;
+ }
+
/**
* get input local parameters map if the param direct is IN
* @return parameters map
diff --git a/dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/main/java/org/apache/dolphinscheduler/plugin/task/api/parameters/K8sTaskParameters.java b/dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/main/java/org/apache/dolphinscheduler/plugin/task/api/parameters/K8sTaskParameters.java
index d3d5e4963b..4f045abe19 100644
--- a/dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/main/java/org/apache/dolphinscheduler/plugin/task/api/parameters/K8sTaskParameters.java
+++ b/dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/main/java/org/apache/dolphinscheduler/plugin/task/api/parameters/K8sTaskParameters.java
@@ -17,19 +17,16 @@
package org.apache.dolphinscheduler.plugin.task.api.parameters;
-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.model.ResourceInfo;
-import org.apache.dolphinscheduler.plugin.task.api.parameters.resource.DataSourceParameters;
import org.apache.dolphinscheduler.plugin.task.api.parameters.resource.ResourceParametersHelper;
import org.apache.commons.lang3.StringUtils;
import java.util.ArrayList;
import java.util.List;
-import java.util.Objects;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
@@ -58,18 +55,12 @@ public class K8sTaskParameters extends AbstractParameters {
public boolean checkParameters() {
return StringUtils.isNotEmpty(image);
}
- public K8sTaskExecutionContext generateExtendedContext(ResourceParametersHelper parametersHelper) {
- DataSourceParameters dataSourceParameters =
- (DataSourceParameters) parametersHelper.getResourceParameters(ResourceType.DATASOURCE, datasource);
- K8sTaskExecutionContext k8sTaskExecutionContext = new K8sTaskExecutionContext();
- k8sTaskExecutionContext.setConnectionParams(
- Objects.nonNull(dataSourceParameters) ? dataSourceParameters.getConnectionParams() : null);
- return k8sTaskExecutionContext;
- }
+
@Override
public List<ResourceInfo> getResourceFilesList() {
return new ArrayList<>();
}
+
@Override
public ResourceParametersHelper getResources() {
ResourceParametersHelper resources = super.getResources();
diff --git a/dolphinscheduler-task-plugin/dolphinscheduler-task-k8s/src/main/java/org/apache/dolphinscheduler/plugin/task/k8s/K8sTask.java b/dolphinscheduler-task-plugin/dolphinscheduler-task-k8s/src/main/java/org/apache/dolphinscheduler/plugin/task/k8s/K8sTask.java
index fceb29e163..fdb39d7c28 100644
--- a/dolphinscheduler-task-plugin/dolphinscheduler-task-k8s/src/main/java/org/apache/dolphinscheduler/plugin/task/k8s/K8sTask.java
+++ b/dolphinscheduler-task-plugin/dolphinscheduler-task-k8s/src/main/java/org/apache/dolphinscheduler/plugin/task/k8s/K8sTask.java
@@ -70,14 +70,16 @@ public class K8sTask extends AbstractK8sTask {
}
k8sTaskExecutionContext =
- k8sTaskParameters.generateExtendedContext(taskExecutionContext.getResourceParametersHelper());
- taskRequest.setK8sTaskExecutionContext(k8sTaskExecutionContext);
+ k8sTaskParameters.generateK8sTaskExecutionContext(taskExecutionContext.getResourceParametersHelper(),
+ k8sTaskParameters.getDatasource());
k8sConnectionParam =
(K8sConnectionParam) DataSourceUtils.buildConnectionParams(DbType.valueOf(k8sTaskParameters.getType()),
k8sTaskExecutionContext.getConnectionParams());
String kubeConfig = k8sConnectionParam.getKubeConfig();
k8sTaskParameters.setNamespace(k8sConnectionParam.getNamespace());
k8sTaskParameters.setKubeConfig(kubeConfig);
+ k8sTaskExecutionContext.setConfigYaml(kubeConfig);
+ taskRequest.setK8sTaskExecutionContext(k8sTaskExecutionContext);
log.info("Initialize k8s task params:{}", JSONUtils.toPrettyJsonString(k8sTaskParameters));
}