🎉 Destination Oracle - Added support for connection via ssh tunnel

.github/workflows/publish-command.yml

@@ -130,6 +130,8 @@ jobs:
           MIXPANEL_INTEGRATION_TEST_CREDS: ${{ secrets.MIXPANEL_INTEGRATION_TEST_CREDS }}
           MSSQL_RDS_TEST_CREDS: ${{ secrets.MSSQL_RDS_TEST_CREDS }}
           PAYPAL_TRANSACTION_CREDS: ${{ secrets.SOURCE_PAYPAL_TRANSACTION_CREDS }}
+          ORACLE_SSH_KEY_TEST_CREDS: ${{ secrets.ORACLE_SSH_KEY_TEST_CREDS }}
+          ORACLE_SSH_PWD_TEST_CREDS: ${{ secrets.ORACLE_SSH_PWD_TEST_CREDS }}
           POSTGRES_SSH_KEY_TEST_CREDS: ${{ secrets.POSTGRES_SSH_KEY_TEST_CREDS }}
           POSTGRES_SSH_PWD_TEST_CREDS: ${{ secrets.POSTGRES_SSH_PWD_TEST_CREDS }}
           POSTHOG_TEST_CREDS: ${{ secrets.POSTHOG_TEST_CREDS }}

.github/workflows/test-command.yml

@@ -126,6 +126,8 @@ jobs:
           MIXPANEL_INTEGRATION_TEST_CREDS: ${{ secrets.MIXPANEL_INTEGRATION_TEST_CREDS }}
           MSSQL_RDS_TEST_CREDS: ${{ secrets.MSSQL_RDS_TEST_CREDS }}
           PAYPAL_TRANSACTION_CREDS: ${{ secrets.SOURCE_PAYPAL_TRANSACTION_CREDS }}
+          ORACLE_SSH_KEY_TEST_CREDS: ${{ secrets.ORACLE_SSH_KEY_TEST_CREDS }}
+          ORACLE_SSH_PWD_TEST_CREDS: ${{ secrets.ORACLE_SSH_PWD_TEST_CREDS }}
           POSTGRES_SSH_KEY_TEST_CREDS: ${{ secrets.POSTGRES_SSH_KEY_TEST_CREDS }}
           POSTGRES_SSH_PWD_TEST_CREDS: ${{ secrets.POSTGRES_SSH_PWD_TEST_CREDS }}
           POSTHOG_TEST_CREDS: ${{ secrets.POSTHOG_TEST_CREDS }}

airbyte-integrations/connectors/destination-oracle/src/main/java/io/airbyte/integrations/destination/oracle/OracleDestination.java

@@ -30,13 +30,17 @@
 import io.airbyte.integrations.base.Destination;
 import io.airbyte.integrations.base.IntegrationRunner;
 import io.airbyte.integrations.base.JavaBaseConstants;
+import io.airbyte.integrations.base.ssh.SshWrappedDestination;
 import io.airbyte.integrations.destination.jdbc.AbstractJdbcDestination;
