He Wang
2 years ago
committed by
GitHub
21 changed files with 525 additions and 2 deletions
@ -0,0 +1,24 @@
|
||||
# OceanBase |
||||
|
||||
 |
||||
|
||||
## Datasource Parameters |
||||
|
||||
| **Datasource** | **Description** | |
||||
|----------------------------|--------------------------------------------------------------| |
||||
| Datasource | Select OCEANBASE. | |
||||
| Datasource name | Enter the name of the DataSource. | |
||||
| Description | Enter a description of the DataSource. | |
||||
| IP/Host name | Enter the OceanBase service IP. | |
||||
| Port | Enter the OceanBase service port. | |
||||
| Username | Set the username for OceanBase connection. | |
||||
| Password | Set the password for OceanBase connection. | |
||||
| Database name | Enter the database name of the OceanBase connection. | |
||||
| Compatible mode | Set the compatible mode of the OceanBase connection. | |
||||
| Jdbc connection parameters | Parameter settings for OceanBase connection, in JSON format. | |
||||
|
||||
## Native Supported |
||||
|
||||
No, you need to import the OceanBase jdbc driver [oceanbase-client](https://mvnrepository.com/artifact/com.oceanbase/oceanbase-client) first, refer to the section example in [datasource-setting](../howto/datasource-setting.md) `DataSource Center` section. |
||||
|
||||
The compatible mode of the datasource can be 'mysql' or 'oracle', if you only use OceanBase with 'mysql' mode, you can also treat OceanBase as MySQL and manage the datasource referring to [mysql datasource](mysql.md) |
||||
@ -0,0 +1,21 @@
|
||||
# OceanBase 数据源 |
||||
|
||||
 |
||||
|
||||
- 数据源：选择 OCEANBASE |
||||
- 数据源名称：输入数据源的名称 |
||||
- 描述：输入数据源的描述 |
||||
- IP 主机名：输入连接 OceanBase 的 IP |
||||
- 端口：输入连接 OceanBase 的端口 |
||||
- 用户名：设置连接 OceanBase 的用户名 |
||||
- 密码：设置连接 OceanBase 的密码 |
||||
- 数据库名：输入连接 OceanBase 的数据库名称 |
||||
- 兼容模式：输入连接 OceanBase 的兼容模式 |
||||
- Jdbc 连接参数：用于 OceanBase 连接的参数设置，以 JSON 形式填写 |
||||
|
||||
## 是否原生支持 |
||||
|
||||
否，使用前需要先引入 OceanBase 的 JDBC 驱动 [oceanbase-client](https://mvnrepository.com/artifact/com.oceanbase/oceanbase-client)，请参考 [数据源配置](../howto/datasource-setting.md) 中的 "数据源中心" 章节。 |
||||
|
||||
OceanBase 数据源的兼容模式可以是 'mysql' 或 'oracle'，如果你只使用 mysql 模式，你也可以选择将 OceanBase 数据源当作 MySQL 数据源来使用，请参考 [MySQL 数据源](mysql.md) |
||||
|
||||
|
After Width: | Height: | Size: 132 KiB |
@ -0,0 +1,43 @@
|
||||
<?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-oceanbase</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> |
||||
</dependencies> |
||||
</project> |
||||
@ -0,0 +1,31 @@
|
||||
/* |
||||
* 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.oceanbase; |
||||
|
||||
import org.apache.dolphinscheduler.spi.datasource.BaseConnectionParam; |
||||
import org.apache.dolphinscheduler.spi.datasource.DataSourceChannel; |
||||
import org.apache.dolphinscheduler.spi.datasource.DataSourceClient; |
||||
import org.apache.dolphinscheduler.spi.enums.DbType; |
||||
|
||||
public class OceanBaseDataSourceChannel implements DataSourceChannel { |
||||
|
||||
@Override |
||||
public DataSourceClient createDataSourceClient(BaseConnectionParam baseConnectionParam, DbType dbType) { |
||||
return new OceanBaseDataSourceClient(baseConnectionParam, dbType); |
||||
} |
||||
} |
||||
@ -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.oceanbase; |
||||
|
||||
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 OceanBaseDataSourceChannelFactory implements DataSourceChannelFactory { |
||||
|
||||
@Override |
||||
public String getName() { |
||||
return "oceanbase"; |
||||
} |
||||
|
||||
@Override |
||||
public DataSourceChannel create() { |
||||
return new OceanBaseDataSourceChannel(); |
||||
} |
||||
} |
||||
@ -0,0 +1,29 @@
|
||||
/* |
||||
* 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.oceanbase; |
||||
|
||||
import org.apache.dolphinscheduler.plugin.datasource.api.client.CommonDataSourceClient; |
||||
import org.apache.dolphinscheduler.spi.datasource.BaseConnectionParam; |
||||
import org.apache.dolphinscheduler.spi.enums.DbType; |
||||
|
||||
public class OceanBaseDataSourceClient extends CommonDataSourceClient { |
||||
|
||||
public OceanBaseDataSourceClient(BaseConnectionParam baseConnectionParam, DbType dbType) { |
||||
super(baseConnectionParam, dbType); |
||||
} |
||||
} |
||||
@ -0,0 +1,28 @@
|
||||
/* |
||||
* 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.oceanbase.param; |
||||
|
||||
import org.apache.dolphinscheduler.spi.datasource.BaseConnectionParam; |
||||
|
||||
import lombok.Data; |
||||
|
||||
@Data |
||||
public class OceanBaseConnectionParam extends BaseConnectionParam { |
||||
|
||||
protected String compatibleMode; |
||||
} |
||||
@ -0,0 +1,39 @@
|
||||
/* |
||||
* 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.oceanbase.param; |
||||
|
||||
import org.apache.dolphinscheduler.plugin.datasource.api.datasource.BaseDataSourceParamDTO; |
||||
import org.apache.dolphinscheduler.spi.enums.DbType; |
||||
|
||||
public class OceanBaseDataSourceParamDTO extends BaseDataSourceParamDTO { |
||||
|
||||
protected String compatibleMode; |
||||
|
||||
public String getCompatibleMode() { |
||||
return compatibleMode; |
||||
} |
||||
|
||||
public void setCompatibleMode(String compatibleMode) { |
||||
this.compatibleMode = compatibleMode; |
||||
} |
||||
|
||||
@Override |
||||
public DbType getType() { |
||||
return DbType.OCEANBASE; |
||||
} |
||||
} |
||||
@ -0,0 +1,191 @@
|
||||
/* |
||||
* 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.oceanbase.param; |
||||
|
||||
import org.apache.dolphinscheduler.common.constants.Constants; |
||||
import org.apache.dolphinscheduler.common.constants.DataSourceConstants; |
||||
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.spi.datasource.ConnectionParam; |
||||
import org.apache.dolphinscheduler.spi.enums.DbType; |
||||
|
||||
import org.apache.commons.collections4.MapUtils; |
||||
|
||||
import java.io.IOException; |
||||
import java.sql.Connection; |
||||
import java.sql.DriverManager; |
||||
import java.sql.SQLException; |
||||
import java.util.ArrayList; |
||||
import java.util.HashMap; |
||||
import java.util.List; |
||||
import java.util.Map; |
||||
|
||||
import lombok.extern.slf4j.Slf4j; |
||||
|
||||
import com.google.auto.service.AutoService; |
||||
|
||||
@Slf4j |
||||
@AutoService(DataSourceProcessor.class) |
||||
public class OceanBaseDataSourceProcessor extends AbstractDataSourceProcessor { |
||||
|
||||
private static final String ALLOW_LOAD_LOCAL_IN_FILE_NAME = "allowLoadLocalInfile"; |
||||
|
||||
private static final String AUTO_DESERIALIZE = "autoDeserialize"; |
||||
|
||||
private static final String ALLOW_LOCAL_IN_FILE_NAME = "allowLocalInfile"; |
||||
|
||||
private static final String ALLOW_URL_IN_LOCAL_IN_FILE_NAME = "allowUrlInLocalInfile"; |
||||
|
||||
private static final String APPEND_PARAMS = |
||||
"allowLoadLocalInfile=false&autoDeserialize=false&allowLocalInfile=false&allowUrlInLocalInfile=false"; |
||||
|
||||
@Override |
||||
public BaseDataSourceParamDTO castDatasourceParamDTO(String paramJson) { |
||||
return JSONUtils.parseObject(paramJson, OceanBaseDataSourceParamDTO.class); |
||||
} |
||||
|
||||
@Override |
||||
public BaseDataSourceParamDTO createDatasourceParamDTO(String connectionJson) { |
||||
OceanBaseConnectionParam connectionParams = (OceanBaseConnectionParam) createConnectionParams(connectionJson); |
||||
OceanBaseDataSourceParamDTO dataSourceParamDTO = new OceanBaseDataSourceParamDTO(); |
||||
dataSourceParamDTO.setUserName(connectionParams.getUser()); |
||||
dataSourceParamDTO.setDatabase(connectionParams.getDatabase()); |
||||
dataSourceParamDTO.setOther(connectionParams.getOther()); |
||||
|
||||
String address = connectionParams.getAddress(); |
||||
String[] hostSeparator = address.split(Constants.DOUBLE_SLASH); |
||||
String[] hostPortArray = hostSeparator[hostSeparator.length - 1].split(Constants.COMMA); |
||||
dataSourceParamDTO.setPort(Integer.parseInt(hostPortArray[0].split(Constants.COLON)[1])); |
||||
dataSourceParamDTO.setHost(hostPortArray[0].split(Constants.COLON)[0]); |
||||
return dataSourceParamDTO; |
||||
} |
||||
|
||||
@Override |
||||
public ConnectionParam createConnectionParams(BaseDataSourceParamDTO datasourceParam) { |
||||
OceanBaseDataSourceParamDTO obDataSourceParamDTO = (OceanBaseDataSourceParamDTO) datasourceParam; |
||||
String address = String.format("%s%s:%s", DataSourceConstants.JDBC_OCEANBASE, obDataSourceParamDTO.getHost(), |
||||
obDataSourceParamDTO.getPort()); |
||||
String jdbcUrl = String.format("%s/%s", address, obDataSourceParamDTO.getDatabase()); |
||||
|
||||
OceanBaseConnectionParam connectionParam = new OceanBaseConnectionParam(); |
||||
connectionParam.setJdbcUrl(jdbcUrl); |
||||
connectionParam.setDatabase(obDataSourceParamDTO.getDatabase()); |
||||
connectionParam.setAddress(address); |
||||
connectionParam.setUser(obDataSourceParamDTO.getUserName()); |
||||
connectionParam.setPassword(PasswordUtils.encodePassword(obDataSourceParamDTO.getPassword())); |
||||
connectionParam.setCompatibleMode(obDataSourceParamDTO.getCompatibleMode()); |
||||
connectionParam.setDriverClassName(getDatasourceDriver()); |
||||
connectionParam.setValidationQuery(getValidationQuery(connectionParam.getCompatibleMode())); |
||||
connectionParam.setOther(obDataSourceParamDTO.getOther()); |
||||
return connectionParam; |
||||
} |
||||
|
||||
@Override |
||||
public ConnectionParam createConnectionParams(String connectionJson) { |
||||
return JSONUtils.parseObject(connectionJson, OceanBaseConnectionParam.class); |
||||
} |
||||
|
||||
@Override |
||||
public String getDatasourceDriver() { |
||||
return DataSourceConstants.COM_OCEANBASE_JDBC_DRIVER; |
||||
} |
||||
|
||||
@Override |
||||
public String getValidationQuery() { |
||||
throw new UnsupportedOperationException("Can't get validation query without compatible mode"); |
||||
} |
||||
|
||||
public String getValidationQuery(String compatibleMode) { |
||||
if (compatibleMode != null) { |
||||
switch (compatibleMode.trim().toLowerCase()) { |
||||
case "mysql": |
||||
return DataSourceConstants.MYSQL_VALIDATION_QUERY; |
||||
case "oracle": |
||||
return DataSourceConstants.ORACLE_VALIDATION_QUERY; |
||||
} |
||||
} |
||||
throw new UnsupportedOperationException("Invalid compatible mode: " + compatibleMode); |
||||
|
||||
} |
||||
|
||||
@Override |
||||
public String getJdbcUrl(ConnectionParam connectionParam) { |
||||
OceanBaseConnectionParam obConnectionParam = (OceanBaseConnectionParam) connectionParam; |
||||
String jdbcUrl = obConnectionParam.getJdbcUrl(); |
||||
if (MapUtils.isNotEmpty(obConnectionParam.getOther())) { |
||||
return String.format("%s?%s&%s", jdbcUrl, transformOther(obConnectionParam.getOther()), APPEND_PARAMS); |
||||
} |
||||
return String.format("%s?%s", jdbcUrl, APPEND_PARAMS); |
||||
} |
||||
|
||||
private String transformOther(Map<String, String> paramMap) { |
||||
if (MapUtils.isEmpty(paramMap)) { |
||||
return null; |
||||
} |
||||
Map<String, String> otherMap = new HashMap<>(); |
||||
paramMap.forEach((k, v) -> { |
||||
if (!checkKeyIsLegitimate(k)) { |
||||
return; |
||||
} |
||||
otherMap.put(k, v); |
||||
}); |
||||
if (MapUtils.isEmpty(otherMap)) { |
||||
return null; |
||||
} |
||||
List<String> otherList = new ArrayList<>(); |
||||
otherMap.forEach((key, value) -> otherList.add(String.format("%s=%s", key, value))); |
||||
return String.join("&", otherList); |
||||
} |
||||
|
||||
private static boolean checkKeyIsLegitimate(String key) { |
||||
return !key.contains(ALLOW_LOAD_LOCAL_IN_FILE_NAME) |
||||
&& !key.contains(AUTO_DESERIALIZE) |
||||
&& !key.contains(ALLOW_LOCAL_IN_FILE_NAME) |
||||
&& !key.contains(ALLOW_URL_IN_LOCAL_IN_FILE_NAME); |
||||
} |
||||
|
||||
@Override |
||||
public Connection getConnection(ConnectionParam connectionParam) throws ClassNotFoundException, SQLException, IOException { |
||||
OceanBaseConnectionParam obConnectionParam = (OceanBaseConnectionParam) connectionParam; |
||||
Class.forName(getDatasourceDriver()); |
||||
String user = obConnectionParam.getUser(); |
||||
if (user.contains(AUTO_DESERIALIZE)) { |
||||
log.warn("sensitive param : {} in username field is filtered", AUTO_DESERIALIZE); |
||||
user = user.replace(AUTO_DESERIALIZE, ""); |
||||
} |
||||
String password = PasswordUtils.decodePassword(obConnectionParam.getPassword()); |
||||
if (password.contains(AUTO_DESERIALIZE)) { |
||||
log.warn("sensitive param : {} in password field is filtered", AUTO_DESERIALIZE); |
||||
password = password.replace(AUTO_DESERIALIZE, ""); |
||||
} |
||||
return DriverManager.getConnection(getJdbcUrl(connectionParam), user, password); |
||||
} |
||||
|
||||
@Override |
||||
public DbType getDbType() { |
||||
return DbType.OCEANBASE; |
||||
} |
||||
|
||||
@Override |
||||
public DataSourceProcessor create() { |
||||
return new OceanBaseDataSourceProcessor(); |
||||
} |
||||
} |
||||
Loading…
Reference in new issue