ssh tunnel for postgres normalization

.github/workflows/publish-command.yml

@@ -86,6 +86,8 @@ jobs:
           BRAINTREE_TEST_CREDS: ${{ secrets.BRAINTREE_TEST_CREDS }}
           CART_TEST_CREDS: ${{ secrets.CART_TEST_CREDS }}
           CHARGEBEE_INTEGRATION_TEST_CREDS: ${{ secrets.CHARGEBEE_INTEGRATION_TEST_CREDS }}
+          DESTINATION_POSTGRES_SSH_KEY_TEST_CREDS: ${{ secrets.DESTINATION_POSTGRES_SSH_KEY_TEST_CREDS }}
+          DESTINATION_POSTGRES_SSH_PWD_TEST_CREDS: ${{ secrets.DESTINATION_POSTGRES_SSH_PWD_TEST_CREDS }}
           DESTINATION_PUBSUB_TEST_CREDS: ${{ secrets.DESTINATION_PUBSUB_TEST_CREDS }}
           DESTINATION_KEEN_TEST_CREDS: ${{ secrets.DESTINATION_KEEN_TEST_CREDS }}
           DESTINATION_KVDB_TEST_CREDS: ${{ secrets.DESTINATION_KVDB_TEST_CREDS }}

.github/workflows/test-command.yml

@@ -86,6 +86,8 @@ jobs:
           BRAINTREE_TEST_CREDS: ${{ secrets.BRAINTREE_TEST_CREDS }}
           CART_TEST_CREDS: ${{ secrets.CART_TEST_CREDS }}
           CHARGEBEE_INTEGRATION_TEST_CREDS: ${{ secrets.CHARGEBEE_INTEGRATION_TEST_CREDS }}
+          DESTINATION_POSTGRES_SSH_KEY_TEST_CREDS: ${{ secrets.DESTINATION_POSTGRES_SSH_KEY_TEST_CREDS }}
+          DESTINATION_POSTGRES_SSH_PWD_TEST_CREDS: ${{ secrets.DESTINATION_POSTGRES_SSH_PWD_TEST_CREDS }}
           DESTINATION_PUBSUB_TEST_CREDS: ${{ secrets.DESTINATION_PUBSUB_TEST_CREDS }}
           DESTINATION_KEEN_TEST_CREDS: ${{ secrets.DESTINATION_KEEN_TEST_CREDS }}
           DESTINATION_KVDB_TEST_CREDS: ${{ secrets.DESTINATION_KVDB_TEST_CREDS }}

airbyte-integrations/bases/base-java/src/main/java/io/airbyte/integrations/base/AirbyteMessageConsumer.java

@@ -24,6 +24,7 @@
 
 package io.airbyte.integrations.base;
 
+import io.airbyte.commons.concurrency.VoidCallable;
 import io.airbyte.commons.functional.CheckedConsumer;
 import io.airbyte.protocol.models.AirbyteMessage;
 
@@ -56,4 +57,29 @@ public interface AirbyteMessageConsumer extends CheckedConsumer<AirbyteMessage,
   @Override
   void close() throws Exception;
 
+  /**
+   * Append a function to be called on {@link AirbyteMessageConsumer#close}.
+   */
+  static AirbyteMessageConsumer appendOnClose(final AirbyteMessageConsumer consumer, final VoidCallable voidCallable) {
+    return new AirbyteMessageConsumer() {
+
+      @Override
+      public void start() throws Exception {
+        consumer.start();
+      }
+
+      @Override
+      public void accept(final AirbyteMessage message) throws Exception {
+        consumer.accept(message);
+      }
+
+      @Override
+      public void close() throws Exception {
+        consumer.close();
+        voidCallable.call();
+      }
+
+    };
+  }
+
 }

airbyte-integrations/bases/base-java/src/main/java/io/airbyte/integrations/base/ssh/SshWrappedDestination.java

