[Ticket 6435] Source-ClickHouse: added ssl support

airbyte-integrations/connectors/source-clickhouse/ReadMe.md

@@ -0,0 +1,5 @@
+for ssl test custom image is used. To push it use
+docker build -t airbyte/clickhouse-with-ssl:dev -f Clickhouse.Dockerfile
+
+
+under the tools\integration-tests-ssl dir

airbyte-integrations/connectors/source-clickhouse/build.gradle

@@ -22,5 +22,5 @@ dependencies {
     integrationTestJavaImplementation project(':airbyte-integrations:bases:standard-source-test')
     integrationTestJavaImplementation project(':airbyte-integrations:connectors:source-clickhouse')
     integrationTestJavaImplementation testFixtures(project(':airbyte-integrations:connectors:source-jdbc'))
-    integrationTestJavaImplementation "org.testcontainers:clickhouse:1.15.3"
+    integrationTestJavaImplementation "org.testcontainers:clickhouse:1.16.0"
 }

airbyte-integrations/connectors/source-clickhouse/src/main/java/io/airbyte/integrations/source/clickhouse/ClickHouseSource.java

@@ -35,6 +35,10 @@ public class ClickHouseSource extends AbstractJdbcSource implements Source {
    * https://clickhouse.tech/docs/en/operations/system-tables/columns/ to fetch the primary keys.
    */
 
+  public static final List<String> SSL_PARAMETERS = List.of(
+      "ssl=true",
+      "sslmode=none");
+
   @Override
   protected Map<String, List<String>> discoverPrimaryKeys(JdbcDatabase database,
                                                           List<TableInfo<CommonField<JDBCType>>> tableInfos) {
@@ -73,13 +77,20 @@ public ClickHouseSource() {
 
   @Override
   public JsonNode toDatabaseConfig(JsonNode config) {
+    final StringBuilder jdbcUrl = new StringBuilder(String.format("jdbc:clickhouse://%s:%s/%s",
+        config.get("host").asText(),
+        config.get("port").asText(),
+        config.get("database").asText()));
+
+    // assume ssl if not explicitly mentioned.
+    if (!config.has("ssl") || config.get("ssl").asBoolean()) {
+      jdbcUrl.append("?").append(String.join("&", SSL_PARAMETERS));
+    }
+
     return Jsons.jsonNode(ImmutableMap.builder()
         .put("username", config.get("username").asText())
         .put("password", config.get("password").asText())
-        .put("jdbc_url", String.format("jdbc:clickhouse://%s:%s/%s",
-            config.get("host").asText(),
-            config.get("port").asText(),
-            config.get("database").asText()))
+        .put("jdbc_url", jdbcUrl.toString())
         .build());
   }
 

airbyte-integrations/connectors/source-clickhouse/src/main/resources/spec.json

@@ -32,6 +32,12 @@
         "description": "Password associated with the username.",
         "type": "string",
         "airbyte_secret": true
+      },
+      "ssl": {
+        "title": "SSL Connection",
+        "description": "Encrypt data using SSL.",
+        "type": "boolean",
+        "default": true
       }
     }
   }

airbyte-integrations/connectors/source-clickhouse/src/test-integration/java/io/airbyte/integrations/io/airbyte/integration_tests/sources/ClickHouseJdbcSourceAcceptanceTest.java

