Add connector metadata as sentry tags (#10475)

* Pass worker metadata to connector

* Fix compilation

* Pass in job id and image from worker

* Remove application version

* Add default job environment variables

* Add back removed comment

* Rename env map to job metadata

* Fix env configs

* Read connector from application

* Use empty string

* Remove println

* Fix unit test

* Fix compilation error

* Introduce constants for worker env

* Add worker env to ENV_VARS_TO_TRANSFER

* Pass into getWorkerMetadata map to all constructions

* Format code

* Format octavia cli

* Fix test compilation

* Fix typos

Author: tuliren

airbyte-cdk/python/airbyte_cdk/models/airbyte_protocol.py

@@ -240,7 +240,7 @@ class Config:
     )
     spec: Optional[ConnectorSpecification] = None
     connectionStatus: Optional[AirbyteConnectionStatus] = None
-    catalog: Optional[AirbyteCatalog] = Field(None, description="catalog message: the calalog")
+    catalog: Optional[AirbyteCatalog] = Field(None, description="catalog message: the catalog")
     record: Optional[AirbyteRecordMessage] = Field(None, description="record message: the record")
     state: Optional[AirbyteStateMessage] = Field(
         None,

airbyte-commons/src/main/java/io/airbyte/commons/version/AirbyteVersion.java

@@ -12,7 +12,7 @@
  */
 public class AirbyteVersion {
 
-  private static final String DEV_VERSION = "dev";
+  public static final String DEV_VERSION = "dev";
   public static final String AIRBYTE_VERSION_KEY_NAME = "airbyte_version";
 
   private final String version;

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

@@ -7,6 +7,7 @@
 import com.google.common.base.Preconditions;
 import com.google.common.base.Splitter;
 import com.google.common.base.Strings;
+import io.airbyte.commons.map.MoreMaps;
 import io.airbyte.commons.version.AirbyteVersion;
 import io.airbyte.config.helpers.LogClientSingleton;
 import io.airbyte.config.helpers.LogConfigs;
@@ -20,6 +21,7 @@
 import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
+import java.util.Map.Entry;
 import java.util.Objects;
 import java.util.Optional;
 import java.util.Set;
@@ -168,6 +170,11 @@ public class EnvConfigs implements Configs {
 
   public static final String DEFAULT_NETWORK = "host";
 
+  public static final Map<String, Function<EnvConfigs, String>> JOB_SHARED_ENVS = Map.of(
+      AIRBYTE_VERSION, (instance) -> instance.getAirbyteVersion().serialize(),
+      AIRBYTE_ROLE, EnvConfigs::getAirbyteRole,
+      WORKER_ENVIRONMENT, (instance) -> instance.getWorkerEnvironment().name());
+
   public static final int DEFAULT_TEMPORAL_HISTORY_RETENTION_IN_DAYS = 30;
 
   private final Function<String, String> getEnv;
@@ -632,11 +639,22 @@ public String getJobMainContainerMemoryLimit() {
     return getEnvOrDefault(JOB_MAIN_CONTAINER_MEMORY_LIMIT, DEFAULT_JOB_MEMORY_REQUIREMENT);
   }
 
+  /**
+   * There are two types of environment variables available to the job container:
+   * <ul>
+   * <li>Exclusive variables prefixed with JOB_DEFAULT_ENV_PREFIX</li>
+   * <li>Shared variables defined in JOB_SHARED_ENVS</li>
+   * </ul>
+   */
   @Override
   public Map<String, String> getJobDefaultEnvMap() {
-    return getAllEnvKeys.get().stream()
+    final Map<String, String> jobPrefixedEnvMap = getAllEnvKeys.get().stream()
         .filter(key -> key.startsWith(JOB_DEFAULT_ENV_PREFIX))
         .collect(Collectors.toMap(key -> key.replace(JOB_DEFAULT_ENV_PREFIX, ""), getEnv));
+    final Map<String, String> jobSharedEnvMap = JOB_SHARED_ENVS.entrySet().stream().collect(Collectors.toMap(
+        Entry::getKey,
+        entry -> Objects.requireNonNullElse(entry.getValue().apply(this), "")));
+    return MoreMaps.merge(jobPrefixedEnvMap, jobSharedEnvMap);
   }
 
   @Override

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

@@ -0,0 +1,16 @@
+/*
+ * Copyright (c) 2021 Airbyte, Inc., all rights reserved.
+ */
+
+package io.airbyte.config;
+
+/**
+ * These extra env variables are created on the fly and passed to the connector.
+ */
+public class WorkerEnvConstants {
+
+  public static final String WORKER_CONNECTOR_IMAGE = "WORKER_CONNECTOR_IMAGE";
+  public static final String WORKER_JOB_ID = "WORKER_JOB_ID";
+  public static final String WORKER_JOB_ATTEMPT = "WORKER_JOB_ATTEMPT";
+
+}

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

@@ -7,6 +7,7 @@
 import static org.junit.jupiter.api.Assertions.*;
 
 import io.airbyte.commons.version.AirbyteVersion;
+import io.airbyte.config.Configs.WorkerEnvironment;
 import java.nio.file.Paths;
 import java.util.HashMap;
 import java.util.List;
@@ -343,16 +344,27 @@ void testCheckJobMemoryLimitTakePrecedentIfSet() {
   }
 
   @Test
-  void testEmptyEnvMapRetrieval() {
-    assertEquals(Map.of(), config.getJobDefaultEnvMap());
+  void testSharedJobEnvMapRetrieval() {
+    envMap.put(EnvConfigs.AIRBYTE_VERSION, "dev");
+    envMap.put(EnvConfigs.WORKER_ENVIRONMENT, WorkerEnvironment.KUBERNETES.name());
+    final Map<String, String> expected = Map.of("AIRBYTE_VERSION", "dev",
+        "AIRBYTE_ROLE", "",
+        "WORKER_ENVIRONMENT", "KUBERNETES");
+    assertEquals(expected, config.getJobDefaultEnvMap());
   }
 
   @Test
-  void testEnvMapRetrieval() {
+  void testAllJobEnvMapRetrieval() {
+    envMap.put(EnvConfigs.AIRBYTE_VERSION, "dev");
+    envMap.put(EnvConfigs.AIRBYTE_ROLE, "UNIT_TEST");
     envMap.put(EnvConfigs.JOB_DEFAULT_ENV_PREFIX + "ENV1", "VAL1");
     envMap.put(EnvConfigs.JOB_DEFAULT_ENV_PREFIX + "ENV2", "VAL\"2WithQuotesand$ymbols");
 
-    final var expected = Map.of("ENV1", "VAL1", "ENV2", "VAL\"2WithQuotesand$ymbols");
+    final Map<String, String> expected = Map.of("ENV1", "VAL1",
+        "ENV2", "VAL\"2WithQuotesand$ymbols",
+        "AIRBYTE_VERSION", "dev",
+        "AIRBYTE_ROLE", "UNIT_TEST",
+        "WORKER_ENVIRONMENT", "DOCKER");
     assertEquals(expected, config.getJobDefaultEnvMap());
   }
 

airbyte-integrations/bases/airbyte-protocol/airbyte_protocol/models/airbyte_protocol.py

@@ -226,7 +226,7 @@ class Config:
     log: Optional[AirbyteLogMessage] = Field(None, description="log message: any kind of logging you want the platform to know about.")
     spec: Optional[ConnectorSpecification] = None
     connectionStatus: Optional[AirbyteConnectionStatus] = None
-    catalog: Optional[AirbyteCatalog] = Field(None, description="catalog message: the calalog")
+    catalog: Optional[AirbyteCatalog] = Field(None, description="catalog message: the catalog")
     record: Optional[AirbyteRecordMessage] = Field(None, description="record message: the record")
     state: Optional[AirbyteStateMessage] = Field(
         None, description="schema message: the state. Must be the last message produced. The platform uses this information"

airbyte-integrations/bases/base-java/build.gradle

@@ -5,6 +5,7 @@ plugins {
 
 dependencies {
     implementation project(':airbyte-protocol:models')
+    implementation project(':airbyte-config:models')
     implementation project(':airbyte-commons-cli')
     implementation project(':airbyte-json-validation')
     api 'io.sentry:sentry:5.6.0'

airbyte-integrations/bases/base-java/src/main/java/io/airbyte/integrations/base/IntegrationRunner.java

@@ -12,13 +12,18 @@
 import io.airbyte.commons.lang.Exceptions.Procedure;
 import io.airbyte.commons.string.Strings;
 import io.airbyte.commons.util.AutoCloseableIterator;
+import io.airbyte.commons.version.AirbyteVersion;
+import io.airbyte.config.Configs.WorkerEnvironment;
+import io.airbyte.config.EnvConfigs;
+import io.airbyte.config.WorkerEnvConstants;
 import io.airbyte.integrations.base.sentry.AirbyteSentry;
 import io.airbyte.protocol.models.AirbyteConnectionStatus;
 import io.airbyte.protocol.models.AirbyteMessage;
 import io.airbyte.protocol.models.AirbyteMessage.Type;
 import io.airbyte.protocol.models.ConfiguredAirbyteCatalog;
 import io.airbyte.validation.json.JsonSchemaValidator;
 import io.sentry.ITransaction;
+import io.sentry.NoOpTransaction;
 import io.sentry.Sentry;
 import io.sentry.SentryLevel;
 import io.sentry.SpanStatus;
@@ -93,16 +98,10 @@ public IntegrationRunner(final Source source) {
   }
 
   public void run(final String[] args) throws Exception {
-    initSentry();
-
     final IntegrationConfig parsed = cliParser.parse(args);
-    final ITransaction transaction = Sentry.startTransaction(
-        integration.getClass().getSimpleName(),
-        parsed.getCommand().toString(),
-        true);
-    LOGGER.info("Sentry transaction event: {}", transaction.getEventId());
+    final ITransaction transaction = createSentryTransaction(integration.getClass(), parsed.getCommand());
     try {
-      runInternal(transaction, parsed);
+      runInternal(parsed);
       transaction.finish(SpanStatus.OK);
     } catch (final Exception e) {
       transaction.setThrowable(e);
@@ -117,7 +116,7 @@ public void run(final String[] args) throws Exception {
     }
   }
 
-  public void runInternal(final ITransaction transaction, final IntegrationConfig parsed) throws Exception {
+  private void runInternal(final IntegrationConfig parsed) throws Exception {
     LOGGER.info("Running integration: {}", integration.getClass().getName());
     LOGGER.info("Command: {}", parsed.getCommand());
     LOGGER.info("Integration config: {}", parsed);
@@ -285,21 +284,76 @@ private static <T> T parseConfig(final Path path, final Class<T> klass) {
     return Jsons.object(jsonNode, klass);
   }
 
-  private static void initSentry() {
+  private static ITransaction createSentryTransaction(final Class<?> connectorClass, final Command command) {
+    if (command == Command.SPEC) {
+      return NoOpTransaction.getInstance();
+    }
+
     final Map<String, String> env = System.getenv();
-    final String connector = env.getOrDefault("APPLICATION", "unknown");
-    final String version = env.getOrDefault("APPLICATION_VERSION", "unknown");
     final boolean enableSentry = Boolean.parseBoolean(env.getOrDefault("ENABLE_SENTRY", "false"));
+    final String sentryDsn = env.getOrDefault("SENTRY_DSN", "");
+    if (!enableSentry || sentryDsn.equals("")) {
+      LOGGER.debug("Skip Sentry transaction because DSN is not available");
+      return NoOpTransaction.getInstance();
+    }
+
+    final String version = parseConnectorVersion(env.getOrDefault("WORKER_CONNECTOR_IMAGE", ""));
+    final String airbyteVersion = env.getOrDefault(EnvConfigs.AIRBYTE_VERSION, "");
+    final String airbyteRole = env.getOrDefault(EnvConfigs.AIRBYTE_ROLE, "");
+    final boolean isDev = version.equals(AirbyteVersion.DEV_VERSION)
+        || airbyteVersion.equals(AirbyteVersion.DEV_VERSION)
+        || airbyteRole.equals("airbyter");
+    if (isDev) {
+      LOGGER.debug("Skip Sentry transaction for dev environment");
+      return NoOpTransaction.getInstance();
+    }
+    final String connector = env.getOrDefault("APPLICATION", "");
+    if (connector.equals("")) {
+      LOGGER.debug("Skip Sentry transaction for unknown connector");
+      return NoOpTransaction.getInstance();
+    }
+    final String workerEnvironment = env.getOrDefault("WORKER_ENVIRONMENT", "");
+    final String workerJobId = env.getOrDefault(WorkerEnvConstants.WORKER_JOB_ID, "");
+    final String workerJobAttempt = env.getOrDefault(WorkerEnvConstants.WORKER_JOB_ATTEMPT, "");
+    if (workerEnvironment.equals("") || workerEnvironment.equals(WorkerEnvironment.DOCKER.name())) {
+      LOGGER.debug("Skip Sentry transaction for unknown or docker deployment");
+      return NoOpTransaction.getInstance();
+    }
 
     // https://docs.sentry.io/platforms/java/configuration/
     Sentry.init(options -> {
-      options.setDsn(enableSentry ? env.getOrDefault("SENTRY_DSN", "") : "");
+      options.setDsn(sentryDsn);
       options.setEnableExternalConfiguration(true);
-      options.setTracesSampleRate(enableSentry ? 1.0 : 0.0);
+      options.setTracesSampleRate(1.0);
       options.setRelease(String.format("%s@%s", connector, version));
+      options.setEnvironment(isDev ? "dev" : "production");
       options.setTag("connector", connector);
       options.setTag("connector_version", version);
+      options.setTag("job_id", workerJobId);
+      options.setTag("job_attempt", workerJobAttempt);
+      options.setTag("airbyte_version", airbyteVersion);
+      options.setTag("worker_environment", workerEnvironment);
     });
+
+    final ITransaction transaction = Sentry.startTransaction(
+        connectorClass.getSimpleName(),
+        command.toString(),
+        true);
+    LOGGER.info("Sentry transaction event: {}", transaction.getEventId());
+    return transaction;
+  }
+
+  /**
+   * @param connectorImage Expected format: [organization/]image[:version]
+   */
+  @VisibleForTesting
+  static String parseConnectorVersion(final String connectorImage) {
+    if (connectorImage == null || connectorImage.equals("")) {
+      return "unknown";
+    }
+
+    final String[] tokens = connectorImage.split(":");
+    return tokens[tokens.length - 1];
   }
 
 }

airbyte-integrations/bases/base-java/src/main/java/io/airbyte/integrations/base/JavaBaseConstants.java

@@ -4,18 +4,20 @@
 
 package io.airbyte.integrations.base;
 
-public class JavaBaseConstants {
+public final class JavaBaseConstants {
 
-  public static String ARGS_CONFIG_KEY = "config";
-  public static String ARGS_CATALOG_KEY = "catalog";
-  public static String ARGS_STATE_KEY = "state";
+  private JavaBaseConstants() {}
 
-  public static String ARGS_CONFIG_DESC = "path to the json configuration file";
-  public static String ARGS_CATALOG_DESC = "input path for the catalog";
-  public static String ARGS_PATH_DESC = "path to the json-encoded state file";
+  public static final String ARGS_CONFIG_KEY = "config";
+  public static final String ARGS_CATALOG_KEY = "catalog";
+  public static final String ARGS_STATE_KEY = "state";
 
-  public static String COLUMN_NAME_AB_ID = "_airbyte_ab_id";
-  public static String COLUMN_NAME_EMITTED_AT = "_airbyte_emitted_at";
-  public static String COLUMN_NAME_DATA = "_airbyte_data";
+  public static final String ARGS_CONFIG_DESC = "path to the json configuration file";
+  public static final String ARGS_CATALOG_DESC = "input path for the catalog";
+  public static final String ARGS_PATH_DESC = "path to the json-encoded state file";
+
+  public static final String COLUMN_NAME_AB_ID = "_airbyte_ab_id";
+  public static final String COLUMN_NAME_EMITTED_AT = "_airbyte_emitted_at";
+  public static final String COLUMN_NAME_DATA = "_airbyte_data";
 
 }

airbyte-integrations/bases/base-java/src/test/java/io/airbyte/integrations/base/IntegrationRunnerTest.java

@@ -368,4 +368,14 @@ private void startSleepingThread(final List<Exception> caughtExceptions, final b
     });
   }
 
+  @Test
+  void testParseConnectorImage() {
+    assertEquals("unknown", IntegrationRunner.parseConnectorVersion(null));
+    assertEquals("unknown", IntegrationRunner.parseConnectorVersion(""));
+    assertEquals("1.0.1-alpha", IntegrationRunner.parseConnectorVersion("airbyte/destination-test:1.0.1-alpha"));
+    assertEquals("dev", IntegrationRunner.parseConnectorVersion("airbyte/destination-test:dev"));
+    assertEquals("1.0.1-alpha", IntegrationRunner.parseConnectorVersion("destination-test:1.0.1-alpha"));
+    assertEquals("1.0.1-alpha", IntegrationRunner.parseConnectorVersion(":1.0.1-alpha"));
+  }
+
 }

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

@@ -13,7 +13,6 @@ FROM airbyte/integration-base-java:dev
 WORKDIR /airbyte
 
 ENV APPLICATION destination-bigquery-denormalized
-ENV APPLICATION_VERSION 0.2.10
 ENV ENABLE_SENTRY true
 
 COPY --from=build /airbyte /airbyte

airbyte-integrations/connectors/destination-bigquery/Dockerfile

@@ -13,7 +13,6 @@ FROM airbyte/integration-base-java:dev
 WORKDIR /airbyte
 
 ENV APPLICATION destination-bigquery
-ENV APPLICATION_VERSION 0.6.11
 ENV ENABLE_SENTRY true
 
 COPY --from=build /airbyte /airbyte

airbyte-integrations/connectors/destination-dev-null/Dockerfile

@@ -13,7 +13,6 @@ FROM airbyte/integration-base-java:dev
 WORKDIR /airbyte
 
 ENV APPLICATION destination-dev-null
-ENV APPLICATION_VERSION 0.2.3
 ENV ENABLE_SENTRY true
 
 COPY --from=build /airbyte /airbyte

airbyte-integrations/connectors/destination-e2e-test/Dockerfile

@@ -13,7 +13,6 @@ FROM airbyte/integration-base-java:dev
 WORKDIR /airbyte
 
 ENV APPLICATION destination-e2e-test
-ENV APPLICATION_VERSION 0.2.2
 ENV ENABLE_SENTRY true
 
 COPY --from=build /airbyte /airbyte

airbyte-integrations/connectors/destination-snowflake/Dockerfile

@@ -18,7 +18,6 @@ COPY build/distributions/${APPLICATION}*.tar ${APPLICATION}.tar
 
 RUN tar xf ${APPLICATION}.tar --strip-components=1
 
-ENV APPLICATION_VERSION 0.4.17
 ENV ENABLE_SENTRY true
 
 LABEL io.airbyte.version=0.4.17

airbyte-integrations/connectors/source-e2e-test-cloud/Dockerfile

@@ -13,7 +13,6 @@ FROM airbyte/integration-base-java:dev
 WORKDIR /airbyte
 
 ENV APPLICATION source-e2e-test-cloud
-ENV APPLICATION_VERSION 2.1.0
 ENV ENABLE_SENTRY true
 
 COPY --from=build /airbyte /airbyte

airbyte-integrations/connectors/source-e2e-test/Dockerfile

@@ -13,7 +13,6 @@ FROM airbyte/integration-base-java:dev
 WORKDIR /airbyte
 
 ENV APPLICATION source-e2e-test
-ENV APPLICATION_VERSION=2.1.0
 ENV ENABLE_SENTRY true
 
 COPY --from=build /airbyte /airbyte

airbyte-workers/src/main/java/io/airbyte/workers/DbtTransformationRunner.java

@@ -104,6 +104,7 @@ public boolean transform(final String jobId,
               resourceRequirements,
               Map.of(KubeProcessFactory.JOB_TYPE, KubeProcessFactory.SYNC_JOB, KubeProcessFactory.SYNC_STEP, KubeProcessFactory.CUSTOM_STEP),
               Collections.emptyMap(),
+              Collections.emptyMap(),
               dbtArguments.toArray(new String[0]));
       LineGobbler.gobble(process.getInputStream(), LOGGER::info, CONTAINER_LOG_MDC_BUILDER);
       LineGobbler.gobble(process.getErrorStream(), LOGGER::error, CONTAINER_LOG_MDC_BUILDER);

airbyte-workers/src/main/java/io/airbyte/workers/normalization/DefaultNormalizationRunner.java

@@ -130,6 +130,7 @@ private boolean runProcess(final String jobId,
           resourceRequirements,
           Map.of(KubeProcessFactory.JOB_TYPE, KubeProcessFactory.SYNC_JOB, KubeProcessFactory.SYNC_STEP, KubeProcessFactory.NORMALISE_STEP),
           Collections.emptyMap(),
+          Collections.emptyMap(),
           args);
 
       LineGobbler.gobble(process.getInputStream(), LOGGER::info, CONTAINER_LOG_MDC_BUILDER);