🎉 New implementation for AWS Secret Manager for issue

airbytehq#10518

Added new implementation AWSSecretManagerPersistence and integration tests AWSSecretManagerPersistenceIntegrationTest

A new implementation of SecretPersistence to support AWS Secret Manager AWSSecretManagerPersistence
New Integration tests as suggested on the open Issue, similar to GCP secret manager

airbyte-config/config-persistence/build.gradle

@@ -20,6 +20,7 @@ dependencies {
     implementation 'commons-io:commons-io:2.7'
     implementation 'com.google.cloud:google-cloud-secretmanager:2.0.5'
     implementation 'com.bettercloud:vault-java-driver:5.1.0'
+    implementation 'com.amazonaws.secretsmanager:aws-secretsmanager-caching-java:1.0.1'
 
     testImplementation 'org.hamcrest:hamcrest-all:1.3'
     testImplementation libs.platform.testcontainers.postgresql

airbyte-config/config-persistence/src/main/java/io/airbyte/config/persistence/split_secrets/AWSSecretManagerPersistence.java

@@ -0,0 +1,105 @@
+/*
+ * Copyright (c) 2022 Airbyte, Inc., all rights reserved.
+ */
+
+package io.airbyte.config.persistence.split_secrets;
+
+import com.amazonaws.secretsmanager.caching.SecretCache;
+import com.amazonaws.services.secretsmanager.AWSSecretsManager;
+import com.amazonaws.services.secretsmanager.AWSSecretsManagerClientBuilder;
+import com.amazonaws.services.secretsmanager.model.CreateSecretRequest;
+import com.amazonaws.services.secretsmanager.model.DeleteSecretRequest;
+import com.amazonaws.services.secretsmanager.model.ResourceNotFoundException;
+import com.amazonaws.services.secretsmanager.model.UpdateSecretRequest;
+import com.google.common.annotations.VisibleForTesting;
+import java.util.Optional;
+import lombok.extern.slf4j.Slf4j;
+
+/**
+ * SecretPersistence implementation for AWS Secret Manager using <a href=
+ * "https://sdk.amazonaws.com/java/api/latest/software/amazon/awssdk/services/secretsmanager/package-summary.html">Java
+ * SDK</a> The current implementation doesn't make use of `SecretCoordinate#getVersion` as this
+ * version is non-compatible with how AWS secret manager deals with versions. In AWS versions is an
+ * internal idiom that can is accessible, but it's a UUID + a tag <a href=
+ * "https://sdk.amazonaws.com/java/api/latest/software/amazon/awssdk/services/secretsmanager/SecretsManagerClient.html#listSecretVersionIds--">more
+ * details.</a>
+ */
+@Slf4j
+public class AWSSecretManagerPersistence implements SecretPersistence {
+
+  private final AWSSecretsManager client;
+
+  @VisibleForTesting
+  protected final SecretCache cache;
+
+  /**
+   * Creates a AWSSecretManagerPersistence using the defaults client and region from the current AWS
+   * credentials. This implementation makes use of SecretCache as optimization to access secrets
+   *
+   * @see SecretCache
+   */
+  public AWSSecretManagerPersistence() {
+    this.client = AWSSecretsManagerClientBuilder.defaultClient();
+    this.cache = new SecretCache(this.client);
+  }
+
+  /**
+   * Creates a AWSSecretManagerPersistence overriding the current region. This implementation makes
+   * use of SecretCache as optimization to access secrets
+   *
+   * @param region AWS region to use
+   * @see SecretCache
+   */
+  public AWSSecretManagerPersistence(final String region) {
+    this.client = AWSSecretsManagerClientBuilder
+        .standard()
+        .withRegion(region)
+        .build();
+    this.cache = new SecretCache(this.client);
+  }
+
+  @Override
+  public Optional<String> read(final SecretCoordinate coordinate) {
+    String secretString = null;
+    try {
+      log.debug("Reading secret {}", coordinate.getCoordinateBase());
+      secretString = cache.getSecretString(coordinate.getCoordinateBase());
+    } catch (ResourceNotFoundException e) {
+      log.warn("Secret {} not found", coordinate.getCoordinateBase());
+    }
+    return Optional.ofNullable(secretString);
+  }
+
+  @Override
+  public void write(final SecretCoordinate coordinate, final String payload) {
+    if (read(coordinate).isPresent()) {
+      log.debug("Secret {} found updating payload.", coordinate.getCoordinateBase());
+      final UpdateSecretRequest request = new UpdateSecretRequest()
+          .withSecretId(coordinate.getCoordinateBase())
+          .withSecretString(payload)
+          .withDescription("Airbyte secret.");
+      client.updateSecret(request);
+    } else {
+      log.debug("Secret {} not found, creating a new one.", coordinate.getCoordinateBase());
+      final CreateSecretRequest secretRequest = new CreateSecretRequest()
+          .withName(coordinate.getCoordinateBase())
+          .withSecretString(payload)
+          .withDescription("Airbyte secret.");
+      client.createSecret(secretRequest);
+    }
+
+  }
+
+  /**
+   * Utility to clean up after integration tests.
+   *
+   * @param coordinate SecretCoordinate to delete.
+   */
+  @VisibleForTesting
+  protected void deleteSecret(final SecretCoordinate coordinate) {
+    client.deleteSecret(new DeleteSecretRequest()
+        .withSecretId(coordinate.getCoordinateBase())
+        .withForceDeleteWithoutRecovery(true));
+  }
+
+}

airbyte-config/config-persistence/src/test-integration/java/io/airbyte/config/persistence/split_secrets/AWSSecretManagerPersistenceIntegrationTest.java

@@ -0,0 +1,58 @@
+/*
+ * Copyright (c) 2022 Airbyte, Inc., all rights reserved.
+ */
+
+package io.airbyte.config.persistence.split_secrets;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.util.Optional;
+import org.apache.commons.lang3.RandomUtils;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+public class AWSSecretManagerPersistenceIntegrationTest {
+
+  public String coordinate_base;
+  private AWSSecretManagerPersistence persistence;
+
+  @BeforeEach
+  void setup() {
+    persistence = new AWSSecretManagerPersistence();
+    coordinate_base = "aws/airbyte/secret/integration/" + RandomUtils.nextInt() % 20000;
+  }
+
+  @Test
+  void testReadWriteUpdate() throws InterruptedException {
+    SecretCoordinate secretCoordinate = new SecretCoordinate(coordinate_base, 1);
+
+    // try reading a non-existent secret
+    Optional<String> firstRead = persistence.read(secretCoordinate);
+    assertTrue(firstRead.isEmpty());
+
+    // write it
+    String payload = "foo-secret";
+    persistence.write(secretCoordinate, payload);
+    persistence.cache.refreshNow(secretCoordinate.getCoordinateBase());
+    Optional<String> read2 = persistence.read(secretCoordinate);
+    assertTrue(read2.isPresent());
+    assertEquals(payload, read2.get());
+
+    // update it
+    final var secondPayload = "bar-secret";
+    final var coordinate2 = new SecretCoordinate(coordinate_base, 2);
+    persistence.write(coordinate2, secondPayload);
+    persistence.cache.refreshNow(secretCoordinate.getCoordinateBase());
+    final var thirdRead = persistence.read(coordinate2);
+    assertTrue(thirdRead.isPresent());
+    assertEquals(secondPayload, thirdRead.get());
+  }
+
+  @AfterEach
+  void tearDown() {
+    persistence.deleteSecret(new SecretCoordinate(coordinate_base, 1));
+  }
+
+}