[Improvement] Refactoring K8S task plugin with connections managed in…

… connection center (apache#14977)

dolphinscheduler-datasource-plugin/dolphinscheduler-datasource-all/pom.xml

@@ -148,5 +148,10 @@
             <artifactId>dolphinscheduler-datasource-sagemaker</artifactId>
             <version>${project.version}</version>
         </dependency>
+        <dependency>
+            <groupId>org.apache.dolphinscheduler</groupId>
+            <artifactId>dolphinscheduler-datasource-k8s</artifactId>
+            <version>${project.version}</version>
+        </dependency>
     </dependencies>
 </project>

dolphinscheduler-datasource-plugin/dolphinscheduler-datasource-k8s/pom.xml

@@ -0,0 +1,50 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+  ~ 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.
+  -->
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+    <parent>
+        <groupId>org.apache.dolphinscheduler</groupId>
+        <artifactId>dolphinscheduler-datasource-plugin</artifactId>
+        <version>dev-SNAPSHOT</version>
+    </parent>
+
+    <artifactId>dolphinscheduler-datasource-k8s</artifactId>
+    <packaging>jar</packaging>
+    <name>${project.artifactId}</name>
+
+    <dependencies>
+        <dependency>
+            <groupId>org.apache.dolphinscheduler</groupId>
+            <artifactId>dolphinscheduler-spi</artifactId>
+            <scope>provided</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.dolphinscheduler</groupId>
+            <artifactId>dolphinscheduler-datasource-api</artifactId>
+            <version>${project.version}</version>
+        </dependency>
+
+        <dependency>
+            <groupId>io.fabric8</groupId>
+            <artifactId>kubernetes-client</artifactId>
+        </dependency>
+    </dependencies>
+
+</project>

dolphinscheduler-datasource-plugin/dolphinscheduler-datasource-k8s/src/main/java/org/apache/dolphinscheduler/plugin/datasource/k8s/K8sClientWrapper.java

@@ -0,0 +1,55 @@
+/*
+ * 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.datasource.k8s;
+
+import lombok.extern.slf4j.Slf4j;
+import io.fabric8.kubernetes.api.model.NamespaceList;
+import io.fabric8.kubernetes.client.Config;
+import io.fabric8.kubernetes.client.KubernetesClient;
+import io.fabric8.kubernetes.client.KubernetesClientBuilder;
+
+@Slf4j
+public class K8sClientWrapper implements AutoCloseable {
+
+    private KubernetesClient client;
+
+    public K8sClientWrapper() {
+    }
+
+    public boolean checkConnect(String kubeConfigYaml, String namespace) {
+        try {
+            Config config = Config.fromKubeconfig(kubeConfigYaml);
+            client = new KubernetesClientBuilder().withConfig(config).build();
+            NamespaceList namespaceList = client.namespaces().list();
+            if (!namespaceList.getItems().stream().anyMatch(ns -> ns.getMetadata().getName().equals(namespace))) {
+                log.info("failed to connect to the K8S cluster, namespace not found\n");
+                return false;
+            }
+            log.info("successfully connected to the K8S cluster");
+            return true;
+        } catch (Exception e) {
+            log.info("failed to connect to the K8S cluster\n");
+            return false;
+        }
+    }
+
+    @Override
+    public void close() throws Exception {
+
+    }
+}

dolphinscheduler-datasource-plugin/dolphinscheduler-datasource-k8s/src/main/java/org/apache/dolphinscheduler/plugin/datasource/k8s/K8sDataSourceChannel.java

@@ -0,0 +1,37 @@
+/*
+ * 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.datasource.k8s;
+
+import org.apache.dolphinscheduler.spi.datasource.AdHocDataSourceClient;
+import org.apache.dolphinscheduler.spi.datasource.BaseConnectionParam;
+import org.apache.dolphinscheduler.spi.datasource.DataSourceChannel;
+import org.apache.dolphinscheduler.spi.datasource.PooledDataSourceClient;
+import org.apache.dolphinscheduler.spi.enums.DbType;
+
+public class K8sDataSourceChannel implements DataSourceChannel {
+
+    @Override
+    public AdHocDataSourceClient createAdHocDataSourceClient(BaseConnectionParam baseConnectionParam, DbType dbType) {
+        throw new UnsupportedOperationException("K8S AdHocDataSourceClient is not supported");
+    }
+
+    @Override
+    public PooledDataSourceClient createPooledDataSourceClient(BaseConnectionParam baseConnectionParam, DbType dbType) {
+        throw new UnsupportedOperationException("K8S AdHocDataSourceClient is not supported");
+    }
+}

dolphinscheduler-datasource-plugin/dolphinscheduler-datasource-k8s/src/main/java/org/apache/dolphinscheduler/plugin/datasource/k8s/K8sDataSourceChannelFactory.java

@@ -0,0 +1,38 @@
+/*
+ * 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.datasource.k8s;
+
+import org.apache.dolphinscheduler.spi.datasource.DataSourceChannel;
+import org.apache.dolphinscheduler.spi.datasource.DataSourceChannelFactory;
+
+import com.google.auto.service.AutoService;
+
+@AutoService(DataSourceChannelFactory.class)
+public class K8sDataSourceChannelFactory implements DataSourceChannelFactory {
+
+    @Override
+    public DataSourceChannel create() {
+        return new K8sDataSourceChannel();
+    }
+
+    @Override
+    public String getName() {
+        return "k8s";
+    }
+
+}

dolphinscheduler-datasource-plugin/dolphinscheduler-datasource-k8s/src/main/java/org/apache/dolphinscheduler/plugin/datasource/k8s/param/K8sConnectionParam.java

@@ -0,0 +1,32 @@
+/*
+ * 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.datasource.k8s.param;
+
+import org.apache.dolphinscheduler.spi.datasource.ConnectionParam;
+
+import lombok.Data;
+
+import com.fasterxml.jackson.annotation.JsonInclude;
+
+@Data
+@JsonInclude(JsonInclude.Include.NON_NULL)
+public class K8sConnectionParam implements ConnectionParam {
+
+    protected String kubeConfig;
+    protected String namespace;
+}

dolphinscheduler-datasource-plugin/dolphinscheduler-datasource-k8s/src/main/java/org/apache/dolphinscheduler/plugin/datasource/k8s/param/K8sDataSourceParamDTO.java

@@ -0,0 +1,35 @@
+/*
+ * 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.datasource.k8s.param;
+
+import org.apache.dolphinscheduler.plugin.datasource.api.datasource.BaseDataSourceParamDTO;
+import org.apache.dolphinscheduler.spi.enums.DbType;
+
+import lombok.Data;
+
+@Data
+public class K8sDataSourceParamDTO extends BaseDataSourceParamDTO {
+
+    protected String kubeConfig;
+    protected String namespace;
+
+    @Override
+    public DbType getType() {
+        return DbType.K8S;
+    }
+}

dolphinscheduler-datasource-plugin/dolphinscheduler-datasource-k8s/src/main/java/org/apache/dolphinscheduler/plugin/datasource/k8s/param/K8sDataSourceProcessor.java

@@ -0,0 +1,128 @@
+/*
+ * 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.datasource.k8s.param;
+
+import org.apache.dolphinscheduler.common.utils.JSONUtils;
+import org.apache.dolphinscheduler.plugin.datasource.api.datasource.BaseDataSourceParamDTO;
+import org.apache.dolphinscheduler.plugin.datasource.api.datasource.DataSourceProcessor;
+import org.apache.dolphinscheduler.plugin.datasource.api.utils.PasswordUtils;
+import org.apache.dolphinscheduler.plugin.datasource.k8s.K8sClientWrapper;
+import org.apache.dolphinscheduler.spi.datasource.ConnectionParam;
+import org.apache.dolphinscheduler.spi.enums.DbType;
+
+import org.apache.commons.lang3.StringUtils;
+
+import java.io.IOException;
+import java.sql.Connection;
+import java.sql.SQLException;
+import java.text.MessageFormat;
+
+import lombok.extern.slf4j.Slf4j;
+
+import com.google.auto.service.AutoService;
+
+@AutoService(DataSourceProcessor.class)
+@Slf4j
+public class K8sDataSourceProcessor implements DataSourceProcessor {
+
+    @Override
+    public BaseDataSourceParamDTO castDatasourceParamDTO(String paramJson) {
+        return JSONUtils.parseObject(paramJson, K8sDataSourceParamDTO.class);
+
+    }
+
+    @Override
+    public void checkDatasourceParam(BaseDataSourceParamDTO datasourceParam) {
+        K8sDataSourceParamDTO k8sDataSourceParamDTO = (K8sDataSourceParamDTO) datasourceParam;
+        if (StringUtils.isEmpty(k8sDataSourceParamDTO.getKubeConfig())) {
+            throw new IllegalArgumentException("sagemaker datasource param is not valid");
+        }
+    }
+
+    @Override
+    public String getDatasourceUniqueId(ConnectionParam connectionParam, DbType dbType) {
+        K8sConnectionParam baseConnectionParam = (K8sConnectionParam) connectionParam;
+        return MessageFormat.format("{0}@{1}@{2}", dbType.getDescp(),
+                PasswordUtils.encodePassword(baseConnectionParam.getKubeConfig()), baseConnectionParam.getNamespace());
+    }
+
+    @Override
+    public BaseDataSourceParamDTO createDatasourceParamDTO(String connectionJson) {
+        K8sConnectionParam connectionParams = (K8sConnectionParam) createConnectionParams(connectionJson);
+        K8sDataSourceParamDTO k8sDataSourceParamDTO = new K8sDataSourceParamDTO();
+        k8sDataSourceParamDTO.setKubeConfig(connectionParams.getKubeConfig());
+        k8sDataSourceParamDTO.setNamespace(connectionParams.getNamespace());
+        return k8sDataSourceParamDTO;
+    }
+
+    @Override
+    public K8sConnectionParam createConnectionParams(BaseDataSourceParamDTO datasourceParam) {
+        K8sDataSourceParamDTO k8sDataSourceParam = (K8sDataSourceParamDTO) datasourceParam;
+        K8sConnectionParam k8sConnectionParam = new K8sConnectionParam();
+        k8sConnectionParam.setKubeConfig(k8sDataSourceParam.getKubeConfig());
+        k8sConnectionParam.setNamespace(k8sDataSourceParam.getNamespace());
+        return k8sConnectionParam;
+    }
+
+    @Override
+    public ConnectionParam createConnectionParams(String connectionJson) {
+        return JSONUtils.parseObject(connectionJson, K8sConnectionParam.class);
+    }
+
+    @Override
+    public String getDatasourceDriver() {
+        return "";
+    }
+
+    @Override
+    public String getValidationQuery() {
+        return "";
+    }
+
+    @Override
+    public String getJdbcUrl(ConnectionParam connectionParam) {
+        return "";
+    }
+
+    @Override
+    public Connection getConnection(ConnectionParam connectionParam) throws ClassNotFoundException, SQLException, IOException {
+        return null;
+    }
+
+    @Override
+    public boolean checkDataSourceConnectivity(ConnectionParam connectionParam) {
+        K8sConnectionParam baseConnectionParam = (K8sConnectionParam) connectionParam;
+        try (
+                K8sClientWrapper k8sClientWrapper = new K8sClientWrapper()) {
+            return k8sClientWrapper.checkConnect(baseConnectionParam.kubeConfig, baseConnectionParam.namespace);
+        } catch (Exception e) {
+            log.error("failed to connect to the K8S cluster", e);
+            return false;
+        }
+    }
+
+    @Override
+    public DbType getDbType() {
+        return DbType.K8S;
+    }
+
+    @Override
+    public DataSourceProcessor create() {
+        return new K8sDataSourceProcessor();
+    }
+}