🎉 introduce automatic migration at the startup of server for docker e…

…nvironment (airbytehq#3980) * introduce automatic migration at the startup of server * handle versions with non-zero patch * it works!!! * add dummy data * cleanup orphan configs * add more assertions * format + add comments * move migration acceptance test to acceptance test directory * add automatic migration test to the build * address review comments * missed out on these * format * add more assertions * format * fix test * format * use default port for temporal * move seed to server + introduce atomice replacement for config * make tests better * remove unwanted changes * move atomic replacement logic behind persistence + pass path to latest seeds * format * update seeds * review comments * update seeds * merge latest seeds with configs * fix bug around latest seed * update seed * update seed * seeds should be populated by separate container * address review comment + change latest definition url * update seeds * format * update seed references * update seed * update seed * update seed * update seed references * update seed references + add Migration Acceptance Test * update seed container in kube + disable automatic migration for kube + update docs * update docs * address review comments from Michel * update doc * temporary commmit to see if build becomes green * delete seeds from airbyte config + undo temp commit
frankfanslc · Jun 29, 2021 · 8877528 · 8877528
1 parent a8513af
commit 8877528
Show file tree

Hide file tree

Showing 214 changed files with 2,686 additions and 153 deletions.
diff --git a/.github/workflows/gradle.yml b/.github/workflows/gradle.yml
@@ -122,6 +122,9 @@ jobs:
       - name: Run End-to-End Acceptance Tests
         run: ./tools/bin/acceptance_test.sh
 
+      - name: Automatic Migration Acceptance Test
+        run: MIGRATION_TEST_VERSION=$(grep VERSION .env | tr  -d "VERSION=") ./gradlew --no-daemon :airbyte-tests:automaticMigrationAcceptanceTest --rerun-tasks --scan -i
+
       - name: Push Core Docker Images
         if: success() && github.ref == 'refs/heads/master'
         run: |

diff --git a/airbyte-commons/src/main/java/io/airbyte/commons/version/AirbyteVersion.java b/airbyte-commons/src/main/java/io/airbyte/commons/version/AirbyteVersion.java
@@ -136,4 +136,17 @@ public String toString() {
         '}';
   }
 
+  public static AirbyteVersion versionWithoutPatch(AirbyteVersion airbyteVersion) {
+    String versionWithoutPatch = "" + airbyteVersion.getMajorVersion()
+        + "."
+        + airbyteVersion.getMinorVersion()
+        + ".0-"
+        + airbyteVersion.getVersion().replace("\n", "").strip().split("-")[1];
+    return new AirbyteVersion(versionWithoutPatch);
+  }
+
+  public static AirbyteVersion versionWithoutPatch(String airbyteVersion) {
+    return versionWithoutPatch(new AirbyteVersion(airbyteVersion));
+  }
+
 }
diff --git a/airbyte-config/init/Dockerfile b/airbyte-config/init/Dockerfile
@@ -5,4 +5,3 @@ WORKDIR /app
 # the sole purpose of this image is to seed the data volume with the default data
 # that the app should have when it is first installed.
 COPY scripts scripts
-COPY src/main/resources/config seed/config
diff --git a/airbyte-config/init/build.gradle b/airbyte-config/init/build.gradle
@@ -7,42 +7,3 @@ dependencies {
 
     implementation project(':airbyte-config:models')
 }
-
-// generate seed for each yaml file.
-task generateSeed {
-    def seeds = [
-            [
-                "sourceDefinitionId",
-                new File(project.projectDir, '/src/main/resources/seed/source_definitions.yaml'),
-                new File(project.projectDir, '/src/main/resources/config/STANDARD_SOURCE_DEFINITION')
-            ],
-            [
-                "destinationDefinitionId",
-                new File(project.projectDir, '/src/main/resources/seed/destination_definitions.yaml'),
-                new File(project.projectDir, '/src/main/resources/config/STANDARD_DESTINATION_DEFINITION')
-            ],
-    ]
-    seeds.each{val ->
-        def name = val[0]
-        def taskName = "generateSeed$name"
-        dependsOn taskName
-        task "$taskName"(type: JavaExec) {
-            classpath = sourceSets.main.runtimeClasspath
-
-            main = 'io.airbyte.config.init.SeedRepository'
-
-            // arguments to pass to the application
-            args '--id-name'
-            args val[0]
-            args '--input-path'
-            args val[1]
-            args '--output-path'
-            args val[2]
-        }
-    }
-}
-
-// we only want to attempt generateSeed if tests have passed.
-generateSeed.dependsOn(check)
-generateSeed.dependsOn(assemble)
-build.dependsOn(generateSeed)
diff --git a/airbyte-config/models/src/main/java/io/airbyte/config/ConfigSchema.java b/airbyte-config/models/src/main/java/io/airbyte/config/ConfigSchema.java
@@ -27,45 +27,102 @@
 import io.airbyte.commons.json.JsonSchemas;
 import java.io.File;
 import java.nio.file.Path;
