Add JobErrorReporter for sending sync job connector failures to Sentry

airbyte-config/config-models/src/main/java/io/airbyte/config/Configs.java

@@ -420,6 +420,17 @@ public interface Configs {
    */
   TrackingStrategy getTrackingStrategy();
 
+  /**
+   * Define whether to send job failure events to Sentry or log-only. Airbyte internal use.
+   */
+  ErrorReportingStrategy getErrorReportingStrategy();
+
+  /**
+   * Determines the Sentry DSN that should be used when reporting connector job failures to Sentry.
+   * Used with SENTRY error reporting strategy. Airbyte internal use.
+   */
+  String getErrorReportingSentryDSN();
+
   // APPLICATIONS
   // Worker
   /**
@@ -522,6 +533,11 @@ enum TrackingStrategy {
     LOGGING
   }
 
+  enum ErrorReportingStrategy {
+    SENTRY,
+    LOGGING
+  }
+
   enum WorkerEnvironment {
     DOCKER,
     KUBERNETES

airbyte-config/config-models/src/main/java/io/airbyte/config/EnvConfigs.java

@@ -50,6 +50,8 @@ public class EnvConfigs implements Configs {
   public static final String CONFIG_ROOT = "CONFIG_ROOT";
   public static final String DOCKER_NETWORK = "DOCKER_NETWORK";
   public static final String TRACKING_STRATEGY = "TRACKING_STRATEGY";
+  public static final String ERROR_REPORTING_STRATEGY = "ERROR_REPORTING_STRATEGY";
+  public static final String ERROR_REPORTING_SENTRY_DSN = "ERROR_REPORTING_SENTRY_DSN";
   public static final String DEPLOYMENT_MODE = "DEPLOYMENT_MODE";
   public static final String DATABASE_USER = "DATABASE_USER";
   public static final String DATABASE_PASSWORD = "DATABASE_PASSWORD";
@@ -751,6 +753,23 @@ public TrackingStrategy getTrackingStrategy() {
     });
   }
 
+  @Override
+  public ErrorReportingStrategy getErrorReportingStrategy() {
+    return getEnvOrDefault(ERROR_REPORTING_STRATEGY, ErrorReportingStrategy.LOGGING, s -> {
+      try {
+        return ErrorReportingStrategy.valueOf(s.toUpperCase());
+      } catch (final IllegalArgumentException e) {
+        LOGGER.info(s + " not recognized, defaulting to " + ErrorReportingStrategy.LOGGING);
+        return ErrorReportingStrategy.LOGGING;
+      }
+    });
+  }
+
+  @Override
+  public String getErrorReportingSentryDSN() {
+    return getEnvOrDefault(ERROR_REPORTING_SENTRY_DSN, "");
+  }
+
   // APPLICATIONS
   // Worker
   @Override

airbyte-config/config-models/src/test/java/io/airbyte/config/EnvConfigsTest.java

@@ -177,6 +177,27 @@ void testTrackingStrategy() {
     assertEquals(Configs.TrackingStrategy.LOGGING, config.getTrackingStrategy());
   }
 
+  @Test
+  void testErrorReportingStrategy() {
+    envMap.put(EnvConfigs.ERROR_REPORTING_STRATEGY, null);
+    assertEquals(Configs.ErrorReportingStrategy.LOGGING, config.getErrorReportingStrategy());
+
+    envMap.put(EnvConfigs.ERROR_REPORTING_STRATEGY, "abc");
+    assertEquals(Configs.ErrorReportingStrategy.LOGGING, config.getErrorReportingStrategy());
+
+    envMap.put(EnvConfigs.ERROR_REPORTING_STRATEGY, "logging");
+    assertEquals(Configs.ErrorReportingStrategy.LOGGING, config.getErrorReportingStrategy());
+
+    envMap.put(EnvConfigs.ERROR_REPORTING_STRATEGY, "sentry");
+    assertEquals(Configs.ErrorReportingStrategy.SENTRY, config.getErrorReportingStrategy());
+
+    envMap.put(EnvConfigs.ERROR_REPORTING_STRATEGY, "LOGGING");
+    assertEquals(Configs.ErrorReportingStrategy.LOGGING, config.getErrorReportingStrategy());
+
+    envMap.put(EnvConfigs.ERROR_REPORTING_STRATEGY, "SENTRY");
+    assertEquals(Configs.ErrorReportingStrategy.SENTRY, config.getErrorReportingStrategy());
+  }
+
   @Test
   void testDeploymentMode() {
     envMap.put(EnvConfigs.DEPLOYMENT_MODE, null);

airbyte-scheduler/scheduler-persistence/build.gradle

@@ -3,6 +3,8 @@ plugins {
 }
 
 dependencies {
+    implementation 'io.sentry:sentry:6.1.0'
+
     implementation project(':airbyte-analytics')
     implementation project(':airbyte-commons-docker')
     implementation project(':airbyte-config:config-models')

airbyte-scheduler/scheduler-persistence/src/main/java/io/airbyte/scheduler/persistence/job_error_reporter/ErrorReportingClient.java

@@ -0,0 +1,21 @@
+/*
+ * Copyright (c) 2022 Airbyte, Inc., all rights reserved.
+ */
+
+package io.airbyte.scheduler.persistence.job_error_reporter;
+
+import io.airbyte.config.FailureReason;
+import io.airbyte.config.StandardWorkspace;
+import java.util.Map;
+
+/**
+ * A generic interface for a client that reports errors
+ */
+public interface ErrorReportingClient {
+
+  /**
+   * Report a job failure reason
+   */
+  void report(StandardWorkspace workspace, final FailureReason reason, final String dockerImage, Map<String, String> metadata);
+
+}

airbyte-scheduler/scheduler-persistence/src/main/java/io/airbyte/scheduler/persistence/job_error_reporter/ErrorReportingClientFactory.java

@@ -0,0 +1,25 @@
+/*
+ * Copyright (c) 2022 Airbyte, Inc., all rights reserved.
+ */
+
+package io.airbyte.scheduler.persistence.job_error_reporter;
+
+import io.airbyte.config.Configs;
+import io.airbyte.config.Configs.ErrorReportingStrategy;
+
+public class ErrorReportingClientFactory {
+
+  /**
+   * Creates an error reporting client based on the desired strategy to use
+   *
+   * @param strategy - which type of error reporting client should be created
+   * @return ErrorReportingClient
+   */
+  public static ErrorReportingClient getClient(final ErrorReportingStrategy strategy, final Configs configs) {
+    return switch (strategy) {
+      case SENTRY -> new SentryErrorReportingClient(configs.getErrorReportingSentryDSN());
+      case LOGGING -> new LoggingErrorReportingClient();
+    };
+  }
+
+}

airbyte-scheduler/scheduler-persistence/src/main/java/io/airbyte/scheduler/persistence/job_error_reporter/JobErrorReporter.java

@@ -0,0 +1,86 @@
+/*
+ * Copyright (c) 2022 Airbyte, Inc., all rights reserved.
+ */
+
+package io.airbyte.scheduler.persistence.job_error_reporter;
+
+import io.airbyte.config.AttemptFailureSummary;
+import io.airbyte.config.FailureReason;
+import io.airbyte.config.FailureReason.FailureOrigin;
+import io.airbyte.config.JobSyncConfig;
+import io.airbyte.config.StandardDestinationDefinition;
+import io.airbyte.config.StandardSourceDefinition;
+import io.airbyte.config.StandardWorkspace;
+import io.airbyte.config.persistence.ConfigRepository;
+import java.util.HashMap;
+import java.util.List;
+import java.util.UUID;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class JobErrorReporter {
+
+  private static final String AIRBYTE_VERSION_META_KEY = "airbyte_version";
+  private static final String FAILURE_ORIGIN_META_KEY = "failure_origin";
+  private static final String FAILURE_TYPE_META_KEY = "failure_type";
+  private static final String CONNECTION_ID_META_KEY = "connection_id";
+  private static final String CONNECTOR_NAME_META_KEY = "connector_name";
+  private static final String CONNECTOR_DEFINITION_ID_META_KEY = "connector_definition_id";
+  private static final String CONNECTOR_RELEASE_STAGE_META_KEY = "connector_release_stage";
+
+  private final ConfigRepository configRepository;
+  private final String airbyteVersion;
+  private final ErrorReportingClient errorReportingClient;
+
+  public JobErrorReporter(
+                          final ConfigRepository configRepository,
+                          final String airbyteVersion,
+                          final ErrorReportingClient errorReportingClient) {
+
+    this.configRepository = configRepository;
+    this.airbyteVersion = airbyteVersion;
+    this.errorReportingClient = errorReportingClient;
+  }
+
+  private static final Logger LOGGER = LoggerFactory.getLogger(JobErrorReporter.class);
+
+  public void reportSyncJobFailure(final UUID connectionId, final AttemptFailureSummary failureSummary, final JobSyncConfig jobSyncConfig) {
+    LOGGER.info("reportSyncJobFailure");
+    final List<FailureReason> traceMessageFailures = failureSummary.getFailures().stream()
+        .filter(failure -> failure.getMetadata() != null && failure.getMetadata().getAdditionalProperties().containsKey("from_trace_message"))
+        .toList();
+
+    final StandardWorkspace workspace = configRepository.getStandardWorkspaceFromConnection(connectionId, true);
+
+    for (final FailureReason failureReason : traceMessageFailures) {
+      final FailureOrigin failureOrigin = failureReason.getFailureOrigin();
+
+      final HashMap<String, String> metadata = new HashMap<>();
+      metadata.put(CONNECTION_ID_META_KEY, connectionId.toString());
+      metadata.put(AIRBYTE_VERSION_META_KEY, airbyteVersion);
+      metadata.put(FAILURE_ORIGIN_META_KEY, failureOrigin.value());
+      metadata.put(FAILURE_TYPE_META_KEY, failureReason.getFailureType().value());
+
+      if (failureOrigin == FailureOrigin.SOURCE) {
+        final StandardSourceDefinition sourceDefinition = configRepository.getSourceDefinitionFromConnection(connectionId);
+        final String dockerImage = jobSyncConfig.getSourceDockerImage();
+
+        metadata.put(CONNECTOR_DEFINITION_ID_META_KEY, sourceDefinition.getSourceDefinitionId().toString());
+        metadata.put(CONNECTOR_NAME_META_KEY, sourceDefinition.getName());
+        metadata.put(CONNECTOR_RELEASE_STAGE_META_KEY, sourceDefinition.getReleaseStage().value());
+
+        errorReportingClient.report(workspace, failureReason, dockerImage, metadata);
+      } else if (failureOrigin == FailureOrigin.DESTINATION) {
+        final StandardDestinationDefinition destinationDefinition = configRepository.getDestinationDefinitionFromConnection(connectionId);
+        final String dockerImage = jobSyncConfig.getDestinationDockerImage();
+
+        metadata.put(CONNECTOR_DEFINITION_ID_META_KEY, destinationDefinition.getDestinationDefinitionId().toString());
+        metadata.put(CONNECTOR_NAME_META_KEY, destinationDefinition.getName());
+        metadata.put(CONNECTOR_RELEASE_STAGE_META_KEY, destinationDefinition.getReleaseStage().value());
+
+        errorReportingClient.report(workspace, failureReason, dockerImage, metadata);
+      }
+    }
+  }
+
+}

airbyte-scheduler/scheduler-persistence/src/main/java/io/airbyte/scheduler/persistence/job_error_reporter/LoggingErrorReportingClient.java

@@ -0,0 +1,26 @@
+/*
+ * Copyright (c) 2022 Airbyte, Inc., all rights reserved.
+ */
+
+package io.airbyte.scheduler.persistence.job_error_reporter;
+
+import io.airbyte.config.FailureReason;
+import io.airbyte.config.StandardWorkspace;
+import java.util.Map;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class LoggingErrorReportingClient implements ErrorReportingClient {
+
+  private static final Logger LOGGER = LoggerFactory.getLogger(LoggingErrorReportingClient.class);
+
+  @Override
+  public void report(final StandardWorkspace workspace, final FailureReason reason, final String dockerImage, final Map<String, String> metadata) {
+    LOGGER.info("Report Job Error -> workspaceId: {}, dockerImage: {}, failureReason: {}, metadata: {}",
+        workspace.getWorkspaceId(),
+        dockerImage,
+        reason,
+        metadata);
+  }
+
+}

airbyte-scheduler/scheduler-persistence/src/main/java/io/airbyte/scheduler/persistence/job_error_reporter/SentryErrorReportingClient.java

@@ -0,0 +1,86 @@
+/*
+ * Copyright (c) 2022 Airbyte, Inc., all rights reserved.
+ */
+
+package io.airbyte.scheduler.persistence.job_error_reporter;
+
+import io.airbyte.config.FailureReason;
+import io.airbyte.config.StandardWorkspace;
+import io.sentry.Sentry;
+import io.sentry.SentryEvent;
+import io.sentry.protocol.Message;
+import io.sentry.protocol.SentryException;
+import io.sentry.protocol.SentryId;
+import io.sentry.protocol.User;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class SentryErrorReportingClient implements ErrorReportingClient {
+
+  private static final Logger LOGGER = LoggerFactory.getLogger(LoggingErrorReportingClient.class);
+
+  public SentryErrorReportingClient(final String sentryDSN) {
+    Sentry.init(options -> {
+      options.setDsn(sentryDSN);
+      options.setEnableUncaughtExceptionHandler(false);
+    });
+  }
+
+  @Override
+  public void report(final StandardWorkspace workspace,
+                     final FailureReason failureReason,
+                     final String dockerImage,
+                     final Map<String, String> metadata) {
+    final SentryEvent event = new SentryEvent();
+
+    // airbyte/source-xyz:1.2.0 -> source-xyz@1.2.0
+    final String release = dockerImage.replace(":", "@").substring(dockerImage.lastIndexOf("/") + 1);
+    event.setRelease(release);
+
+    // add connector to event fingerprint to ensure separate grouping per connector
+    final String[] releaseParts = release.split("@");
+    if (releaseParts.length > 0) {
+      event.setFingerprints(List.of("{{ default }}", releaseParts[0]));
+    }
+
+    // set workspace as the user in sentry to get impact and priority
+    final User sentryUser = new User();
+    sentryUser.setId(String.valueOf(workspace.getWorkspaceId()));
+    sentryUser.setUsername(workspace.getName());
+    event.setUser(sentryUser);
+
+    // set metadata as tags
+    event.setTags(metadata);
+
+    final Message message = new Message();
+    message.setFormatted(failureReason.getInternalMessage());
+    event.setMessage(message);
+
+    // don't attach current thread stacktrace to the event
+    event.setThreads(new ArrayList<>());
+    event.setPlatform("other");
+
+    // attach failure reason stack trace
+    final String failureStackTrace = failureReason.getStacktrace();
+    if (failureStackTrace != null) {
+      final List<SentryException> parsedExceptions = SentryExceptionHelper.buildSentryExceptions(failureStackTrace);
+      event.setExceptions(parsedExceptions);
+    }
+
+    Sentry.withScope(scope -> {
+      final Map<String, String> failureReasonContext = new HashMap<>();
+      failureReasonContext.put("internalMessage", failureReason.getInternalMessage());
+      failureReasonContext.put("externalMessage", failureReason.getExternalMessage());
+      failureReasonContext.put("stacktrace", failureReason.getStacktrace());
+      scope.setContexts("Failure Reason", failureReasonContext);
+
+      final SentryId eventId = Sentry.captureEvent(event);
+      LOGGER.info("SENT SENTRY EVENT: {}", eventId);
+    });
+  }
+
+}