Add datadog tracking to record schema validation errors

Author: alovew

airbyte-container-orchestrator/src/main/java/io/airbyte/container_orchestrator/ReplicationJobOrchestrator.java

@@ -97,7 +97,8 @@ public Optional<String> runJob() throws Exception {
         new NamespacingMapper(syncInput.getNamespaceDefinition(), syncInput.getNamespaceFormat(), syncInput.getPrefix()),
         new DefaultAirbyteDestination(workerConfigs, destinationLauncher),
         new AirbyteMessageTracker(),
-        new RecordSchemaValidator(WorkerUtils.mapStreamNamesToSchemas(syncInput)));
+        new RecordSchemaValidator(WorkerUtils.mapStreamNamesToSchemas(syncInput)),
+        configs);
 
     log.info("Running replication worker...");
     final Path jobRoot = WorkerUtils.getJobRoot(configs.getWorkspaceRoot(), jobRunConfig.getJobId(), jobRunConfig.getAttemptId());

airbyte-metrics/lib/src/main/java/io/airbyte/metrics/lib/OssMetricsRegistry.java

@@ -91,7 +91,10 @@ public enum OssMetricsRegistry implements MetricsRegistry {
       "oldest running job in seconds"),
   OVERALL_JOB_RUNTIME_IN_LAST_HOUR_BY_TERMINAL_STATE_SECS(MetricEmittingApps.METRICS_REPORTER,
       "overall_job_runtime_in_last_hour_by_terminal_state_secs",
-      "overall job runtime - scheduling and execution for all attempts - for jobs that reach terminal states in the last hour. tagged by terminal states.");
+      "overall job runtime - scheduling and execution for all attempts - for jobs that reach terminal states in the last hour. tagged by terminal states."),
+  NUM_RECORD_SCHEMA_VALIDATION_ERRORS(MetricEmittingApps.WORKER,
+      "record_schema_validation_error",
+      "number of record schema validation errors");
 
   private final MetricEmittingApp application;
   private final String metricName;

airbyte-workers/src/main/java/io/airbyte/workers/general/DefaultReplicationWorker.java

@@ -4,6 +4,7 @@
 
 package io.airbyte.workers.general;
 
+import io.airbyte.config.Configs;
 import io.airbyte.config.FailureReason;
 import io.airbyte.config.ReplicationAttemptSummary;
 import io.airbyte.config.ReplicationOutput;
@@ -14,6 +15,10 @@
 import io.airbyte.config.SyncStats;
 import io.airbyte.config.WorkerDestinationConfig;
 import io.airbyte.config.WorkerSourceConfig;
+import io.airbyte.metrics.lib.DatadogClientConfiguration;
+import io.airbyte.metrics.lib.DogStatsDMetricSingleton;
+import io.airbyte.metrics.lib.MetricEmittingApps;
+import io.airbyte.metrics.lib.OssMetricsRegistry;
 import io.airbyte.protocol.models.AirbyteMessage;
 import io.airbyte.protocol.models.AirbyteRecordMessage;
 import io.airbyte.workers.*;
