Handle data from DataDog of type:rate

proxy/src/main/java/com/wavefront/agent/listeners/DataDogPortUnificationHandler.java

@@ -306,6 +306,13 @@ private boolean reportMetric(final JsonNode metric, @Nullable final AtomicIntege
       if (deviceNode != null) {
         tags.put("device", deviceNode.textValue());
       }
+      int interval = 1; // If the metric is of type rate its value needs to be multiplied by the specified interval
+      JsonNode type = metric.get("type");
+      if (type != null) {
+        if (type.textValue().equals("rate")) {
+          interval = metric.get("interval") == null ? 1 : metric.get("interval").intValue();
+        }
+      }
       JsonNode pointsNode = metric.get("points");
       if (pointsNode == null) {
         pointHandler.reject((ReportPoint) null, "Skipping - 'points' field missing.");
@@ -314,7 +321,7 @@ private boolean reportMetric(final JsonNode metric, @Nullable final AtomicIntege
       for (JsonNode node : pointsNode) {
         if (node.size() == 2) {
           reportValue(metricName, hostName, tags, node.get(1), node.get(0).longValue() * 1000,
-              pointCounter);
+              pointCounter, interval);
         } else {
           pointHandler.reject((ReportPoint) null,
               "WF-300: Inconsistent point value size (expected: 2)");
@@ -460,6 +467,11 @@ private boolean reportSystemMetrics(final JsonNode metrics,
 
   private void reportValue(String metricName, String hostName, Map<String, String> tags,
                            JsonNode valueNode, long timestamp, AtomicInteger pointCounter) {
+    reportValue(metricName, hostName, tags, valueNode, timestamp, pointCounter, 1);
+  }
+
+  private void reportValue(String metricName, String hostName, Map<String, String> tags,
+                           JsonNode valueNode, long timestamp, AtomicInteger pointCounter, int interval) {
     if (valueNode == null || valueNode.isNull()) return;
     double value;
     if (valueNode.isTextual()) {
@@ -476,6 +488,8 @@ private void reportValue(String metricName, String hostName, Map<String, String>
       value = valueNode.asLong();
     }
 
+    value = value * interval; // interval will normally be 1 unless the metric was a rate type with a specified interval
+
     ReportPoint point = ReportPoint.newBuilder().
         setTable("dummy").
         setMetric(metricName).

proxy/src/test/java/com/wavefront/agent/PushAgentTest.java

@@ -581,6 +581,14 @@ public void testDataDogUnifiedPortHandler() throws Exception {
         setAnnotations(ImmutableMap.of("device", "eth0")).                    
         build());
     expectLastCall().once();
+    mockPointHandler.report(ReportPoint.newBuilder().
+            setTable("dummy").
+            setMetric("test.metric").
+            setHost("testhost").
+            setTimestamp(1531176936000L).
+            setValue(400.0d).
+            build());
+    expectLastCall().once();
     replay(mockPointHandler);
     gzippedHttpPost("http://localhost:" + ddPort + "/api/v1/series", getResource("ddTestTimeseries.json"));
     verify(mockPointHandler);

proxy/src/test/resources/com.wavefront.agent/ddTestTimeseries.json

@@ -37,6 +37,19 @@
       "source_type_name": "System",
       "metric": "system.net.packets_in.count",
       "device": "eth0"
+    },
+    {
+      "metric":"test.metric",
+      "points":[
+        [
+          1531176936,
+          20
+        ]
+      ],
+      "type":"rate",
+      "interval": 20,
+      "host":"testhost",
+      "tags":null
     }
   ]
 }