statsd-emitter (apache#2410)

META-INF/services/io.druid.initialization.DruidModule

@@ -0,0 +1 @@
+io.druid.emitter.statsd.StatsDEmitterModule

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

@@ -0,0 +1,46 @@
+---
+layout: doc_page
+---
+
+# StatsD Emitter
+
+To use this extension, make sure to [include](../../operations/including-extensions.html) `statsd-emitter` extension.
+
+## Introduction
+
+This extension emits druid metrics to a StatsD server.
+(https://github.com/etsy/statsd)
+(https://github.com/armon/statsite)
+
+## Configuration
+
+All the configuration parameters for the StatsD emitter are under `druid.emitter.statsd`.
+
+|property|description|required?|default|
+|--------|-----------|---------|-------|
+|`druid.emitter.statsd.hostname`|The hostname of the StatsD server.|yes|none|
+|`druid.emitter.statsd.port`|The port of the StatsD server.|yes|none|
+|`druid.emitter.statsd.prefix`|Optional metric name prefix.|no|""|
+|`druid.emitter.statsd.separator`|Metric name separator|no|.|  
+|`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 to StatsD Event Converter
+
+Each metric sent to StatsD must specify a type, one of `[timer, counter, guage]`. StatsD Emitter expects this mapping to
+be provided as a JSON file.  Additionally, this mapping specifies which dimensions should be included for each metric.
+If the user does not specify their own JSON file, a default mapping is used.  All
+metrics are expected to be mapped. Metrics which are not mapped will log an error.
+StatsD metric path is organized using the following schema:
+`<druid metric name> : { "dimensions" : <dimension list>, "type" : <StatsD type>}`
+e.g.
+`query/time" : { "dimensions" : ["dataSource", "type"], "type" : "timer"}`
+
+For metrics which are emitted from multiple services with different dimensions, the metric name is prefixed with 
+the service name. 
+e.g.
+`"coordinator-segment/count" : { "dimensions" : ["dataSource"], "type" : "gauge" },
+ "historical-segment/count" : { "dimensions" : ["dataSource", "tier", "priority"], "type" : "gauge" }`
+
+For most use-cases, the default mapping is sufficient.
+

extensions-contrib/statsd-emitter/pom.xml

@@ -0,0 +1,90 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+ ~ Licensed to Metamarkets Group Inc. (Metamarkets) under one
+ ~ or more contributor license agreements. See the NOTICE file
+ ~ distributed with this work for additional information
+ ~ regarding copyright ownership. Metamarkets licenses this file
+ ~ to you under the Apache License, Version 2.0 (the
+ ~ "License"); you may not use this file except in compliance
+ ~ with the License. You may obtain a copy of the License at
+ ~
+ ~ http://www.apache.org/licenses/LICENSE-2.0
+ ~
+ ~ Unless required by applicable law or agreed to in writing,
+ ~ software distributed under the License is distributed on an
+ ~ "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ ~ KIND, either express or implied. See the License for the
+ ~ specific language governing permissions and limitations
+ ~ under the License.
+  -->
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+  <parent>
+    <artifactId>druid</artifactId>
+    <groupId>io.druid</groupId>
+    <version>0.9.1-SNAPSHOT</version>
+    <relativePath>../../pom.xml</relativePath>
+  </parent>
+  <modelVersion>4.0.0</modelVersion>
+
+  <groupId>io.druid.extensions.contrib</groupId>
+  <artifactId>statsd-emitter</artifactId>
+  <name>statsd-emitter</name>
+  <description>Extension support for emitting Druid metrics to StatsD</description>
+
+  <dependencies>
+    <dependency>
+      <groupId>io.druid</groupId>
+      <artifactId>druid-common</artifactId>
+      <version>${project.parent.version}</version>
+      <scope>provided</scope>
+    </dependency>
+    <dependency>
+      <groupId>io.druid</groupId>
+      <artifactId>druid-api</artifactId>
+      <version>${project.parent.version}</version>
+      <scope>provided</scope>
+    </dependency>
+    <dependency>
+      <groupId>com.metamx</groupId>
+      <artifactId>emitter</artifactId>
+      <scope>provided</scope>
+    </dependency>
+    <dependency>
+      <groupId>com.timgroup</groupId>
+      <artifactId>java-statsd-client</artifactId>
+      <version>3.0.1</version>
+    </dependency>
+    <dependency>
+      <groupId>junit</groupId>
+      <artifactId>junit</artifactId>
+      <scope>test</scope>
+    </dependency>
+    <dependency>
+      <groupId>org.easymock</groupId>
+      <artifactId>easymock</artifactId>
+      <scope>test</scope>
+    </dependency>
+    <dependency>
+      <groupId>pl.pragmatists</groupId>
+      <artifactId>JUnitParams</artifactId>
+      <version>1.0.4</version>
+      <scope>test</scope>
+    </dependency>
+    <dependency>
+      <groupId>io.druid</groupId>
+      <artifactId>druid-server</artifactId>
+      <version>${project.parent.version}</version>
+      <type>test-jar</type>
+      <scope>test</scope>
+    </dependency>
+    <dependency>
+      <groupId>io.druid</groupId>
+      <artifactId>druid-processing</artifactId>
+      <version>${project.parent.version}</version>
+      <type>test-jar</type>
+      <scope>test</scope>
+    </dependency>
+  </dependencies>
+</project>

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

@@ -0,0 +1,91 @@
+/*
+ *  Licensed to Metamarkets Group Inc. (Metamarkets) under one
+ *  or more contributor license agreements. See the NOTICE file
+ *  distributed with this work for additional information
+ *  regarding copyright ownership. Metamarkets licenses this file
+ *  to you under the Apache License, Version 2.0 (the
+ *  "License"); you may not use this file except in compliance
+ *  with the License. You may obtain a copy of the License at
+ *
+ *  http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *  Unless required by applicable law or agreed to in writing,
+ *  software distributed under the License is distributed on an
+ *  "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ *  KIND, either express or implied. See the License for the
+ *  specific language governing permissions and limitations
+ *  under the License.
+ */
+
+package io.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.metamx.common.ISE;
+import com.metamx.common.logger.Logger;
+
+import java.io.File;
+import java.io.FileInputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.util.Map;
+
+/**
+ */
+public class DimensionConverter
+{
+
+  private final static Logger log = new Logger(DimensionConverter.class);
+  private Map<String, StatsDMetric> metricMap;
+
+  public DimensionConverter(ObjectMapper mapper, String dimensionMapPath)
+  {
+    metricMap = readMap(mapper, dimensionMapPath);
+  }
+
+  public StatsDMetric.Type addFilteredUserDims(String service, String metric, Map<String, Object> userDims, ImmutableList.Builder<String> builder)
+  {
+     /*
+        Find the metric in the map. If we cant find it try to look it up prefixed by the service name.
+        This is because some metrics are reported differently, but with the same name, from different services.
+       */
+    StatsDMetric statsDMetric = null;
+    if (metricMap.containsKey(metric)) {
+      statsDMetric = metricMap.get(metric);
+    } else if (metricMap.containsKey(service + "-" + metric)) {
+      statsDMetric = metricMap.get(service + "-" + metric);
+    }
+    if (statsDMetric != null) {
+      for (String dim : statsDMetric.dimensions) {
+        if (userDims.containsKey(dim)) {
+          builder.add(userDims.get(dim).toString());
+        }
+      }
+      return statsDMetric.type;
+    } else {
+      return null;
+    }
+  }
+
+  private Map<String, StatsDMetric> readMap(ObjectMapper mapper, String dimensionMapPath)
+  {
+    try {
+      InputStream is;
+      if (Strings.isNullOrEmpty(dimensionMapPath)) {
+        log.info("Using default metric dimension and types");
+        is = this.getClass().getClassLoader().getResourceAsStream("defaultMetricDimensions.json");
+      } else {
+        log.info("Using metric dimensions at types at [%s]", dimensionMapPath);
+        is = new FileInputStream(new File(dimensionMapPath));
+      }
+      return mapper.reader(new TypeReference<Map<String, StatsDMetric>>()
+      {
+      }).readValue(is);
+    }
+    catch (IOException e) {
+      throw new ISE(e, "Failed to parse metric dimensions and types");
+    }
+  }
+}

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

@@ -0,0 +1,128 @@
+/*
+ *  Licensed to Metamarkets Group Inc. (Metamarkets) under one
+ *  or more contributor license agreements. See the NOTICE file
+ *  distributed with this work for additional information
+ *  regarding copyright ownership. Metamarkets licenses this file
+ *  to you under the Apache License, Version 2.0 (the
+ *  "License"); you may not use this file except in compliance
+ *  with the License. You may obtain a copy of the License at
+ *
+ *  http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *  Unless required by applicable law or agreed to in writing,
+ *  software distributed under the License is distributed on an
+ *  "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ *  KIND, either express or implied. See the License for the
+ *  specific language governing permissions and limitations
+ *  under the License.
+ */
+
+package io.druid.emitter.statsd;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.common.base.Joiner;
+import com.google.common.collect.ImmutableList;
+import com.metamx.common.logger.Logger;
+import com.metamx.emitter.core.Emitter;
+import com.metamx.emitter.core.Event;
+import com.metamx.emitter.service.ServiceMetricEvent;
+import com.timgroup.statsd.NonBlockingStatsDClient;
+import com.timgroup.statsd.StatsDClient;
+import com.timgroup.statsd.StatsDClientErrorHandler;
+
+import java.io.IOException;
+import java.util.Map;
+
+/**
+ */
+public class StatsDEmitter implements Emitter
+{
+
+  private final static Logger log = new Logger(StatsDEmitter.class);
+  private final static String DRUID_METRIC_SEPARATOR = "\\/";
+  private final static String STATSD_SEPARATOR = ":|\\|";
+
+  private final StatsDClient statsd;
+  private final StatsDEmitterConfig config;
+  private final DimensionConverter converter;
+
+  public StatsDEmitter(StatsDEmitterConfig config, ObjectMapper mapper) {
+    this.config = config;
+    this.converter = new DimensionConverter(mapper, config.getDimensionMapPath());
+    statsd = new NonBlockingStatsDClient(
+        config.getPrefix(),
+        config.getHostname(),
+        config.getPort(),
+        new StatsDClientErrorHandler()
+        {
+          private int exceptionCount = 0;
+          @Override
+          public void handle(Exception exception)
+          {
+            if (exceptionCount % 1000 == 0) {
+              log.error(exception, "Error sending metric to StatsD.");
+            }
+            exceptionCount += 1;
+          }
+        }
+    );
+  }
+
+
+  @Override
+  public void start() {}
+
+  @Override
+  public void emit(Event event)
+  {
+    if (event instanceof ServiceMetricEvent) {
+      ServiceMetricEvent metricEvent = (ServiceMetricEvent) event;
+      String host = metricEvent.getHost();
+      String service = metricEvent.getService();
+      String metric = metricEvent.getMetric();
+      Map<String, Object> userDims = metricEvent.getUserDims();
+      Number value = metricEvent.getValue();
+
+      ImmutableList.Builder<String> nameBuilder = new ImmutableList.Builder<>();
+      if (config.getIncludeHost()) {
+        nameBuilder.add(host);
+      }
+      nameBuilder.add(service);
+      nameBuilder.add(metric);
+
+      StatsDMetric.Type metricType = converter.addFilteredUserDims(service, metric, userDims, nameBuilder);
+
+      if (metricType != null) {
+
+        String fullName = Joiner.on(config.getSeparator())
+                                .join(nameBuilder.build())
+                                .replaceAll(DRUID_METRIC_SEPARATOR, config.getSeparator())
+                                .replaceAll(STATSD_SEPARATOR, config.getSeparator());
+
+        switch (metricType) {
+          case count:
+            statsd.count(fullName, value.longValue());
+            break;
+          case timer:
+            statsd.time(fullName, value.longValue());
+            break;
+          case gauge:
+            statsd.gauge(fullName, value.longValue());
+            break;
+        }
+      } else {
+        log.error("Metric=[%s] has no StatsD type mapping", metric);
+      }
+    }
+  }
+
+  @Override
+  public void flush() throws IOException {}
+
+  @Override
+  public void close() throws IOException
+  {
+    statsd.stop();
+  }
+
+}