Export temporal metrics to datadog

airbyte-metrics/metrics-lib/build.gradle

@@ -11,6 +11,7 @@ dependencies {
     implementation libs.otel.semconv
     implementation libs.otel.sdk
     implementation libs.otel.sdk.testing
+    implementation libs.micrometer.statsd
     implementation platform(libs.otel.bom)
     implementation("io.opentelemetry:opentelemetry-api")
     implementation("io.opentelemetry:opentelemetry-sdk")

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

@@ -6,6 +6,10 @@
 
 import io.airbyte.config.Configs;
 import io.airbyte.config.EnvConfigs;
+import io.micrometer.core.instrument.Clock;
+import io.micrometer.core.instrument.MeterRegistry;
+import io.micrometer.statsd.StatsdConfig;
+import io.micrometer.statsd.StatsdMeterRegistry;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -16,7 +20,7 @@ public class MetricClientFactory {
 
   private static final Logger LOGGER = LoggerFactory.getLogger(MetricClientFactory.class);
 
-  private static final String DATADOG_METRIC_CLIENT = "datadog";
+  static final String DATADOG_METRIC_CLIENT = "datadog";
   private static final String OTEL_METRIC_CLIENT = "otel";
 
   private static final Configs configs = new EnvConfigs();
@@ -67,6 +71,47 @@ public static synchronized void initialize(final MetricEmittingApp metricEmittin
     }
   }
 
+  private static StatsdConfig getDatadogStatsDConfig() {
+    return new StatsdConfig() {
+
+      /**
+       * @return
+       */
+      @Override
+      public String host() {
+        return configs.getDDAgentHost();
+      }
+
+      /**
+       * @param key Key to lookup in the config.
+       * @return
+       */
+      @Override
+      public String get(String key) {
+        return null;
+      }
+
+    };
+  }
+
+  /**
+   *
+   * Returns a meter registry to be consumed by temporal configs.
+   *
+   */
+  public static MeterRegistry getMeterRegistry() {
+
+    if (DATADOG_METRIC_CLIENT.equals(configs.getMetricClient())) {
+      StatsdConfig config = getDatadogStatsDConfig();
+      return new StatsdMeterRegistry(config, Clock.SYSTEM);
+    }
+
+    // To support open telemetry, we need to use a different type of Config. For now we simply return
+    // null - in this case, we do not register any metric emitting mechanism in temporal and thus
+    // users will not receive temporal related metrics.
+    return null;
+  }
+
   private static DogStatsDMetricClient initializeDatadogMetricClient(
                                                                      final MetricEmittingApp metricEmittingApp) {
     final DogStatsDMetricClient client = new DogStatsDMetricClient();

airbyte-metrics/metrics-lib/src/test/java/io/airbyte/metrics/lib/MetricClientFactoryTest.java

@@ -6,6 +6,7 @@
 
 import static org.hamcrest.CoreMatchers.instanceOf;
 import static org.hamcrest.MatcherAssert.assertThat;
+import static org.junit.jupiter.api.Assertions.assertNull;
 
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.Assertions;
@@ -43,4 +44,10 @@ void testMetricClientFactoryCreateMultipleTimesThrows() {
     });
   }
 
+  @Test
+  @DisplayName("Should not return null if metric client not specified;")
+  void testMicroMeterRegistryRuturnsNullForEmptyClientConfig() {
+    assertNull(MetricClientFactory.getMeterRegistry());
+  }
+
 }

airbyte-workers/build.gradle

