-
Notifications
You must be signed in to change notification settings - Fork 4.1k
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
🎉 Destination MySQl - Added support for connection via ssh (aka basti…
…on server) (#6317) * updated mysql tests * updated mysql tests * added mysql ssh tunnel tests by key * fixed remarks * fixed remarks * updated DatabricksStreamCopier * switch to custom file for ssh config in normalization * updated MySQL SSH tests * bump version * get local port properly * updated assertSameValue for MySQL ssh tunnel * updated image version and documentation * updated code style * updated CI credentials * updated normalization documentation Co-authored-by: George Claireaux <george@claireaux.co.uk>
- Loading branch information
1 parent
96aedd3
commit 8fa1571
Showing
13 changed files
with
303 additions
and
12 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
36 changes: 36 additions & 0 deletions
36
.../java/io/airbyte/integrations/destination/mysql/SshKeyMySQLDestinationAcceptanceTest.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,36 @@ | ||
| /* | ||
| * MIT License | ||
| * | ||
| * Copyright (c) 2020 Airbyte | ||
| * | ||
| * Permission is hereby granted, free of charge, to any person obtaining a copy | ||
| * of this software and associated documentation files (the "Software"), to deal | ||
| * in the Software without restriction, including without limitation the rights | ||
| * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell | ||
| * copies of the Software, and to permit persons to whom the Software is | ||
| * furnished to do so, subject to the following conditions: | ||
| * | ||
| * The above copyright notice and this permission notice shall be included in all | ||
| * copies or substantial portions of the Software. | ||
| * | ||
| * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR | ||
| * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, | ||
| * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE | ||
| * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER | ||
| * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, | ||
| * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE | ||
| * SOFTWARE. | ||
| */ | ||
|
|
||
| package io.airbyte.integrations.destination.mysql; | ||
|
|
||
| import java.nio.file.Path; | ||
|
|
||
| public class SshKeyMySQLDestinationAcceptanceTest extends SshMySQLDestinationAcceptanceTest { | ||
|
|
||
| @Override | ||
| public Path getConfigFilePath() { | ||
| return Path.of("secrets/ssh-key-config.json"); | ||
| } | ||
|
|
||
| } |
192 changes: 192 additions & 0 deletions
192
...ion/java/io/airbyte/integrations/destination/mysql/SshMySQLDestinationAcceptanceTest.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,192 @@ | ||
| /* | ||
| * MIT License | ||
| * | ||
| * Copyright (c) 2020 Airbyte | ||
| * | ||
| * Permission is hereby granted, free of charge, to any person obtaining a copy | ||
| * of this software and associated documentation files (the "Software"), to deal | ||
| * in the Software without restriction, including without limitation the rights | ||
| * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell | ||
| * copies of the Software, and to permit persons to whom the Software is | ||
| * furnished to do so, subject to the following conditions: | ||
| * | ||
| * The above copyright notice and this permission notice shall be included in all | ||
| * copies or substantial portions of the Software. | ||
| * | ||
| * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR | ||
| * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, | ||
| * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE | ||
| * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER | ||
| * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, | ||
| * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE | ||
| * SOFTWARE. | ||
| */ | ||
|
|
||
| package io.airbyte.integrations.destination.mysql; | ||
|
|
||
| import static org.junit.jupiter.api.Assertions.assertEquals; | ||
|
|
||
| import com.fasterxml.jackson.databind.JsonNode; | ||
| import com.fasterxml.jackson.databind.node.ObjectNode; | ||
| import io.airbyte.commons.functional.CheckedFunction; | ||
| import io.airbyte.commons.io.IOs; | ||
| import io.airbyte.commons.json.Jsons; | ||
| import io.airbyte.db.Database; | ||
| import io.airbyte.db.Databases; | ||
| import io.airbyte.integrations.base.JavaBaseConstants; | ||
| import io.airbyte.integrations.base.ssh.SshTunnel; | ||
| import io.airbyte.integrations.destination.ExtendedNameTransformer; | ||
| import io.airbyte.integrations.standardtest.destination.DestinationAcceptanceTest; | ||
| import java.nio.file.Path; | ||
| import java.util.ArrayList; | ||
| import java.util.List; | ||
| import java.util.stream.Collectors; | ||
| import org.apache.commons.lang3.RandomStringUtils; | ||
| import org.jooq.JSONFormat; | ||
|
|
||
| /** | ||
| * Abstract class that allows us to avoid duplicating testing logic for testing SSH with a key file | ||
| * or with a password. | ||
| */ | ||
| public abstract class SshMySQLDestinationAcceptanceTest extends DestinationAcceptanceTest { | ||
|
|
||
| private static final JSONFormat JSON_FORMAT = new JSONFormat().recordFormat(JSONFormat.RecordFormat.OBJECT); | ||
|
|
||
| private final ExtendedNameTransformer namingResolver = new MySQLNameTransformer(); | ||
| private String schemaName; | ||
|
|
||
| public abstract Path getConfigFilePath(); | ||
|
|
||
| @Override | ||
| protected String getImageName() { | ||
| return "airbyte/destination-mysql:dev"; | ||
| } | ||
|
|
||
| @Override | ||
| protected JsonNode getConfig() { | ||
| var config = getConfigFromSecretsFile(); | ||
| ((ObjectNode) config).put("database", schemaName); | ||
| return config; | ||
| } | ||
|
|
||
| private JsonNode getConfigFromSecretsFile() { | ||
| return Jsons.deserialize(IOs.readFile(getConfigFilePath())); | ||
| } | ||
|
|
||
| @Override | ||
| protected JsonNode getFailCheckConfig() { | ||
| final JsonNode clone = Jsons.clone(getConfig()); | ||
| ((ObjectNode) clone).put("password", "wrong password"); | ||
| return clone; | ||
| } | ||
|
|
||
| @Override | ||
| protected List<JsonNode> retrieveRecords(final TestDestinationEnv env, | ||
| final String streamName, | ||
| final String namespace, | ||
| final JsonNode streamSchema) | ||
| throws Exception { | ||
| return retrieveRecordsFromTable(namingResolver.getRawTableName(streamName), namespace) | ||
| .stream() | ||
| .map(r -> Jsons.deserialize(r.get(JavaBaseConstants.COLUMN_NAME_DATA).asText())) | ||
| .collect(Collectors.toList()); | ||
| } | ||
|
|
||
| @Override | ||
| protected boolean supportsNormalization() { | ||
| return true; | ||
| } | ||
|
|
||
| @Override | ||
| protected boolean supportsDBT() { | ||
| return true; | ||
| } | ||
|
|
||
| @Override | ||
| protected boolean implementsNamespaces() { | ||
| return true; | ||
| } | ||
|
|
||
| @Override | ||
| protected List<JsonNode> retrieveNormalizedRecords(final TestDestinationEnv env, | ||
| final String streamName, | ||
| final String namespace) | ||
| throws Exception { | ||
| var tableName = namingResolver.getIdentifier(streamName); | ||
| String schema = namespace != null ? namingResolver.getIdentifier(namespace) : namingResolver.getIdentifier(schemaName); | ||
| return retrieveRecordsFromTable(tableName, schema); | ||
| } | ||
|
|
||
| @Override | ||
| protected List<String> resolveIdentifier(final String identifier) { | ||
| final List<String> result = new ArrayList<>(); | ||
| final String resolved = namingResolver.getIdentifier(identifier); | ||
| result.add(identifier); | ||
| result.add(resolved); | ||
| if (!resolved.startsWith("\"")) { | ||
| result.add(resolved.toLowerCase()); | ||
| result.add(resolved.toUpperCase()); | ||
| } | ||
| return result; | ||
| } | ||
|
|
||
| private static Database getDatabaseFromConfig(final JsonNode config) { | ||
| return Databases.createMySqlDatabase( | ||
| config.get("username").asText(), | ||
| config.get("password").asText(), | ||
| String.format("jdbc:mysql://%s:%s", | ||
| config.get("host").asText(), | ||
| config.get("port").asText())); | ||
| } | ||
|
|
||
| private List<JsonNode> retrieveRecordsFromTable(String tableName, String schemaName) throws Exception { | ||
| var schema = schemaName == null ? this.schemaName : schemaName; | ||
| return SshTunnel.sshWrap( | ||
| getConfig(), | ||
| MySQLDestination.HOST_KEY, | ||
| MySQLDestination.PORT_KEY, | ||
| (CheckedFunction<JsonNode, List<JsonNode>, Exception>) mangledConfig -> getDatabaseFromConfig(mangledConfig) | ||
| .query( | ||
| ctx -> ctx | ||
| .fetch(String.format("SELECT * FROM %s.%s ORDER BY %s ASC;", schema, tableName.toLowerCase(), | ||
| JavaBaseConstants.COLUMN_NAME_EMITTED_AT)) | ||
| .stream() | ||
| .map(r -> r.formatJSON(JSON_FORMAT)) | ||
| .map(Jsons::deserialize) | ||
| .collect(Collectors.toList()))); | ||
| } | ||
|
|
||
| @Override | ||
| protected void setup(final TestDestinationEnv testEnv) throws Exception { | ||
| schemaName = RandomStringUtils.randomAlphabetic(8).toLowerCase(); | ||
| var config = getConfig(); | ||
| SshTunnel.sshWrap( | ||
| config, | ||
| MySQLDestination.HOST_KEY, | ||
| MySQLDestination.PORT_KEY, | ||
| mangledConfig -> { | ||
| getDatabaseFromConfig(mangledConfig).query(ctx -> ctx.fetch(String.format("CREATE DATABASE %s;", schemaName))); | ||
| }); | ||
| } | ||
|
|
||
| @Override | ||
| protected void tearDown(final TestDestinationEnv testEnv) throws Exception { | ||
| SshTunnel.sshWrap( | ||
| getConfig(), | ||
| MySQLDestination.HOST_KEY, | ||
| MySQLDestination.PORT_KEY, | ||
| mangledConfig -> { | ||
| getDatabaseFromConfig(mangledConfig).query(ctx -> ctx.fetch(String.format("DROP DATABASE %s", schemaName))); | ||
| }); | ||
| } | ||
|
|
||
| protected void assertSameValue(JsonNode expectedValue, JsonNode actualValue) { | ||
| if (expectedValue.isBoolean()) { | ||
| // Boolean in MySQL are stored as TINYINT (0 or 1) so we force them to boolean values here | ||
| assertEquals(expectedValue.asBoolean(), actualValue.asBoolean()); | ||
| } else { | ||
| assertEquals(expectedValue, actualValue); | ||
| } | ||
| } | ||
|
|
||
| } |
36 changes: 36 additions & 0 deletions
36
.../io/airbyte/integrations/destination/mysql/SshPasswordMySQLDestinationAcceptanceTest.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,36 @@ | ||
| /* | ||
| * MIT License | ||
| * | ||
| * Copyright (c) 2020 Airbyte | ||
| * | ||
| * Permission is hereby granted, free of charge, to any person obtaining a copy | ||
| * of this software and associated documentation files (the "Software"), to deal | ||
| * in the Software without restriction, including without limitation the rights | ||
| * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell | ||
| * copies of the Software, and to permit persons to whom the Software is | ||
| * furnished to do so, subject to the following conditions: | ||
| * | ||
| * The above copyright notice and this permission notice shall be included in all | ||
| * copies or substantial portions of the Software. | ||
| * | ||
| * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR | ||
| * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, | ||
| * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE | ||
| * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER | ||
| * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, | ||
| * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE | ||
| * SOFTWARE. | ||
| */ | ||
|
|
||
| package io.airbyte.integrations.destination.mysql; | ||
|
|
||
| import java.nio.file.Path; | ||
|
|
||
| public class SshPasswordMySQLDestinationAcceptanceTest extends SshMySQLDestinationAcceptanceTest { | ||
|
|
||
| @Override | ||
| public Path getConfigFilePath() { | ||
| return Path.of("secrets/ssh-pwd-config.json"); | ||
| } | ||
|
|
||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.