@@ -76,7 +76,7 @@ public String primaryKeyClause(final List<String> columns) {
   @Override
   @BeforeEach
   public void setup() throws Exception {
-    db = new ClickHouseContainer("yandex/clickhouse-server:21.3.10.1-alpine");
+    db = new ClickHouseContainer("yandex/clickhouse-server:21.8.8.29-alpine");
     db.start();
 
     config = Jsons.jsonNode(ImmutableMap.builder()
@@ -85,6 +85,7 @@ public void setup() throws Exception {
         .put("database", SCHEMA_NAME)
         .put("username", db.getUsername())
         .put("password", db.getPassword())
+        .put("ssl", false)
         .build());
 
     super.setup();

airbyte-integrations/connectors/source-clickhouse/src/test-integration/java/io/airbyte/integrations/io/airbyte/integration_tests/sources/ClickHouseJdbcStressTest.java

@@ -31,7 +31,7 @@ public Optional<String> getDefaultSchemaName() {
   @Override
   @BeforeEach
   public void setup() throws Exception {
-    db = new ClickHouseContainer("yandex/clickhouse-server:21.3.10.1-alpine");
+    db = new ClickHouseContainer("yandex/clickhouse-server:21.8.8.29-alpine");
     db.start();
 
     config = Jsons.jsonNode(ImmutableMap.builder()
@@ -40,6 +40,7 @@ public void setup() throws Exception {
         .put("database", SCHEMA_NAME)
         .put("username", db.getUsername())
         .put("password", db.getPassword())
+        .put("ssl", false)
         .build());
 
     super.setup();

airbyte-integrations/connectors/source-clickhouse/src/test-integration/java/io/airbyte/integrations/io/airbyte/integration_tests/sources/ClickHouseSourceAcceptanceTest.java

@@ -86,7 +86,7 @@ protected List<String> getRegexTests() {
 
   @Override
   protected void setupEnvironment(TestDestinationEnv environment) throws Exception {
-    db = new ClickHouseContainer("yandex/clickhouse-server:21.3.10.1-alpine");
+    db = new ClickHouseContainer("yandex/clickhouse-server:21.8.8.29-alpine");
     db.start();
 
     config = Jsons.jsonNode(ImmutableMap.builder()
@@ -95,6 +95,7 @@ protected void setupEnvironment(TestDestinationEnv environment) throws Exception
         .put("database", SCHEMA_NAME)
         .put("username", db.getUsername())
         .put("password", db.getPassword())
+        .put("ssl", false)
         .build());
 
     JdbcDatabase database = Databases.createJdbcDatabase(

airbyte-integrations/connectors/source-clickhouse/src/test-integration/java/io/airbyte/integrations/io/airbyte/integration_tests/sources/SslClickHouseJdbcSourceAcceptanceTest.java

@@ -0,0 +1,98 @@
+/*
+ * Copyright (c) 2021 Airbyte, Inc., all rights reserved.
+ */
+
+package io.airbyte.integrations.io.airbyte.integration_tests.sources;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.google.common.collect.ImmutableMap;
+import io.airbyte.commons.json.Jsons;
+import io.airbyte.integrations.source.clickhouse.ClickHouseSource;
+import io.airbyte.integrations.source.jdbc.AbstractJdbcSource;
+import io.airbyte.integrations.source.jdbc.test.JdbcSourceAcceptanceTest;
+import java.sql.SQLException;
+import java.util.List;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.testcontainers.containers.GenericContainer;
+
+public class SslClickHouseJdbcSourceAcceptanceTest extends JdbcSourceAcceptanceTest {
+
+  private static final String SCHEMA_NAME = "default";
+  private GenericContainer db;
+  private JsonNode config;
+
+  @Override
+  public boolean supportsSchemas() {
+    return false;
+  }
+
+  @Override
+  public JsonNode getConfig() {
+    return Jsons.clone(config);
+  }
+
+  @Override
+  public String getDriverClass() {
+    return ClickHouseSource.DRIVER_CLASS;
+  }
+
+  @Override
+  public String createTableQuery(final String tableName, final String columnClause, final String primaryKeyClause) {
+    // ClickHouse requires Engine to be mentioned as part of create table query.
+    // Refer : https://clickhouse.tech/docs/en/engines/table-engines/ for more information
+    return String.format("CREATE TABLE %s(%s) %s",
+        tableName, columnClause, primaryKeyClause.equals("") ? "Engine = TinyLog"
+            : "ENGINE = MergeTree() ORDER BY " + primaryKeyClause + " PRIMARY KEY "
+                + primaryKeyClause);
+  }
+
+  @Override
+  @AfterEach
+  public void tearDown() throws SQLException {
+    db.close();
+    db.stop();
+    super.tearDown();
+  }
+
+  @Override
+  public String primaryKeyClause(final List<String> columns) {
+    if (columns.isEmpty()) {
+      return "";
+    }
+
+    final StringBuilder clause = new StringBuilder();
+    clause.append("(");
+    for (int i = 0; i < columns.size(); i++) {
+      clause.append(columns.get(i));
+      if (i != (columns.size() - 1)) {
+        clause.append(",");
+      }
+    }
+    clause.append(")");
+    return clause.toString();
+  }
+
+  @Override
+  @BeforeEach
+  public void setup() throws Exception {
+    db = new GenericContainer("airbyte/clickhouse-with-ssl:dev").withExposedPorts(8443);
+    db.start();
+
+    config = Jsons.jsonNode(ImmutableMap.builder()
+        .put("host", db.getHost())
+        .put("port", db.getFirstMappedPort())
+        .put("database", SCHEMA_NAME)
+        .put("username", "default")
+        .put("password", "")
+        .build());
+
+    super.setup();
+  }
+
+  @Override
+  public AbstractJdbcSource getJdbcSource() {
+    return new ClickHouseSource();
+  }
+
+}

tools/integrations-test-ssl/Clickhouse.Dockerfile

@@ -0,0 +1,9 @@
+FROM yandex/clickhouse-server:latest
+
+EXPOSE 8443
+EXPOSE 9000
+EXPOSE 8123
+EXPOSE 9009
+
+COPY clickhouse_certs.sh /docker-entrypoint-initdb.d/
+

tools/integrations-test-ssl/clickhouse_certs.sh

@@ -0,0 +1,15 @@
+echo "Preparing certs"
+
+openssl req -subj "/CN=my.host.name" -new \
+    -newkey rsa:2048 -days 365 -nodes -x509 \
+    -keyout /etc/clickhouse-server/server.key \
+    -out /etc/clickhouse-server/server.crt
+
+openssl dhparam -out /etc/clickhouse-server/dhparam.pem 1024
+
+chown $(id -u clickhouse):$(id -g clickhouse) /etc/clickhouse-server/server.{key,crt}
+
+echo "<yandex><https_port>8443</https_port></yandex>" > /etc/clickhouse-server/config.d/https.xml
+
+
+echo "Finished preparing certs"