+import java.util.function.Function;
 
 public enum ConfigSchema {
 
   // workspace
-  STANDARD_WORKSPACE("StandardWorkspace.yaml"),
+  STANDARD_WORKSPACE("StandardWorkspace.yaml", StandardWorkspace.class, standardWorkspace -> {
+    return standardWorkspace.getWorkspaceId().toString();
+  }),
 
   // source
-  STANDARD_SOURCE_DEFINITION("StandardSourceDefinition.yaml"),
-  SOURCE_CONNECTION("SourceConnection.yaml"),
+  STANDARD_SOURCE_DEFINITION("StandardSourceDefinition.yaml", StandardSourceDefinition.class,
+      standardSourceDefinition -> {
+        return standardSourceDefinition.getSourceDefinitionId().toString();
+      }),
+  SOURCE_CONNECTION("SourceConnection.yaml", SourceConnection.class,
+      sourceConnection -> {
+        return sourceConnection.getSourceId().toString();
+      }),
 
   // destination
-  STANDARD_DESTINATION_DEFINITION("StandardDestinationDefinition.yaml"),
-  DESTINATION_CONNECTION("DestinationConnection.yaml"),
+  STANDARD_DESTINATION_DEFINITION("StandardDestinationDefinition.yaml",
+      StandardDestinationDefinition.class, standardDestinationDefinition -> {
+        return standardDestinationDefinition.getDestinationDefinitionId().toString();
+      }),
+  DESTINATION_CONNECTION("DestinationConnection.yaml", DestinationConnection.class,
+      destinationConnection -> {
+        return destinationConnection.getDestinationId().toString();
+      }),
 
   // sync
-  STANDARD_SYNC("StandardSync.yaml"),
-  STANDARD_SYNC_OPERATION("StandardSyncOperation.yaml"),
-  STANDARD_SYNC_SUMMARY("StandardSyncSummary.yaml"),
+  STANDARD_SYNC("StandardSync.yaml", StandardSync.class, standardSync -> {
+    return standardSync.getConnectionId().toString();
+  }),
+  STANDARD_SYNC_OPERATION("StandardSyncOperation.yaml", StandardSyncOperation.class,
+      standardSyncOperation -> {
+        return standardSyncOperation.getOperationId().toString();
+      }),
+  STANDARD_SYNC_SUMMARY("StandardSyncSummary.yaml", StandardSyncSummary.class,
+      standardSyncSummary -> {
+        throw new RuntimeException("StandardSyncSummary doesn't have an id");
+      }),
 
   // worker
-  STANDARD_SYNC_INPUT("StandardSyncInput.yaml"),
-  NORMALIZATION_INPUT("NormalizationInput.yaml"),
-  OPERATOR_DBT_INPUT("OperatorDbtInput.yaml"),
+  STANDARD_SYNC_INPUT("StandardSyncInput.yaml", StandardSyncInput.class,
+      standardSyncInput -> {
+        throw new RuntimeException("StandardSyncInput doesn't have an id");
+      }),
+  NORMALIZATION_INPUT("NormalizationInput.yaml", NormalizationInput.class,
+      normalizationInput -> {
+        throw new RuntimeException("NormalizationInput doesn't have an id");
+      }),
+  OPERATOR_DBT_INPUT("OperatorDbtInput.yaml", OperatorDbtInput.class,
+      operatorDbtInput -> {
+        throw new RuntimeException("OperatorDbtInput doesn't have an id");
+      }),
 
-  STANDARD_SYNC_OUTPUT("StandardSyncOutput.yaml"),
-  REPLICATION_OUTPUT("ReplicationOutput.yaml"),
+  STANDARD_SYNC_OUTPUT("StandardSyncOutput.yaml", StandardSyncOutput.class,
+      standardWorkspace -> {
+        throw new RuntimeException("StandardSyncOutput doesn't have an id");
+      }),
+  REPLICATION_OUTPUT("ReplicationOutput.yaml", ReplicationOutput.class,
+      standardWorkspace -> {
+        throw new RuntimeException("ReplicationOutput doesn't have an id");
+      }),
 
-  STATE("State.yaml");
+  STATE("State.yaml", State.class, standardWorkspace -> {
+    throw new RuntimeException("State doesn't have an id");
+  });
 
   static final Path KNOWN_SCHEMAS_ROOT = JsonSchemas.prepareSchemas("types", ConfigSchema.class);
 
   private final String schemaFilename;
+  private final Class<?> className;
+  private final Function<?, String> extractId;
 
-  ConfigSchema(final String schemaFilename) {
+  <T> ConfigSchema(final String schemaFilename,
+                   Class<T> className,
+                   Function<T, String> extractId) {
     this.schemaFilename = schemaFilename;
+    this.className = className;
+    this.extractId = extractId;
   }
 
   public File getFile() {
     return KNOWN_SCHEMAS_ROOT.resolve(schemaFilename).toFile();
   }
 
+  public <T> Class<T> getClassName() {
+    return (Class<T>) className;
+  }
+
+  public <T> String getId(T object) {
+    if (getClassName().isInstance(object)) {
+      return ((Function<T, String>) extractId).apply(object);
+    }
+    throw new RuntimeException("Object: " + object + " is not instance of class " + getClassName().getName());
+  }
+
 }
diff --git a/...yte-config/persistence/src/main/java/io/airbyte/config/persistence/ConfigPersistence.java b/...yte-config/persistence/src/main/java/io/airbyte/config/persistence/ConfigPersistence.java
@@ -24,10 +24,13 @@
 
 package io.airbyte.config.persistence;
 
+import com.fasterxml.jackson.databind.JsonNode;
 import io.airbyte.config.ConfigSchema;
 import io.airbyte.validation.json.JsonValidationException;
 import java.io.IOException;
 import java.util.List;
+import java.util.Map;
+import java.util.stream.Stream;
 
 public interface ConfigPersistence {
 
@@ -37,4 +40,8 @@ public interface ConfigPersistence {
 
   <T> void writeConfig(ConfigSchema configType, String configId, T config) throws JsonValidationException, IOException;
 
+  <T> void replaceAllConfigs(Map<ConfigSchema, Stream<T>> configs, boolean dryRun) throws IOException;
+
+  Map<String, Stream<JsonNode>> dumpConfigs() throws IOException;
+
 }
diff --git a/airbyte-config/persistence/src/main/java/io/airbyte/config/persistence/ConfigRepository.java b/airbyte-config/persistence/src/main/java/io/airbyte/config/persistence/ConfigRepository.java
@@ -24,6 +24,7 @@
 
 package io.airbyte.config.persistence;
 
+import com.fasterxml.jackson.databind.JsonNode;
 import io.airbyte.commons.lang.MoreBooleans;
 import io.airbyte.config.ConfigSchema;
 import io.airbyte.config.DestinationConnection;
@@ -37,7 +38,9 @@
 import java.io.IOException;
 import java.util.ArrayList;
 import java.util.List;
+import java.util.Map;
 import java.util.UUID;
+import java.util.stream.Stream;
 
 public class ConfigRepository {
 
@@ -203,4 +206,12 @@ public List<StandardSyncOperation> listStandardSyncOperations() throws IOExcepti
     return persistence.listConfigs(ConfigSchema.STANDARD_SYNC_OPERATION, StandardSyncOperation.class);
   }
 
+  public <T> void replaceAllConfigs(Map<ConfigSchema, Stream<T>> configs, boolean dryRun) throws IOException {
+    persistence.replaceAllConfigs(configs, dryRun);
+  }
+
+  public Map<String, Stream<JsonNode>> dumpConfigs() throws IOException {
+    return persistence.dumpConfigs();
+  }
+
 }