+import java.util.List;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 public class OracleDestination extends AbstractJdbcDestination implements Destination {
 
   private static final Logger LOGGER = LoggerFactory.getLogger(OracleDestination.class);
+  public static final List<String> HOST_KEY = List.of("host");
+  public static final List<String> PORT_KEY = List.of("port");
 
   public static final String DRIVER_CLASS = "oracle.jdbc.OracleDriver";
 
@@ -66,7 +70,7 @@ public JsonNode toJdbcConfig(JsonNode config) {
   }
 
   public static void main(String[] args) throws Exception {
-    final Destination destination = new OracleDestination();
+    final Destination destination = new SshWrappedDestination(new OracleDestination(), HOST_KEY, PORT_KEY);
     LOGGER.info("starting destination: {}", OracleDestination.class);
     new IntegrationRunner(destination).run(args);
     LOGGER.info("completed destination: {}", OracleDestination.class);

airbyte-integrations/connectors/destination-oracle/src/test-integration/java/io/airbyte/integrations/destination/oracle/SshKeyOracleDestinationAcceptanceTest.java

@@ -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.oracle;
+
+import java.nio.file.Path;
+
+public class SshKeyOracleDestinationAcceptanceTest extends SshOracleDestinationAcceptanceTest {
+
+  @Override
+  public Path getConfigFilePath() {
+    return Path.of("secrets/ssh-key-config.json");
+  }
+
+}

airbyte-integrations/connectors/destination-oracle/src/test-integration/java/io/airbyte/integrations/destination/oracle/SshOracleDestinationAcceptanceTest.java

@@ -0,0 +1,149 @@
+/*
+ * 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.oracle;
+
+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.List;
+import java.util.stream.Collectors;
+import org.apache.commons.lang3.RandomStringUtils;
+import org.jooq.JSONFormat;
+
+public abstract class SshOracleDestinationAcceptanceTest extends DestinationAcceptanceTest {
+
+  private static final JSONFormat JSON_FORMAT = new JSONFormat().recordFormat(JSONFormat.RecordFormat.OBJECT);
+
+  private final ExtendedNameTransformer namingResolver = new OracleNameTransformer();
+
+  private final String schemaName = "TEST_" + RandomStringUtils.randomAlphabetic(6).toUpperCase();
+
+  public abstract Path getConfigFilePath();
+
+  @Override
+  protected String getImageName() {
+    return "airbyte/destination-oracle:dev";
+  }
+
+  @Override
+  protected JsonNode getConfig() {
+    final JsonNode config = getConfigFromSecretsFile();
+    ((ObjectNode) config).put("schema", schemaName);
+    return config;
+  }
+
+  private JsonNode getConfigFromSecretsFile() {
+    return Jsons.deserialize(IOs.readFile(getConfigFilePath()));
+  }
+
+  @Override
+  protected JsonNode getFailCheckConfig() throws Exception {
+    final JsonNode clone = Jsons.clone(getConfig());
+    ((ObjectNode) clone).put("password", "wrong password");
+    return clone;
+  }
+
+  @Override
+  protected List<JsonNode> retrieveRecords(TestDestinationEnv testEnv, String streamName, String namespace, JsonNode streamSchema) throws Exception {
+    List<JsonNode> jsonNodes = retrieveRecordsFromTable(namingResolver.getRawTableName(streamName), namespace);
+    return jsonNodes
+        .stream()
+        .map(r -> Jsons.deserialize(r.get(JavaBaseConstants.COLUMN_NAME_DATA.toUpperCase()).asText()))
+        .collect(Collectors.toList());
+  }
+
+  private List<JsonNode> retrieveRecordsFromTable(final String tableName, final String schemaName) throws Exception {
+    final JsonNode config = getConfig();
+    return SshTunnel.sshWrap(
+        config,
+        OracleDestination.HOST_KEY,
+        OracleDestination.PORT_KEY,
+        (CheckedFunction<JsonNode, List<JsonNode>, Exception>) mangledConfig -> getDatabaseFromConfig(mangledConfig)
+            .query(
+                ctx -> ctx
+                    .fetch(String.format("SELECT * FROM %s.%s ORDER BY %s ASC", schemaName, tableName, OracleDestination.COLUMN_NAME_EMITTED_AT)))
+            .stream()
+            .map(r -> r.formatJSON(JSON_FORMAT))
+            .map(Jsons::deserialize)
+            .collect(Collectors.toList()));
+  }
+
+  @Override
+  protected void setup(TestDestinationEnv testEnv) throws Exception {
+    SshTunnel.sshWrap(
+        getConfig(),
+        OracleDestination.HOST_KEY,
+        OracleDestination.PORT_KEY,
+        mangledConfig -> {
+          Database databaseFromConfig = getDatabaseFromConfig(mangledConfig);
+          databaseFromConfig.query(ctx -> ctx.fetch(String.format("CREATE USER %s IDENTIFIED BY %s", schemaName, schemaName)));
+          databaseFromConfig.query(ctx -> ctx.fetch(String.format("GRANT CREATE TABLE TO %s", schemaName)));
+          databaseFromConfig.query(ctx -> ctx.fetch(String.format("ALTER USER %s quota unlimited on USERS", schemaName)));
+        });
+  }
+
+  private Database getDatabaseFromConfig(final JsonNode config) {
+    return Databases.createDatabase(
+        config.get("username").asText(),
+        config.get("password").asText(),
+        String.format("jdbc:oracle:thin:@//%s:%s/%s",
+            config.get("host").asText(),
+            config.get("port").asText(),
+            config.get("sid").asText()),
+        "oracle.jdbc.driver.OracleDriver",
+        null);
+  }
+
+  @Override
+  protected void tearDown(TestDestinationEnv testEnv) throws Exception {
+    SshTunnel.sshWrap(
+        getConfig(),
+        OracleDestination.HOST_KEY,
+        OracleDestination.PORT_KEY,
+        mangledConfig -> {
+          getDatabaseFromConfig(mangledConfig).query(ctx -> ctx.fetch(String.format("DROP USER %s CASCADE", schemaName)));
+        });
+  }
+
+  @Override
+  protected boolean supportsDBT() {
+    return true;
+  }
+
+  @Override
+  protected boolean implementsNamespaces() {
+    return true;
+  }
+
+}

airbyte-integrations/connectors/destination-oracle/src/test-integration/java/io/airbyte/integrations/destination/oracle/SshPasswordOracleDestinationAcceptanceTest.java

@@ -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.oracle;
+
+import java.nio.file.Path;
+
+public class SshPasswordOracleDestinationAcceptanceTest extends SshOracleDestinationAcceptanceTest {
+
+  @Override
+  public Path getConfigFilePath() {
+    return Path.of("secrets/ssh-pwd-config.json");
+  }
+
+}

docs/integrations/destinations/oracle.md

@@ -62,6 +62,23 @@ You should now have all the requirements needed to configure Oracle as a destina
 * **Username**
 * **Password**
 * **Database**
+* 
+## Connection to Oracle via an SSH Tunnel
+
+Airbyte has the ability to connect to a Oracle instance via an SSH Tunnel. The reason you might want to do this because it is not possible (or against security policy) to connect to the database directly (e.g. it does not have a public IP address).
+
+When using an SSH tunnel, you are configuring Airbyte to connect to an intermediate server (a.k.a. a bastion sever) that _does_ have direct access to the database. Airbyte connects to the bastion and then asks the bastion to connect directly to the server.
+
+Using this feature requires additional configuration, when creating the source. We will talk through what each piece of configuration means.
+1. Configure all fields for the source as you normally would, except `SSH Tunnel Method`.
+2. `SSH Tunnel Method` defaults to `No Tunnel` (meaning a direct connection). If you want to use an SSH Tunnel choose `SSH Key Authentication` or `Password Authentication`.
+    1. Choose `Key Authentication` if you will be using an RSA private key as your secret for establishing the SSH Tunnel (see below for more information on generating this key).
+    2. Choose `Password Authentication` if you will be using a password as your secret for establishing the SSH Tunnel.
+3. `SSH Tunnel Jump Server Host` refers to the intermediate (bastion) server that Airbyte will connect to. This should be a hostname or an IP Address.
+4. `SSH Connection Port` is the port on the bastion server with which to make the SSH connection. The default port for SSH connections is `22`, so unless you have explicitly changed something, go with the default.
+5. `SSH Login Username` is the username that Airbyte should use when connection to the bastion server. This is NOT the Oracle username.
+6. If you are using `Password Authentication`, then `SSH Login Username` should be set to the password of the User from the previous step. If you are using `SSH Key Authentication` leave this blank. Again, this is not the Oracle password, but the password for the OS-user that Airbyte is using to perform commands on the bastion.
+7. If you are using `SSH Key Authentication`, then `SSH Private Key` should be set to the RSA Private Key that you are using to create the SSH connection. This should be the full contents of the key file starting with `-----BEGIN RSA PRIVATE KEY-----` and ending with `-----END RSA PRIVATE KEY-----`.
 
 ## Changelog
 | Version | Date | Pull Request | Subject |

tools/bin/ci_credentials.sh

@@ -29,6 +29,8 @@ write_standard_creds destination-gcs "$DESTINATION_GCS_CREDS"
 write_standard_creds destination-kvdb "$DESTINATION_KVDB_TEST_CREDS"
 write_standard_creds destination-keen "$DESTINATION_KEEN_TEST_CREDS"
 
+write_standard_creds destination-oracle "$ORACLE_SSH_KEY_TEST_CREDS" "ssh-key-config.json"
+write_standard_creds destination-oracle "$ORACLE_SSH_PWD_TEST_CREDS" "ssh-pwd-config.json"
 write_standard_creds destination-postgres "$DESTINATION_POSTGRES_SSH_PWD_TEST_CREDS" "ssh-pwd-config.json"
 write_standard_creds destination-postgres "$DESTINATION_POSTGRES_SSH_KEY_TEST_CREDS" "ssh-key-config.json"
 write_standard_creds destination-postgres "$DESTINATION_PUBSUB_TEST_CREDS" "credentials.json"