Merge pull request apache#33 from Shopify/fix_datadog_metric_names

Fix Datadog metric name sanitization

Author: charlescng

core/src/main/java/com/bealetech/metrics/reporting/Statsd.java

@@ -28,6 +28,9 @@ public static enum StatType { COUNTER, TIMER, GAUGE }
     private final String host;
     private final int port;
 
+    private String prefix = "spark";
+    private String appPrefix = "spark.app-";
+
     private boolean prependNewline = false;
 
     private ByteArrayOutputStream outputData;
@@ -54,6 +57,38 @@ public void connect() throws IllegalStateException, SocketException {
         this.writer = new BufferedWriter(new OutputStreamWriter(outputData));
     }
 
+    public void setNamePrefix(String namePrefix) {
+        prefix = namePrefix;
+        appPrefix = namePrefix + ".app-";
+    }
+
+    private String buildMetricName(String rawName) throws IllegalArgumentException {
+        rawName = WHITESPACE.matcher(rawName).replaceAll("-");
+
+        // e.g. spark.worker.executors
+        if (!rawName.startsWith(appPrefix)) {
+            return rawName;
+        }
+
+        String[] parts = rawName.split("\\.");
+        if (parts.length < 5) {
+            throw new IllegalArgumentException("A spark app metric name must contain at least 4 parts: " + rawName);
+        }
+
+        StringBuilder stringBuilder = new StringBuilder(prefix);
+        if ("driver".equals(parts[2])) {
+            // e.g. spark.app-20141209201233-0145.driver.BlockManager.memory.maxMem_MB
+            stringBuilder.append(rawName.substring(rawName.indexOf(".driver.")));
+        } else if ("executor".equals(parts[3])) {
+            // e.g. spark.app-20141209201027-0139.31.executor.filesystem.file.read_bytes
+            stringBuilder.append(rawName.substring(rawName.indexOf(".executor.")));
+        } else {
+            throw new IllegalArgumentException("Unrecognized metric name pattern: " + rawName);
+        }
+
+        return stringBuilder.toString();
+    }
+
     public void send(String name, String value, StatType statType) throws IOException {
         String statTypeStr = "";
         switch (statType) {
@@ -67,32 +102,14 @@ public void send(String name, String value, StatType statType) throws IOExceptio
                 statTypeStr = "ms";
                 break;
         }
-        name = sanitizeString(name);
-        String tags = null;
-        List<String> parts = new ArrayList<String>(Arrays.asList(name.split("\\.")));
-        String prefix = parts.remove(0);
-        String source = parts.remove(0);
-        if (source.equals("executor")) { // "spark.executor.0.filesystem.file.largeRead_ops" (ExecutorSource)
-            String executorId = parts.remove(0);
-            tags = String.format("#executor:%s", executorId);
-            while (parts.size() > 3) {
-                parts.remove(parts.size() -1);
-            }
-            name = String.format("%s.%s.%s", prefix, source, StringUtils.join(parts, "_"));
-        } else if (source.equals("application")) { // "spark.application.Apriori.1394489355680.runtime_ms" (ApplicationSource)
-            String applicationName = parts.remove(0);
-            String currentTime = parts.remove(0);
-            String metricName = parts.remove(0);
-            tags = String.format("#application:%s", applicationName);
-            name = String.format("%s.%s.", prefix, source) + metricName;
-        } else {
-            String realSource = parts.remove(0);
-            // "spark.OrdersModel.DAGScheduler.stage.failedStages" (DAGSchedulerSource)
-            // "spark.OrdersModel.BlockManager.memory.maxMem_MB" (BlockManagerSource)
-            if (realSource.equals("DAGScheduler") || realSource.equals("BlockManager")) {
-                tags = String.format("#application:%s", source);
-                name = String.format("%s.application.%s.", prefix, realSource) + String.format("%s_%s", parts.toArray());
-            }
+
+        String tags = null; // TODO: Would be nice to get the job name and job user as tags
+
+        try {
+            name = buildMetricName(name);
+        } catch (IllegalArgumentException e) {
+            logger.error("Error sending to Statsd:", e);
+            return; // Drop metrics that we can't process so we don't push metrics with app names (e.g. 20141209201233-0145)
         }
 
         try {
@@ -129,10 +146,6 @@ public void close() throws IOException {
         this.writer = null;
     }
 
-    private String sanitizeString(String s) {
-        return WHITESPACE.matcher(s).replaceAll("-");
-    }
-
     private DatagramPacket newPacket(ByteArrayOutputStream out) {
         byte[] dataBuffer;
 

core/src/main/java/com/bealetech/metrics/reporting/StatsdReporter.java

@@ -122,6 +122,7 @@ public StatsdReporter(MetricRegistry registry,
         super(registry, "statsd-reporter", filter, rateUnit, durationUnit);
 
         this.statsd = statsd;
+        this.statsd.setNamePrefix(prefix);
         this.prefix = prefix;
     }