Browse Source
* Revert "Fix k8sTaskExecutionContext setting configYaml (#15116)" This reverts commit3.2.1-releasece11674668. * Revert "[Improvement] Refactoring K8S task plugin with connections managed in connection center (#14977)" This reverts commitc532fea00e.
Jay Chung
3 months ago
committed by
GitHub
159 changed files with 333 additions and 827 deletions
@ -1,49 +0,0 @@
|
||||
<?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>3.2.2-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> |
||||
@ -1,55 +0,0 @@
|
||||
/* |
||||
* 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 { |
||||
|
||||
} |
||||
} |
||||
@ -1,37 +0,0 @@
|
||||
/* |
||||
* 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"); |
||||
} |
||||
} |
||||
@ -1,38 +0,0 @@
|
||||
/* |
||||
* 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"; |
||||
} |
||||
|
||||
} |
||||
@ -1,32 +0,0 @@
|
||||
/* |
||||
* 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; |
||||
} |
||||
@ -1,35 +0,0 @@
|
||||
/* |
||||
* 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; |
||||
} |
||||
} |
||||
@ -1,129 +0,0 @@
|
||||
/* |
||||
* 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.AbstractDataSourceProcessor; |
||||
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 extends AbstractDataSourceProcessor { |
||||
|
||||
@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(); |
||||
} |
||||
} |
||||
@ -1,107 +0,0 @@
|
||||
/* |
||||
* 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.plugin.datasource.k8s.param.K8sConnectionParam; |
||||
import org.apache.dolphinscheduler.plugin.datasource.k8s.param.K8sDataSourceParamDTO; |
||||
import org.apache.dolphinscheduler.plugin.datasource.k8s.param.K8sDataSourceProcessor; |
||||
import org.apache.dolphinscheduler.spi.enums.DbType; |
||||
|
||||
import org.junit.jupiter.api.Assertions; |
||||
import org.junit.jupiter.api.BeforeEach; |
||||
import org.junit.jupiter.api.Test; |
||||
import org.junit.jupiter.api.extension.ExtendWith; |
||||
import org.mockito.MockedConstruction; |
||||
import org.mockito.Mockito; |
||||
import org.mockito.junit.jupiter.MockitoExtension; |
||||
|
||||
@ExtendWith(MockitoExtension.class) |
||||
public class K8sDataSourceProcessorTest { |
||||
|
||||
private K8sDataSourceProcessor k8sDataSourceProcessor; |
||||
|
||||
private String connectJson = |
||||
"{\"namespace\":\"namespace\",\"kubeConfig\":\"kubeConfig\"}"; |
||||
|
||||
@BeforeEach |
||||
public void init() { |
||||
k8sDataSourceProcessor = new K8sDataSourceProcessor(); |
||||
} |
||||
|
||||
@Test |
||||
void testCheckDatasourceParam() { |
||||
K8sDataSourceParamDTO k8sDataSourceParamDTO = new K8sDataSourceParamDTO(); |
||||
Assertions.assertThrows(IllegalArgumentException.class, |
||||
() -> k8sDataSourceProcessor.checkDatasourceParam(k8sDataSourceParamDTO)); |
||||
k8sDataSourceParamDTO.setNamespace("namespace"); |
||||
Assertions.assertThrows(IllegalArgumentException.class, |
||||
() -> k8sDataSourceProcessor.checkDatasourceParam(k8sDataSourceParamDTO)); |
||||
k8sDataSourceParamDTO.setKubeConfig("kubeConfig"); |
||||
Assertions |
||||
.assertDoesNotThrow( |
||||
() -> k8sDataSourceProcessor.checkDatasourceParam(k8sDataSourceParamDTO)); |
||||
} |
||||
|
||||
@Test |
||||
void testGetDatasourceUniqueId() { |
||||
K8sConnectionParam k8sConnectionParam = new K8sConnectionParam(); |
||||
k8sConnectionParam.setNamespace("namespace"); |
||||
k8sConnectionParam.setKubeConfig("kubeConfig"); |
||||
Assertions.assertEquals("k8s@kubeConfig@namespace", |
||||
k8sDataSourceProcessor.getDatasourceUniqueId(k8sConnectionParam, DbType.K8S)); |
||||
|
||||
} |
||||
|
||||
@Test |
||||
void testCreateDatasourceParamDTO() { |
||||
K8sDataSourceParamDTO k8sDataSourceParamDTO = |
||||
(K8sDataSourceParamDTO) k8sDataSourceProcessor.createDatasourceParamDTO(connectJson); |
||||
Assertions.assertEquals("namespace", k8sDataSourceParamDTO.getNamespace()); |
||||
Assertions.assertEquals("kubeConfig", k8sDataSourceParamDTO.getKubeConfig()); |
||||
} |
||||
|
||||
@Test |
||||
void testCreateConnectionParams() { |
||||
K8sDataSourceParamDTO k8sDataSourceParamDTO = |
||||
(K8sDataSourceParamDTO) k8sDataSourceProcessor.createDatasourceParamDTO(connectJson); |
||||
K8sConnectionParam k8sConnectionParam = |
||||
k8sDataSourceProcessor.createConnectionParams(k8sDataSourceParamDTO); |
||||
Assertions.assertEquals("namespace", k8sConnectionParam.getNamespace()); |
||||
Assertions.assertEquals("kubeConfig", k8sConnectionParam.getKubeConfig()); |
||||
} |
||||
|
||||
@Test |
||||
void testTestConnection() { |
||||
K8sDataSourceParamDTO k8sDataSourceParamDTO = |
||||
(K8sDataSourceParamDTO) k8sDataSourceProcessor.createDatasourceParamDTO(connectJson); |
||||
K8sConnectionParam connectionParam = |
||||
k8sDataSourceProcessor.createConnectionParams(k8sDataSourceParamDTO); |
||||
Assertions.assertFalse(k8sDataSourceProcessor.checkDataSourceConnectivity(connectionParam)); |
||||
|
||||
try ( |
||||
MockedConstruction<K8sClientWrapper> k8sClientWrapperMockedConstruction = |
||||
Mockito.mockConstruction(K8sClientWrapper.class, (mock, context) -> { |
||||
Mockito.when( |
||||
mock.checkConnect(connectionParam.getKubeConfig(), connectionParam.getNamespace())) |
||||
.thenReturn(true); |
||||
})) { |
||||
Assertions.assertTrue(k8sDataSourceProcessor.checkDataSourceConnectivity(connectionParam)); |
||||
} |
||||
|
||||
} |
||||
} |
||||
Some files were not shown because too many files have changed in this diff Show More
Loading…
Reference in new issue