feat: clone sources and destinations API (#10396)

* feat: clone sources and destinations API

* fix: source is pulled without secrets

Author: harshithmullapudi

airbyte-api/src/main/openapi/config.yaml

@@ -494,6 +494,29 @@ paths:
                 $ref: "#/components/schemas/SourceReadList"
         "422":
           $ref: "#/components/responses/InvalidInputResponse"
+  /v1/sources/clone:
+    post:
+      tags:
+        - source
+      summary: Clone source
+      operationId: cloneSource
+      requestBody:
+        content:
+          application/json:
+            schema:
+              $ref: "#/components/schemas/SourceIdRequestBody"
+        required: true
+      responses:
+        "200":
+          description: Successful operation
+          content:
+            application/json:
+              schema:
+                $ref: "#/components/schemas/SourceRead"
+        "404":
+          $ref: "#/components/responses/NotFoundResponse"
+        "422":
+          $ref: "#/components/responses/InvalidInputResponse"
   /v1/sources/delete:
     post:
       tags:
@@ -892,6 +915,29 @@ paths:
           $ref: "#/components/responses/NotFoundResponse"
         "422":
           $ref: "#/components/responses/InvalidInputResponse"
+  /v1/destinations/clone:
+    post:
+      tags:
+        - destination
+      summary: Clone destination
+      operationId: cloneDestination
+      requestBody:
+        content:
+          application/json:
+            schema:
+              $ref: "#/components/schemas/DestinationIdRequestBody"
+        required: true
+      responses:
+        "200":
+          description: Successful operation
+          content:
+            application/json:
+              schema:
+                $ref: "#/components/schemas/DestinationRead"
+        "404":
+          $ref: "#/components/responses/NotFoundResponse"
+        "422":
+          $ref: "#/components/responses/InvalidInputResponse"
   /v1/connections/create:
     post:
       tags:

airbyte-server/src/main/java/io/airbyte/server/apis/ConfigurationApi.java

@@ -403,6 +403,11 @@ public void deleteSource(final SourceIdRequestBody sourceIdRequestBody) {
     });
   }
 
