Properly report metrics to datadog with the appropriate tags

This requires a much more recent version of our metrics-datadog library but
does result in the right values being reported into datadog.

build.gradle

@@ -35,6 +35,7 @@ repositories {
     jcenter()
 
     maven { url 'https://dl.bintray.com/rtyler/maven' }
+    maven { url 'https://dl.bintray.com/lookout/systems' }
 }
 
 dependencies {
@@ -55,7 +56,7 @@ dependencies {
     /* Needed for command line options parsing */
     compile 'commons-cli:commons-cli:1.2+'
 
-    compile 'com.timgroup:java-statsd-client:3.1.2+'
+    compile 'com.github.lookout:metrics-datadog:0.1.3'
 
     ['metrics-core', 'metrics-graphite'].each { artifactName ->
         compile "io.dropwizard.metrics:${artifactName}:3.1.0"

src/main/groovy/com/github/lookout/verspaetung/Main.groovy

@@ -16,11 +16,14 @@ import org.apache.curator.retry.ExponentialBackoffRetry
 import org.apache.curator.framework.CuratorFrameworkFactory
 import org.apache.curator.framework.CuratorFramework
 import org.apache.curator.framework.recipes.cache.TreeCache
+import org.coursera.metrics.datadog.DatadogReporter
+import org.coursera.metrics.datadog.transport.UdpTransport
 import org.slf4j.Logger
 import org.slf4j.LoggerFactory
 
 import com.codahale.metrics.*
 
+
 class Main {
     private static final String METRICS_PREFIX = 'verspaetung'
 
@@ -126,20 +129,24 @@ class Main {
                                       .convertDurationsTo(TimeUnit.MILLISECONDS)
                                       .build()
         }
+        else {
+            UdpTransport transport = new UdpTransport.Builder()
+                                                     .withPrefix(statsdPrefix)
+                                                     .build()
+
+            reporter = DatadogReporter.forRegistry(registry)
+                                      .withEC2Host()
+                                      .withTransport(transport)
+                                      .build()
+        }
 
         /* Start the reporter if we've got it */
         reporter?.start(1, TimeUnit.SECONDS)
 
         logger.info("Starting wait loop...")
-        while (true) {
-            Thread.sleep(1 * 1000)
+        synchronized(this) {
+            wait()
         }
-
-        logger.info("exiting..")
-        poller.die()
-        poller.join()
-
-        return
     }
 
     static void registerMetricFor(KafkaConsumer consumer,
@@ -153,7 +160,8 @@ class Main {
         ConsumerGauge gauge = new ConsumerGauge(consumer,
                                                 consumerOffsets,
                                                 topicOffsets)
-        this.registry.register(gauge.name, gauge)
+        consumerGauges.put(consumer, gauge)
+        this.registry.register(gauge.nameForRegistry, gauge)
     }
 
 

src/main/groovy/com/github/lookout/verspaetung/metrics/ConsumerGauge.groovy

@@ -3,16 +3,16 @@ package com.github.lookout.verspaetung.metrics
 import java.util.AbstractMap
 import com.codahale.metrics.Gauge
 import groovy.transform.TypeChecked
+import org.coursera.metrics.datadog.Tagged
 
 import com.github.lookout.verspaetung.KafkaConsumer
 import com.github.lookout.verspaetung.TopicPartition
 
-
 /**
  * Dropwizard Metrics Gauge for reporting the value of a given KafkaConsumer
  */
 @TypeChecked
-class ConsumerGauge implements Gauge<Integer> {
+class ConsumerGauge implements Gauge<Integer>, Tagged {
     protected KafkaConsumer consumer
     protected AbstractMap<KafkaConsumer, Integer> consumers
     protected AbstractMap<TopicPartition, Long> topics
@@ -37,7 +37,30 @@ class ConsumerGauge implements Gauge<Integer> {
         return ((Integer)this.topics[topicPartition]) - this.consumers[consumer]
     }
 
+    @Override
+    List<String> getTags() {
+        return ["partition:${this.consumer.partition}",
+                "topic:${this.consumer.topic}",
+                "consumer-group:${this.consumer.name}"
+                ].collect { s -> s.strings.join('') }
+    }
+
+    /**
+     * return a unique name for this gauge
+     */
+    String getNameForRegistry() {
+        return "${this.consumer.topic}.${this.consumer.partition}.${this.consumer.name}"
+    }
+
+    @Override
     String getName() {
-        return "verspaetung.${this.consumer.topic}.${this.consumer.partition}.${this.consumer.name}"
+        return this.consumer.topic
+
+        /* need to return this if we're just using the console or statsd
+         * reporters
+
+        return "${this.consumer.topic}.${this.consumer.partition}.${this.consumer.name}"
+
+         */
     }
 }

src/main/groovy/com/github/lookout/verspaetung/metrics/HeartbeatGauge.groovy

@@ -2,7 +2,6 @@ package com.github.lookout.verspaetung.metrics
 
 import com.codahale.metrics.*
 
-
 /**
  * A simple gauge that will always just return 1 indicating that the process is
  * alive