@@ -77,14 +82,16 @@ public class DefaultReplicationWorker implements ReplicationWorker {
   private final AtomicBoolean cancelled;
   private final AtomicBoolean hasFailed;
   private final RecordSchemaValidator recordSchemaValidator;
+  private final Configs configs;
 
   public DefaultReplicationWorker(final String jobId,
                                   final int attempt,
                                   final AirbyteSource source,
                                   final AirbyteMapper mapper,
                                   final AirbyteDestination destination,
                                   final MessageTracker messageTracker,
-                                  final RecordSchemaValidator recordSchemaValidator) {
+                                  final RecordSchemaValidator recordSchemaValidator,
+                                  final Configs configs) {
     this.jobId = jobId;
     this.attempt = attempt;
     this.source = source;
@@ -93,11 +100,22 @@ public DefaultReplicationWorker(final String jobId,
     this.messageTracker = messageTracker;
     this.executors = Executors.newFixedThreadPool(2);
     this.recordSchemaValidator = recordSchemaValidator;
+    this.configs = configs;
 
     this.cancelled = new AtomicBoolean(false);
     this.hasFailed = new AtomicBoolean(false);
   }
 
+  public DefaultReplicationWorker(final String jobId,
+                                  final int attempt,
+                                  final AirbyteSource source,
+                                  final AirbyteMapper mapper,
+                                  final AirbyteDestination destination,
+                                  final MessageTracker messageTracker,
+                                  final RecordSchemaValidator recordSchemaValidator) {
+    this(jobId, attempt, source, mapper, destination, messageTracker, recordSchemaValidator, null);
+  }
+
   /**
    * Run executes two threads. The first pipes data from STDOUT of the source to STDIN of the
    * destination. The second listen on STDOUT of the destination. The goal of this second thread is to
@@ -111,7 +129,7 @@ public DefaultReplicationWorker(final String jobId,
    * @throws WorkerException
    */
   @Override
-  public ReplicationOutput run(final StandardSyncInput syncInput, final Path jobRoot) throws WorkerException {
+  public final ReplicationOutput run(final StandardSyncInput syncInput, final Path jobRoot) throws WorkerException {
     LOGGER.info("start sync worker. job id: {} attempt id: {}", jobId, attempt);
 
     // todo (cgardens) - this should not be happening in the worker. this is configuration information
@@ -153,7 +171,7 @@ public ReplicationOutput run(final StandardSyncInput syncInput, final Path jobRo
             });
 
         final CompletableFuture<?> replicationThreadFuture = CompletableFuture.runAsync(
-            getReplicationRunnable(source, destination, cancelled, mapper, messageTracker, mdc, recordSchemaValidator),
+            getReplicationRunnable(source, destination, cancelled, mapper, messageTracker, mdc, recordSchemaValidator, configs),
             executors).whenComplete((msg, ex) -> {
               if (ex != null) {
                 if (ex.getCause() instanceof SourceException) {
@@ -292,7 +310,8 @@ private static Runnable getReplicationRunnable(final AirbyteSource source,
                                                  final AirbyteMapper mapper,
                                                  final MessageTracker messageTracker,
                                                  final Map<String, String> mdc,
-                                                 final RecordSchemaValidator recordSchemaValidator) {
+                                                 final RecordSchemaValidator recordSchemaValidator,
+                                                 final Configs configs) {
     return () -> {
       MDC.setContextMap(mdc);
       LOGGER.info("Replication thread started.");
@@ -354,8 +373,15 @@ private static Runnable getReplicationRunnable(final AirbyteSource source,
         }
         LOGGER.info("Total records read: {} ({})", recordsRead, FileUtils.byteCountToDisplaySize(messageTracker.getTotalBytesEmitted()));
         if (!validationErrors.isEmpty()) {
+          DogStatsDMetricSingleton.initialize(MetricEmittingApps.WORKER, new DatadogClientConfiguration(configs));
           validationErrors.forEach((stream, errorPair) -> {
             LOGGER.warn("Schema validation errors found for stream {}. Error messages: {}", stream, errorPair.getLeft());
+            final String[] validationErrorMetadata = {
+                "docker_repo:airbyte/test", // dockerImage.split(":")[0]
+                "docker_version:0.0.0", // dockerImage.split(":")[1]
+                "stream:" + stream
+            };
+            DogStatsDMetricSingleton.count(OssMetricsRegistry.NUM_RECORD_SCHEMA_VALIDATION_ERRORS, 1, validationErrorMetadata);
           });
         }
 

airbyte-workers/src/main/java/io/airbyte/workers/temporal/sync/OrchestratorConstants.java

@@ -41,6 +41,7 @@ public class OrchestratorConstants {
           EnvConfigs.JOB_MAIN_CONTAINER_MEMORY_LIMIT,
           EnvConfigs.JOB_DEFAULT_ENV_MAP,
           EnvConfigs.LOCAL_ROOT,
+          EnvConfigs.PUBLISH_METRICS,
           LOG_LEVEL,
           LogClientSingleton.GCS_LOG_BUCKET,
           LogClientSingleton.GOOGLE_APPLICATION_CREDENTIALS,