[MVP] Integrate sentry to all java-based connectors

airbyte-integrations/connectors/destination-bigquery-denormalized/Dockerfile

@@ -13,6 +13,7 @@ FROM airbyte/integration-base-java:dev
 WORKDIR /airbyte
 
 ENV APPLICATION destination-bigquery-denormalized
+ENV SENTRY_DSN "https://[email protected]/6145965"
 
 COPY --from=build /airbyte /airbyte
 

airbyte-integrations/connectors/destination-bigquery-denormalized/build.gradle

@@ -12,6 +12,7 @@ application {
 dependencies {
     implementation 'com.google.cloud:google-cloud-bigquery:1.122.2'
     implementation 'org.apache.commons:commons-lang3:3.11'
+    implementation 'io.sentry:sentry-log4j2:5.5.3'
 
     implementation project(':airbyte-config:models')
     implementation project(':airbyte-integrations:bases:base-java')

airbyte-integrations/connectors/destination-bigquery-denormalized/src/main/java/io/airbyte/integrations/destination/bigquery/BigQueryDenormalizedDestination.java

@@ -11,6 +11,9 @@
 import io.airbyte.integrations.destination.bigquery.formatter.DefaultBigQueryDenormalizedRecordFormatter;
 import io.airbyte.integrations.destination.bigquery.formatter.GcsBigQueryDenormalizedRecordFormatter;
 import io.airbyte.integrations.destination.bigquery.uploader.UploaderType;
+import io.sentry.ITransaction;
+import io.sentry.Sentry;
+import io.sentry.SpanStatus;
 import java.util.Map;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -19,6 +22,18 @@ public class BigQueryDenormalizedDestination extends BigQueryDestination {
 
   private static final Logger LOGGER = LoggerFactory.getLogger(BigQueryDenormalizedDestination.class);
 
+  public static void initSentry() {
+    Sentry.init(options -> {
+      // allow setting properties from env variables see https://docs.sentry.io/platforms/java/configuration/
+      options.setEnableExternalConfiguration(true);
+      // To set a uniform sample rate
+      options.setTracesSampleRate(1.0);
+    });
+    Sentry.configureScope(scope -> {
+      scope.setTag("connector", "destination-bigquery-denormalized");
+    });
+  }
+
   @Override
   protected String getTargetTableName(final String streamName) {
     // This BigQuery destination does not write to a staging "raw" table but directly to a normalized
@@ -46,10 +61,19 @@ protected boolean isDefaultAirbyteTmpTableSchema() {
   }
 
   public static void main(final String[] args) throws Exception {
+    initSentry();
     final Destination destination = new BigQueryDenormalizedDestination();
-    LOGGER.info("starting destination: {}", BigQueryDenormalizedDestination.class);
-    new IntegrationRunner(destination).run(args);
-    LOGGER.info("completed destination: {}", BigQueryDenormalizedDestination.class);
+    ITransaction transaction = Sentry.startTransaction("IntegrationRunner()", "run");
+    try {
+      LOGGER.info("starting destination: {}", BigQueryDestination.class);
+      new IntegrationRunner(destination).run(args);
+    } catch (Exception e) {
+      transaction.setThrowable(e);
+      transaction.setStatus(SpanStatus.INTERNAL_ERROR);
+    } finally {
+      transaction.finish();
+      LOGGER.info("completed destination: {}", BigQueryDestination.class);
+    }
   }
 
 }

airbyte-integrations/connectors/destination-bigquery-denormalized/src/main/resources/log4j2.xml

@@ -0,0 +1,11 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<Configuration status="warn" packages="org.apache.logging.log4j.core,io.sentry.log4j2">
+  <Appenders>
+    <Sentry name="Sentry"/>
+  </Appenders>
+  <Loggers>
+    <Root level="ERROR">
+      <AppenderRef ref="Sentry"/>
+    </Root>
+  </Loggers>
+</Configuration>

airbyte-integrations/connectors/destination-bigquery/Dockerfile

@@ -13,6 +13,7 @@ FROM airbyte/integration-base-java:dev
 WORKDIR /airbyte
 
 ENV APPLICATION destination-bigquery
+ENV SENTRY_DSN "https://[email protected]/6145965"
 
 COPY --from=build /airbyte /airbyte
 

airbyte-integrations/connectors/destination-bigquery/build.gradle

@@ -12,6 +12,7 @@ application {
 dependencies {
     implementation 'com.google.cloud:google-cloud-bigquery:1.122.2'
     implementation 'org.apache.commons:commons-lang3:3.11'
+    implementation 'io.sentry:sentry-log4j2:5.5.3'
 
     // csv
     implementation 'org.apache.commons:commons-csv:1.4'

airbyte-integrations/connectors/destination-bigquery/src/main/java/io/airbyte/integrations/destination/bigquery/BigQueryDestination.java

@@ -34,6 +34,9 @@
 import io.airbyte.protocol.models.AirbyteStream;
 import io.airbyte.protocol.models.ConfiguredAirbyteCatalog;
 import io.airbyte.protocol.models.ConfiguredAirbyteStream;
+import io.sentry.ITransaction;
+import io.sentry.Sentry;
+import io.sentry.SpanStatus;
 import java.io.ByteArrayInputStream;
 import java.io.IOException;
 import java.util.HashMap;
@@ -53,6 +56,18 @@ public BigQueryDestination() {
     namingResolver = new BigQuerySQLNameTransformer();
   }
 
+  public static void initSentry() {
+    Sentry.init(options -> {
+      // allow setting properties from env variables see https://docs.sentry.io/platforms/java/configuration/
+      options.setEnableExternalConfiguration(true);
+      // To set a uniform sample rate
+      options.setTracesSampleRate(1.0);
+    });
+    Sentry.configureScope(scope -> {
+      scope.setTag("connector", "destination-bigquery");
+    });
+  }
+
   @Override
   public AirbyteConnectionStatus check(final JsonNode config) {
     try {
@@ -124,32 +139,29 @@ protected BigQuery getBigQuery(final JsonNode config) {
    * 1. Create a temporary table for each stream
    * </p>
    * <p>
-   * 2. Write records to each stream directly (the bigquery client handles managing when to push the
-   * records over the network)
+   * 2. Write records to each stream directly (the bigquery client handles managing when to push the records over the network)
    * </p>
    * <p>
-   * 4. Once all records have been written close the writers, so that any remaining records are
-   * flushed.
+   * 4. Once all records have been written close the writers, so that any remaining records are flushed.
    * </p>
    * <p>
    * 5. Copy the temp tables to the final table name (overwriting if necessary).
    * </p>
    *
-   * @param config - integration-specific configuration object as json. e.g. { "username": "airbyte",
-   *        "password": "super secure" }
+   * @param config  - integration-specific configuration object as json. e.g. { "username": "airbyte", "password": "super secure" }
    * @param catalog - schema of the incoming messages.
    * @return consumer that writes singer messages to the database.
    */
   @Override
   public AirbyteMessageConsumer getConsumer(final JsonNode config,
-                                            final ConfiguredAirbyteCatalog catalog,
-                                            final Consumer<AirbyteMessage> outputRecordCollector)
+      final ConfiguredAirbyteCatalog catalog,
+      final Consumer<AirbyteMessage> outputRecordCollector)
       throws IOException {
     return getRecordConsumer(getUploaderMap(config, catalog), outputRecordCollector);
   }
 
   protected Map<AirbyteStreamNameNamespacePair, AbstractBigQueryUploader<?>> getUploaderMap(final JsonNode config,
-                                                                                            final ConfiguredAirbyteCatalog catalog)
+      final ConfiguredAirbyteCatalog catalog)
       throws IOException {
     final BigQuery bigquery = getBigQuery(config);
 
@@ -176,9 +188,8 @@ protected Map<AirbyteStreamNameNamespacePair, AbstractBigQueryUploader<?>> getUp
   }
 
   /**
-   * BigQuery might have different structure of the Temporary table. If this method returns TRUE,
-   * temporary table will have only three common Airbyte attributes. In case of FALSE, temporary table
-   * structure will be in line with Airbyte message JsonSchema.
+   * BigQuery might have different structure of the Temporary table. If this method returns TRUE, temporary table will have only three common Airbyte
+   * attributes. In case of FALSE, temporary table structure will be in line with Airbyte message JsonSchema.
    *
    * @return use default AirbyteSchema or build using JsonSchema
    */
@@ -197,15 +208,23 @@ protected String getTargetTableName(final String streamName) {
   }
 
   protected AirbyteMessageConsumer getRecordConsumer(final Map<AirbyteStreamNameNamespacePair, AbstractBigQueryUploader<?>> writeConfigs,
-                                                     final Consumer<AirbyteMessage> outputRecordCollector) {
+      final Consumer<AirbyteMessage> outputRecordCollector) {
     return new BigQueryRecordConsumer(writeConfigs, outputRecordCollector);
   }
 
   public static void main(final String[] args) throws Exception {
+    ITransaction transaction = Sentry.startTransaction("IntegrationRunner()", "run");
     final Destination destination = new BigQueryDestination();
-    LOGGER.info("starting destination: {}", BigQueryDestination.class);
-    new IntegrationRunner(destination).run(args);
-    LOGGER.info("completed destination: {}", BigQueryDestination.class);
+    try {
+      LOGGER.info("starting destination: {}", BigQueryDestination.class);
+      new IntegrationRunner(destination).run(args);
+    } catch (Exception e) {
+      transaction.setThrowable(e);
+      transaction.setStatus(SpanStatus.INTERNAL_ERROR);
+    } finally {
+      transaction.finish();
+      LOGGER.info("completed destination: {}", BigQueryDestination.class);
+    }
   }
 
 }

airbyte-integrations/connectors/destination-bigquery/src/main/resources/log4j2.xml

@@ -0,0 +1,11 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<Configuration status="warn" packages="org.apache.logging.log4j.core,io.sentry.log4j2">
+  <Appenders>
+    <Sentry name="Sentry"/>
+  </Appenders>
+  <Loggers>
+    <Root level="ERROR">
+      <AppenderRef ref="Sentry"/>
+    </Root>
+  </Loggers>
+</Configuration>

airbyte-integrations/connectors/destination-snowflake/Dockerfile

@@ -11,6 +11,7 @@ WORKDIR /airbyte
 
 ENV APPLICATION destination-snowflake
 
+ENV SENTRY_DSN "https://[email protected]/6145965"
 # Needed for JDK17 (in turn, needed on M1 macs) - see https://github.com/snowflakedb/snowflake-jdbc/issues/589#issuecomment-983944767
 ENV DESTINATION_SNOWFLAKE_OPTS "--add-opens java.base/java.nio=ALL-UNNAMED"
 

airbyte-integrations/connectors/destination-snowflake/build.gradle

@@ -30,6 +30,7 @@ dependencies {
     implementation 'org.apache.commons:commons-csv:1.4'
     implementation 'com.github.alexmojaki:s3-stream-upload:2.2.2'
     implementation "io.aesy:datasize:1.0.0"
+    implementation 'io.sentry:sentry-log4j2:5.5.3'
 
     implementation project(':airbyte-config:models')
     implementation project(':airbyte-db:lib')

airbyte-integrations/connectors/destination-snowflake/src/main/java/io/airbyte/integrations/destination/snowflake/SnowflakeDatabase.java

@@ -23,7 +23,8 @@ public class SnowflakeDatabase {
   private static final Duration QUERY_TIMEOUT = Duration.ofHours(3);
   private static final SnowflakeSQLNameTransformer nameTransformer = new SnowflakeSQLNameTransformer();
 
-  public static Connection getConnection(final JsonNode config) throws SQLException {
+  public static Connection getConnection(final JsonNode config)
+      throws SQLException {
     final String connectUrl = String.format("jdbc:snowflake://%s", config.get("host").asText());
 
     final Properties properties = new Properties();

airbyte-integrations/connectors/destination-snowflake/src/main/java/io/airbyte/integrations/destination/snowflake/SnowflakeDestination.java

@@ -9,6 +9,9 @@
 import io.airbyte.integrations.base.Destination;
 import io.airbyte.integrations.base.IntegrationRunner;
 import io.airbyte.integrations.destination.jdbc.copy.SwitchingDestination;
+import io.sentry.ITransaction;
+import io.sentry.Sentry;
+import io.sentry.SpanStatus;
 import java.util.Map;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -17,6 +20,18 @@ public class SnowflakeDestination extends SwitchingDestination<SnowflakeDestinat
 
   private static final Logger LOGGER = LoggerFactory.getLogger(SnowflakeDestination.class);
 
+  public static void initSentry() {
+    Sentry.init(options -> {
+      // allow setting properties from env variables see https://docs.sentry.io/platforms/java/configuration/
+      options.setEnableExternalConfiguration(true);
+      // To set a uniform sample rate
+      options.setTracesSampleRate(1.0);
+    });
+    Sentry.configureScope(scope -> {
+      scope.setTag("connector", "destination-snowflake");
+    });
+  }
+
   enum DestinationType {
     INSERT,
     COPY_S3,
@@ -65,10 +80,24 @@ public static Map<DestinationType, Destination> getTypeToDestination() {
   }
 
   public static void main(final String[] args) throws Exception {
+    initSentry();
+    try {
+      throw new Exception("This is hello from Sentry");
+    } catch (Exception e) {
+      LOGGER.error(e.getMessage());
+    }
     final Destination destination = new SnowflakeDestination();
-    LOGGER.info("starting destination: {}", SnowflakeDestination.class);
-    new IntegrationRunner(destination).run(args);
-    LOGGER.info("completed destination: {}", SnowflakeDestination.class);
+    ITransaction transaction = Sentry.startTransaction("IntegrationRunner()", "run");
+    try {
+      LOGGER.info("starting destination: {}", SnowflakeDestination.class);
+      new IntegrationRunner(destination).run(args);
+    } catch (Exception e) {
+      transaction.setThrowable(e);
+      transaction.setStatus(SpanStatus.INTERNAL_ERROR);
+    } finally {
+      transaction.finish();
+      LOGGER.info("completed destination: {}", SnowflakeDestination.class);
+    }
   }
 
 }

airbyte-integrations/connectors/destination-snowflake/src/main/resources/log4j2.xml

@@ -0,0 +1,11 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<Configuration status="warn" packages="org.apache.logging.log4j.core,io.sentry.log4j2">
+  <Appenders>
+    <Sentry name="Sentry"/>
+  </Appenders>
+  <Loggers>
+    <Root level="ERROR">
+      <AppenderRef ref="Sentry"/>
+    </Root>
+  </Loggers>
+</Configuration>