@@ -0,0 +1,81 @@
+/*
+ * 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.base.ssh;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.node.ObjectNode;
+import io.airbyte.commons.json.Jsons;
+import io.airbyte.commons.resources.MoreResources;
+import io.airbyte.integrations.base.AirbyteMessageConsumer;
+import io.airbyte.integrations.base.Destination;
+import io.airbyte.protocol.models.AirbyteConnectionStatus;
+import io.airbyte.protocol.models.AirbyteMessage;
+import io.airbyte.protocol.models.ConfiguredAirbyteCatalog;
+import io.airbyte.protocol.models.ConnectorSpecification;
+import java.util.List;
+import java.util.function.Consumer;
+
+/**
+ * Decorates a Destination with an SSH Tunnel using the standard configuration that Airbyte uses for
+ * configuring SSH.
+ */
+public class SshWrappedDestination implements Destination {
+
+  private final Destination delegate;
+  private final List<String> hostKey;
+  private final List<String> portKey;
+
+  public SshWrappedDestination(final Destination delegate,
+                               final List<String> hostKey,
+                               final List<String> portKey) {
+    this.delegate = delegate;
+    this.hostKey = hostKey;
+    this.portKey = portKey;
+  }
+
+  @Override
+  public ConnectorSpecification spec() throws Exception {
+    // inject the standard ssh configuration into the spec.
+    final ConnectorSpecification originalSpec = delegate.spec();
+    final ObjectNode propNode = (ObjectNode) originalSpec.getConnectionSpecification().get("properties");
+    propNode.set("tunnel_method", Jsons.deserialize(MoreResources.readResource("ssh-tunnel-spec.json")));
+    return originalSpec;
+  }
+
+  @Override
+  public AirbyteConnectionStatus check(final JsonNode config) throws Exception {
+    return SshTunnel.sshWrap(config, hostKey, portKey, delegate::check);
+  }
+
+  @Override
+  public AirbyteMessageConsumer getConsumer(final JsonNode config,
+                                            final ConfiguredAirbyteCatalog catalog,
+                                            final Consumer<AirbyteMessage> outputRecordCollector)
+      throws Exception {
+    final SshTunnel tunnel = SshTunnel.getInstance(config, hostKey, portKey);
+    return AirbyteMessageConsumer.appendOnClose(delegate.getConsumer(tunnel.getConfigInTunnel(), catalog, outputRecordCollector), tunnel::close);
+  }
+
+}

airbyte-integrations/bases/base-java/src/main/java/io/airbyte/integrations/base/ssh/readme.md

@@ -1,13 +1,14 @@
-# Developing an SSH Source
+# Developing an SSH Connector
 
 ## Goal
-Easy development of any source that needs the ability to connect to a resource via SSH Tunnel.
+Easy development of any connector that needs the ability to connect to a resource via SSH Tunnel.
 
 ## Overview
 Our SSH connector support is designed to be easy to plug into any existing connector. There are a few major pieces to consider:
 1. Add SSH Configuration to the Spec - for SSH, we need to take in additional configuration, so we need to inject extra fields into the connector configuration.
 2. Add SSH Logic to the Connector - before the connector code begins to execute we need to start an SSH tunnel. This library provides logic to create that tunnel (and clean it up).
 3. Acceptance Testing - it is a good practice to include acceptance testing for the SSH version of a connector for at least one of the SSH types (password or ssh key). While unit testing for the SSH functionality exists in this package (coming soon), high-level acceptance testing to make sure this feature works with the individual connector belongs in the connector.
+4. Normalization Support for Destinations - if the connector is a destination and supports normalization, there's a small change required in the normalization code to update the config so that dbt uses the right credentials for the SSH tunnel.
 
 ## How To
 
@@ -21,6 +22,15 @@ Our SSH connector support is designed to be easy to plug into any existing conne
 ### Acceptance Testing
 1. The only difference between existing acceptance testing and acceptance testing with SSH is that the configuration that is used for testing needs to contain additional fields. You can see the `Postgres Source ssh key creds` in lastpass to see an example of what that might look like. Those credentials leverage an existing bastion host in our test infrastructure. (As future work, we want to get rid of the need to use a static bastion server and instead do it in docker so we can run it all locally.)
 
+### Normalization Support for Destinations
+1. The core functionality for ssh tunnelling with normalization is already in place but you'll need to add a small tweak to `transform_config/transform.py` in the normalization module. Find the function `transform_{connector}()` and add at the start:
+    ```
+    if TransformConfig.is_ssh_tunnelling(config):
+        config = TransformConfig.get_ssh_altered_config(config, port_key="port", host_key="host")
+    ```
+    Replace port_key and host_key as necessary. Look at `transform_postgres()` to see an example.
+2. If your `host_key="host"` and `port_key="port"` then step 1 should be sufficient. However if the key names differ for your connector, you will also need to add some logic into `sshtunneling.sh` (within airbyte-workers) to handle this, as currently it assumes that the keys are exactly `host` and `port`.
+
 ## Misc
 
 ### How to wrap the protocol in an SSH Tunnel

