🎉 Destination MSSQL: Added support for connection via SSH tunnels (#5…

airbyte-config/init/src/main/resources/config/STANDARD_DESTINATION_DEFINITION/d4353156-9217-4cad-8dd7-c108fd4f74cf.json

@@ -2,6 +2,6 @@
   "destinationDefinitionId": "d4353156-9217-4cad-8dd7-c108fd4f74cf",
   "name": "MS SQL Server",
   "dockerRepository": "airbyte/destination-mssql",
-  "dockerImageTag": "0.1.8",
+  "dockerImageTag": "0.1.9",
   "documentationUrl": "https://docs.airbyte.io/integrations/destinations/mssql"
 }

airbyte-config/init/src/main/resources/seed/destination_definitions.yaml

@@ -68,7 +68,7 @@
 - destinationDefinitionId: d4353156-9217-4cad-8dd7-c108fd4f74cf
   name: MS SQL Server
   dockerRepository: airbyte/destination-mssql
-  dockerImageTag: 0.1.8
+  dockerImageTag: 0.1.9
   documentationUrl: https://docs.airbyte.io/integrations/destinations/mssql
 - destinationDefinitionId: 3986776d-2319-4de9-8af8-db14c0996e72
   name: Oracle (Alpha)

airbyte-integrations/connectors/destination-mssql/Dockerfile

@@ -8,5 +8,5 @@ COPY build/distributions/${APPLICATION}*.tar ${APPLICATION}.tar
 
 RUN tar xf ${APPLICATION}.tar --strip-components=1
 
-LABEL io.airbyte.version=0.1.8
+LABEL io.airbyte.version=0.1.9
 LABEL io.airbyte.name=airbyte/destination-mssql

airbyte-integrations/connectors/destination-mssql/build.gradle

