Move to DogStatsD.

airbyte-metrics/build.gradle

@@ -10,4 +10,6 @@ dependencies {
     implementation 'io.prometheus:simpleclient_hotspot:0.12.0' // basic client instrumentation
     implementation 'io.prometheus:simpleclient_httpserver:0.12.0' // basic server to serve prometheus port
     implementation 'io.prometheus:simpleclient_pushgateway:0.12.0' // push libs for basic server
+
+    implementation 'com.datadoghq:java-dogstatsd-client:4.0.0'
 }

airbyte-metrics/src/main/java/io/airbyte/metrics/DogstatsdMetricSingleton.java

@@ -0,0 +1,128 @@
+/*
+ * Copyright (c) 2021 Airbyte, Inc., all rights reserved.
+ */
+
+package io.airbyte.metrics;
+
+import com.timgroup.statsd.NonBlockingStatsDClientBuilder;
+import com.timgroup.statsd.StatsDClient;
+import java.util.Map;
+import lombok.extern.slf4j.Slf4j;
+import org.slf4j.MDC;
+
+/**
+ * Light wrapper around the DogsStatsD client to make using the client slightly more ergonomic.
+ * <p>
+ * This class mainly exists to help Airbyte instrument/debug application on Airbyte Cloud.
+ * <p>
+ * Open source users are free to turn this on and consume the same metrics.
+ */
+@Slf4j
+public class DogstatsdMetricSingleton {
+
+  private static DogstatsdMetricSingleton instance;
+  private final StatsDClient statsDClient;
+  private final boolean instancePublish;
+
+  public DogstatsdMetricSingleton(final String appName, final boolean publish) {
+    instancePublish = publish;
+    statsDClient = new NonBlockingStatsDClientBuilder()
+        .prefix(appName)
+        .hostname(System.getenv("DD_AGENT_HOST"))
+        .port(Integer.parseInt(System.getenv("DD_DOGSTATSD_PORT")))
+        .build();
+  }
+
+  public static synchronized DogstatsdMetricSingleton getInstance() {
+    if (instance == null) {
+      throw new RuntimeException("You must initialize configuration with the initializeMonitoringServiceDaemon() method before getting an instance.");
+    }
+    return instance;
+  }
+
+  public synchronized static void initialize(final String appName, final Map<String, String> mdc, final boolean publish) {
+    if (instance != null) {
+      throw new RuntimeException("You cannot initialize configuration more than once.");
+    }
+    if (publish) {
+      MDC.setContextMap(mdc);
+      log.info("Starting DogStatsD client..");
+      // The second constructor argument ('true') makes this server start as a separate daemon thread.
+      // http://prometheus.github.io/client_java/io/prometheus/client/exporter/HTTPServer.html#HTTPServer-int-boolean-
+      instance = new DogstatsdMetricSingleton(appName, publish);
+    }
+  }
+
+  /**
+   * Increment or decrement a counter.
+   *
+   * @param name of counter.
+   * @param amt to adjust.
+   * @param tags
+   */
+  public void count(final String name, final double amt, final String... tags) {
+    if (instancePublish) {
+      log.info("publishing count, name: {}, value: {}", name, amt);
+      statsDClient.count(name, amt, tags);
+    }
+  }
+
+  /**
+   * Record the latest value for a gauge.
+   *
+   * @param name of gauge.
+   * @param val to record.
+   * @param tags
+   */
+  public void gauge(final String name, final double val, final String... tags) {
+    if (instancePublish) {
+      log.info("publishing gauge, name: {}, value: {}", name, val);
+      statsDClient.gauge(name, val, tags);
+    }
+  }
+
+  /**
+   * Submit a single execution time aggregated locally by the Agent. Use this if approximate stats are
+   * sufficient.
+   *
+   * @param name of histogram.
+   * @param val of time to record.
+   * @param tags
+   */
+  public void recordTimeLocal(final String name, final double val, final String... tags) {
+    if (instancePublish) {
+      log.info("recording histogram, name: {}, value: {}", name, val);
+      statsDClient.histogram(name, val, tags);
+    }
+  }
+
+  /**
+   * Submit a single execution time aggregated globally by Datadog. Use this for precise stats.
+   *
+   * @param name of distribution.
+   * @param val of time to record.
+   * @param tags
+   */
+  public void recordTimeGlobal(final String name, final double val, final String... tags) {
+    if (instancePublish) {
+      log.info("recording distribution, name: {}, value: {}", name, val);
+      statsDClient.distribution(name, val, tags);
+    }
+  }
+
+  /**
+   * Wrapper of {@link #recordTimeGlobal(String, double, String...)} with a runnable for convenience.
+   *
+   * @param name
+   * @param runnable
+   * @param tags
+   */
+  public void recordTimeGlobal(final String name, final Runnable runnable, final String... tags) {
+    final long start = System.currentTimeMillis();
+    runnable.run();
+    final long end = System.currentTimeMillis();
+    final long val = end - start;
+    recordTimeGlobal(name, val, tags);
+  }
+
+}

airbyte-metrics/src/main/java/io/airbyte/metrics/MetricSingleton.java → airbyte-metrics/src/main/java/io/airbyte/metrics/PrometheusMetricSingleton.java

@@ -27,20 +27,21 @@
  * <p>
  * Open source users are free to turn this on and consume the same metrics.
  */