airbyte-integrations/bases/base-normalization/.dockerignore

@@ -1,6 +1,7 @@
 *
 !Dockerfile
 !entrypoint.sh
+!build/sshtunneling.sh
 !setup.py
 !normalization
 !dbt-project-template

airbyte-integrations/bases/base-normalization/Dockerfile

@@ -1,8 +1,27 @@
 FROM fishtownanalytics/dbt:0.19.0
+
+USER root
+WORKDIR /tmp
+RUN apt-get update && apt-get install -y \
+    wget \
+    unzip \
+    libaio-dev \
+    libaio1
+RUN mkdir -p /opt/oracle
+RUN wget https://download.oracle.com/otn_software/linux/instantclient/19600/instantclient-basic-linux.x64-19.6.0.0.0dbru.zip
+RUN unzip instantclient-basic-linux.x64-19.6.0.0.0dbru.zip -d /opt/oracle
+ENV ORACLE_HOME /opt/oracle/instantclient_19_6
+ENV LD_LIBRARY_PATH=$LD_LIBRARY_PATH:$ORACLE_HOME
+ENV TNS_ADMIN /opt/oracle/instantclient_19_6/network/admin
+RUN pip install cx_Oracle
+
 COPY --from=airbyte/base-airbyte-protocol-python:0.1.1 /airbyte /airbyte
 
+RUN apt-get update && apt-get install -y jq sshpass
+
 WORKDIR /airbyte
 COPY entrypoint.sh .
+COPY build/sshtunneling.sh .
 
 WORKDIR /airbyte/normalization_code
 COPY normalization ./normalization
@@ -14,8 +33,10 @@ RUN pip install .
 
 WORKDIR /airbyte/normalization_code
 RUN pip install .
+RUN pip install dbt-oracle==0.4.3
 RUN pip install git+https://github.com/dbeatty10/dbt-mysql@96655ea9f7fca7be90c9112ce8ffbb5aac1d3716#egg=dbt-mysql
 
+
 WORKDIR /airbyte/normalization_code/dbt-template/
 # Download external dbt dependencies
 RUN dbt deps
@@ -24,5 +45,5 @@ WORKDIR /airbyte
 ENV AIRBYTE_ENTRYPOINT "/airbyte/entrypoint.sh"
 ENTRYPOINT ["/airbyte/entrypoint.sh"]
 
-LABEL io.airbyte.version=0.1.42
+LABEL io.airbyte.version=0.1.44
 LABEL io.airbyte.name=airbyte/normalization

airbyte-integrations/bases/base-normalization/README.md

@@ -55,6 +55,7 @@ allowed characters, if quotes are needed or not, and the length limitations:
 - [redshift](../../../docs/integrations/destinations/redshift.md)
 - [snowflake](../../../docs/integrations/destinations/snowflake.md)
 - [mysql](../../../docs/integrations/destinations/mysql.md)
+- [oracle](../../../docs/integrations/destinations/oracle.md)
 
 Rules about truncations, for example for both of these strings which are too long for the postgres 64 limit:
 - `Aaaa_Bbbb_Cccc_Dddd_Eeee_Ffff_Gggg_Hhhh_Iiii`

airbyte-integrations/bases/base-normalization/build.gradle

@@ -8,9 +8,28 @@ airbytePython {
 }
 
 dependencies {
+    implementation project(':airbyte-workers')
     implementation files(project(':airbyte-integrations:bases:airbyte-protocol').airbyteDocker.outputs)
 }
 