@@ -21,6 +21,7 @@ dependencies {
     testImplementation "org.testcontainers:mssqlserver:1.15.3"
 
     integrationTestJavaImplementation project(':airbyte-integrations:bases:standard-destination-test')
+    integrationTestJavaImplementation project(':airbyte-integrations:connectors:destination-mssql')
 
     implementation files(project(':airbyte-integrations:bases:base-java').airbyteDocker.outputs)
     integrationTestJavaImplementation files(project(':airbyte-integrations:bases:base-normalization').airbyteDocker.outputs)

airbyte-integrations/connectors/destination-mssql/src/main/java/io/airbyte/integrations/destination/mssql/MSSQLDestination.java

@@ -9,6 +9,7 @@
 import io.airbyte.commons.json.Jsons;
 import io.airbyte.integrations.base.Destination;
 import io.airbyte.integrations.base.IntegrationRunner;
+import io.airbyte.integrations.base.ssh.SshWrappedDestination;
 import io.airbyte.integrations.destination.jdbc.AbstractJdbcDestination;
 import java.io.File;
 import java.util.ArrayList;
@@ -22,6 +23,8 @@ public class MSSQLDestination extends AbstractJdbcDestination implements Destina
   private static final Logger LOGGER = LoggerFactory.getLogger(MSSQLDestination.class);
 
   public static final String DRIVER_CLASS = "com.microsoft.sqlserver.jdbc.SQLServerDriver";
+  public static final List<String> HOST_KEY = List.of("host");
+  public static final List<String> PORT_KEY = List.of("port");
 
   public MSSQLDestination() {
     super(DRIVER_CLASS, new MSSQLNameTransformer(), new SqlServerOperations());
@@ -89,7 +92,7 @@ private void readSsl(JsonNode config, List<String> additionalParameters) {
   }
 
   public static void main(String[] args) throws Exception {
-    final Destination destination = new MSSQLDestination();
+    final Destination destination = new SshWrappedDestination(new MSSQLDestination(), HOST_KEY, PORT_KEY);
     LOGGER.info("starting destination: {}", MSSQLDestination.class);
     new IntegrationRunner(destination).run(args);
     LOGGER.info("completed destination: {}", MSSQLDestination.class);

airbyte-integrations/connectors/destination-mssql/src/test-integration/java/io/airbyte/integrations/destination/mssql/SshKeyMSSQLDestinationAcceptanceTest.java

@@ -0,0 +1,15 @@
+/*
+ * Copyright (c) 2021 Airbyte, Inc., all rights reserved.
+ */
+
+package io.airbyte.integrations.destination.mssql;
+
+import io.airbyte.integrations.base.ssh.SshTunnel;
+
+public class SshKeyMSSQLDestinationAcceptanceTest extends SshMSSQLDestinationAcceptanceTest {
+
+  @Override
+  public SshTunnel.TunnelMethod getTunnelMethod() {
+    return SshTunnel.TunnelMethod.SSH_KEY_AUTH;
+  }
+}

airbyte-integrations/connectors/destination-mssql/src/test-integration/java/io/airbyte/integrations/destination/mssql/SshMSSQLDestinationAcceptanceTest.java

@@ -0,0 +1,178 @@
+/*
+ * Copyright (c) 2021 Airbyte, Inc., all rights reserved.
+ */
+
+package io.airbyte.integrations.destination.mssql;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.node.ObjectNode;
+import com.google.common.collect.ImmutableMap;
+import io.airbyte.commons.functional.CheckedFunction;
+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.SshBastionContainer;
+import io.airbyte.integrations.base.ssh.SshTunnel;
+import io.airbyte.integrations.destination.ExtendedNameTransformer;
+import io.airbyte.integrations.standardtest.destination.DestinationAcceptanceTest;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Objects;
+import java.util.stream.Collectors;
+import org.apache.commons.lang3.RandomStringUtils;
+import org.jooq.JSONFormat;
+import org.jooq.JSONFormat.RecordFormat;
+import org.testcontainers.containers.JdbcDatabaseContainer;
+import org.testcontainers.containers.MSSQLServerContainer;
+import org.testcontainers.containers.Network;
+
+/**
+ * Abstract class that allows us to avoid duplicating testing logic for testing SSH with a key file or with a password.
+ */
+public abstract class SshMSSQLDestinationAcceptanceTest extends DestinationAcceptanceTest {
+
+  private static final JSONFormat JSON_FORMAT = new JSONFormat().recordFormat(RecordFormat.OBJECT);
+
+  private final ExtendedNameTransformer namingResolver = new ExtendedNameTransformer();
+
+  private final String schemaName = RandomStringUtils.randomAlphabetic(8).toLowerCase();
+  private static final String database = "test";
+  private static MSSQLServerContainer<?> db;
+  private final SshBastionContainer bastion = new SshBastionContainer();
+
+  public abstract SshTunnel.TunnelMethod getTunnelMethod();
+
+  @Override
+  protected String getImageName() {
+    return "airbyte/destination-mssql:dev";
+  }
+
+  @Override
+  protected JsonNode getConfig() throws Exception {
+    return bastion.getTunnelConfig(getTunnelMethod(), getMSSQLDbConfigBuilder(db));
+  }
+
+  @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(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 supportsDBT() {
+    return true;
+  }
+
+  @Override
+  protected boolean implementsNamespaces() {
+    return true;
+  }
+
+  @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;
+  }
+
+  public ImmutableMap.Builder<Object, Object> getMSSQLDbConfigBuilder(JdbcDatabaseContainer<?> db) {
+    return ImmutableMap.builder()
+        .put("host", Objects.requireNonNull(db.getContainerInfo().getNetworkSettings()
+            .getNetworks()
+            .get(((Network.NetworkImpl) bastion.getNetWork()).getName())
+            .getIpAddress()))
+        .put("username", db.getUsername())
+        .put("password", db.getPassword())
+        .put("port", db.getExposedPorts().get(0))
+        .put("database", database)
+        .put("schema", schemaName)
+        .put("ssl", false);
+  }
+
+  private static Database getDatabaseFromConfig(final JsonNode config) {
+    return Databases.createDatabase(
+        config.get("username").asText(),
+        config.get("password").asText(),
+        String.format("jdbc:sqlserver://%s:%s",
+            config.get("host").asText(),
+            config.get("port").asInt()),
+        "com.microsoft.sqlserver.jdbc.SQLServerDriver",
+        null);
+  }
+
+  private List<JsonNode> retrieveRecordsFromTable(final String tableName, final String schemaName) throws Exception {
+    var schema = schemaName == null ? this.schemaName : schemaName;
+    final JsonNode config = getConfig();
+    return SshTunnel.sshWrap(
+        config,
+        MSSQLDestination.HOST_KEY,
+        MSSQLDestination.PORT_KEY,
+        (CheckedFunction<JsonNode, List<JsonNode>, Exception>) mangledConfig -> getDatabaseFromConfig(mangledConfig)
+            .query(
+                ctx -> ctx
+                    .fetch(String.format("USE %s;"
+                            + "SELECT * FROM %s.%s ORDER BY %s ASC;",
+                        database, 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 {
+    startTestContainers();
+
+    SshTunnel.sshWrap(
+        getConfig(),
+        MSSQLDestination.HOST_KEY,
+        MSSQLDestination.PORT_KEY,
+        mangledConfig -> {
+          getDatabaseFromConfig(mangledConfig).query(ctx -> {
+            ctx.fetch(String.format("CREATE DATABASE %s;", database));
+            ctx.fetch(String.format("USE %s;", database));
+            ctx.fetch(String.format("CREATE SCHEMA %s;", schemaName));
+
+            return null;
+          });
+        });
+  }
+
+  private void startTestContainers() {
+    bastion.initAndStartBastion();
+    initAndStartJdbcContainer();
+  }
+
+  private void initAndStartJdbcContainer() {
+    db = new MSSQLServerContainer<>("mcr.microsoft.com/mssql/server:2017-latest")
+        .withNetwork(bastion.getNetWork())
+        .acceptLicense();
+    db.start();
+  }
+
+  @Override
+  protected void tearDown(final TestDestinationEnv testEnv) {
+    bastion.stopAndCloseContainers(db);
+  }
+
+}

airbyte-integrations/connectors/destination-mssql/src/test-integration/java/io/airbyte/integrations/destination/mssql/SshPasswordMSSQLDestinationAcceptanceTest.java

@@ -0,0 +1,16 @@
+/*
+ * Copyright (c) 2021 Airbyte, Inc., all rights reserved.
+ */
+
+package io.airbyte.integrations.destination.mssql;
+
+import io.airbyte.integrations.base.ssh.SshTunnel;
+
+public class SshPasswordMSSQLDestinationAcceptanceTest extends SshMSSQLDestinationAcceptanceTest {
+
+  @Override
+  public SshTunnel.TunnelMethod getTunnelMethod() {
+    return SshTunnel.TunnelMethod.SSH_PASSWORD_AUTH;
+  }
+
+}