Kube Queueing POC

.github/workflows/gradle.yml

@@ -286,41 +286,37 @@ jobs:
 
       - name: Run End-to-End Frontend Tests
         run: ./tools/bin/e2e_test.sh
+  test_kube:
+    runs-on: ubuntu-latest
+    name: Run Acceptance Tests (Kube)
+    steps:
+      - name: Checkout Airbyte
+        uses: actions/checkout@v2
 
-# DISABLED UNTIL WE HAVE TEMPORAL ON KUBE
-#  test_kube:
-#    runs-on: ubuntu-latest
-#    steps:
-#      - name: Checkout Airbyte
-#        uses: actions/checkout@v2
-#
-#      - uses: actions/setup-java@v1
-#        with:
-#          java-version: '14'
-#
-#      - uses: actions/setup-node@v1
-#        with:
-#          node-version: '14.7'
-#
-#      - uses: actions/setup-python@v2
-#        with:
-#          python-version: '3.7'
-#
-#      - name: Setup Minikube
-#        uses: manusa/actions-setup-minikube@v2.3.0
-#        with:
-#          minikube version: 'v1.16.0'
-#          kubernetes version: 'v1.19.2'
-#
-#      - name: Install socat
-#        run: sudo apt-get install socat
-#
-#      - name: Build Core Docker Images and Run Tests
-#        run: CORE_ONLY=true ./gradlew --no-daemon composeBuild test -x :airbyte-webapp:test --scan
-#        env:
-#          GIT_REVISION: ${{ github.sha }}
-#          CORE_ONLY: true
-#
-#      - name: Run Kubernetes End-to-End Acceptance Tests
-#        run: |
-#          ./tools/bin/acceptance_test_kube.sh
+      - uses: actions/setup-java@v1
+        with:
+          java-version: '14'
+
+      - uses: actions/setup-node@v1
+        with:
+          node-version: '14.7'
+
+      - uses: actions/setup-python@v2
+        with:
+          python-version: '3.7'
+
+      - name: Setup Minikube
+        uses: manusa/actions-setup-minikube@v2.4.1
+        with:
+          minikube version: 'v1.21.0-beta.0'
+          kubernetes version: 'v1.20.7'
+
+      - name: Install socat (required for port forwarding)
+        run: sudo apt-get install socat
+
+      - name: Build Core Docker Images and Run Tests
+        run: CORE_ONLY=true ./gradlew --no-daemon build --scan --rerun-tasks
+
+      - name: Run Kubernetes End-to-End Acceptance Tests
+        run: |
+          IS_MINIKUBE=true ./tools/bin/acceptance_test_kube.sh

airbyte-commons-docker/build.gradle

@@ -0,0 +1,11 @@
+plugins {
+    id "java-library"
+}
+
+dependencies {
+    implementation 'org.apache.commons:commons-compress:1.20'
+    implementation 'com.github.docker-java:docker-java:3.2.8'
+    implementation 'com.github.docker-java:docker-java-transport-httpclient5:3.2.8'
+
+    testImplementation 'org.apache.commons:commons-lang3:3.11'
+}

airbyte-commons/src/main/java/io/airbyte/commons/docker/DockerUtils.java → airbyte-commons-docker/src/main/java/io/airbyte/commons/docker/DockerUtils.java

@@ -24,10 +24,36 @@
 
 package io.airbyte.commons.docker;
 
+import com.github.dockerjava.api.DockerClient;
+import com.github.dockerjava.api.command.BuildImageResultCallback;
+import com.github.dockerjava.core.DefaultDockerClientConfig;
+import com.github.dockerjava.core.DockerClientConfig;
+import com.github.dockerjava.core.DockerClientImpl;
+import com.github.dockerjava.httpclient5.ApacheDockerHttpClient;
+import com.github.dockerjava.transport.DockerHttpClient;
+import java.io.File;
+import java.util.Set;
+
 public class DockerUtils {
 
+  private static final DockerClientConfig CONFIG = DefaultDockerClientConfig.createDefaultConfigBuilder().build();
+  private static final DockerHttpClient HTTP_CLIENT = new ApacheDockerHttpClient.Builder()
+      .dockerHost(CONFIG.getDockerHost())
+      .sslConfig(CONFIG.getSSLConfig())
+      .maxConnections(100)
+      .build();
+  private static final DockerClient DOCKER_CLIENT = DockerClientImpl.getInstance(CONFIG, HTTP_CLIENT);
+
   public static String getTaggedImageName(String dockerRepository, String tag) {
     return String.join(":", dockerRepository, tag);
   }
 
+  public static String buildImage(String dockerFilePath, String tag) {
+    return DOCKER_CLIENT.buildImageCmd()
+        .withDockerfile(new File(dockerFilePath))
+        .withTags(Set.of(tag))
+        .exec(new BuildImageResultCallback())
+        .awaitImageId();
+  }
+
 }

