Fix minor DBT Cloud Errors. (#18147)

Fixes URL formatting for dbt cloud integration. Note, without a trailing / the dbt Cloud API returns a 308. Some other small fixes: - Retries around the dbt Cloud invocation - Updates the regex to handle the new URL format.
airbytehq · Oct 19, 2022 · b18f950 · b18f950
1 parent 66e0055
commit b18f950
Show file tree

Hide file tree

Showing 2 changed files with 22 additions and 15 deletions.
diff --git a/airbyte-webapp/src/packages/cloud/services/dbtCloud.ts b/airbyte-webapp/src/packages/cloud/services/dbtCloud.ts
@@ -30,7 +30,7 @@ const toDbtCloudJob = (operation: OperationRead): DbtCloudJob => {
   const { operationId } = operation;
   const { executionUrl } = operation.operatorConfiguration.webhook || {};
 
-  const matches = (executionUrl || "").match(/\/accounts\/([^/]+)\/jobs\/([^]+)\//);
+  const matches = (executionUrl || "").match(/\/accounts\/([^/]+)\/jobs\/([^]+)\/run\//);
 
   if (!matches) {
     throw new Error(`Cannot extract dbt cloud job params from executionUrl ${executionUrl}`);
@@ -81,7 +81,7 @@ export const useDbtIntegration = (connection: WebBackendConnectionRead) => {
 
   const saveJobs = (jobs: DbtCloudJob[]) => {
     // TODO dynamically use the workspace's configured dbt cloud domain when it gets returned by backend
-    const urlForJob = (job: DbtCloudJob) => `${dbtCloudDomain}/api/v2/accounts/${job.account}/jobs/${job.job}/run`;
+    const urlForJob = (job: DbtCloudJob) => `${dbtCloudDomain}/api/v2/accounts/${job.account}/jobs/${job.job}/run/`;
 
     return connectionService.update({
       connectionId: connection.connectionId,

diff --git a/...-workers/src/main/java/io/airbyte/workers/temporal/sync/WebhookOperationActivityImpl.java b/...-workers/src/main/java/io/airbyte/workers/temporal/sync/WebhookOperationActivityImpl.java
@@ -11,7 +11,6 @@
 import io.airbyte.config.WebhookOperationConfigs;
 import io.airbyte.config.persistence.split_secrets.SecretsHydrator;
 import jakarta.inject.Singleton;
-import java.io.IOException;
 import java.net.URI;
 import java.net.http.HttpClient;
 import java.net.http.HttpRequest;
@@ -24,6 +23,7 @@
 public class WebhookOperationActivityImpl implements WebhookOperationActivity {
 
   private static final Logger LOGGER = LoggerFactory.getLogger(WebhookOperationActivityImpl.class);
+  private static final int MAX_RETRIES = 3;
 
   private final HttpClient httpClient;
 
@@ -38,6 +38,7 @@ public WebhookOperationActivityImpl(final HttpClient httpClient, final SecretsHy
   @Override
   public boolean invokeWebhook(OperatorWebhookInput input) {
     LOGGER.debug("Webhook operation input: {}", input);
+    LOGGER.debug("Found webhook config: {}", input.getWorkspaceWebhookConfigs());
     final JsonNode fullWebhookConfigJson = secretsHydrator.hydrate(input.getWorkspaceWebhookConfigs());
     final WebhookOperationConfigs webhookConfigs = Jsons.object(fullWebhookConfigJson, WebhookOperationConfigs.class);
     final Optional<WebhookConfig> webhookConfig =
@@ -48,7 +49,6 @@ public boolean invokeWebhook(OperatorWebhookInput input) {
 
     LOGGER.info("Invoking webhook operation {}", webhookConfig.get().getName());
 
-    LOGGER.debug("Found webhook config: {}", webhookConfig.get());
     final HttpRequest.Builder requestBuilder = HttpRequest.newBuilder()
         .uri(URI.create(input.getExecutionUrl()));
     if (input.getExecutionBody() != null) {
@@ -59,18 +59,25 @@ public boolean invokeWebhook(OperatorWebhookInput input) {
           .header("Content-Type", "application/json")
           .header("Authorization", "Bearer " + webhookConfig.get().getAuthToken()).build();
     }
-    try {
-      HttpResponse<String> response = this.httpClient.send(requestBuilder.build(), HttpResponse.BodyHandlers.ofString());
-      LOGGER.debug("Webhook response: {}", response == null ? null : response.body());
-      LOGGER.info("Webhook response status: {}", response == null ? "empty response" : response.statusCode());
-      // Return true if the request was successful.
-      boolean isSuccessful = response != null && response.statusCode() >= 200 && response.statusCode() <= 300;
-      LOGGER.info("Webhook {} execution status {}", webhookConfig.get().getName(), isSuccessful ? "successful" : "failed");
-      return isSuccessful;
-    } catch (IOException | InterruptedException e) {
-      LOGGER.error(e.getMessage());
-      throw new RuntimeException(e);
+
+    Exception finalException = null;
+    // TODO(mfsiega-airbyte): replace this loop with retries configured on the HttpClient impl.
+    for (int i = 0; i < MAX_RETRIES; i++) {
+      try {
+        HttpResponse<String> response = this.httpClient.send(requestBuilder.build(), HttpResponse.BodyHandlers.ofString());
+        LOGGER.debug("Webhook response: {}", response == null ? null : response.body());
+        LOGGER.info("Webhook response status: {}", response == null ? "empty response" : response.statusCode());
+        // Return true if the request was successful.
+        boolean isSuccessful = response != null && response.statusCode() >= 200 && response.statusCode() <= 300;
+        LOGGER.info("Webhook {} execution status {}", webhookConfig.get().getName(), isSuccessful ? "successful" : "failed");
+        return isSuccessful;
+      } catch (Exception e) {
+        LOGGER.warn(e.getMessage());
+        finalException = e;
+      }
     }
+    // If we ever get here, it means we exceeded MAX_RETRIES without returning in the happy path.
+    throw new RuntimeException(finalException);
   }
 
 }