🎉 Migrate config persistence to database

.env

@@ -1,6 +1,6 @@
 VERSION=0.27.2-alpha
 
-# Airbyte Internal Database, see https://docs.airbyte.io/operator-guides/configuring-airbyte-db
+# Airbyte Internal Job Database, see https://docs.airbyte.io/operator-guides/configuring-airbyte-db
 DATABASE_USER=docker
 DATABASE_PASSWORD=docker
 DATABASE_HOST=db
@@ -9,6 +9,15 @@ DATABASE_DB=airbyte
 # translate manually DATABASE_URL=jdbc:postgresql://${DATABASE_HOST}:${DATABASE_PORT/${DATABASE_DB}
 DATABASE_URL=jdbc:postgresql://db:5432/airbyte
 
+# By default, the Config Database is the same as the Job Database.
+# Uncomment the following variable if a different Config Database is needed.
+# CONFIG_DATABASE_USER=docker
+# CONFIG_DATABASE_PASSWORD=docker
+# CONFIG_DATABASE_HOST=db
+# CONFIG_DATABASE_PORT=5432
+# CONFIG_DATABASE_DB=airbyte
+# CONFIG_DATABASE_URL=jdbc:postgresql://db:5432/airbyte
+
 # When using the airbyte-db via default docker image:
 CONFIG_ROOT=/data
 DATA_DOCKER_MOUNT=airbyte_data

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

@@ -0,0 +1,6 @@
+- workspaceId: 5ae6b09b-fdec-41af-aaf7-7d94cfc33ef6
+  name: default
+  slug: default
+  initialSetupComplete: false
+  displaySetupWizard: true
+  tombstone: false

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

