Move config volume wait for fs persistence construction

airbytehq · Jul 20, 2021 · f1be721 · f1be721
1 parent 667ba9b
commit f1be721
Show file tree

Hide file tree

Showing 4 changed files with 20 additions and 35 deletions.
diff --git a/...fig/persistence/src/main/java/io/airbyte/config/persistence/ConfigPersistenceBuilder.java b/...fig/persistence/src/main/java/io/airbyte/config/persistence/ConfigPersistenceBuilder.java
@@ -80,19 +80,6 @@ ConfigPersistence create() throws IOException, InterruptedException {
     return getDbPersistenceWithFileSeed();
   }
 
-  ConfigPersistence getFileSystemPersistence() throws InterruptedException {
-    Path configRoot = configs.getConfigRoot();
-    LOGGER.info("Constructing file system config persistence (root: {})", configRoot);
-    while (true) {
-      if (Files.exists(configRoot.resolve("config"))) {
-        LOGGER.info("Config volume is ready");
-        return FileSystemConfigPersistence.createWithValidation(configRoot);
-      }
-      LOGGER.warn("Config volume is not ready yet");
-      Thread.sleep(3000);
-    }
-  }
-
   /**
    * Create the database config persistence and load it with the initial seed from the YAML seed files
    * if the database should be initialized.
@@ -109,7 +96,7 @@ ConfigPersistence getDbPersistenceWithYamlSeed() throws IOException {
    */
   ConfigPersistence getDbPersistenceWithFileSeed() throws IOException, InterruptedException {
     LOGGER.info("Creating db-based config persistence, and loading seed and existing data from files");
-    ConfigPersistence fsConfigPersistence = getFileSystemPersistence();
+    ConfigPersistence fsConfigPersistence = FileSystemConfigPersistence.createWithValidation(configs.getConfigRoot());
     return getDbPersistence(fsConfigPersistence);
   }
 

