AMBARI-9305 AMS: certain metric charts look incorrect (dsen)

ambari-metrics/ambari-metrics-common/pom.xml

@@ -50,5 +50,11 @@
       <artifactId>jackson-mapper-asl</artifactId>
       <version>1.8.0</version>
     </dependency>
+    <dependency>
+      <groupId>junit</groupId>
+      <artifactId>junit</artifactId>
+      <scope>test</scope>
+      <version>4.10</version>
+    </dependency>
   </dependencies>
 </project>

ambari-metrics/ambari-metrics-common/src/main/java/org/apache/hadoop/metrics2/sink/timeline/cache/TimelineMetricsCache.java

@@ -17,15 +17,18 @@
  */
 package org.apache.hadoop.metrics2.sink.timeline.cache;
 
+import com.google.common.base.Optional;
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.apache.hadoop.classification.InterfaceAudience;
 import org.apache.hadoop.classification.InterfaceStability;
+import org.apache.hadoop.metrics2.MetricType;
 import org.apache.hadoop.metrics2.sink.timeline.TimelineMetric;
 
 import java.util.HashMap;
 import java.util.LinkedHashMap;
 import java.util.Map;
+import java.util.TreeMap;
 
 @InterfaceAudience.Public
 @InterfaceStability.Evolving
