Use config-secrets module for secrets hydration (#9547)

jdpgrailsdev · jdpgrailsdev · commit 36dd6409dab0 · 2023-11-06T10:56:26.000-05:00
diff --git a/airbyte-workers/build.gradle b/airbyte-workers/build.gradle
@@ -48,8 +48,8 @@ dependencies {
     implementation libs.sts
     implementation libs.aws.java.sdk.s3
     implementation libs.aws.java.sdk.sts
-    implementation 'com.google.auth:google-auth-library-oauth2-http:1.4.0'
-    implementation 'com.auth0:java-jwt:3.19.2'
+    implementation libs.google.auth.library.oauth2.http
+    implementation libs.java.jwt
     implementation libs.kubernetes.client
     implementation libs.guava
     implementation(libs.temporal.sdk) {
@@ -75,6 +75,7 @@ dependencies {
     implementation project(':airbyte-commons-with-dependencies')
     implementation project(':airbyte-config:config-models')
     implementation project(':airbyte-config:config-persistence')
+    implementation project(':airbyte-config:config-secrets')
     implementation project(':airbyte-config:specs')
     implementation project(':airbyte-config:init')
     implementation project(':airbyte-db:jooq')
diff --git a/airbyte-workers/src/main/java/io/airbyte/workers/config/SecretPersistenceBeanFactory.java b/airbyte-workers/src/main/java/io/airbyte/workers/config/SecretPersistenceBeanFactory.java
diff --git a/airbyte-workers/src/main/java/io/airbyte/workers/temporal/check/connection/CheckConnectionActivityImpl.java b/airbyte-workers/src/main/java/io/airbyte/workers/temporal/check/connection/CheckConnectionActivityImpl.java
@@ -30,7 +30,7 @@
 import io.airbyte.config.StandardCheckConnectionOutput.Status;
 import io.airbyte.config.helpers.LogConfigs;
 import io.airbyte.config.helpers.ResourceRequirementsUtils;
-import io.airbyte.config.persistence.split_secrets.SecretsHydrator;
+import io.airbyte.config.secrets.hydration.SecretsHydrator;
 import io.airbyte.metrics.lib.ApmTraceUtils;
 import io.airbyte.metrics.lib.MetricClientFactory;
 import io.airbyte.metrics.lib.OssMetricsRegistry;
diff --git a/airbyte-workers/src/main/java/io/airbyte/workers/temporal/discover/catalog/DiscoverCatalogActivityImpl.java b/airbyte-workers/src/main/java/io/airbyte/workers/temporal/discover/catalog/DiscoverCatalogActivityImpl.java
@@ -27,7 +27,7 @@
 import io.airbyte.config.StandardDiscoverCatalogInput;
 import io.airbyte.config.helpers.LogConfigs;
 import io.airbyte.config.helpers.ResourceRequirementsUtils;
-import io.airbyte.config.persistence.split_secrets.SecretsHydrator;
+import io.airbyte.config.secrets.hydration.SecretsHydrator;
 import io.airbyte.metrics.lib.ApmTraceUtils;
 import io.airbyte.metrics.lib.MetricClient;
 import io.airbyte.metrics.lib.OssMetricsRegistry;
diff --git a/airbyte-workers/src/main/java/io/airbyte/workers/temporal/sync/DbtTransformationActivityImpl.java b/airbyte-workers/src/main/java/io/airbyte/workers/temporal/sync/DbtTransformationActivityImpl.java
@@ -23,7 +23,7 @@
 import io.airbyte.config.OperatorDbtInput;
 import io.airbyte.config.ResourceRequirements;
 import io.airbyte.config.helpers.LogConfigs;
-import io.airbyte.config.persistence.split_secrets.SecretsHydrator;
+import io.airbyte.config.secrets.hydration.SecretsHydrator;
 import io.airbyte.featureflag.FeatureFlagClient;
 import io.airbyte.metrics.lib.ApmTraceUtils;
 import io.airbyte.metrics.lib.MetricClient;
diff --git a/airbyte-workers/src/main/java/io/airbyte/workers/temporal/sync/NormalizationActivityImpl.java b/airbyte-workers/src/main/java/io/airbyte/workers/temporal/sync/NormalizationActivityImpl.java
@@ -35,7 +35,7 @@
 import io.airbyte.config.StandardSyncInput;
 import io.airbyte.config.StandardSyncOutput;
 import io.airbyte.config.helpers.LogConfigs;
-import io.airbyte.config.persistence.split_secrets.SecretsHydrator;
+import io.airbyte.config.secrets.hydration.SecretsHydrator;
 import io.airbyte.featureflag.FeatureFlagClient;
 import io.airbyte.featureflag.RemoveLargeSyncInputs;
 import io.airbyte.featureflag.Workspace;
diff --git a/airbyte-workers/src/main/java/io/airbyte/workers/temporal/sync/ReplicationActivityImpl.java b/airbyte-workers/src/main/java/io/airbyte/workers/temporal/sync/ReplicationActivityImpl.java
@@ -45,7 +45,7 @@
 import io.airbyte.config.StateWrapper;
 import io.airbyte.config.helpers.LogConfigs;
 import io.airbyte.config.helpers.StateMessageHelper;
-import io.airbyte.config.persistence.split_secrets.SecretsHydrator;
+import io.airbyte.config.secrets.hydration.SecretsHydrator;
 import io.airbyte.featureflag.FeatureFlagClient;
 import io.airbyte.featureflag.RemoveLargeSyncInputs;
 import io.airbyte.featureflag.ResetBackfillState;
@@ -335,16 +335,16 @@ protected ReplicationInput getHydratedReplicationInput(final ReplicationActivity
         .withState(state);
   }
 
-  private State getUpdatedStateForBackfill(State state,
-                                           RefreshSchemaActivityOutput schemaRefreshOutput,
+  private State getUpdatedStateForBackfill(final State state,
+                                           final RefreshSchemaActivityOutput schemaRefreshOutput,
                                            final UUID workspaceId,
                                            final UUID connectionId,
                                            final ConfiguredAirbyteCatalog catalog)
       throws Exception {
     if (schemaRefreshOutput != null && schemaRefreshOutput.getAppliedDiff() != null) {
       final var streamsToBackfill = BackfillHelper.getStreamsToBackfill(schemaRefreshOutput.getAppliedDiff(), catalog);
       LOGGER.debug("Backfilling streams: {}", String.join(", ", streamsToBackfill.stream().map(StreamDescriptor::getName).toList()));
-      State resetState = BackfillHelper.clearStateForStreamsToBackfill(state, streamsToBackfill);
+      final State resetState = BackfillHelper.clearStateForStreamsToBackfill(state, streamsToBackfill);
       // persist the state
       // this will be behind a separate feature flag since it's a destructive operation.
       if (resetState != null && featureFlagClient.boolVariation(ResetBackfillState.INSTANCE, new Workspace(workspaceId))) {
@@ -359,7 +359,7 @@ private State getUpdatedStateForBackfill(State state,
   }
 
   @NotNull
-  private ConfiguredAirbyteCatalog retrieveCatalog(ReplicationActivityInput replicationActivityInput) throws Exception {
+  private ConfiguredAirbyteCatalog retrieveCatalog(final ReplicationActivityInput replicationActivityInput) throws Exception {
     final ConnectionRead connectionInfo =
         AirbyteApiClient
             .retryWithJitterThrows(
@@ -384,7 +384,7 @@ private void persistState(final State resetState, final UUID connectionId) throw
         "create or update the state");
   }
 
-  private State retrieveState(ReplicationActivityInput replicationActivityInput) throws Exception {
+  private State retrieveState(final ReplicationActivityInput replicationActivityInput) throws Exception {
     final ConnectionState connectionState = AirbyteApiClient.retryWithJitterThrows(
         () -> airbyteApiClient.getStateApi().getState(new ConnectionIdRequestBody().connectionId(replicationActivityInput.getConnectionId())),
         "retrieve the state");
@@ -395,7 +395,8 @@ private State retrieveState(ReplicationActivityInput replicationActivityInput) t
     return state;
   }
 
-  private void updateCatalogForReset(ReplicationActivityInput replicationActivityInput, ConfiguredAirbyteCatalog catalog) throws Exception {
+  private void updateCatalogForReset(final ReplicationActivityInput replicationActivityInput, final ConfiguredAirbyteCatalog catalog)
+      throws Exception {
     final JobOptionalRead jobInfo = AirbyteApiClient.retryWithJitterThrows(
         () -> airbyteApiClient.getJobsApi().getLastReplicationJob(
             new ConnectionIdRequestBody().connectionId(replicationActivityInput.getConnectionId())),
@@ -411,7 +412,7 @@ private void updateCatalogForReset(ReplicationActivityInput replicationActivityI
 
   // Simple converter from StandardSyncInput to ReplicationInput.
   // TODO: remove when the workflow version that passes a StandardSyncInput is removed.
-  private ReplicationInput getReplicationInputFromSyncInput(StandardSyncInput hydratedSyncInput,
+  private ReplicationInput getReplicationInputFromSyncInput(final StandardSyncInput hydratedSyncInput,
                                                             final JobRunConfig jobRunConfig,
                                                             final IntegrationLauncherConfig sourceLauncherConfig,
                                                             final IntegrationLauncherConfig destinationLauncherConfig) {
diff --git a/airbyte-workers/src/main/java/io/airbyte/workers/temporal/sync/WebhookOperationActivityImpl.java b/airbyte-workers/src/main/java/io/airbyte/workers/temporal/sync/WebhookOperationActivityImpl.java
@@ -13,7 +13,7 @@
 import io.airbyte.config.OperatorWebhookInput;
 import io.airbyte.config.WebhookConfig;
 import io.airbyte.config.WebhookOperationConfigs;
-import io.airbyte.config.persistence.split_secrets.SecretsHydrator;
+import io.airbyte.config.secrets.hydration.SecretsHydrator;
 import io.airbyte.metrics.lib.ApmTraceUtils;
 import io.airbyte.metrics.lib.MetricClientFactory;
 import io.airbyte.metrics.lib.OssMetricsRegistry;
diff --git a/airbyte-workers/src/test/java/io/airbyte/workers/config/DataPlaneActivityInitializationMicronautTest.java b/airbyte-workers/src/test/java/io/airbyte/workers/config/DataPlaneActivityInitializationMicronautTest.java
@@ -8,7 +8,7 @@
 import static org.mockito.Mockito.mock;
 
 import io.airbyte.commons.temporal.config.WorkerMode;
-import io.airbyte.config.persistence.split_secrets.SecretPersistence;
+import io.airbyte.config.secrets.persistence.SecretPersistence;
 import io.airbyte.workers.temporal.scheduling.activities.ConfigFetchActivity;
 import io.airbyte.workers.temporal.scheduling.activities.ConfigFetchActivityImpl;
 import io.airbyte.workers.temporal.sync.DbtTransformationActivity;
diff --git a/airbyte-workers/src/test/java/io/airbyte/workers/temporal/sync/NormalizationActivityImplTest.java b/airbyte-workers/src/test/java/io/airbyte/workers/temporal/sync/NormalizationActivityImplTest.java
@@ -22,7 +22,7 @@
 import io.airbyte.config.Configs;
 import io.airbyte.config.NormalizationInput;
 import io.airbyte.config.helpers.LogConfigs;
-import io.airbyte.config.persistence.split_secrets.SecretsHydrator;
+import io.airbyte.config.secrets.hydration.SecretsHydrator;
 import io.airbyte.config.storage.CloudStorageConfigs;
 import io.airbyte.featureflag.FeatureFlagClient;
 import io.airbyte.featureflag.RemoveLargeSyncInputs;
@@ -55,20 +55,20 @@ class NormalizationActivityImplTest {
   private static WorkerConfigsProvider mWorkerConfigsProvider;
   private static ProcessFactory mProcessFactory;
   private static SecretsHydrator mSecretsHydrator;
-  private static Path WORKSPACE_ROOT = Path.of("/unused/path");
-  private static Configs.WorkerEnvironment WORKER_ENVIRONMENT = Configs.WorkerEnvironment.KUBERNETES;
+  private static final Path WORKSPACE_ROOT = Path.of("/unused/path");
+  private static final Configs.WorkerEnvironment WORKER_ENVIRONMENT = Configs.WorkerEnvironment.KUBERNETES;
   private static CloudStorageConfigs mCloudStorageConfigs;
   private static LogConfigs LOG_CONFIGS;
-  private static String AIRBYTE_VERSION = "1.0";
-  private static Integer SERVER_PORT = 8888;
+  private static final String AIRBYTE_VERSION = "1.0";
+  private static final Integer SERVER_PORT = 8888;
   private static AirbyteConfigValidator mAirbyteConfigValidator;
   private static AirbyteApiClient mAirbyteApiClient;
   private static ConnectionApi mConnectionApi;
   private static FeatureFlagClient mFeatureFlagClient;
   private static NormalizationActivityImpl normalizationActivityImpl;
   private static NormalizationActivity normalizationActivity;
-  private static JobRunConfig JOB_RUN_CONFIG = new JobRunConfig().withJobId("unused").withAttemptId(0L);
-  private static IntegrationLauncherConfig DESTINATION_CONFIG = new IntegrationLauncherConfig()
+  private static final JobRunConfig JOB_RUN_CONFIG = new JobRunConfig().withJobId("unused").withAttemptId(0L);
+  private static final IntegrationLauncherConfig DESTINATION_CONFIG = new IntegrationLauncherConfig()
       .withDockerImage("unused")
       .withNormalizationDockerImage("unused:unused");
   private static WorkerConfigs mWorkerConfigs;
diff --git a/airbyte-workers/src/test/java/io/airbyte/workers/temporal/sync/ReplicationActivityTest.java b/airbyte-workers/src/test/java/io/airbyte/workers/temporal/sync/ReplicationActivityTest.java
@@ -40,7 +40,7 @@
 import io.airbyte.config.SyncResourceRequirements;
 import io.airbyte.config.helpers.LogConfigs;
 import io.airbyte.config.helpers.StateMessageHelper;
-import io.airbyte.config.persistence.split_secrets.SecretsHydrator;
+import io.airbyte.config.secrets.hydration.SecretsHydrator;
 import io.airbyte.featureflag.FeatureFlagClient;
 import io.airbyte.featureflag.RemoveLargeSyncInputs;
 import io.airbyte.featureflag.TestClient;
diff --git a/airbyte-workers/src/test/java/io/airbyte/workers/temporal/sync/WebhookOperationActivityTest.java b/airbyte-workers/src/test/java/io/airbyte/workers/temporal/sync/WebhookOperationActivityTest.java
@@ -13,7 +13,7 @@
 import io.airbyte.config.OperatorWebhookInput;
 import io.airbyte.config.WebhookConfig;
 import io.airbyte.config.WebhookOperationConfigs;
-import io.airbyte.config.persistence.split_secrets.SecretsHydrator;
+import io.airbyte.config.secrets.hydration.SecretsHydrator;
 import java.io.IOException;
 import java.net.http.HttpClient;
 import java.net.http.HttpResponse;
@@ -52,7 +52,7 @@ void webhookActivityInvokesConfiguredWebhook() throws IOException, InterruptedEx
         .withWebhookConfigId(WEBHOOK_ID);
     // TODO(mfsiega-airbyte): make these matchers more specific.
     when(httpClient.send(any(), any())).thenReturn(mockHttpResponse);
-    boolean success = webhookActivity.invokeWebhook(input);
+    final boolean success = webhookActivity.invokeWebhook(input);
     assertTrue(success);
   }
 
diff --git a/deps.toml b/deps.toml
@@ -76,6 +76,7 @@ failsafe-okhttp = { module = "dev.failsafe:failsafe-okhttp", version.ref = "fail
 findsecbugs-plugin = { module = "com.h3xstream.findsecbugs:findsecbugs-plugin", version = "1.12.0" }
 flyway-core = { module = "org.flywaydb:flyway-core", version.ref = "flyway" }
 glassfish = { module = "org.glassfish.jersey:jackson-bom", version.ref = "glassfish_version" }
+google-auth-library-oauth2-http = { module = "com.google.auth:google-auth-library-oauth2-http", version = "1.4.0" }
 google-cloud-storage = { module = "com.google.cloud:google-cloud-storage", version = "2.17.2" }
 google-cloud-storage-secretmanager = { module = "com.google.cloud:google-cloud-secretmanager", version = "2.0.5" }
 guava = { module = "com.google.guava:guava", version = "31.1-jre" }