+// we need to access the sshtunneling script from airbyte-workers for ssh support
+task copySshScript(type: Copy, dependsOn: [project(':airbyte-workers').processResources]) {
+    from "${project(':airbyte-workers').buildDir}/resources/main"
+    into "${buildDir}"
+    include "sshtunneling.sh"
+}
+
+// make sure the copy task above worked (if it fails, it fails silently annoyingly)
+task checkSshScriptCopy(type: Task, dependsOn: copySshScript) {
+    doFirst {
+        assert file("${buildDir}/sshtunneling.sh").exists() :
+                "Copy of sshtunneling.sh failed, check that it is present in airbyte-workers."
+    }
+}
+
+test.dependsOn checkSshScriptCopy
+assemble.dependsOn checkSshScriptCopy
+
 installReqs.dependsOn(":airbyte-integrations:bases:airbyte-protocol:installReqs")
 integrationTest.dependsOn(build)
 
@@ -24,6 +43,8 @@ task("customIntegrationTestPython", type: PythonTask, dependsOn: installTestReqs
     dependsOn ':airbyte-integrations:connectors:destination-postgres:airbyteDocker'
     dependsOn ':airbyte-integrations:connectors:destination-redshift:airbyteDocker'
     dependsOn ':airbyte-integrations:connectors:destination-snowflake:airbyteDocker'
+    dependsOn ':airbyte-integrations:connectors:destination-oracle:airbyteDocker'
+
 }
 
 integrationTest.dependsOn("customIntegrationTestPython")

airbyte-integrations/bases/base-normalization/dbt-project-template/dbt_project.yml

@@ -1,3 +1,6 @@
+# This file is necessary to install dbt-utils with dbt deps
+# the content will be overwritten by the transform function
+
 # Name your package! Package names should contain only lowercase characters
 # and underscores. A good package name should reflect your organization's
 # name or the intended use of these models

airbyte-integrations/bases/base-normalization/dbt-project-template/macros/cross_db_utils/array.sql

@@ -20,6 +20,10 @@
     cross join unnest({{ array_col }}) as {{ array_col }}
 {%- endmacro %}
 
+{% macro oracle__cross_join_unnest(stream_name, array_col) -%}
+    {% do exceptions.warn("Normalization does not support unnesting for Oracle yet.") %}
+{%- endmacro %}
+
 {% macro postgres__cross_join_unnest(stream_name, array_col) -%}
     cross join jsonb_array_elements(
         case jsonb_typeof({{ array_col }})
@@ -66,6 +70,10 @@
     _airbyte_nested_data
 {%- endmacro %}
 
+{% macro oracle__unnested_column_value(column_col) -%}
+    {{ column_col }}
+{%- endmacro %}
+
 {# unnest_cte -------------------------------------------------     #}
 
 {% macro unnest_cte(table_name, stream_name, column_col) -%}

airbyte-integrations/bases/base-normalization/dbt-project-template/macros/cross_db_utils/datatypes.sql

@@ -16,6 +16,10 @@
     jsonb
 {% endmacro %}
 
+{%- macro oracle__type_json() -%}
+    varchar2(4000)
+{%- endmacro -%}
+
 {% macro snowflake__type_json() %}
     variant
 {% endmacro %}
@@ -31,24 +35,39 @@
     char
 {%- endmacro -%}
 
+{%- macro oracle__type_string() -%}
+    varchar2(4000)
+{%- endmacro -%}
+
 
 {# float ------------------------------------------------- #}
 {% macro mysql__type_float() %}
     float
 {% endmacro %}
 
+{% macro oracle__type_float() %}
+    float
+{% endmacro %}
+
 
 {# int  ------------------------------------------------- #}
 {% macro default__type_int() %}
     signed
 {% endmacro %}
 
+{% macro oracle__type_int() %}
+    int
+{% endmacro %}
 
 {# bigint ------------------------------------------------- #}
 {% macro mysql__type_bigint() %}
     signed
 {% endmacro %}
 
+{% macro oracle__type_bigint() %}
+    numeric
+{% endmacro %}
+
 
 {# numeric ------------------------------------------------- #}
 {% macro mysql__type_numeric() %}
@@ -81,6 +100,10 @@
     char
 {%- endmacro -%}
 
+{% macro oracle__type_timestamp_with_timezone() %}
+    varchar2(4000)
+{% endmacro %}
+
 
 {# date  -------------------------------------------------     #}
 
@@ -91,3 +114,7 @@
 {% macro default__type_date() %}
     date
 {% endmacro %}
+
+{% macro oracle__type_date() %}
+    varchar2(4000)
+{% endmacro %}