+  @Override
+  public SourceRead cloneSource(final SourceIdRequestBody sourceIdRequestBody) {
+    return execute(() -> sourceHandler.cloneSource(sourceIdRequestBody));
+  }
+
   @Override
   public CheckConnectionRead checkConnectionToSource(final SourceIdRequestBody sourceIdRequestBody) {
     return execute(() -> schedulerHandler.checkSourceConnectionFromSourceId(sourceIdRequestBody));
@@ -507,6 +512,11 @@ public DestinationRead getDestination(final DestinationIdRequestBody destination
     return execute(() -> destinationHandler.getDestination(destinationIdRequestBody));
   }
 
+  @Override
+  public DestinationRead cloneDestination(final DestinationIdRequestBody destinationIdRequestBody) {
+    return execute(() -> destinationHandler.cloneDestination(destinationIdRequestBody));
+  }
+
   @Override
   public CheckConnectionRead checkConnectionToDestination(final DestinationIdRequestBody destinationIdRequestBody) {
     return execute(() -> schedulerHandler.checkDestinationConnectionFromDestinationId(destinationIdRequestBody));

airbyte-server/src/main/java/io/airbyte/server/handlers/DestinationHandler.java

@@ -19,6 +19,7 @@
 import io.airbyte.commons.json.Jsons;
 import io.airbyte.config.DestinationConnection;
 import io.airbyte.config.StandardDestinationDefinition;
+import io.airbyte.config.StandardSourceDefinition;
 import io.airbyte.config.persistence.ConfigNotFoundException;
 import io.airbyte.config.persistence.ConfigRepository;
 import io.airbyte.config.persistence.split_secrets.JsonSecretsProcessor;
@@ -153,6 +154,28 @@ public DestinationRead getDestination(final DestinationIdRequestBody destination
     return buildDestinationRead(destinationIdRequestBody.getDestinationId());
   }
 
+  public DestinationRead cloneDestination(final DestinationIdRequestBody destinationIdRequestBody)
+          throws JsonValidationException, IOException, ConfigNotFoundException {
+    // read destination configuration from db
+    final DestinationRead destinationToClone = buildDestinationReadWithSecrets(destinationIdRequestBody.getDestinationId());
+    final ConnectorSpecification spec = getSpec(destinationToClone.getDestinationDefinitionId());
+
+    // persist
+    final UUID destinationId = uuidGenerator.get();
+    final String copyText = " (Copy)";
+    final String destinationName = destinationToClone.getName() + copyText;
+    persistDestinationConnection(
+            destinationName,
+            destinationToClone.getDestinationDefinitionId(),
+            destinationToClone.getWorkspaceId(),
+            destinationId,
+            destinationToClone.getConnectionConfiguration(),
+            false);
+
+    // read configuration from db
+    return buildDestinationRead(destinationId, spec);
+  }
+
   public DestinationReadList listDestinationsForWorkspace(final WorkspaceIdRequestBody workspaceIdRequestBody)
       throws ConfigNotFoundException, IOException, JsonValidationException {
     final List<DestinationRead> reads = Lists.newArrayList();
@@ -249,6 +272,16 @@ private DestinationRead buildDestinationRead(final UUID destinationId, final Con
     return toDestinationRead(dci, standardDestinationDefinition);
   }
 
+  private DestinationRead buildDestinationReadWithSecrets(final UUID destinationId)
+          throws ConfigNotFoundException, IOException, JsonValidationException {
+
+    // remove secrets from config before returning the read
+    final DestinationConnection dci = Jsons.clone(configRepository.getDestinationConnectionWithSecrets(destinationId));
+    final StandardDestinationDefinition standardDestinationDefinition =
+            configRepository.getStandardDestinationDefinition(dci.getDestinationDefinitionId());
+    return toDestinationRead(dci, standardDestinationDefinition);
+  }
+
   protected static DestinationRead toDestinationRead(final DestinationConnection destinationConnection,
                                                      final StandardDestinationDefinition standardDestinationDefinition) {
     return new DestinationRead()

airbyte-server/src/main/java/io/airbyte/server/handlers/SourceHandler.java

@@ -115,6 +115,30 @@ public SourceRead getSource(final SourceIdRequestBody sourceIdRequestBody)
     return buildSourceRead(sourceIdRequestBody.getSourceId());
   }
 
+  public SourceRead cloneSource(final SourceIdRequestBody sourceIdRequestBody)
+      throws JsonValidationException, IOException, ConfigNotFoundException {
+    // read source configuration from db
+    final SourceRead sourceToClone = buildSourceReadWithSecrets(sourceIdRequestBody.getSourceId());
+
+    // persist
+    final UUID sourceId = uuidGenerator.get();
+    final StandardSourceDefinition sourceDef = configRepository.getSourceDefinitionFromSource(sourceIdRequestBody.getSourceId());
+    final ConnectorSpecification spec = sourceDef.getSpec();
+    final String copyText = " (Copy)";
+    final String sourceName = sourceToClone.getName() + copyText;
+    persistSourceConnection(
+        sourceName,
+        sourceToClone.getSourceDefinitionId(),
+        sourceToClone.getWorkspaceId(),
+        sourceId,
+        false,
+        sourceToClone.getConnectionConfiguration(),
+        spec);
+
+    // read configuration from db
+    return buildSourceRead(sourceId, spec);
+  }
+
   public SourceReadList listSourcesForWorkspace(final WorkspaceIdRequestBody workspaceIdRequestBody)
       throws ConfigNotFoundException, IOException, JsonValidationException {
 
@@ -220,6 +244,15 @@ private SourceRead buildSourceRead(final UUID sourceId, final ConnectorSpecifica
     return toSourceRead(sourceConnection, standardSourceDefinition);
   }
 
+  private SourceRead buildSourceReadWithSecrets(final UUID sourceId)
+      throws ConfigNotFoundException, IOException, JsonValidationException {
+    // read configuration from db
+    final SourceConnection sourceConnection = configRepository.getSourceConnectionWithSecrets(sourceId);
+    final StandardSourceDefinition standardSourceDefinition = configRepository
+        .getStandardSourceDefinition(sourceConnection.getSourceDefinitionId());
+    return toSourceRead(sourceConnection, standardSourceDefinition);
+  }
+
   private void validateSource(final ConnectorSpecification spec, final JsonNode implementationJson)
       throws JsonValidationException {
     validator.ensure(spec.getConnectionSpecification(), implementationJson);

airbyte-server/src/test/java/io/airbyte/server/handlers/DestinationHandlerTest.java

@@ -251,4 +251,39 @@ void testSearchDestinations() throws JsonValidationException, ConfigNotFoundExce
     assertEquals(0, actualDestinationRead.getDestinations().size());
   }
 
+  @Test
+  void testCloneDestination() throws JsonValidationException, ConfigNotFoundException, IOException {
+    final DestinationConnection clonedConnection = DestinationHelpers.generateDestination(standardDestinationDefinition.getDestinationDefinitionId());
+    final DestinationRead expectedDestinationRead = new DestinationRead()
+            .name(clonedConnection.getName())
+            .destinationDefinitionId(standardDestinationDefinition.getDestinationDefinitionId())
+            .workspaceId(clonedConnection.getWorkspaceId())
+            .destinationId(clonedConnection.getDestinationId())
+            .connectionConfiguration(clonedConnection.getConfiguration())
+            .destinationName(standardDestinationDefinition.getName());
+    final DestinationRead destinationRead = new DestinationRead()
+            .name(destinationConnection.getName())
+            .destinationDefinitionId(standardDestinationDefinition.getDestinationDefinitionId())
+            .workspaceId(destinationConnection.getWorkspaceId())
+            .destinationId(destinationConnection.getDestinationId())
+            .connectionConfiguration(destinationConnection.getConfiguration())
+            .destinationName(standardDestinationDefinition.getName());
+
+    final DestinationIdRequestBody destinationIdRequestBody = new DestinationIdRequestBody().destinationId(destinationRead.getDestinationId());
+
+    when(uuidGenerator.get()).thenReturn(clonedConnection.getDestinationId());
+    when(configRepository.getDestinationConnectionWithSecrets(destinationConnection.getDestinationId())).thenReturn(destinationConnection);
+    when(configRepository.getDestinationConnection(clonedConnection.getDestinationId())).thenReturn(clonedConnection);
+
+    when(configRepository.getStandardDestinationDefinition(destinationDefinitionSpecificationRead.getDestinationDefinitionId()))
+            .thenReturn(standardDestinationDefinition);
+    when(configRepository.getDestinationDefinitionFromDestination(destinationConnection.getDestinationId())).thenReturn(standardDestinationDefinition);
+    when(secretsProcessor.maskSecrets(destinationConnection.getConfiguration(), destinationDefinitionSpecificationRead.getConnectionSpecification()))
+            .thenReturn(destinationConnection.getConfiguration());
+
+    final DestinationRead actualDestinationRead = destinationHandler.cloneDestination(destinationIdRequestBody);
+
+    assertEquals(expectedDestinationRead, actualDestinationRead);
+  }
+
 }

airbyte-server/src/test/java/io/airbyte/server/handlers/SourceHandlerTest.java

@@ -179,12 +179,37 @@ void testGetSource() throws JsonValidationException, ConfigNotFoundException, IO
     verify(secretsProcessor).maskSecrets(sourceConnection.getConfiguration(), sourceDefinitionSpecificationRead.getConnectionSpecification());
   }
 
+  @Test
+  void testCloneSource() throws JsonValidationException, ConfigNotFoundException, IOException {
+    final SourceConnection clonedConnection = SourceHelpers.generateSource(standardSourceDefinition.getSourceDefinitionId());
+    final SourceRead expectedClonedSourceRead = SourceHelpers.getSourceRead(clonedConnection, standardSourceDefinition);
+    final SourceRead sourceRead = SourceHelpers.getSourceRead(sourceConnection, standardSourceDefinition);
+
+    final SourceIdRequestBody sourceIdRequestBody = new SourceIdRequestBody().sourceId(sourceRead.getSourceId());
+
+    when(uuidGenerator.get()).thenReturn(clonedConnection.getSourceId());
+    when(configRepository.getSourceConnectionWithSecrets(sourceConnection.getSourceId())).thenReturn(sourceConnection);
+    when(configRepository.getSourceConnection(clonedConnection.getSourceId())).thenReturn(clonedConnection);
+
+    when(configRepository.getStandardSourceDefinition(sourceDefinitionSpecificationRead.getSourceDefinitionId()))
+        .thenReturn(standardSourceDefinition);
+    when(configRepository.getSourceDefinitionFromSource(sourceConnection.getSourceId())).thenReturn(standardSourceDefinition);
+    when(secretsProcessor.maskSecrets(sourceConnection.getConfiguration(), sourceDefinitionSpecificationRead.getConnectionSpecification()))
+        .thenReturn(sourceConnection.getConfiguration());
+
+    final SourceRead actualSourceRead = sourceHandler.cloneSource(sourceIdRequestBody);
+
+    assertEquals(expectedClonedSourceRead, actualSourceRead);
+  }
+
   @Test
   void testListSourcesForWorkspace() throws JsonValidationException, ConfigNotFoundException, IOException {
     final SourceRead expectedSourceRead = SourceHelpers.getSourceRead(sourceConnection, standardSourceDefinition);
     final WorkspaceIdRequestBody workspaceIdRequestBody = new WorkspaceIdRequestBody().workspaceId(sourceConnection.getWorkspaceId());
 
     when(configRepository.getSourceConnection(sourceConnection.getSourceId())).thenReturn(sourceConnection);
+    when(configRepository.getSourceConnection(sourceConnection.getSourceId())).thenReturn(sourceConnection);
+
     when(configRepository.listSourceConnection()).thenReturn(Lists.newArrayList(sourceConnection));
     when(configRepository.getStandardSourceDefinition(sourceDefinitionSpecificationRead.getSourceDefinitionId()))
         .thenReturn(standardSourceDefinition);