-public class MetricSingleton {
+@Deprecated
+public class PrometheusMetricSingleton {
 
-  private static final Logger LOGGER = LoggerFactory.getLogger(MetricSingleton.class);
-  private static MetricSingleton instance;
+  private static final Logger LOGGER = LoggerFactory.getLogger(PrometheusMetricSingleton.class);
+  private static PrometheusMetricSingleton instance;
 
   private final Map<String, Gauge> nameToGauge = new HashMap<>();
   private final Map<String, Counter> nameToCounter = new HashMap<>();
   private final Map<String, Histogram> nameToHistogram = new HashMap<>();
 
   private HTTPServer monitoringDaemon;
 
-  private MetricSingleton() {}
+  private PrometheusMetricSingleton() {}
 
-  public static synchronized MetricSingleton getInstance() {
+  public static synchronized PrometheusMetricSingleton getInstance() {
     if (instance == null) {
       throw new RuntimeException("You must initialize configuration with the initializeMonitoringServiceDaemon() method before getting an instance.");
     }
@@ -193,7 +194,7 @@ public synchronized static void initializeMonitoringServiceDaemon(final String m
     if (instance != null) {
       throw new RuntimeException("You cannot initialize configuration more than once.");
     }
-    instance = new MetricSingleton();
+    instance = new PrometheusMetricSingleton();
     if (publish) {
       try {
         MDC.setContextMap(mdc);

airbyte-metrics/src/test/java/io/airbyte/metrics/MetricSingletonTest.java → airbyte-metrics/src/test/java/io/airbyte/metrics/PrometheusMetricSingletonTest.java

@@ -25,7 +25,7 @@
  * Since Prometheus publishes metrics at a specific port, we can test our wrapper by querying the
  * port and validating the response.
  */
-public class MetricSingletonTest {
+public class PrometheusMetricSingletonTest {
 
   private static final HttpClient HTTP_CLIENT = HttpClient.newBuilder()
       .version(HttpClient.Version.HTTP_1_1)
@@ -41,25 +41,26 @@ public static void setUp() throws IOException {
       availPort = socket.getLocalPort();
     }
 
-    MetricSingleton.initializeMonitoringServiceDaemon(String.valueOf(availPort), Map.of(), true);
+    PrometheusMetricSingleton.initializeMonitoringServiceDaemon(String.valueOf(availPort), Map.of(), true);
   }
 
   @AfterAll
   public static void tearDown() {
-    MetricSingleton.getInstance().closeMonitoringServiceDaemon();
+    PrometheusMetricSingleton.getInstance().closeMonitoringServiceDaemon();
   }
 
   @Nested
   class Validation {
 
     @Test
     public void testNameWithDashFails() {
-      assertThrows(RuntimeException.class, () -> MetricSingleton.getInstance().incrementCounter("bad-name", 0.0, "name with dashes are not allowed"));
+      assertThrows(RuntimeException.class,
+          () -> PrometheusMetricSingleton.getInstance().incrementCounter("bad-name", 0.0, "name with dashes are not allowed"));
     }
 
     @Test
     public void testNoDescriptionFails() {
-      assertThrows(RuntimeException.class, () -> MetricSingleton.getInstance().incrementCounter("good_name", 0.0, null));
+      assertThrows(RuntimeException.class, () -> PrometheusMetricSingleton.getInstance().incrementCounter("good_name", 0.0, null));
     }
 
   }
@@ -69,7 +70,7 @@ public void testCounter() throws InterruptedException, IOException {
     final var metricName = "test_counter";
     final var rand = new Random();
     for (int i = 0; i < 5; i++) {
-      MetricSingleton.getInstance().incrementCounter(metricName, rand.nextDouble() * 2, "testing counter");
+      PrometheusMetricSingleton.getInstance().incrementCounter(metricName, rand.nextDouble() * 2, "testing counter");
       Thread.sleep(500);
     }
 
@@ -82,7 +83,7 @@ public void testGauge() throws InterruptedException, IOException {
     final var metricName = "test_gauge";
     final var rand = new Random();
     for (int i = 0; i < 5; i++) {
-      MetricSingleton.getInstance().incrementCounter(metricName, rand.nextDouble() * 2, "testing gauge");
+      PrometheusMetricSingleton.getInstance().incrementCounter(metricName, rand.nextDouble() * 2, "testing gauge");
       Thread.sleep(500);
     }
 
@@ -95,7 +96,7 @@ public void testTimer() throws InterruptedException, IOException {
     final var metricName = "test_timer";
     final var rand = new Random();
     for (int i = 0; i < 5; i++) {
-      MetricSingleton.getInstance().recordTime(metricName, rand.nextDouble() * 2, "testing time");
+      PrometheusMetricSingleton.getInstance().recordTime(metricName, rand.nextDouble() * 2, "testing time");
       Thread.sleep(500);
     }
 

airbyte-scheduler/app/src/main/java/io/airbyte/scheduler/app/SchedulerApp.java

@@ -29,7 +29,7 @@
 import io.airbyte.db.Database;
 import io.airbyte.db.instance.configs.ConfigsDatabaseInstance;
 import io.airbyte.db.instance.jobs.JobsDatabaseInstance;
-import io.airbyte.metrics.MetricSingleton;
+import io.airbyte.metrics.PrometheusMetricSingleton;
 import io.airbyte.scheduler.models.Job;
 import io.airbyte.scheduler.models.JobStatus;
 import io.airbyte.scheduler.persistence.DefaultJobPersistence;
@@ -273,7 +273,7 @@ public static void main(final String[] args) throws IOException, InterruptedExce
     final TemporalClient temporalClient = TemporalClient.production(temporalHost, workspaceRoot, configs);
 
     final Map<String, String> mdc = MDC.getCopyOfContextMap();
-    MetricSingleton.initializeMonitoringServiceDaemon("8082", mdc, configs.getPublishMetrics());
+    PrometheusMetricSingleton.initializeMonitoringServiceDaemon("8082", mdc, configs.getPublishMetrics());
 
     LOGGER.info("Launching scheduler...");
     new SchedulerApp(