@@ -47,6 +47,12 @@ public interface Configs {
 
   String getDatabaseUrl();
 
+  String getConfigDatabaseUser();
+
+  String getConfigDatabasePassword();
+
+  String getConfigDatabaseUrl();
+
   String getWebappUrl();
 
   String getWorkspaceDockerMount();

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

@@ -52,6 +52,9 @@ public class EnvConfigs implements Configs {
   public static final String DATABASE_USER = "DATABASE_USER";
   public static final String DATABASE_PASSWORD = "DATABASE_PASSWORD";
   public static final String DATABASE_URL = "DATABASE_URL";
+  public static final String CONFIG_DATABASE_USER = "CONFIG_DATABASE_USER";
+  public static final String CONFIG_DATABASE_PASSWORD = "CONFIG_DATABASE_PASSWORD";
+  public static final String CONFIG_DATABASE_URL = "CONFIG_DATABASE_URL";
   public static final String WEBAPP_URL = "WEBAPP_URL";
   private static final String MINIMUM_WORKSPACE_RETENTION_DAYS = "MINIMUM_WORKSPACE_RETENTION_DAYS";
   private static final String MAXIMUM_WORKSPACE_RETENTION_DAYS = "MAXIMUM_WORKSPACE_RETENTION_DAYS";
@@ -127,6 +130,24 @@ public String getDatabaseUrl() {
     return getEnsureEnv(DATABASE_URL);
   }
 
+  @Override
+  public String getConfigDatabaseUser() {
+    // Default to reuse the job database
+    return getEnvOrDefault(CONFIG_DATABASE_USER, getDatabaseUser());
+  }
+
+  @Override
+  public String getConfigDatabasePassword() {
+    // Default to reuse the job database
+    return getEnvOrDefault(CONFIG_DATABASE_PASSWORD, getDatabasePassword());
+  }
+
+  @Override
+  public String getConfigDatabaseUrl() {
+    // Default to reuse the job database
+    return getEnvOrDefault(CONFIG_DATABASE_URL, getDatabaseUrl());
+  }
+
   @Override
   public String getWebappUrl() {
     return getEnsureEnv(WEBAPP_URL);
@@ -255,10 +276,10 @@ private long getEnvOrDefault(String key, long defaultValue) {
 
   private <T> T getEnvOrDefault(String key, T defaultValue, Function<String, T> parser) {
     final String value = getEnv.apply(key);
-    if (value != null) {
+    if (value != null && !value.isEmpty()) {
       return parser.apply(value);
     } else {
-      LOGGER.info(key + " not found, defaulting to " + defaultValue);
+      LOGGER.info(key + " not found or empty, defaulting to " + defaultValue);
       return defaultValue;
     }
   }

airbyte-config/persistence/build.gradle

@@ -1,6 +1,10 @@
 dependencies {
     implementation group: 'commons-io', name: 'commons-io', version: '2.7'
 
+    implementation project(':airbyte-db')
     implementation project(':airbyte-config:models')
-    implementation project(":airbyte-json-validation")
+    implementation project(':airbyte-config:init')
+    implementation project(':airbyte-json-validation')
+
+    testImplementation "org.testcontainers:postgresql:1.15.1"
 }

airbyte-config/persistence/src/main/java/io/airbyte/config/persistence/AirbyteConfigsTable.java

@@ -0,0 +1,47 @@
+/*
+ * MIT License
+ *
+ * Copyright (c) 2020 Airbyte
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy
+ * of this software and associated documentation files (the "Software"), to deal
+ * in the Software without restriction, including without limitation the rights
+ * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+ * copies of the Software, and to permit persons to whom the Software is
+ * furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in all
+ * copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+ * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+ * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+ * SOFTWARE.
+ */
+
+package io.airbyte.config.persistence;
+
+import static org.jooq.impl.DSL.field;
+import static org.jooq.impl.DSL.table;
+
+import java.sql.Timestamp;
+import org.jooq.Field;
+import org.jooq.JSONB;
+import org.jooq.Record;
+import org.jooq.Table;
+
+public class AirbyteConfigsTable {
+
+  public static final String AIRBYTE_CONFIGS_TABLE_SCHEMA = "airbyte_configs_table.sql";
+
+  public static final Table<Record> AIRBYTE_CONFIGS = table("airbyte_configs");
+  public static final Field<String> CONFIG_ID = field("config_id", String.class);
+  public static final Field<String> CONFIG_TYPE = field("config_type", String.class);
+  public static final Field<JSONB> CONFIG_BLOB = field("config_blob", JSONB.class);
+  public static final Field<Timestamp> CREATED_AT = field("created_at", Timestamp.class);
+  public static final Field<Timestamp> UPDATED_AT = field("updated_at", Timestamp.class);
+
+}

airbyte-config/persistence/src/main/java/io/airbyte/config/persistence/ConfigPersistenceFactory.java

@@ -0,0 +1,151 @@
+/*
+ * MIT License
+ *
+ * Copyright (c) 2020 Airbyte
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy
+ * of this software and associated documentation files (the "Software"), to deal
+ * in the Software without restriction, including without limitation the rights
+ * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+ * copies of the Software, and to permit persons to whom the Software is
+ * furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in all
+ * copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+ * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+ * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+ * SOFTWARE.
+ */
+
+package io.airbyte.config.persistence;
+
+import static io.airbyte.config.persistence.AirbyteConfigsTable.AIRBYTE_CONFIGS_TABLE_SCHEMA;
+
+import io.airbyte.commons.resources.MoreResources;
+import io.airbyte.config.Configs;
+import io.airbyte.db.Database;
+import io.airbyte.db.Databases;
+import java.io.IOException;
+import java.nio.file.Path;
+import java.util.function.Function;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * By default, this factory returns a database config persistence. it can still return a file system
+ * config persistence for testing purpose. This legacy feature should be removed after the file to
+ * database migration is completely done.
+ */
+public class ConfigPersistenceFactory {
+
+  private static final Logger LOGGER = LoggerFactory.getLogger(ConfigPersistenceFactory.class);
+
+  private final Configs configs;
+  private final boolean setupDatabase;
+  private final boolean useConfigDatabase;
+
+  /**
+   * @param setupDatabase initialize the database and load data; this is necessary because this method
+   *        has multiple callers, and we want to setup the database only once to prevent race
+   *        conditions.
+   */
+  private ConfigPersistenceFactory(Configs configs, boolean setupDatabase, boolean useConfigDatabase) {
+    this.configs = configs;
+    this.setupDatabase = setupDatabase;
+    this.useConfigDatabase = useConfigDatabase;
+  }
+
+  public static Builder build(Configs configs) {
+    return new Builder(configs);
+  }
+
+  public static class Builder {
+
+    private final Configs configs;
+    private boolean setupDatabase = true;
+    private boolean useConfigDatabase = true;
+
+    private Builder(Configs configs) {
+      this.configs = configs;
+    }
+
+    public Builder setupDatabase(boolean setupDatabase) {
+      this.setupDatabase = setupDatabase;
+      return this;
+    }
+
+    public Builder useConfigDatabase(boolean useConfigDatabase) {
+      this.useConfigDatabase = useConfigDatabase;
+      return this;
+    }
+
+    public ConfigPersistenceFactory get() {
+      return new ConfigPersistenceFactory(configs, setupDatabase, useConfigDatabase);
+    }
+
+  }
+
+  /**
+   * Create a config persistence based on the configs.
+   * <p/>
+   * If config root is defined, create a database config persistence and copy the configs from the
+   * file-based config persistence. Otherwise, seed the database from the yaml files.
+   */
+  public ConfigPersistence create() throws IOException {
+    if (!useConfigDatabase) {
+      Path configRoot = configs.getConfigRoot();
+      LOGGER.info("Use file system config persistence (root: {})", configRoot);
+      return FileSystemConfigPersistence.createWithValidation(configRoot);
+    }
+
+    if (configs.getConfigRoot() == null) {
+      // This branch will only be true in a future Airbyte version, in which
+      // the config root is no longer required and everything lives in the database.
+      return createDbPersistenceWithYamlSeed();
+    }
+
+    return createDbPersistenceWithFileSeed();
+  }
+
+  ConfigPersistence createDbPersistenceWithYamlSeed() throws IOException {
+    ConfigPersistence seedConfigPersistence = new YamlSeedConfigPersistence();
+    return createDbPersistence(seedConfigPersistence);
+  }
+
+  ConfigPersistence createDbPersistenceWithFileSeed() throws IOException {
+    Path configRoot = configs.getConfigRoot();
+    ConfigPersistence fsConfigPersistence = FileSystemConfigPersistence.createWithValidation(configRoot);
+    return createDbPersistence(fsConfigPersistence);
+  }
+
+  ConfigPersistence createDbPersistence(ConfigPersistence seedConfigPersistence) throws IOException {
+    LOGGER.info("Use database config persistence.");
+
+    // When setupDatabase is true, it means the database will be initialized after we
+    // connect to the database. So the database itself is considered ready as long as
+    // the connection is alive. Otherwise, the database is expected to have full data.
+    Function<Database, Boolean> isReady = setupDatabase
+        ? Databases.IS_CONFIG_DATABASE_CONNECTED
+        : Databases.IS_CONFIG_DATABASE_LOADED_WITH_DATA;
+
+    Database database = Databases.createPostgresDatabaseWithRetry(
+        configs.getConfigDatabaseUser(),
+        configs.getConfigDatabasePassword(),
+        configs.getConfigDatabaseUrl(),
+        isReady);
+
+    DatabaseConfigPersistence dbConfigPersistence = new DatabaseConfigPersistence(database);
+    if (setupDatabase) {
+      dbConfigPersistence.initialize(MoreResources.readResource(AIRBYTE_CONFIGS_TABLE_SCHEMA));
+      dbConfigPersistence.loadData(seedConfigPersistence);
+    }
+
+    return new ValidatingConfigPersistence(dbConfigPersistence);
+  }
+
+}