Add structured dbt cloud information to the operations api (#19395)

mfsiega-airbyte · akashkulk · commit 174a740d4a02 · 2022-11-17T15:06:47.000-08:00
* add structured dbt cloud information to the operations api

* remove unused webhook features, test updates

* update tests to use structured dbt cloud operation api

* add missing webhook operator type
diff --git a/airbyte-api/src/main/openapi/config.yaml b/airbyte-api/src/main/openapi/config.yaml
@@ -3778,19 +3778,35 @@ components:
           type: string
     OperatorWebhook:
       type: object
-      required:
-        - executionUrl
       properties:
-        executionUrl:
-          type: string
-          description: The URL to call to execute the webhook operation via POST request.
-        executionBody:
-          type: string
-          description: If populated, this will be sent with the POST request.
         webhookConfigId:
           type: string
           format: uuid
           description: The id of the webhook configs to use from the workspace.
+        webhookType:
+          type: string
+          enum:
+            - dbtCloud
+        dbtCloud:
+          type: object
+          required:
+            - accountId
+            - jobId
+          properties:
+            accountId:
+              type: integer
+              description: The account id associated with the job
+            jobId:
+              type: integer
+              description: The job id associated with the job
+        executionUrl:
+          type: string
+          description: DEPRECATED. Populate dbtCloud instead.
+          deprecated: true
+        executionBody:
+          type: string
+          description: DEPRECATED. Populate dbtCloud instead.
+          deprecated: true
     CheckOperationRead:
       type: object
       required:
diff --git a/airbyte-server/src/main/java/io/airbyte/server/converters/OperationsConverter.java b/airbyte-server/src/main/java/io/airbyte/server/converters/OperationsConverter.java
@@ -4,17 +4,22 @@
 
 package io.airbyte.server.converters;
 
+import static io.airbyte.api.model.generated.OperatorWebhook.WebhookTypeEnum.DBTCLOUD;
+
 import com.google.common.base.Preconditions;
 import io.airbyte.api.model.generated.OperationRead;
 import io.airbyte.api.model.generated.OperatorConfiguration;
 import io.airbyte.api.model.generated.OperatorNormalization.OptionEnum;
+import io.airbyte.api.model.generated.OperatorWebhookDbtCloud;
 import io.airbyte.commons.enums.Enums;
 import io.airbyte.config.OperatorDbt;
 import io.airbyte.config.OperatorNormalization;
 import io.airbyte.config.OperatorNormalization.Option;
 import io.airbyte.config.OperatorWebhook;
 import io.airbyte.config.StandardSyncOperation;
 import io.airbyte.config.StandardSyncOperation.OperatorType;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
 
 public class OperationsConverter {
 
@@ -44,10 +49,7 @@ public static void populateOperatorConfigFromApi(final OperatorConfiguration ope
       case WEBHOOK -> {
         Preconditions.checkArgument(operatorConfig.getWebhook() != null);
         // TODO(mfsiega-airbyte): check that the webhook config id references a real webhook config.
-        standardSyncOperation.withOperatorWebhook(new OperatorWebhook()
-            .withExecutionUrl(operatorConfig.getWebhook().getExecutionUrl())
-            .withExecutionBody(operatorConfig.getWebhook().getExecutionBody())
-            .withWebhookConfigId(operatorConfig.getWebhook().getWebhookConfigId()));
+        standardSyncOperation.withOperatorWebhook(webhookOperatorFromConfig(operatorConfig.getWebhook()));
         // Null out the other configs, since it's mutually exclusive. We need to do this if it's an update.
         standardSyncOperation.withOperatorNormalization(null);
         standardSyncOperation.withOperatorDbt(null);
@@ -82,10 +84,7 @@ public static OperationRead operationReadFromPersistedOperation(final StandardSy
       }
       case WEBHOOK -> {
         Preconditions.checkArgument(standardSyncOperation.getOperatorWebhook() != null);
-        operatorConfiguration.webhook(new io.airbyte.api.model.generated.OperatorWebhook()
-            .webhookConfigId(standardSyncOperation.getOperatorWebhook().getWebhookConfigId())
-            .executionUrl(standardSyncOperation.getOperatorWebhook().getExecutionUrl())
-            .executionBody(standardSyncOperation.getOperatorWebhook().getExecutionBody()));
+        operatorConfiguration.webhook(webhookOperatorFromPersistence(standardSyncOperation.getOperatorWebhook()));
       }
     }
     return new OperationRead()
@@ -95,4 +94,68 @@ public static OperationRead operationReadFromPersistedOperation(final StandardSy
         .operatorConfiguration(operatorConfiguration);
   }
 
+  private static OperatorWebhook webhookOperatorFromConfig(io.airbyte.api.model.generated.OperatorWebhook webhookConfig) {
+    final var operatorWebhook = new OperatorWebhook().withWebhookConfigId(webhookConfig.getWebhookConfigId());
+    // TODO(mfsiega-airbyte): remove this once the frontend is sending the new format.
+    if (webhookConfig.getWebhookType() == null) {
+      return operatorWebhook
+          .withExecutionUrl(webhookConfig.getExecutionUrl())
+          .withExecutionBody(webhookConfig.getExecutionBody());
+    }
+    switch (webhookConfig.getWebhookType()) {
+      case DBTCLOUD -> {
+        return operatorWebhook
+            .withExecutionUrl(DbtCloudOperationConverter.getExecutionUrlFrom(webhookConfig.getDbtCloud()))
+            .withExecutionBody(DbtCloudOperationConverter.getDbtCloudExecutionBody());
+      }
+      // Future webhook operator types added here.
+    }
+    throw new IllegalArgumentException("Unsupported webhook operation type");
+  }
+
+  private static io.airbyte.api.model.generated.OperatorWebhook webhookOperatorFromPersistence(final OperatorWebhook persistedWebhook) {
+    final io.airbyte.api.model.generated.OperatorWebhook webhookOperator = new io.airbyte.api.model.generated.OperatorWebhook()
+        .webhookConfigId(persistedWebhook.getWebhookConfigId());
+    OperatorWebhookDbtCloud dbtCloudOperator = DbtCloudOperationConverter.parseFrom(persistedWebhook);
+    if (dbtCloudOperator != null) {
+      webhookOperator.webhookType(DBTCLOUD).dbtCloud(dbtCloudOperator);
+      // TODO(mfsiega-airbyte): remove once frontend switches to new format.
+      // Dual-write deprecated webhook format.
+      webhookOperator.executionUrl(DbtCloudOperationConverter.getExecutionUrlFrom(dbtCloudOperator));
+      webhookOperator.executionBody(DbtCloudOperationConverter.getDbtCloudExecutionBody());
+    } else {
+      throw new IllegalArgumentException("Unexpected webhook operator config");
+    }
+    return webhookOperator;
+  }
+
+  private static class DbtCloudOperationConverter {
+
+    // See https://docs.getdbt.com/dbt-cloud/api-v2 for documentation on dbt Cloud API endpoints.
+    final static Pattern dbtUrlPattern = Pattern.compile("^https://cloud\\.getdbt\\.com/api/v2/accounts/(\\d+)/jobs/(\\d+)/run/$");
+    private static final int ACCOUNT_REGEX_GROUP = 1;
+    private static final int JOB_REGEX_GROUP = 2;
+
+    private static OperatorWebhookDbtCloud parseFrom(OperatorWebhook persistedWebhook) {
+      Matcher dbtCloudUrlMatcher = dbtUrlPattern.matcher(persistedWebhook.getExecutionUrl());
+      final var dbtCloudConfig = new OperatorWebhookDbtCloud();
+      if (dbtCloudUrlMatcher.matches()) {
+        dbtCloudConfig.setAccountId(Integer.valueOf(dbtCloudUrlMatcher.group(ACCOUNT_REGEX_GROUP)));
+        dbtCloudConfig.setJobId(Integer.valueOf(dbtCloudUrlMatcher.group(JOB_REGEX_GROUP)));
+        return dbtCloudConfig;
+      }
+      return null;
+    }
+
+    private static String getExecutionUrlFrom(final OperatorWebhookDbtCloud dbtCloudConfig) {
+      return String.format("https://cloud.getdbt.com/api/v2/accounts/%d/jobs/%d/run/", dbtCloudConfig.getAccountId(),
+          dbtCloudConfig.getJobId());
+    }
+
+    private static String getDbtCloudExecutionBody() {
+      return "{\"cause\": \"airbyte\"}";
+    }
+
+  }
+
 }
diff --git a/airbyte-server/src/test/java/io/airbyte/server/handlers/OperationsHandlerTest.java b/airbyte-server/src/test/java/io/airbyte/server/handlers/OperationsHandlerTest.java
@@ -5,6 +5,7 @@
 package io.airbyte.server.handlers;
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 import static org.mockito.ArgumentMatchers.eq;
 import static org.mockito.Mockito.mock;
@@ -24,6 +25,8 @@
 import io.airbyte.api.model.generated.OperatorNormalization.OptionEnum;
 import io.airbyte.api.model.generated.OperatorType;
 import io.airbyte.api.model.generated.OperatorWebhook;
+import io.airbyte.api.model.generated.OperatorWebhook.WebhookTypeEnum;
+import io.airbyte.api.model.generated.OperatorWebhookDbtCloud;
 import io.airbyte.commons.enums.Enums;
 import io.airbyte.config.OperatorNormalization.Option;
 import io.airbyte.config.StandardSync;
@@ -45,10 +48,12 @@ class OperationsHandlerTest {
 
   private static final String WEBHOOK_OPERATION_NAME = "fake-operation-name";
   private static final UUID WEBHOOK_CONFIG_ID = UUID.randomUUID();
-  private static final String WEBHOOK_EXECUTION_URL = "fake-execution-url";
-  private static final String WEBHOOK_EXECUTION_BODY = "fake-execution-body";
   private static final UUID WEBHOOK_OPERATION_ID = UUID.randomUUID();
-  public static final String NEW_EXECUTION_URL = "new-execution-url";
+  private static final Integer DBT_CLOUD_WEBHOOK_ACCOUNT_ID = 123;
+  private static final Integer DBT_CLOUD_WEBHOOK_JOB_ID = 456;
+  private static final Integer NEW_DBT_CLOUD_WEBHOOK_ACCOUNT_ID = 789;
+  public static final String EXECUTION_BODY = "{\"cause\": \"airbyte\"}";
+  public static final String EXECUTION_URL_TEMPLATE = "https://cloud.getdbt.com/api/v2/accounts/%d/jobs/%d/run/";
   private ConfigRepository configRepository;
   private Supplier<UUID> uuidGenerator;
   private OperationsHandler operationsHandler;
@@ -104,8 +109,10 @@ void testCreateWebhookOperation() throws JsonValidationException, ConfigNotFound
     when(uuidGenerator.get()).thenReturn(WEBHOOK_OPERATION_ID);
     final OperatorWebhook webhookConfig = new OperatorWebhook()
         .webhookConfigId(WEBHOOK_CONFIG_ID)
-        .executionUrl(WEBHOOK_EXECUTION_URL)
-        .executionBody(WEBHOOK_EXECUTION_BODY);
+        .webhookType(WebhookTypeEnum.DBTCLOUD)
+        .dbtCloud(new OperatorWebhookDbtCloud()
+            .accountId(DBT_CLOUD_WEBHOOK_ACCOUNT_ID)
+            .jobId(DBT_CLOUD_WEBHOOK_JOB_ID));
     final OperationCreate operationCreate = new OperationCreate()
         .workspaceId(standardSyncOperation.getWorkspaceId())
         .name(WEBHOOK_OPERATION_NAME)
@@ -119,8 +126,9 @@ void testCreateWebhookOperation() throws JsonValidationException, ConfigNotFound
         .withOperatorType(StandardSyncOperation.OperatorType.WEBHOOK)
         .withOperatorWebhook(new io.airbyte.config.OperatorWebhook()
             .withWebhookConfigId(WEBHOOK_CONFIG_ID)
-            .withExecutionUrl(WEBHOOK_EXECUTION_URL)
-            .withExecutionBody(WEBHOOK_EXECUTION_BODY))
+            .withExecutionUrl(String.format(EXECUTION_URL_TEMPLATE, DBT_CLOUD_WEBHOOK_ACCOUNT_ID,
+                DBT_CLOUD_WEBHOOK_JOB_ID))
+            .withExecutionBody(EXECUTION_BODY))
         .withTombstone(false);
 
     when(configRepository.getStandardSyncOperation(WEBHOOK_OPERATION_ID)).thenReturn(expectedPersistedOperation);
@@ -131,7 +139,12 @@ void testCreateWebhookOperation() throws JsonValidationException, ConfigNotFound
     assertEquals(WEBHOOK_OPERATION_ID, actualOperationRead.getOperationId());
     assertEquals(WEBHOOK_OPERATION_NAME, actualOperationRead.getName());
     assertEquals(OperatorType.WEBHOOK, actualOperationRead.getOperatorConfiguration().getOperatorType());
-    assertEquals(webhookConfig, actualOperationRead.getOperatorConfiguration().getWebhook());
+
+    // NOTE: we expect the server to dual-write on read until the frontend moves to the new format.
+    final OperatorWebhook expectedWebhookConfigRead =
+        webhookConfig.executionUrl(String.format(EXECUTION_URL_TEMPLATE, DBT_CLOUD_WEBHOOK_ACCOUNT_ID,
+            DBT_CLOUD_WEBHOOK_JOB_ID)).executionBody(EXECUTION_BODY);
+    assertEquals(expectedWebhookConfigRead, actualOperationRead.getOperatorConfiguration().getWebhook());
 
     verify(configRepository).writeStandardSyncOperation(eq(expectedPersistedOperation));
   }
@@ -189,36 +202,58 @@ void testUpdateWebhookOperation() throws JsonValidationException, ConfigNotFound
     when(uuidGenerator.get()).thenReturn(WEBHOOK_OPERATION_ID);
     final OperatorWebhook webhookConfig = new OperatorWebhook()
         .webhookConfigId(WEBHOOK_CONFIG_ID)
-        .executionUrl(NEW_EXECUTION_URL)
-        .executionBody(WEBHOOK_EXECUTION_BODY);
+        .webhookType(WebhookTypeEnum.DBTCLOUD)
+        .dbtCloud(new OperatorWebhookDbtCloud()
+            .accountId(NEW_DBT_CLOUD_WEBHOOK_ACCOUNT_ID)
+            .jobId(DBT_CLOUD_WEBHOOK_JOB_ID));
     final OperationUpdate operationUpdate = new OperationUpdate()
         .name(WEBHOOK_OPERATION_NAME)
         .operationId(WEBHOOK_OPERATION_ID)
         .operatorConfiguration(new OperatorConfiguration()
             .operatorType(OperatorType.WEBHOOK).webhook(webhookConfig));
 
+    final var persistedWebhook = new io.airbyte.config.OperatorWebhook()
+        .withWebhookConfigId(WEBHOOK_CONFIG_ID)
+        .withExecutionUrl(String.format(EXECUTION_URL_TEMPLATE, DBT_CLOUD_WEBHOOK_ACCOUNT_ID,
+            DBT_CLOUD_WEBHOOK_JOB_ID))
+        .withExecutionBody(EXECUTION_BODY);
+
+    final var updatedWebhook = new io.airbyte.config.OperatorWebhook()
+        .withWebhookConfigId(WEBHOOK_CONFIG_ID)
+        .withExecutionUrl(String.format(EXECUTION_URL_TEMPLATE, NEW_DBT_CLOUD_WEBHOOK_ACCOUNT_ID,
+            DBT_CLOUD_WEBHOOK_JOB_ID))
+        .withExecutionBody(EXECUTION_BODY);
+
     final StandardSyncOperation persistedOperation = new StandardSyncOperation()
         .withWorkspaceId(standardSyncOperation.getWorkspaceId())
         .withOperationId(WEBHOOK_OPERATION_ID)
         .withName(WEBHOOK_OPERATION_NAME)
         .withOperatorType(StandardSyncOperation.OperatorType.WEBHOOK)
-        .withOperatorWebhook(new io.airbyte.config.OperatorWebhook()
-            .withWebhookConfigId(WEBHOOK_CONFIG_ID)
-            .withExecutionUrl(WEBHOOK_EXECUTION_URL)
-            .withExecutionBody(WEBHOOK_EXECUTION_BODY));
+        .withOperatorWebhook(persistedWebhook);
 
-    when(configRepository.getStandardSyncOperation(WEBHOOK_OPERATION_ID)).thenReturn(persistedOperation);
+    final StandardSyncOperation updatedOperation = new StandardSyncOperation()
+        .withWorkspaceId(standardSyncOperation.getWorkspaceId())
+        .withOperationId(WEBHOOK_OPERATION_ID)
+        .withName(WEBHOOK_OPERATION_NAME)
+        .withOperatorType(StandardSyncOperation.OperatorType.WEBHOOK)
+        .withOperatorWebhook(updatedWebhook);
+
+    when(configRepository.getStandardSyncOperation(WEBHOOK_OPERATION_ID)).thenReturn(persistedOperation).thenReturn(updatedOperation);
 
     final OperationRead actualOperationRead = operationsHandler.updateOperation(operationUpdate);
 
     assertEquals(WEBHOOK_OPERATION_ID, actualOperationRead.getOperationId());
     assertEquals(WEBHOOK_OPERATION_NAME, actualOperationRead.getName());
     assertEquals(OperatorType.WEBHOOK, actualOperationRead.getOperatorConfiguration().getOperatorType());
-    assertEquals(webhookConfig, actualOperationRead.getOperatorConfiguration().getWebhook());
+    final OperatorWebhook expectedWebhookConfigRead =
+        webhookConfig.executionUrl(String.format(EXECUTION_URL_TEMPLATE, NEW_DBT_CLOUD_WEBHOOK_ACCOUNT_ID,
+            DBT_CLOUD_WEBHOOK_JOB_ID)).executionBody(EXECUTION_BODY);
+    assertEquals(expectedWebhookConfigRead, actualOperationRead.getOperatorConfiguration().getWebhook());
 
     verify(configRepository)
         .writeStandardSyncOperation(persistedOperation.withOperatorWebhook(persistedOperation.getOperatorWebhook().withExecutionUrl(
-            NEW_EXECUTION_URL)));
+            String.format(EXECUTION_URL_TEMPLATE, NEW_DBT_CLOUD_WEBHOOK_ACCOUNT_ID,
+                DBT_CLOUD_WEBHOOK_JOB_ID))));
   }
 
   @Test