@@ -37,6 +40,7 @@ public class TimelineMetricsCache {
   public static final int MAX_EVICTION_TIME_MILLIS = 59000; // ~ 1 min
   private final int maxRecsPerName;
   private final int maxEvictionTimeInMillis;
+  private final Map<String, Double> counterMetricLastValue = new HashMap<String, Double>();
 
   public TimelineMetricsCache(int maxRecsPerName, int maxEvictionTimeInMillis) {
     this.maxRecsPerName = maxRecsPerName;
@@ -151,4 +155,26 @@ public class TimelineMetricsCache {
   public void putTimelineMetric(TimelineMetric timelineMetric) {
     timelineMetricCache.put(timelineMetric.getMetricName(), timelineMetric);
   }
+
+  private void transformMetricValuesToDerivative(TimelineMetric timelineMetric) {
+    String metricName = timelineMetric.getMetricName();
+    double firstValue = timelineMetric.getMetricValues().size() > 0
+        ? timelineMetric.getMetricValues().entrySet().iterator().next().getValue() : 0;
+    double previousValue = Optional.fromNullable(counterMetricLastValue.get(metricName)).or(firstValue);
+    Map<Long, Double> metricValues = timelineMetric.getMetricValues();
+    Map<Long, Double>   newMetricValues = new TreeMap<Long, Double>();
+    for (Map.Entry<Long, Double> entry : metricValues.entrySet()) {
+      newMetricValues.put(entry.getKey(), entry.getValue() - previousValue);
+      previousValue = entry.getValue();
+    }
+    timelineMetric.setMetricValues(newMetricValues);
+    counterMetricLastValue.put(metricName, previousValue);
+  }
+
+  public void putTimelineMetric(TimelineMetric timelineMetric, MetricType type) {
+    if (type == MetricType.COUNTER) {
+      transformMetricValuesToDerivative(timelineMetric);
+    }
+    putTimelineMetric(timelineMetric);
+  }
 }

ambari-metrics/ambari-metrics-common/src/test/java/org/apache/hadoop/metrics2/sink/timeline/cache/TimelineMetricsCacheTest.java

@@ -0,0 +1,88 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF 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 org.apache.hadoop.metrics2.sink.timeline.cache;
+
+import static com.sun.tools.javac.util.Assert.assertEquals;
+
+import org.apache.hadoop.metrics2.MetricType;
+import org.apache.hadoop.metrics2.sink.timeline.TimelineMetric;
+import org.junit.Test;
+
+import java.util.Map;
+import java.util.TreeMap;
+
+public class TimelineMetricsCacheTest {
+
+  private static final long DEFAULT_START_TIME = 1411023766;
+  private static final String METRIC_NAME = "Test name";
+  private static final double delta = 0.00001;
+
+  private final TimelineMetricsCache timelineMetricsCache = new TimelineMetricsCache(
+      TimelineMetricsCache.MAX_RECS_PER_NAME_DEFAULT,
+      TimelineMetricsCache.MAX_EVICTION_TIME_MILLIS);
+
+  @Test
+  public void testPutGetCounterTimelineMetric() throws Exception {
+    TimelineMetric metric = createTimelineMetric(new TreeMap<Long, Double>() {{
+      put(1L, 10.0);
+    }}, DEFAULT_START_TIME);
+    timelineMetricsCache.putTimelineMetric(metric, MetricType.COUNTER);
+    metric = createTimelineMetric(new TreeMap<Long, Double>() {{
+      put(2L, 10.0);
+      put(3L, 20.0);
+      put(4L, 30.0);
+    }}, DEFAULT_START_TIME + 2 * TimelineMetricsCache.MAX_EVICTION_TIME_MILLIS);
+    timelineMetricsCache.putTimelineMetric(metric, MetricType.COUNTER);
+    TimelineMetric cachedMetric
+        = timelineMetricsCache.getTimelineMetric(METRIC_NAME);
+    assertEquals(0, cachedMetric.getMetricValues().get(1L), delta);
+    assertEquals(0, cachedMetric.getMetricValues().get(2L), delta);
+    assertEquals(10, cachedMetric.getMetricValues().get(3L), delta);
+    assertEquals(10, cachedMetric.getMetricValues().get(4L), delta);
+
+    metric = createTimelineMetric(new TreeMap<Long, Double>() {{
+      put(5L, 100.0);
+      put(6L, 120.0);
+      put(7L, 230.0);
+    }}, DEFAULT_START_TIME + 3 * TimelineMetricsCache.MAX_EVICTION_TIME_MILLIS);
+    timelineMetricsCache.putTimelineMetric(metric, MetricType.COUNTER);
+    metric = createTimelineMetric(new TreeMap<Long, Double>() {{
+      put(8L, 300.0);
+    }}, DEFAULT_START_TIME + 5 * TimelineMetricsCache.MAX_EVICTION_TIME_MILLIS);
+
+    timelineMetricsCache.putTimelineMetric(metric, MetricType.COUNTER);
+    cachedMetric = timelineMetricsCache.getTimelineMetric(METRIC_NAME);
+    assertEquals(70, cachedMetric.getMetricValues().get(5L), delta);
+    assertEquals(20, cachedMetric.getMetricValues().get(6L), delta);
+    assertEquals(110, cachedMetric.getMetricValues().get(7L), delta);
+    assertEquals(70, cachedMetric.getMetricValues().get(8L), delta);
+  }
+
+  private TimelineMetric createTimelineMetric(Map<Long, Double> metricValues,
+                                              long startTime) {
+    TimelineMetric timelineMetric = new TimelineMetric();
+    timelineMetric.setMetricName(METRIC_NAME);
+    timelineMetric.setHostName("Test hostName");
+    timelineMetric.setAppId("test serviceName");
+    timelineMetric.setStartTime(startTime);
+    timelineMetric.setType("Number");
+    timelineMetric.setMetricValues(metricValues);
+    return timelineMetric;
+  }
+
+}

ambari-metrics/ambari-metrics-hadoop-sink/src/main/java/org/apache/hadoop/metrics2/sink/timeline/HadoopTimelineMetricsSink.java

@@ -164,7 +164,7 @@ public class HadoopTimelineMetricsSink extends AbstractTimelineMetricsSink imple
         timelineMetric.setType(ClassUtils.getShortCanonicalName(value, "Number"));
         timelineMetric.getMetricValues().put(startTime, value.doubleValue());
         // Put intermediate values into the cache until it is time to send
-        metricsCache.putTimelineMetric(timelineMetric);
+        metricsCache.putTimelineMetric(timelineMetric, metric.type());
 
         // Retrieve all values from cache if it is time to send
         TimelineMetric cachedMetric = metricsCache.getTimelineMetric(name);