diff --git a/.../persistence/src/main/java/io/airbyte/config/persistence/FileSystemConfigPersistence.java b/.../persistence/src/main/java/io/airbyte/config/persistence/FileSystemConfigPersistence.java
@@ -47,8 +47,9 @@
 public class FileSystemConfigPersistence implements ConfigPersistence {
 
   private static final Logger LOGGER = LoggerFactory.getLogger(FileSystemConfigPersistence.class);
-  private static final String CONFIG_DIR = "config";
+  public static final String CONFIG_DIR = "config";
   private static final String TMP_DIR = "tmp_storage";
+  private static final int INTERVAL_WAITING_SECONDS = 3;
 
   private static final Object lock = new Object();
 
@@ -57,7 +58,18 @@ public class FileSystemConfigPersistence implements ConfigPersistence {
   // root for where configs are stored
   private final Path configRoot;
 
-  public static ConfigPersistence createWithValidation(final Path storageRoot) {
+  public static ConfigPersistence createWithValidation(final Path storageRoot) throws InterruptedException {
+    LOGGER.info("Constructing file system config persistence (root: {})", storageRoot);
+
+    Path configRoot = storageRoot.resolve(CONFIG_DIR);
+    int totalWaitingSeconds = 0;
+    while (!Files.exists(configRoot)) {
+      LOGGER.warn("Config volume is not ready yet (waiting time: {} s)", totalWaitingSeconds);
+      Thread.sleep(INTERVAL_WAITING_SECONDS * 1000);
+      totalWaitingSeconds += INTERVAL_WAITING_SECONDS;
+    }
+    LOGGER.info("Config volume is ready (waiting time: {} s)", totalWaitingSeconds);
+
     return new ValidatingConfigPersistence(new FileSystemConfigPersistence(storageRoot));
   }
 

diff --git a/...persistence/src/test/java/io/airbyte/config/persistence/ConfigPersistenceBuilderTest.java b/...persistence/src/test/java/io/airbyte/config/persistence/ConfigPersistenceBuilderTest.java
@@ -155,20 +155,6 @@ public void testCreateDbPersistenceWithoutSetupDatabase() throws Exception {
         dbPersistence.dumpConfigs());
   }
 
-  @Test
-  public void testCreateFileSystemConfigPersistence() throws Exception {
-    Path testRoot = Path.of("/tmp/cpf_test_file_system");
-    Path rootPath = Files.createTempDirectory(Files.createDirectories(testRoot), ConfigPersistenceBuilderTest.class.getName());
-    ConfigPersistence seedPersistence = new FileSystemConfigPersistence(rootPath);
-    writeSource(seedPersistence, SOURCE_GITHUB);
-    writeDestination(seedPersistence, DESTINATION_S3);
-
-    when(configs.getConfigRoot()).thenReturn(rootPath);
-
-    ConfigPersistence filePersistence = new ConfigPersistenceBuilder(configs, false).getFileSystemPersistence();
-    assertSameConfigDump(seedPersistence.dumpConfigs(), filePersistence.dumpConfigs());
-  }
-
   /**
    * This test mimics the file -> db config persistence migration process.
    */
@@ -188,10 +174,10 @@ public void testMigrateFromFileToDbPersistence() throws Exception {
     // first run uses file system config persistence, and adds an extra workspace
     Path testRoot = Path.of("/tmp/cpf_test_migration");
     Path storageRoot = Files.createTempDirectory(Files.createDirectories(testRoot), ConfigPersistenceBuilderTest.class.getName());
-    Files.createDirectories(storageRoot.resolve("config"));
+    Files.createDirectories(storageRoot.resolve(FileSystemConfigPersistence.CONFIG_DIR));
     when(configs.getConfigRoot()).thenReturn(storageRoot);
 
-    ConfigPersistence filePersistence = new ConfigPersistenceBuilder(configs, false).getFileSystemPersistence();
+    ConfigPersistence filePersistence = FileSystemConfigPersistence.createWithValidation(storageRoot);
 
     filePersistence.replaceAllConfigs(seedConfigs, false);
     filePersistence.writeConfig(ConfigSchema.STANDARD_WORKSPACE, extraWorkspace.getWorkspaceId().toString(), extraWorkspace);

diff --git a/airbyte-server/src/test/java/io/airbyte/server/migration/RunMigrationTest.java b/airbyte-server/src/test/java/io/airbyte/server/migration/RunMigrationTest.java
@@ -117,7 +117,7 @@ public void testRunMigration() {
     }
   }
 
-  private void assertPreMigrationConfigs(Path configRoot, JobPersistence jobPersistence) throws IOException, JsonValidationException {
+  private void assertPreMigrationConfigs(Path configRoot, JobPersistence jobPersistence) throws Exception {
     assertDatabaseVersion(jobPersistence, INITIAL_VERSION);
     ConfigRepository configRepository = new ConfigRepository(FileSystemConfigPersistence.createWithValidation(configRoot));
     Map<String, StandardSourceDefinition> sourceDefinitionsBeforeMigration = configRepository.listStandardSources().stream()
@@ -132,7 +132,7 @@ private void assertDatabaseVersion(JobPersistence jobPersistence, String version
     assertEquals(versionFromDb.get(), version);
   }
 
-  private void assertPostMigrationConfigs(Path importRoot) throws IOException, JsonValidationException, ConfigNotFoundException {
+  private void assertPostMigrationConfigs(Path importRoot) throws Exception {
     final ConfigRepository configRepository = new ConfigRepository(FileSystemConfigPersistence.createWithValidation(importRoot));
     final StandardSyncOperation standardSyncOperation = assertSyncOperations(configRepository);
     assertStandardSyncs(configRepository, standardSyncOperation);
@@ -293,7 +293,7 @@ private void assertDestinations(ConfigRepository configRepository) throws JsonVa
     }
   }
 
-  private void runMigration(JobPersistence jobPersistence, Path configRoot) throws IOException {
+  private void runMigration(JobPersistence jobPersistence, Path configRoot) throws Exception {
     try (RunMigration runMigration = new RunMigration(
         INITIAL_VERSION,
         jobPersistence,