Feature: Ability to save/load secrets through google secrets manager.

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

@@ -48,4 +48,6 @@ public interface AirbyteConfig {
    */
   File getConfigSchemaFile();
 
+  boolean isSecret();
+
 }

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

@@ -35,44 +35,42 @@ public enum ConfigSchema implements AirbyteConfig {
   STANDARD_WORKSPACE("StandardWorkspace.yaml",
       StandardWorkspace.class,
       standardWorkspace -> standardWorkspace.getWorkspaceId().toString(),
-      "workspaceId"),
+      "workspaceId", false),
 
   // source
   STANDARD_SOURCE_DEFINITION("StandardSourceDefinition.yaml",
       StandardSourceDefinition.class,
       standardSourceDefinition -> standardSourceDefinition.getSourceDefinitionId().toString(),
-      "sourceDefinitionId"),
+      "sourceDefinitionId", false),
   SOURCE_CONNECTION("SourceConnection.yaml",
       SourceConnection.class,
       sourceConnection -> sourceConnection.getSourceId().toString(),
-      "sourceId"),
+      "sourceId", true),
 
   // destination
   STANDARD_DESTINATION_DEFINITION("StandardDestinationDefinition.yaml",
       StandardDestinationDefinition.class,
       standardDestinationDefinition -> standardDestinationDefinition.getDestinationDefinitionId().toString(),
-      "destinationDefinitionId"),
+      "destinationDefinitionId", false),
   DESTINATION_CONNECTION("DestinationConnection.yaml",
       DestinationConnection.class,
       destinationConnection -> destinationConnection.getDestinationId().toString(),
-      "destinationId"),
+      "destinationId", true),
 
   // sync
   STANDARD_SYNC("StandardSync.yaml",
       StandardSync.class,
       standardSync -> standardSync.getConnectionId().toString(),
-      "connectionId"),
+      "connectionId", false),
   STANDARD_SYNC_OPERATION("StandardSyncOperation.yaml",
       StandardSyncOperation.class,
-      standardSyncOperation -> standardSyncOperation.getOperationId().toString(),
-      "operationId"),
-
+      standardSyncOperation -> standardSyncOperation.getOperationId().toString(), "operationId", false),
   SOURCE_OAUTH_PARAM("SourceOAuthParameter.yaml", SourceOAuthParameter.class,
       sourceOAuthParameter -> sourceOAuthParameter.getOauthParameterId().toString(),
-      "oauthParameterId"),
+      "oauthParameterId", false),
   DESTINATION_OAUTH_PARAM("DestinationOAuthParameter.yaml", DestinationOAuthParameter.class,
       destinationOAuthParameter -> destinationOAuthParameter.getOauthParameterId().toString(),
-      "oauthParameterId"),
+      "oauthParameterId", false),
 
   STANDARD_SYNC_SUMMARY("StandardSyncSummary.yaml", StandardSyncSummary.class),
 