@@ -20,6 +20,7 @@ dependencies {
     implementation 'org.eclipse.jetty:jetty-servlet:9.4.31.v20200723'
     implementation 'org.quartz-scheduler:quartz:2.3.2'
     implementation libs.flyway.core
+    implementation libs.micrometer.statsd
 
     implementation project(':airbyte-analytics')
     implementation project(':airbyte-commons-docker')

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

@@ -233,8 +233,10 @@ private void registerSync(final WorkerFactory factory) {
     final PersistStateActivityImpl persistStateActivity = new PersistStateActivityImpl(statePersistence, featureFlags);
 
     final Worker syncWorker = factory.newWorker(TemporalJobType.SYNC.name(), getWorkerOptions(maxWorkers.getMaxSyncWorkers()));
+
     syncWorker.registerWorkflowImplementationTypes(SyncWorkflowImpl.class);
     syncWorker.registerActivitiesImplementations(replicationActivity, normalizationActivity, dbtTransformationActivity, persistStateActivity);
+
   }
 
   private void registerDiscover(final WorkerFactory factory) {
@@ -427,6 +429,7 @@ private static void launchWorkerApp(final Configs configs, final DSLContext conf
         new OAuthConfigSupplier(configRepository, trackingClient));
 
     final WorkflowServiceStubs temporalService = TemporalUtils.createTemporalService();
+
     final WorkflowClient workflowClient = TemporalUtils.createWorkflowClient(temporalService, TemporalUtils.getNamespace());
     final StreamResetPersistence streamResetPersistence = new StreamResetPersistence(configDatabase);
 

airbyte-workers/src/main/java/io/airbyte/workers/temporal/TemporalUtils.java

@@ -5,10 +5,15 @@
 package io.airbyte.workers.temporal;
 
 import com.google.common.annotations.VisibleForTesting;
+import com.uber.m3.tally.RootScopeBuilder;
+import com.uber.m3.tally.Scope;
+import com.uber.m3.tally.StatsReporter;
 import io.airbyte.commons.lang.Exceptions;
 import io.airbyte.config.Configs;
 import io.airbyte.config.EnvConfigs;
+import io.airbyte.metrics.lib.MetricClientFactory;
 import io.airbyte.scheduler.models.JobRunConfig;
+import io.micrometer.core.instrument.MeterRegistry;
 import io.temporal.activity.ActivityExecutionContext;
 import io.temporal.api.common.v1.WorkflowExecution;
 import io.temporal.api.namespace.v1.NamespaceConfig;
@@ -21,6 +26,7 @@
 import io.temporal.client.WorkflowOptions;
 import io.temporal.client.WorkflowStub;
 import io.temporal.common.RetryOptions;
+import io.temporal.common.reporter.MicrometerClientStatsReporter;
 import io.temporal.serviceclient.SimpleSslContextBuilder;
 import io.temporal.serviceclient.WorkflowServiceStubs;
 import io.temporal.serviceclient.WorkflowServiceStubsOptions;
@@ -90,15 +96,25 @@ public static WorkflowServiceStubs createTemporalService() {
   private static WorkflowServiceStubsOptions getCloudTemporalOptions() {
     final InputStream clientCert = new ByteArrayInputStream(configs.getTemporalCloudClientCert().getBytes(StandardCharsets.UTF_8));
     final InputStream clientKey = new ByteArrayInputStream(configs.getTemporalCloudClientKey().getBytes(StandardCharsets.UTF_8));
+    WorkflowServiceStubsOptions.Builder optionBuilder;
     try {
-      return WorkflowServiceStubsOptions.newBuilder()
+      optionBuilder = WorkflowServiceStubsOptions.newBuilder()
           .setSslContext(SimpleSslContextBuilder.forPKCS8(clientCert, clientKey).build())
-          .setTarget(configs.getTemporalCloudHost())
-          .build();
+          .setTarget(configs.getTemporalCloudHost());
     } catch (final SSLException e) {
       log.error("SSL Exception occurred attempting to establish Temporal Cloud options.");
       throw new RuntimeException(e);
     }
+
+    MeterRegistry registry = MetricClientFactory.getMeterRegistry();
+    if (registry != null) {
+      StatsReporter reporter = new MicrometerClientStatsReporter(registry);
+      Scope scope = new RootScopeBuilder()
+          .reporter(reporter)
+          .reportEvery(com.uber.m3.util.Duration.ofSeconds(10));
+      optionBuilder.setMetricsScope(scope);
+    }
+    return optionBuilder.build();
   }
 
   @VisibleForTesting

deps.toml

@@ -86,6 +86,8 @@ otel-bom = {module = "io.opentelemetry:opentelemetry-bom", version = "1.14.0"}
 otel-semconv = {module = "io.opentelemetry:opentelemetry-semconv", version = "1.14.0-alpha"}
 otel-sdk = {module = "io.opentelemetry:opentelemetry-sdk-metrics", version = "1.14.0"}
 otel-sdk-testing = {module = "io.opentelemetry:opentelemetry-sdk-metrics-testing", version = "1.13.0-alpha"}
+micrometer-statsd = {module = "io.micrometer:micrometer-registry-statsd", version = "1.9.2"}
+
 quartz-scheduler = {module="org.quartz-scheduler:quartz", version = "2.3.2"}
 
 [bundles]