airbyte-commons/build.gradle

@@ -3,7 +3,6 @@ plugins {
 }
 
 dependencies {
-    testImplementation 'org.apache.commons:commons-lang3:3.11'
-
-    implementation group: 'org.apache.commons', name: 'commons-compress', version: '1.20'
+    implementation 'org.apache.commons:commons-compress:1.20'
+    implementation 'org.apache.commons:commons-lang3:3.11'
 }

airbyte-commons/src/main/java/io/airbyte/commons/string/Strings.java

@@ -26,6 +26,7 @@
 
 import com.google.common.collect.Streams;
 import java.util.stream.Collectors;
+import org.apache.commons.lang3.RandomStringUtils;
 
 public class Strings {
 
@@ -35,4 +36,8 @@ public static String join(Iterable<?> iterable, CharSequence separator) {
         .collect(Collectors.joining(separator));
   }
 
+  public static String addRandomSuffix(String base, String separator, int suffixLength) {
+    return base + separator + RandomStringUtils.randomAlphabetic(suffixLength).toLowerCase();
+  }
+
 }

airbyte-config/models/src/main/java/io/airbyte/config/Configs.java

@@ -25,6 +25,7 @@
 package io.airbyte.config;
 
 import java.nio.file.Path;
+import java.util.Set;
 
 public interface Configs {
 
@@ -62,6 +63,8 @@ public interface Configs {
 
   String getTemporalHost();
 
+  Set<Integer> getTemporalWorkerPorts();
+
   enum TrackingStrategy {
     SEGMENT,
     LOGGING

airbyte-config/models/src/main/java/io/airbyte/config/EnvConfigs.java

@@ -26,8 +26,11 @@
 
 import com.google.common.base.Preconditions;
 import java.nio.file.Path;
+import java.util.Arrays;
 import java.util.Optional;
+import java.util.Set;
 import java.util.function.Function;
+import java.util.stream.Collectors;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -53,6 +56,7 @@ public class EnvConfigs implements Configs {
   private static final String MAXIMUM_WORKSPACE_RETENTION_DAYS = "MAXIMUM_WORKSPACE_RETENTION_DAYS";
   private static final String MAXIMUM_WORKSPACE_SIZE_MB = "MAXIMUM_WORKSPACE_SIZE_MB";
   private static final String TEMPORAL_HOST = "TEMPORAL_HOST";
+  private static final String TEMPORAL_WORKER_PORTS = "TEMPORAL_WORKER_PORTS";
 
   private static final long DEFAULT_MINIMUM_WORKSPACE_RETENTION_DAYS = 1;
   private static final long DEFAULT_MAXIMUM_WORKSPACE_RETENTION_DAYS = 60;
@@ -166,6 +170,13 @@ public String getTemporalHost() {
     return getEnvOrDefault(TEMPORAL_HOST, "airbyte-temporal:7233");
   }
 
+  @Override
+  public Set<Integer> getTemporalWorkerPorts() {
+    return Arrays.stream(getEnvOrDefault(TEMPORAL_WORKER_PORTS, "").split(","))
+        .map(Integer::valueOf)
+        .collect(Collectors.toSet());
+  }
+
   private String getEnvOrDefault(String key, String defaultValue) {
     return getEnvOrDefault(key, defaultValue, Function.identity());
   }

airbyte-db/src/test/java/io/airbyte/db/PostgresUtilsTest.java

@@ -31,12 +31,12 @@
 import com.google.common.collect.ImmutableMap;
 import io.airbyte.commons.io.IOs;
 import io.airbyte.commons.json.Jsons;
+import io.airbyte.commons.string.Strings;
 import io.airbyte.db.jdbc.DefaultJdbcDatabase;
 import io.airbyte.db.jdbc.JdbcDatabase;
 import io.airbyte.test.utils.PostgreSQLContainerHelper;
 import java.sql.SQLException;
 import org.apache.commons.dbcp2.BasicDataSource;
-import org.apache.commons.lang3.RandomStringUtils;
 import org.junit.jupiter.api.BeforeAll;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
@@ -58,7 +58,7 @@ static void init() {
 
   @BeforeEach
   void setup() throws Exception {
-    final String dbName = "db_" + RandomStringUtils.randomAlphabetic(10).toLowerCase();
+    final String dbName = Strings.addRandomSuffix("db", "_", 10);
 
     final JsonNode config = getConfig(PSQL_DB, dbName);
 

airbyte-db/src/test/java/io/airbyte/db/jdbc/TestDefaultJdbcDatabase.java

@@ -31,13 +31,13 @@
 import com.google.common.collect.Lists;
 import io.airbyte.commons.io.IOs;
 import io.airbyte.commons.json.Jsons;
+import io.airbyte.commons.string.Strings;
 import io.airbyte.db.Databases;
 import io.airbyte.test.utils.PostgreSQLContainerHelper;
 import java.sql.SQLException;
 import java.util.List;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
-import org.apache.commons.lang3.RandomStringUtils;
 import org.junit.jupiter.api.AfterAll;
 import org.junit.jupiter.api.BeforeAll;
 import org.junit.jupiter.api.BeforeEach;
@@ -65,7 +65,7 @@ static void init() {
 
   @BeforeEach
   void setup() throws Exception {
-    final String dbName = "db_" + RandomStringUtils.randomAlphabetic(10).toLowerCase();
+    final String dbName = Strings.addRandomSuffix("db", "_", 10);
 
     config = getConfig(PSQL_DB, dbName);
 

airbyte-db/src/test/java/io/airbyte/db/jdbc/TestJdbcUtils.java

@@ -34,6 +34,7 @@
 import io.airbyte.commons.io.IOs;
 import io.airbyte.commons.json.Jsons;
 import io.airbyte.commons.stream.MoreStreams;
+import io.airbyte.commons.string.Strings;
 import io.airbyte.protocol.models.JsonSchemaPrimitive;
 import io.airbyte.test.utils.PostgreSQLContainerHelper;
 import java.math.BigDecimal;
@@ -48,7 +49,6 @@
 import java.util.Map;
 import java.util.stream.Collectors;
 import org.apache.commons.dbcp2.BasicDataSource;
-import org.apache.commons.lang3.RandomStringUtils;
 import org.junit.jupiter.api.BeforeAll;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
@@ -75,7 +75,7 @@ static void init() {
 
   @BeforeEach
   void setup() throws Exception {
-    final String dbName = "db_" + RandomStringUtils.randomAlphabetic(10).toLowerCase();
+    final String dbName = Strings.addRandomSuffix("db", "_", 10);
 
     final JsonNode config = getConfig(PSQL_DB, dbName);
 

airbyte-db/src/test/java/io/airbyte/db/jdbc/TestStreamingJdbcDatabase.java

@@ -38,6 +38,7 @@
 import io.airbyte.commons.functional.CheckedConsumer;
 import io.airbyte.commons.io.IOs;
 import io.airbyte.commons.json.Jsons;
+import io.airbyte.commons.string.Strings;
 import io.airbyte.test.utils.PostgreSQLContainerHelper;
 import java.sql.Connection;
 import java.sql.PreparedStatement;
@@ -47,7 +48,6 @@
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
 import org.apache.commons.dbcp2.BasicDataSource;
-import org.apache.commons.lang3.RandomStringUtils;
 import org.junit.jupiter.api.AfterAll;
 import org.junit.jupiter.api.BeforeAll;
 import org.junit.jupiter.api.BeforeEach;
@@ -79,7 +79,7 @@ static void init() {
   void setup() throws Exception {
     jdbcStreamingQueryConfiguration = mock(JdbcStreamingQueryConfiguration.class);
 
-    final String dbName = "db_" + RandomStringUtils.randomAlphabetic(10).toLowerCase();
+    final String dbName = Strings.addRandomSuffix("db", "_", 10);
 
     final JsonNode config = getConfig(PSQL_DB, dbName);
 

airbyte-integrations/connectors/destination-bigquery/src/test-integration/java/io/airbyte/integrations/destination/bigquery/BigQueryDestinationAcceptanceTest.java

@@ -43,6 +43,7 @@
 import com.google.common.collect.ImmutableMap;
 import com.google.common.collect.Maps;
 import io.airbyte.commons.json.Jsons;
+import io.airbyte.commons.string.Strings;
 import io.airbyte.integrations.base.JavaBaseConstants;
 import io.airbyte.integrations.destination.StandardNameTransformer;
 import io.airbyte.integrations.standardtest.destination.DestinationAcceptanceTest;
@@ -55,7 +56,6 @@
 import java.util.UUID;
 import java.util.stream.Collectors;
 import java.util.stream.StreamSupport;
-import org.apache.commons.lang3.RandomStringUtils;
 import org.apache.commons.lang3.tuple.ImmutablePair;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -190,7 +190,8 @@ protected void setup(TestDestinationEnv testEnv) throws Exception {
     final String projectId = credentialsJson.get(CONFIG_PROJECT_ID).asText();
     final String datasetLocation = "US";
 
-    final String datasetId = "airbyte_tests_" + RandomStringUtils.randomAlphanumeric(8);
+    final String datasetId = Strings.addRandomSuffix("airbyte_tests", "_", 8);
+
     config = Jsons.jsonNode(ImmutableMap.builder()
         .put(CONFIG_PROJECT_ID, projectId)
         .put(CONFIG_CREDS, credentialsJsonString)

airbyte-integrations/connectors/destination-bigquery/src/test-integration/java/io/airbyte/integrations/destination/bigquery/BigQueryDestinationTest.java

@@ -43,6 +43,7 @@
 import com.google.common.collect.Lists;
 import io.airbyte.commons.json.Jsons;
 import io.airbyte.commons.resources.MoreResources;
+import io.airbyte.commons.string.Strings;
 import io.airbyte.integrations.base.AirbyteMessageConsumer;
 import io.airbyte.integrations.base.Destination;
 import io.airbyte.integrations.base.JavaBaseConstants;
@@ -69,7 +70,6 @@
 import java.util.Set;
 import java.util.stream.Collectors;
 import java.util.stream.StreamSupport;
-import org.apache.commons.lang3.RandomStringUtils;
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
@@ -136,7 +136,7 @@ void setup(TestInfo info) throws IOException {
         .build()
         .getService();
 
-    final String datasetId = "airbyte_tests_" + RandomStringUtils.randomAlphanumeric(8);
+    final String datasetId = Strings.addRandomSuffix("airbyte_tests", "_", 8);
     final String datasetLocation = "EU";
     MESSAGE_USERS1.getRecord().setNamespace(datasetId);
     MESSAGE_USERS2.getRecord().setNamespace(datasetId);

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

@@ -28,6 +28,7 @@
 import com.fasterxml.jackson.databind.node.ObjectNode;
 import com.google.common.collect.ImmutableMap;
 import io.airbyte.commons.json.Jsons;
+import io.airbyte.commons.string.Strings;
 import io.airbyte.db.Database;
 import io.airbyte.db.Databases;
 import io.airbyte.integrations.base.JavaBaseConstants;
@@ -37,7 +38,6 @@
 import java.util.ArrayList;
 import java.util.List;
 import java.util.stream.Collectors;
-import org.apache.commons.lang3.RandomStringUtils;
 import org.jooq.JSONFormat;
 import org.jooq.JSONFormat.RecordFormat;
 import org.junit.jupiter.api.AfterAll;
@@ -166,7 +166,7 @@ private static Database getDatabase(JsonNode config) {
   @Override
   protected void setup(TestDestinationEnv testEnv) throws SQLException {
     configWithoutDbName = getConfig(db);
-    final String dbName = "db_" + RandomStringUtils.randomAlphabetic(10).toLowerCase();
+    final String dbName = Strings.addRandomSuffix("db", "_", 10);
 
     final Database database = getDatabase(configWithoutDbName);
     database.query(ctx -> {

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

@@ -28,6 +28,7 @@
 import com.fasterxml.jackson.databind.node.ObjectNode;
 import com.google.common.collect.ImmutableMap;
 import io.airbyte.commons.json.Jsons;
+import io.airbyte.commons.string.Strings;
 import io.airbyte.db.Database;
 import io.airbyte.db.Databases;
 import io.airbyte.integrations.base.JavaBaseConstants;
@@ -37,7 +38,6 @@
 import java.util.ArrayList;
 import java.util.List;
 import java.util.stream.Collectors;
-import org.apache.commons.lang3.RandomStringUtils;
 import org.jooq.JSONFormat;
 import org.jooq.JSONFormat.RecordFormat;
 import org.junit.jupiter.api.AfterAll;
@@ -175,7 +175,7 @@ private static Database getDatabase(JsonNode config) {
   @Override
   protected void setup(TestDestinationEnv testEnv) throws SQLException {
     configWithoutDbName = getConfig(db);
-    final String dbName = "db_" + RandomStringUtils.randomAlphabetic(10).toLowerCase();
+    final String dbName = Strings.addRandomSuffix("db", "_", 10);
 
     final Database database = getDatabase(configWithoutDbName);
     database.query(ctx -> {