@@ -90,21 +88,25 @@ public enum ConfigSchema implements AirbyteConfig {
   private final Class<?> className;
   private final Function<?, String> extractId;
   private final String idFieldName;
+  private final boolean isSecret;
 
   <T> ConfigSchema(final String schemaFilename,
                    Class<T> className,
                    Function<T, String> extractId,
-                   String idFieldName) {
+                   String idFieldName,
+                   boolean isSecret) {
     this.schemaFilename = schemaFilename;
     this.className = className;
     this.extractId = extractId;
     this.idFieldName = idFieldName;
+    this.isSecret = isSecret;
   }
 
   <T> ConfigSchema(final String schemaFilename,
                    Class<T> className) {
     this.schemaFilename = schemaFilename;
     this.className = className;
+    this.isSecret = false;
     this.extractId = object -> {
       throw new RuntimeException(className.getSimpleName() + " doesn't have an id");
     };
@@ -120,6 +122,11 @@ public <T> Class<T> getClassName() {
     return (Class<T>) className;
   }
 
+  @Override
+  public boolean isSecret() {
+    return isSecret;
+  }
+
   @Override
   public <T> String getId(T object) {
     if (getClassName().isInstance(object)) {

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

@@ -30,6 +30,7 @@
 import java.io.IOException;
 import java.util.List;
 import java.util.Map;
+import java.util.Set;
 import java.util.stream.Stream;
 
 public interface ConfigPersistence {
@@ -46,6 +47,10 @@ public interface ConfigPersistence {
 
   Map<String, Stream<JsonNode>> dumpConfigs() throws IOException;
 
-  void loadData(ConfigPersistence seedPersistence) throws IOException;
+  void loadData(ConfigPersistence seedPersistence, Set<String> connectorRepositoriesInUse) throws IOException;
+
+  Set<String> getRepositoriesFromDefinitionIds(Set<String> usedConnectorDefinitionIds) throws IOException;
+
+  Set<String> listDefinitionIdsInUseByConnectors() throws IOException;
 
 }

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

@@ -42,19 +42,31 @@
 import io.airbyte.validation.json.JsonValidationException;
 import java.io.IOException;
 import java.util.ArrayList;
+import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
 import java.util.Optional;
+import java.util.Set;
 import java.util.UUID;
 import java.util.stream.Stream;
 
 public class ConfigRepository {
 
   private final ConfigPersistence persistence;
+  private final ConfigPersistence secretsPersistence;
 
+  public ConfigRepository(final ConfigPersistence persistence, final ConfigPersistence secretsPersistence) {
+    this.persistence = persistence;
+    this.secretsPersistence = secretsPersistence;
+  }
+
+  /**
+   * This should only be used for backward compatibility where a secret store isn't yet in use.
+   */
   public ConfigRepository(final ConfigPersistence persistence) {
     this.persistence = persistence;
+    this.secretsPersistence = persistence;
   }
 
   public StandardWorkspace getStandardWorkspace(final UUID workspaceId, final boolean includeTombstone)
@@ -168,7 +180,7 @@ public void writeStandardDestinationDefinition(final StandardDestinationDefiniti
   }
 
   public SourceConnection getSourceConnection(final UUID sourceId) throws JsonValidationException, IOException, ConfigNotFoundException {
-    return persistence.getConfig(ConfigSchema.SOURCE_CONNECTION, sourceId.toString(), SourceConnection.class);
+    return secretsPersistence.getConfig(ConfigSchema.SOURCE_CONNECTION, sourceId.toString(), SourceConnection.class);
   }
 
   public void writeSourceConnection(final SourceConnection source, final ConnectorSpecification connectorSpecification)
@@ -177,16 +189,16 @@ public void writeSourceConnection(final SourceConnection source, final Connector
     final JsonSchemaValidator validator = new JsonSchemaValidator();
     validator.ensure(connectorSpecification.getConnectionSpecification(), source.getConfiguration());
 
-    persistence.writeConfig(ConfigSchema.SOURCE_CONNECTION, source.getSourceId().toString(), source);
+    secretsPersistence.writeConfig(ConfigSchema.SOURCE_CONNECTION, source.getSourceId().toString(), source);
   }
 
   public List<SourceConnection> listSourceConnection() throws JsonValidationException, IOException {
-    return persistence.listConfigs(ConfigSchema.SOURCE_CONNECTION, SourceConnection.class);
+    return secretsPersistence.listConfigs(ConfigSchema.SOURCE_CONNECTION, SourceConnection.class);
   }
 
   public DestinationConnection getDestinationConnection(final UUID destinationId)
       throws JsonValidationException, IOException, ConfigNotFoundException {
-    return persistence.getConfig(ConfigSchema.DESTINATION_CONNECTION, destinationId.toString(), DestinationConnection.class);
+    return secretsPersistence.getConfig(ConfigSchema.DESTINATION_CONNECTION, destinationId.toString(), DestinationConnection.class);
   }
 
   public void writeDestinationConnection(final DestinationConnection destinationConnection, final ConnectorSpecification connectorSpecification)
@@ -195,11 +207,11 @@ public void writeDestinationConnection(final DestinationConnection destinationCo
     final JsonSchemaValidator validator = new JsonSchemaValidator();
     validator.ensure(connectorSpecification.getConnectionSpecification(), destinationConnection.getConfiguration());
 
-    persistence.writeConfig(ConfigSchema.DESTINATION_CONNECTION, destinationConnection.getDestinationId().toString(), destinationConnection);
+    secretsPersistence.writeConfig(ConfigSchema.DESTINATION_CONNECTION, destinationConnection.getDestinationId().toString(), destinationConnection);
   }
 
   public List<DestinationConnection> listDestinationConnection() throws JsonValidationException, IOException {
-    return persistence.listConfigs(ConfigSchema.DESTINATION_CONNECTION, DestinationConnection.class);
+    return secretsPersistence.listConfigs(ConfigSchema.DESTINATION_CONNECTION, DestinationConnection.class);
   }
 
   public StandardSync getStandardSync(final UUID connectionId) throws JsonValidationException, IOException, ConfigNotFoundException {
@@ -277,15 +289,58 @@ public List<DestinationOAuthParameter> listDestinationOAuthParam() throws JsonVa
   }
 
   public void replaceAllConfigs(final Map<AirbyteConfig, Stream<?>> configs, final boolean dryRun) throws IOException {
-    persistence.replaceAllConfigs(configs, dryRun);
+    // if we're using a single persistence layer, this is the easy route.
+    if (persistence == secretsPersistence) {
+      persistence.replaceAllConfigs(configs, dryRun);
+      return;
+    }
+
+    // Or if we're using secrets storage, split into sets by where they should be stored.
+    Map<AirbyteConfig, Stream<?>> secretConfigs = new LinkedHashMap<AirbyteConfig, Stream<?>>();
+    Map<AirbyteConfig, Stream<?>> nonsecretConfigs = new LinkedHashMap<AirbyteConfig, Stream<?>>();
+    for (AirbyteConfig configType : configs.keySet()) {
+      if (configType.isSecret()) {
+        secretConfigs.put(configType, configs.get(configType));
+      } else {
+        nonsecretConfigs.put(configType, configs.get(configType));
+      }
+    }
+    // And store each type in its own persistence store.
+    persistence.replaceAllConfigs(nonsecretConfigs, dryRun);
+    secretsPersistence.replaceAllConfigs(secretConfigs, dryRun);
   }
 
   public Map<String, Stream<JsonNode>> dumpConfigs() throws IOException {
-    return persistence.dumpConfigs();
+    Map<String, Stream<JsonNode>> bothConfigTypes = persistence.dumpConfigs();
+    // Secrets and nonsecrets might be stored separately, and if they are, go looking for the rest.
+    // No need to do that if they're in the same single store, because we don't need duplicates or
+    // double work.
+    if (secretsPersistence != persistence) {
+      bothConfigTypes.putAll(secretsPersistence.dumpConfigs());
+    }
+    return bothConfigTypes;
+  }
+
+  /**
+   * During migration, this runs through the list of connectors, determining their docker image and
+   * version, for purposes of skipping upgrading in-use definitions.
+   *
+   * @return
+   */
+  public Set<String> listDefinitionsInUseByConnectors() throws IOException {
+    // Once we have secrets coordinates as references instead of storing the whole config, this lookup
+    // will be simpler again.
+    Set<String> definitionIds;
+    if (secretsPersistence != persistence) {
+      definitionIds = secretsPersistence.listDefinitionIdsInUseByConnectors();
+    } else {
+      definitionIds = persistence.listDefinitionIdsInUseByConnectors();
+    }
+    return persistence.getRepositoriesFromDefinitionIds(definitionIds);
   }
 
-  public void loadData(ConfigPersistence seedPersistence) throws IOException {
-    persistence.loadData(seedPersistence);
+  public void loadData(ConfigPersistence seedPersistence, Set<String> connectorRepositoriesInUse) throws IOException {
+    persistence.loadData(seedPersistence, connectorRepositoriesInUse);
   }
 
 }

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

@@ -100,13 +100,24 @@ public void migrateFileConfigs(Configs serverConfigs) throws IOException {
   }
 
   @Override
-  public void loadData(ConfigPersistence seedConfigPersistence) throws IOException {
+  public void loadData(ConfigPersistence seedConfigPersistence, Set<String> connectorRepositoriesInUse) throws IOException {
     database.transaction(ctx -> {
-      updateConfigsFromSeed(ctx, seedConfigPersistence);
+      final boolean isInitialized = ctx.fetchExists(AIRBYTE_CONFIGS);
+      if (isInitialized) {
+        updateConfigsFromSeed(ctx, seedConfigPersistence, connectorRepositoriesInUse);
+      } else {
+        copyConfigsFromSeed(ctx, seedConfigPersistence);
+      }
       return null;
     });
   }
 
+  /*
+   * public void loadData(ConfigPersistence seedConfigPersistence) throws IOException {
+   * database.transaction(ctx -> { updateConfigsFromSeed(ctx, seedConfigPersistence); return null; });
+   * }
+   */
+
   public ValidatingConfigPersistence withValidation() {
     return new ValidatingConfigPersistence(this);
   }
@@ -312,11 +323,10 @@ private ConnectorCounter(int newCount, int updateCount) {
   }
 
   @VisibleForTesting
-  void updateConfigsFromSeed(DSLContext ctx, ConfigPersistence seedConfigPersistence) throws SQLException {
-    LOGGER.info("Updating connector definitions from the seed if necessary...");
+  void updateConfigsFromSeed(DSLContext ctx, ConfigPersistence seedConfigPersistence, Set<String> connectorRepositoriesInUse) throws SQLException {
+    LOGGER.info("Config database has been initialized; updating connector definitions from the seed if necessary...");
 
     try {
-      Set<String> connectorRepositoriesInUse = getConnectorRepositoriesInUse(ctx);
       LOGGER.info("Connectors in use: {}", connectorRepositoriesInUse);
 
       Map<String, ConnectorInfo> connectorRepositoryToInfoMap = getConnectorRepositoryToInfoMap(ctx);
@@ -474,31 +484,50 @@ Map<String, ConnectorInfo> getConnectorRepositoryToInfoMap(DSLContext ctx) {
    *         connectors can be added manually by users, and their config ids are not always the same
    *         as those in the seed.
    */
-  private Set<String> getConnectorRepositoriesInUse(DSLContext ctx) {
+  public Set<String> listDefinitionIdsInUseByConnectors() throws IOException {
     Set<String> usedConnectorDefinitionIds = new HashSet<>();
-    // query for used source definitions
-    usedConnectorDefinitionIds.addAll(ctx
-        .select(field("config_blob ->> 'sourceDefinitionId'", SQLDataType.VARCHAR))
-        .from(AIRBYTE_CONFIGS)
-        .where(AIRBYTE_CONFIGS.CONFIG_TYPE.eq(ConfigSchema.SOURCE_CONNECTION.name()))
-        .fetch().stream()
-        .flatMap(row -> Stream.of(row.value1()))
-        .collect(Collectors.toSet()));
-    // query for used destination definitions
-    usedConnectorDefinitionIds.addAll(ctx
-        .select(field("config_blob ->> 'destinationDefinitionId'", SQLDataType.VARCHAR))
-        .from(AIRBYTE_CONFIGS)
-        .where(AIRBYTE_CONFIGS.CONFIG_TYPE.eq(ConfigSchema.DESTINATION_CONNECTION.name()))
-        .fetch().stream()
-        .flatMap(row -> Stream.of(row.value1()))
-        .collect(Collectors.toSet()));
+    database.transaction(ctx -> {
+      // query for used source definitions
+      usedConnectorDefinitionIds.addAll(ctx
+          .select(field("config_blob ->> 'sourceDefinitionId'", SQLDataType.VARCHAR))
+          .from(AIRBYTE_CONFIGS)
+          .where(AIRBYTE_CONFIGS.CONFIG_TYPE.eq(ConfigSchema.SOURCE_CONNECTION.name()))
+          .fetch().stream()
+          .flatMap(row -> Stream.of(row.value1()))
+          .collect(Collectors.toSet()));
+      // query for used destination definitions
+      usedConnectorDefinitionIds.addAll(ctx
+          .select(field("config_blob ->> 'destinationDefinitionId'", SQLDataType.VARCHAR))
+          .from(AIRBYTE_CONFIGS)
+          .where(AIRBYTE_CONFIGS.CONFIG_TYPE.eq(ConfigSchema.DESTINATION_CONNECTION.name()))
+          .fetch().stream()
+          .flatMap(row -> Stream.of(row.value1()))
+          .collect(Collectors.toSet()));
+      return null;
+    });
+    return usedConnectorDefinitionIds;
+  }
+
+  public Set<String> getRepositoriesFromDefinitionIds(Set<String> usedConnectorDefinitionIds) throws IOException {
+    final Set<String> reps = new HashSet<String>();
+    // reps.addAll(ctx.select(field("config_blob ->> 'dockerRepository'", SQLDataType.VARCHAR))
+    database.transaction(ctx -> {
+      System.out.println(ctx.select(field("config_blob ->> 'dockerRepository'", SQLDataType.VARCHAR))
+          .from(AIRBYTE_CONFIGS).fetch());
+      return null;
+    });
+
+    database.transaction(ctx -> {
+      reps.addAll(ctx.select(field("config_blob ->> 'dockerRepository'", SQLDataType.VARCHAR))
+          .from(AIRBYTE_CONFIGS)
+          .where(AIRBYTE_CONFIGS.CONFIG_ID.in(usedConnectorDefinitionIds))
+          .fetch().stream()
+          .map(Record1::value1)
+          .collect(Collectors.toSet()));
+      return null;
+    });
+    return reps;
 
-    return ctx.select(field("config_blob ->> 'dockerRepository'", SQLDataType.VARCHAR))
-        .from(AIRBYTE_CONFIGS)
-        .where(AIRBYTE_CONFIGS.CONFIG_ID.in(usedConnectorDefinitionIds))
-        .fetch().stream()
-        .map(Record1::value1)
-        .collect(Collectors.toSet());
   }
 
 }