@@ -313,4 +348,39 @@ void testEnumConversion() {
         io.airbyte.config.OperatorNormalization.Option.class));
   }
 
+  @Test
+  void testDbtCloudRegex() {
+    // Validate that a non-url is rejected.
+    assertThrows(IllegalArgumentException.class, () -> checkDbtCloudUrl("not-a-url"));
+    // Validate that the URL is anchored to the beginning.
+    assertThrows(IllegalArgumentException.class,
+        () -> checkDbtCloudUrl("some-nonsense-" + String.format(EXECUTION_URL_TEMPLATE, DBT_CLOUD_WEBHOOK_ACCOUNT_ID,
+            DBT_CLOUD_WEBHOOK_JOB_ID)));
+    // Validate that the URL is anchored to the end.
+    assertThrows(IllegalArgumentException.class,
+        () -> checkDbtCloudUrl(String.format(EXECUTION_URL_TEMPLATE, DBT_CLOUD_WEBHOOK_ACCOUNT_ID,
+            DBT_CLOUD_WEBHOOK_JOB_ID) + "-some-nonsense"));
+    // Validate that the account id must be an integer.
+    assertThrows(IllegalArgumentException.class, () -> checkDbtCloudUrl("https://cloud.getdbt.com/api/v2/accounts/abc/jobs/123/run/"));
+    // Validate that the job id must be an integer.
+    assertThrows(IllegalArgumentException.class, () -> checkDbtCloudUrl("https://cloud.getdbt.com/api/v2/accounts/123/jobs/abc/run/"));
+  }
+
+  private void checkDbtCloudUrl(final String urlToCheck) throws JsonValidationException, ConfigNotFoundException, IOException {
+    final StandardSyncOperation persistedOperation = new StandardSyncOperation()
+        .withWorkspaceId(standardSyncOperation.getWorkspaceId())
+        .withOperationId(WEBHOOK_OPERATION_ID)
+        .withName(WEBHOOK_OPERATION_NAME)
+        .withOperatorType(StandardSyncOperation.OperatorType.WEBHOOK)
+        .withOperatorWebhook(new io.airbyte.config.OperatorWebhook()
+            .withWebhookConfigId(WEBHOOK_CONFIG_ID)
+            .withExecutionUrl(urlToCheck)
+            .withExecutionBody(EXECUTION_BODY))
+        .withTombstone(false);
+    when(configRepository.getStandardSyncOperation(WEBHOOK_OPERATION_ID)).thenReturn(persistedOperation);
+
+    final OperationIdRequestBody operationIdRequestBody = new OperationIdRequestBody().operationId(WEBHOOK_OPERATION_ID);
+    operationsHandler.getOperation(operationIdRequestBody);
+  }
+
 }
