Support DogStatsD style tags in statsd-emitter (#6605)

* Replace StatsD client library

The [Datadog package][1] is a StatsD compatible drop-in replacement for the
client library, but it seems to be [better maintained][2] and has support for
Datadog DogStatsD specific features, which will be made use of in a subsequent
commit.

The `count`, `time`, and `gauge` methods are actually exactly compatible with
the previous library and the modifications shouldn't be required, but EasyMock
seems to have a hard time dealing with the variable arguments added by the
DogStatsD library and causes tests to fail if no arguments are provided for the
last String vararg. Passing an empty array fixes the test failures.

[1]: https://github.com/DataDog/java-dogstatsd-client
[2]: https://github.com/tim-group/java-statsd-client/issues/37#issuecomment-248698856

* Retain dimension key information for StatsD metrics

This doesn't change behavior, but allows separating dimensions from the metric
name in subsequent commits.

There is a possible order change for values from
`dimsBuilder.build().values()`, but from the tests it looks like it doesn't
affect actual behavior and the order of user dimensions is also retained.

* Support DogStatsD style tags in statsd-emitter

Datadog [doesn't support name-encoded dimensions and uses a concept of _tags_
instead.][1] This change allows Datadog users to send the metrics without
having to encode the various dimensions in the metric names. This enables
building graphs and monitors with and without aggregation across various
dimensions from the same data.

As tests in this commit verify, the behavior remains the same for users who
don't enable the `druid.emitter.statsd.dogstatsd` configuration flag.

[1]: https://www.datadoghq.com/blog/the-power-of-tagged-metrics/#tags-decouple-collection-and-reporting

* Disable convertRange behavior for DogStatsD users

DogStatsD, unlike regular StatsD, supports floating-point values, so this
behavior is unnecessary. It would be possible to still support `convertRange`,
even with `dogstatsd` enabled, but that would mean that people using the
default mapping would have some of the gauges unnecessarily converted.

`time` is in milliseconds and doesn't support floating-point values.

docs/content/development/extensions-contrib/statsd.md

@@ -44,6 +44,7 @@ All the configuration parameters for the StatsD emitter are under `druid.emitter
 |`druid.emitter.statsd.includeHost`|Flag to include the hostname as part of the metric name.|no|false|  
 |`druid.emitter.statsd.dimensionMapPath`|JSON file defining the StatsD type, and desired dimensions for every Druid metric|no|Default mapping provided. See below.|  
 |`druid.emitter.statsd.blankHolder`|The blank character replacement as statsD does not support path with blank character|no|"-"|  
+|`druid.emitter.statsd.dogstatsd`|Flag to enable [DogStatsD](https://docs.datadoghq.com/developers/dogstatsd/) support. Causes dimensions to be included as tags, not as a part of the metric name. `convertRange` fields will be ignored.|no|false|
 
 ### Druid to StatsD Event Converter
 

extensions-contrib/statsd-emitter/pom.xml

@@ -41,9 +41,9 @@
       <scope>provided</scope>
     </dependency>
     <dependency>
-      <groupId>com.timgroup</groupId>
+      <groupId>com.datadoghq</groupId>
-      <artifactId>java-statsd-client</artifactId>
+      <artifactId>java-dogstatsd-client</artifactId>
-      <version>3.0.1</version>
+      <version>2.6.1</version>
     </dependency>
     <dependency>
       <groupId>junit</groupId>

extensions-contrib/statsd-emitter/src/main/java/org/apache/druid/emitter/statsd/DimensionConverter.java

@@ -22,7 +22,7 @@ package org.apache.druid.emitter.statsd;
 import com.fasterxml.jackson.core.type.TypeReference;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.google.common.base.Strings;
-import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
 import org.apache.druid.java.util.common.ISE;
 import org.apache.druid.java.util.common.logger.Logger;
 
@@ -49,7 +49,7 @@ public class DimensionConverter
       String service,
       String metric,
       Map<String, Object> userDims,
-      ImmutableList.Builder<String> builder
+      ImmutableMap.Builder<String, String> builder
   )
   {
      /*
@@ -65,7 +65,7 @@ public class DimensionConverter
     if (statsDMetric != null) {
       for (String dim : statsDMetric.dimensions) {
         if (userDims.containsKey(dim)) {
-          builder.add(userDims.get(dim).toString());
+          builder.put(dim, userDims.get(dim).toString());
         }
       }
       return statsDMetric;

extensions-contrib/statsd-emitter/src/main/java/org/apache/druid/emitter/statsd/StatsDEmitter.java

@@ -22,6 +22,7 @@ package org.apache.druid.emitter.statsd;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.google.common.base.Joiner;
 import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
 import com.timgroup.statsd.NonBlockingStatsDClient;
 import com.timgroup.statsd.StatsDClient;
 import com.timgroup.statsd.StatsDClientErrorHandler;
@@ -33,6 +34,7 @@ import org.apache.druid.java.util.emitter.service.ServiceMetricEvent;
 
 import java.util.Map;
 import java.util.regex.Pattern;
+import java.util.List;
 
 /**
  */
@@ -43,6 +45,7 @@ public class StatsDEmitter implements Emitter
   private static final char DRUID_METRIC_SEPARATOR = '/';
   private static final Pattern STATSD_SEPARATOR = Pattern.compile("[:|]");
   private static final Pattern BLANK = Pattern.compile("\\s+");
+  private static final String[] EMPTY_ARRAY = new String[0];
 
   static final StatsDEmitter of(StatsDEmitterConfig config, ObjectMapper mapper)
   {
@@ -50,6 +53,7 @@ public class StatsDEmitter implements Emitter
         config.getPrefix(),
         config.getHostname(),
         config.getPort(),
+        EMPTY_ARRAY,
         new StatsDClientErrorHandler()
         {
           private int exceptionCount = 0;
@@ -93,32 +97,70 @@ public class StatsDEmitter implements Emitter
       Number value = metricEvent.getValue();
 
       ImmutableList.Builder<String> nameBuilder = new ImmutableList.Builder<>();
-      if (config.getIncludeHost()) {
-        nameBuilder.add(host);
-      }
       nameBuilder.add(service);
       nameBuilder.add(metric);
 
-      StatsDMetric statsDMetric = converter.addFilteredUserDims(service, metric, userDims, nameBuilder);
+      ImmutableMap.Builder<String, String> dimsBuilder = new ImmutableMap.Builder<>();
+      StatsDMetric statsDMetric = converter.addFilteredUserDims(service, metric, userDims, dimsBuilder);
 
       if (statsDMetric != null) {
+        List<String> fullNameList;
+        String[] tags;
+        if (config.getDogstatsd()) {
+          if (config.getIncludeHost()) {
+            dimsBuilder.put("hostname", host);
+          }
 
-        String fullName = Joiner.on(config.getSeparator()).join(nameBuilder.build());
+          fullNameList = nameBuilder.build();
+          tags = dimsBuilder.build().entrySet()
+            .stream()
+            .map(e -> e.getKey() + ":" + e.getValue())
+            .toArray(String[]::new);
+        } else {
+          ImmutableList.Builder<String> fullNameBuilder = new ImmutableList.Builder<>();
+          if (config.getIncludeHost()) {
+            fullNameBuilder.add(host);
+          }
+          fullNameBuilder.addAll(nameBuilder.build());
+          fullNameBuilder.addAll(dimsBuilder.build().values());
+
+          fullNameList = fullNameBuilder.build();
+          tags = EMPTY_ARRAY;
+        }
+
+        String fullName = Joiner.on(config.getSeparator()).join(fullNameList);
         fullName = StringUtils.replaceChar(fullName, DRUID_METRIC_SEPARATOR, config.getSeparator());
         fullName = STATSD_SEPARATOR.matcher(fullName).replaceAll(config.getSeparator());
         fullName = BLANK.matcher(fullName).replaceAll(config.getBlankHolder());
 
-        long val = statsDMetric.convertRange ? Math.round(value.doubleValue() * 100) : value.longValue();
+        if (config.getDogstatsd() && (value instanceof Float || value instanceof Double)) {
-        switch (statsDMetric.type) {
+          switch (statsDMetric.type) {
-          case count:
+            case count:
-            statsd.count(fullName, val);
+              statsd.count(fullName, value.doubleValue(), tags);
-            break;
+              break;
-          case timer:
+            case timer:
-            statsd.time(fullName, val);
+              statsd.time(fullName, value.longValue(), tags);
-            break;
+              break;
-          case gauge:
+            case gauge:
-            statsd.gauge(fullName, val);
+              statsd.gauge(fullName, value.doubleValue(), tags);
-            break;
+              break;
+          }
+        } else {
+          long val = statsDMetric.convertRange && !config.getDogstatsd() ?
+              Math.round(value.doubleValue() * 100) :
+              value.longValue();
+
+          switch (statsDMetric.type) {
+            case count:
+              statsd.count(fullName, val, tags);
+              break;
+            case timer:
+              statsd.time(fullName, val, tags);
+              break;
+            case gauge:
+              statsd.gauge(fullName, val, tags);
+              break;
+          }
         }
       } else {
         log.debug("Metric=[%s] has no StatsD type mapping", statsDMetric);

extensions-contrib/statsd-emitter/src/main/java/org/apache/druid/emitter/statsd/StatsDEmitterConfig.java

@@ -42,6 +42,8 @@ public class StatsDEmitterConfig
   private final String dimensionMapPath;
   @JsonProperty
   private final String blankHolder;
+  @JsonProperty
+  private final Boolean dogstatsd;
 
   @JsonCreator
   public StatsDEmitterConfig(
@@ -51,7 +53,8 @@ public class StatsDEmitterConfig
       @JsonProperty("separator") String separator,
       @JsonProperty("includeHost") Boolean includeHost,
       @JsonProperty("dimensionMapPath") String dimensionMapPath,
-      @JsonProperty("blankHolder") String blankHolder
+      @JsonProperty("blankHolder") String blankHolder,
+      @JsonProperty("dogstatsd") Boolean dogstatsd
   )
   {
     this.hostname = Preconditions.checkNotNull(hostname, "StatsD hostname cannot be null.");
@@ -61,6 +64,7 @@ public class StatsDEmitterConfig
     this.includeHost = includeHost != null ? includeHost : false;
     this.dimensionMapPath = dimensionMapPath;
     this.blankHolder = blankHolder != null ? blankHolder : "-";
+    this.dogstatsd = dogstatsd != null ? dogstatsd : false;
   }
 
   @Override
@@ -90,7 +94,10 @@ public class StatsDEmitterConfig
     if (includeHost != null ? !includeHost.equals(that.includeHost) : that.includeHost != null) {
       return false;
     }
-    return dimensionMapPath != null ? dimensionMapPath.equals(that.dimensionMapPath) : that.dimensionMapPath == null;
+    if (dimensionMapPath != null ? !dimensionMapPath.equals(that.dimensionMapPath) : that.dimensionMapPath != null) {
+      return false;
+    }
+    return dogstatsd != null ? dogstatsd.equals(that.dogstatsd) : that.dogstatsd == null;
 
   }
 
@@ -104,6 +111,7 @@ public class StatsDEmitterConfig
     result = 31 * result + (includeHost != null ? includeHost.hashCode() : 0);
     result = 31 * result + (dimensionMapPath != null ? dimensionMapPath.hashCode() : 0);
     result = 31 * result + (blankHolder != null ? blankHolder.hashCode() : 0);
+    result = 31 * result + (dogstatsd != null ? dogstatsd.hashCode() : 0);
     return result;
   }
 
@@ -148,4 +156,10 @@ public class StatsDEmitterConfig
   {
     return blankHolder;
   }
+
+  @JsonProperty
+  public Boolean getDogstatsd()
+  {
+    return dogstatsd;
+  }
 }

extensions-contrib/statsd-emitter/src/test/java/org/apache/druid/emitter/statsd/DimensionConverterTest.java

@@ -20,7 +20,7 @@
 package org.apache.druid.emitter.statsd;
 
 import com.fasterxml.jackson.databind.ObjectMapper;
-import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
 import org.apache.druid.java.util.common.DateTimes;
 import org.apache.druid.java.util.emitter.service.ServiceMetricEvent;
 import org.junit.Test;
@@ -49,7 +49,7 @@ public class DimensionConverterTest
         .build(DateTimes.nowUtc(), "query/time", 10)
         .build("broker", "brokerHost1");
 
-    ImmutableList.Builder<String> actual = new ImmutableList.Builder<>();
+    ImmutableMap.Builder<String, String> actual = new ImmutableMap.Builder<>();
     StatsDMetric statsDMetric = dimensionConverter.addFilteredUserDims(
         event.getService(),
         event.getMetric(),
@@ -57,9 +57,9 @@ public class DimensionConverterTest
         actual
     );
     assertEquals("correct StatsDMetric.Type", StatsDMetric.Type.timer, statsDMetric.type);
-    ImmutableList.Builder<String> expected = new ImmutableList.Builder<>();
+    ImmutableMap.Builder<String, String> expected = new ImmutableMap.Builder<>();
-    expected.add("data-source");
+    expected.put("dataSource", "data-source");
-    expected.add("groupBy");
+    expected.put("type", "groupBy");
     assertEquals("correct Dimensions", expected.build(), actual.build());
   }
 }

extensions-contrib/statsd-emitter/src/test/java/org/apache/druid/emitter/statsd/StatsDEmitterTest.java

@@ -38,11 +38,30 @@ public class StatsDEmitterTest
   {
     StatsDClient client = createMock(StatsDClient.class);
     StatsDEmitter emitter = new StatsDEmitter(
-        new StatsDEmitterConfig("localhost", 8888, null, null, null, null, null),
+        new StatsDEmitterConfig("localhost", 8888, null, null, null, null, null, null),
         new ObjectMapper(),
         client
     );
-    client.gauge("broker.query.cache.total.hitRate", 54);
+    client.gauge("broker.query.cache.total.hitRate", 54, new String[0]);
+    replay(client);
+    emitter.emit(new ServiceMetricEvent.Builder()
+                     .setDimension("dataSource", "data-source")
+                     .build(DateTimes.nowUtc(), "query/cache/total/hitRate", 0.54)
+                     .build("broker", "brokerHost1")
+    );
+    verify(client);
+  }
+
+  @Test
+  public void testConvertRangeWithDogstatsd()
+  {
+    StatsDClient client = createMock(StatsDClient.class);
+    StatsDEmitter emitter = new StatsDEmitter(
+        new StatsDEmitterConfig("localhost", 8888, null, null, null, null, null, true),
+        new ObjectMapper(),
+        client
+    );
+    client.gauge("broker.query.cache.total.hitRate", 0.54, new String[0]);
     replay(client);
     emitter.emit(new ServiceMetricEvent.Builder()
                      .setDimension("dataSource", "data-source")
@@ -57,11 +76,11 @@ public class StatsDEmitterTest
   {
     StatsDClient client = createMock(StatsDClient.class);
     StatsDEmitter emitter = new StatsDEmitter(
-        new StatsDEmitterConfig("localhost", 8888, null, null, null, null, null),
+        new StatsDEmitterConfig("localhost", 8888, null, null, null, null, null, null),
         new ObjectMapper(),
         client
     );
-    client.time("broker.query.time.data-source.groupBy", 10);
+    client.time("broker.query.time.data-source.groupBy", 10, new String[0]);
     replay(client);
     emitter.emit(new ServiceMetricEvent.Builder()
                      .setDimension("dataSource", "data-source")
@@ -85,11 +104,40 @@ public class StatsDEmitterTest
   {
     StatsDClient client = createMock(StatsDClient.class);
     StatsDEmitter emitter = new StatsDEmitter(
-        new StatsDEmitterConfig("localhost", 8888, null, "#", true, null, null),
+        new StatsDEmitterConfig("localhost", 8888, null, "#", true, null, null, null),
         new ObjectMapper(),
         client
     );
-    client.time("brokerHost1#broker#query#time#data-source#groupBy", 10);
+    client.time("brokerHost1#broker#query#time#data-source#groupBy", 10, new String[0]);
+    replay(client);
+    emitter.emit(new ServiceMetricEvent.Builder()
+                     .setDimension("dataSource", "data-source")
+                     .setDimension("type", "groupBy")
+                     .setDimension("interval", "2013/2015")
+                     .setDimension("some_random_dim1", "random_dim_value1")
+                     .setDimension("some_random_dim2", "random_dim_value2")
+                     .setDimension("hasFilters", "no")
+                     .setDimension("duration", "P1D")
+                     .setDimension("remoteAddress", "194.0.90.2")
+                     .setDimension("id", "ID")
+                     .setDimension("context", "{context}")
+                     .build(DateTimes.nowUtc(), "query/time", 10)
+                     .build("broker", "brokerHost1")
+    );
+    verify(client);
+  }
+
+  @Test
+  public void testDogstatsdEnabled()
+  {
+    StatsDClient client = createMock(StatsDClient.class);
+    StatsDEmitter emitter = new StatsDEmitter(
+        new StatsDEmitterConfig("localhost", 8888, null, "#", true, null, null, true),
+        new ObjectMapper(),
+        client
+    );
+    client.time("broker#query#time", 10,
+        new String[] {"dataSource:data-source", "type:groupBy", "hostname:brokerHost1"});
     replay(client);
     emitter.emit(new ServiceMetricEvent.Builder()
                      .setDimension("dataSource", "data-source")
@@ -113,11 +161,11 @@ public class StatsDEmitterTest
   {
     StatsDClient client = createMock(StatsDClient.class);
     StatsDEmitter emitter = new StatsDEmitter(
-        new StatsDEmitterConfig("localhost", 8888, null, null, true, null, null),
+        new StatsDEmitterConfig("localhost", 8888, null, null, true, null, null, null),
         new ObjectMapper(),
         client
     );
-    client.count("brokerHost1.broker.jvm.gc.count.G1-GC", 1);
+    client.count("brokerHost1.broker.jvm.gc.count.G1-GC", 1, new String[0]);
     replay(client);
     emitter.emit(new ServiceMetricEvent.Builder()
                      .setDimension("gcName", "G1 GC")