diff --git a/airbyte-tests/src/acceptanceTests/java/io/airbyte/test/acceptance/BasicAcceptanceTests.java b/airbyte-tests/src/acceptanceTests/java/io/airbyte/test/acceptance/BasicAcceptanceTests.java
@@ -66,6 +66,8 @@
 import io.airbyte.api.client.model.generated.OperatorConfiguration;
 import io.airbyte.api.client.model.generated.OperatorType;
 import io.airbyte.api.client.model.generated.OperatorWebhook;
+import io.airbyte.api.client.model.generated.OperatorWebhook.WebhookTypeEnum;
+import io.airbyte.api.client.model.generated.OperatorWebhookDbtCloud;
 import io.airbyte.api.client.model.generated.SourceDefinitionIdRequestBody;
 import io.airbyte.api.client.model.generated.SourceDefinitionIdWithWorkspaceId;
 import io.airbyte.api.client.model.generated.SourceDefinitionRead;
@@ -465,10 +467,9 @@ void testWebhookOperationExecutesSuccessfully() throws Exception {
             .operatorType(OperatorType.WEBHOOK)
             .webhook(new OperatorWebhook()
                 .webhookConfigId(workspaceRead.getWebhookConfigs().get(0).getId())
-                // NOTE: reqres.in is free service that hosts a REST API intended for testing frontend/client code.
-                // We use it here as an endpoint that will accept an HTTP POST.
-                .executionUrl("https://reqres.in/api/users")
-                .executionBody("{\"name\": \"morpheus\", \"job\": \"leader\"}"))));
+                // NOTE: this dbt Cloud config won't actually work, but the sync should still succeed.
+                .webhookType(WebhookTypeEnum.DBTCLOUD)
+                .dbtCloud(new OperatorWebhookDbtCloud().accountId(123).jobId(456)))));
     // create a connection with the new operation.
     final UUID sourceId = testHarness.createPostgresSource().getSourceId();
     final UUID destinationId = testHarness.createPostgresDestination().getDestinationId();
diff --git a/docs/reference/api/generated-api-html/index.html b/docs/reference/api/generated-api-html/index.html
