AMBARI-13641. Various metrics show no data available - when HA enabled. (swagle)

ambari-server/src/main/java/org/apache/ambari/server/controller/metrics/timeline/cache/TimelineMetricCache.java

@@ -69,7 +69,7 @@ public class TimelineMetricCache extends UpdatingSelfPopulatingCache {
       if (LOG.isDebugEnabled()) {
         LOG.debug("Returning value from cache: " + value);
       }
-      timelineMetrics.setMetrics(new ArrayList<TimelineMetric>(value.getTimelineMetrics().values()));
+      timelineMetrics = value.getTimelineMetrics();
     }
 
     if (LOG.isDebugEnabled()) {

ambari-server/src/main/java/org/apache/ambari/server/controller/metrics/timeline/cache/TimelineMetricCacheEntryFactory.java

@@ -94,16 +94,10 @@ public class TimelineMetricCacheEntryFactory implements UpdatingCacheEntryFactor
     TimelineMetricsCacheValue value = null;
 
     if (timelineMetrics != null && !timelineMetrics.getMetrics().isEmpty()) {
-      Map<String, TimelineMetric> cacheValue =
-        new HashMap<String, TimelineMetric>(timelineMetrics.getMetrics().size());
-      for (TimelineMetric metric : timelineMetrics.getMetrics()) {
-        cacheValue.put(metric.getMetricName(), metric);
-      }
-
       value = new TimelineMetricsCacheValue(
         metricCacheKey.getTemporalInfo().getStartTime(),
         metricCacheKey.getTemporalInfo().getEndTime(),
-        cacheValue, // Null or empty should prompt a refresh
+        timelineMetrics, // Null or empty should prompt a refresh
         Precision.getPrecision(metricCacheKey.getTemporalInfo().getStartTimeMillis(),
           metricCacheKey.getTemporalInfo().getEndTimeMillis()) //Initial Precision
       );
@@ -210,30 +204,57 @@ public class TimelineMetricCacheEntryFactory implements UpdatingCacheEntryFactor
       TimelineMetricsCacheValue timelineMetricsCacheValue,
       Long requestedStartTime, Long requestedEndTime, boolean removeAll) {
 
-    Map<String, TimelineMetric> existingTimelineMetricMap = timelineMetricsCacheValue.getTimelineMetrics();
+    TimelineMetrics existingTimelineMetrics = timelineMetricsCacheValue.getTimelineMetrics();
 
-    // NOTE: Metrics names so far are unique, the Map optimization avoids
-    // multiple iterations of the List
-    for (TimelineMetric timelineMetric : newMetrics.getMetrics()) {
-      if (LOG.isTraceEnabled()) {
-        TreeMap<Long, Double> sortedMetrics = new TreeMap<Long, Double>(timelineMetric.getMetricValues());
+    // Remove values that do not fit before adding new data
+    updateExistingMetricValues(existingTimelineMetrics, requestedStartTime,
+      requestedEndTime, removeAll);
 
-        LOG.trace("New metric: " + timelineMetric.getMetricName() +
-          " # " + timelineMetric.getMetricValues().size() + ", startTime = " +
-          sortedMetrics.firstKey() + ", endTime = " + sortedMetrics.lastKey());
-      }
+    if (newMetrics != null && !newMetrics.getMetrics().isEmpty()) {
+      for (TimelineMetric timelineMetric : newMetrics.getMetrics()) {
+        if (LOG.isTraceEnabled()) {
+          TreeMap<Long, Double> sortedMetrics = new TreeMap<Long, Double>(timelineMetric.getMetricValues());
 
+          LOG.trace("New metric: " + timelineMetric.getMetricName() +
+            " # " + timelineMetric.getMetricValues().size() + ", startTime = " +
+            sortedMetrics.firstKey() + ", endTime = " + sortedMetrics.lastKey());
+        }
 
-      TimelineMetric existingMetric = existingTimelineMetricMap.get(timelineMetric.getMetricName());
+        TimelineMetric existingMetric = null;
 
-      if (existingMetric != null) {
+        for (TimelineMetric metric : existingTimelineMetrics.getMetrics()) {
+          if (metric.equalsExceptTime(timelineMetric)) {
+            existingMetric = metric;
+          }
+        }
+
+        if (existingMetric != null) {
+          // Add new ones
+          existingMetric.getMetricValues().putAll(timelineMetric.getMetricValues());
 
-        if(removeAll) {
-          existingMetric.setMetricValues(new TreeMap<Long, Double>());
+          if (LOG.isTraceEnabled()) {
+            TreeMap<Long, Double> sortedMetrics = new TreeMap<Long, Double>(existingMetric.getMetricValues());
+            LOG.trace("Merged metric: " + timelineMetric.getMetricName() + ", " +
+              "Final size: " + existingMetric.getMetricValues().size() + ", startTime = " +
+              sortedMetrics.firstKey() + ", endTime = " + sortedMetrics.lastKey());
+          }
+        } else {
+          existingTimelineMetrics.getMetrics().add(timelineMetric);
         }
+      }
+    }
+  }
+
+  // Remove out of band data from the cache
+  private void updateExistingMetricValues(TimelineMetrics existingMetrics,
+      Long requestedStartTime, Long requestedEndTime, boolean removeAll) {
 
+    for (TimelineMetric existingMetric : existingMetrics.getMetrics()) {
+      if(removeAll) {
+        existingMetric.setMetricValues(new TreeMap<Long, Double>());
+      } else {
         Map<Long, Double> existingMetricValues = existingMetric.getMetricValues();
-        LOG.trace("Existing metric: " + timelineMetric.getMetricName() +
+        LOG.trace("Existing metric: " + existingMetric.getMetricName() +
           " # " + existingMetricValues.size());
 
         Iterator<Map.Entry<Long, Double>> valueIterator = existingMetricValues.entrySet().iterator();
@@ -243,22 +264,10 @@ public class TimelineMetricCacheEntryFactory implements UpdatingCacheEntryFactor
         while (valueIterator.hasNext()) {
           Map.Entry<Long, Double> metricEntry = valueIterator.next();
           if (metricEntry.getKey() < requestedStartTime
-              || metricEntry.getKey() > requestedEndTime) {
+            || metricEntry.getKey() > requestedEndTime) {
             valueIterator.remove();
           }
         }
-
-        // Add new ones
-        existingMetricValues.putAll(timelineMetric.getMetricValues());
-
-        if (LOG.isTraceEnabled()) {
-          TreeMap<Long, Double> sortedMetrics = new TreeMap<Long, Double>(existingMetricValues);
-          LOG.trace("Merged metric: " + timelineMetric.getMetricName() + ", " +
-            "Final size: " + existingMetricValues.size() + ", startTime = " +
-            sortedMetrics.firstKey() + ", endTime = " + sortedMetrics.lastKey());
-        }
-      } else {
-        existingTimelineMetricMap.put(timelineMetric.getMetricName(), timelineMetric);
       }
     }
   }

ambari-server/src/main/java/org/apache/ambari/server/controller/metrics/timeline/cache/TimelineMetricsCacheSizeOfEngine.java

@@ -23,6 +23,7 @@ import net.sf.ehcache.pool.impl.DefaultSizeOfEngine;
 import net.sf.ehcache.pool.sizeof.ReflectionSizeOf;
 import net.sf.ehcache.pool.sizeof.SizeOf;
 import org.apache.hadoop.metrics2.sink.timeline.TimelineMetric;
+import org.apache.hadoop.metrics2.sink.timeline.TimelineMetrics;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import java.util.Map;
@@ -89,14 +90,11 @@ public class TimelineMetricsCacheSizeOfEngine implements SizeOfEngine {
 
   private long getTimelineMetricCacheValueSize(TimelineMetricsCacheValue value) {
     long size = 16; // startTime + endTime
-    Map<String, TimelineMetric> metrics = value.getTimelineMetrics();
+    TimelineMetrics metrics = value.getTimelineMetrics();
     size += 8; // Object reference
 
     if (metrics != null) {
-      for (Map.Entry<String, TimelineMetric> metricEntry : metrics.entrySet()) {
-        size += reflectionSizeOf.sizeOf(metricEntry.getKey());
-
-        TimelineMetric metric = metricEntry.getValue();
+      for (TimelineMetric metric : metrics.getMetrics()) {
 
         if (timelineMetricPrimitivesApproximation == 0) {
           timelineMetricPrimitivesApproximation += reflectionSizeOf.sizeOf(metric.getMetricName());

ambari-server/src/main/java/org/apache/ambari/server/controller/metrics/timeline/cache/TimelineMetricsCacheValue.java

@@ -19,6 +19,7 @@ package org.apache.ambari.server.controller.metrics.timeline.cache;
 
 import org.apache.hadoop.metrics2.sink.timeline.Precision;
 import org.apache.hadoop.metrics2.sink.timeline.TimelineMetric;
+import org.apache.hadoop.metrics2.sink.timeline.TimelineMetrics;
 
 import java.util.Date;
 import java.util.Map;
@@ -30,17 +31,18 @@ import java.util.Map;
 public class TimelineMetricsCacheValue {
   private Long startTime;
   private Long endTime;
-  private Map<String, TimelineMetric> timelineMetrics;
+  // Same metrics could be requested across hosts
+  private TimelineMetrics timelineMetrics = new TimelineMetrics();
   private Precision precision;
 
-  public TimelineMetricsCacheValue(Long startTime, Long endTime, Map<String, TimelineMetric> timelineMetrics, Precision precision) {
+  public TimelineMetricsCacheValue(Long startTime, Long endTime, TimelineMetrics timelineMetrics, Precision precision) {
     this.startTime = startTime;
     this.endTime = endTime;
     this.timelineMetrics = timelineMetrics;
     this.precision = precision;
   }
 
-  public Map<String, TimelineMetric> getTimelineMetrics() {
+  public TimelineMetrics getTimelineMetrics() {
     return timelineMetrics;
   }
 
@@ -48,7 +50,7 @@ public class TimelineMetricsCacheValue {
    * Map of metricName to metric values. Works on the assumption that metric
    * name is unique
    */
-  public void setTimelineMetrics(Map<String, TimelineMetric> timelineMetrics) {
+  public void setTimelineMetrics(TimelineMetrics timelineMetrics) {
     this.timelineMetrics = timelineMetrics;
   }
 
@@ -79,19 +81,21 @@ public class TimelineMetricsCacheValue {
   @Override
   public String toString() {
     StringBuilder sb = new StringBuilder("TimelineMetricsCacheValue {" +
-      "metricNames = " + timelineMetrics.keySet() +
       ", startTime = " + new Date(getMillisecondsTime(startTime)) +
       ", endTime = " + new Date(getMillisecondsTime(endTime)) +
       ", precision = " + precision +
       ", timelineMetrics =");
 
-    for (TimelineMetric metric : timelineMetrics.values()) {
+    for (TimelineMetric metric : timelineMetrics.getMetrics()) {
       sb.append(" { ");
       sb.append(metric.getMetricName());
+      sb.append(", ");
+      sb.append(metric.getHostName());
       sb.append(" # ");
       sb.append(metric.getMetricValues().size());
       sb.append(" }");
     }
+
     sb.append("}");
     return sb.toString();
   }

ambari-server/src/test/java/org/apache/ambari/server/controller/metrics/timeline/cache/TimelineMetricCacheSizingTest.java

@@ -22,6 +22,7 @@ import net.sf.ehcache.pool.sizeof.SizeOf;
 import org.apache.ambari.server.controller.internal.TemporalInfoImpl;
 import org.apache.ambari.server.controller.spi.TemporalInfo;
 import org.apache.hadoop.metrics2.sink.timeline.TimelineMetric;
+import org.apache.hadoop.metrics2.sink.timeline.TimelineMetrics;
 import org.junit.Assert;
 import org.junit.Test;
 import java.util.HashMap;
@@ -47,7 +48,7 @@ public class TimelineMetricCacheSizingTest {
     // JSON dser gives a LinkedHashMap
     TreeMap<Long, Double> valueMap = new TreeMap<>();
     long now = System.currentTimeMillis();
-    for (int i = 0; i < 25000; i++) {
+    for (int i = 0; i < 50000; i++) {
       valueMap.put(new Long(now + i), new Double(1.0 + i));
     }
 
@@ -83,16 +84,16 @@ public class TimelineMetricCacheSizingTest {
       "jvm.JvmMetrics.MemHeapCommittedM&appId=RESOURCEMANAGER&" +
       "startTime=1439522640000&endTime=1440127440000&precision=hours");
 
-    Map<String, TimelineMetric> metricMap = new HashMap<>();
-    metricMap.put(metric1, getSampleTimelineMetric(metric1));
-    metricMap.put(metric2, getSampleTimelineMetric(metric2));
-    metricMap.put(metric3, getSampleTimelineMetric(metric3));
-    metricMap.put(metric4, getSampleTimelineMetric(metric4));
-    metricMap.put(metric5, getSampleTimelineMetric(metric5));
-    metricMap.put(metric6, getSampleTimelineMetric(metric6));
+    TimelineMetrics metrics = new TimelineMetrics();
+    metrics.getMetrics().add(getSampleTimelineMetric(metric1));
+    metrics.getMetrics().add(getSampleTimelineMetric(metric2));
+    metrics.getMetrics().add(getSampleTimelineMetric(metric3));
+    metrics.getMetrics().add(getSampleTimelineMetric(metric4));
+    metrics.getMetrics().add(getSampleTimelineMetric(metric5));
+    metrics.getMetrics().add(getSampleTimelineMetric(metric6));
 
     TimelineMetricsCacheValue value = new TimelineMetricsCacheValue(now -
-      1000, now, metricMap, null);
+      1000, now, metrics, null);
 
     TimelineMetricsCacheSizeOfEngine customSizeOfEngine = new TimelineMetricsCacheSizeOfEngine();
 

ambari-server/src/test/java/org/apache/ambari/server/controller/metrics/timeline/cache/TimelineMetricCacheTest.java

@@ -134,7 +134,9 @@ public class TimelineMetricCacheTest {
     replay(configuration);
 
     final long now = System.currentTimeMillis();
-    Map<String, TimelineMetric> valueMap = new HashMap<String, TimelineMetric>();
+
+    TimelineMetrics metrics = new TimelineMetrics();
+
     TimelineMetric timelineMetric = new TimelineMetric();
     timelineMetric.setMetricName("cpu_user");
     timelineMetric.setAppId("app1");
@@ -143,7 +145,8 @@ public class TimelineMetricCacheTest {
     metricValues.put(now + 200, 2.0);
     metricValues.put(now + 300, 3.0);
     timelineMetric.setMetricValues(metricValues);
-    valueMap.put("cpu_user", timelineMetric);
+
+    metrics.getMetrics().add(timelineMetric);
 
     TimelineMetricCacheEntryFactory cacheEntryFactory = createMock(TimelineMetricCacheEntryFactory.class);
 
@@ -152,7 +155,7 @@ public class TimelineMetricCacheTest {
       "app1",
       new TemporalInfoImpl(now, now + 1000, 1)
     );
-    TimelineMetricsCacheValue value = new TimelineMetricsCacheValue(now, now + 1000, valueMap, null);
+    TimelineMetricsCacheValue value = new TimelineMetricsCacheValue(now, now + 1000, metrics, null);
     TimelineAppMetricCacheKey testKey = new TimelineAppMetricCacheKey(
       Collections.singleton("cpu_user"),
       "app1",
@@ -169,7 +172,7 @@ public class TimelineMetricCacheTest {
     TimelineMetricCache cache = cacheProvider.getTimelineMetricsCache();
 
     // call to get
-    TimelineMetrics metrics = cache.getAppTimelineMetricsFromCache(queryKey);
+    metrics = cache.getAppTimelineMetricsFromCache(queryKey);
     List<TimelineMetric> metricsList = metrics.getMetrics();
     Assert.assertEquals(1, metricsList.size());
     TimelineMetric metric = metricsList.iterator().next();
@@ -317,12 +320,12 @@ public class TimelineMetricCacheTest {
     metricValues.put(now + 600, 3.0);
     timelineMetric2.setMetricValues(metricValues);
 
+    TimelineMetrics existingMetrics = new TimelineMetrics();
+    existingMetrics.getMetrics().add(timelineMetric1);
+    existingMetrics.getMetrics().add(timelineMetric2);
+
     TimelineMetricsCacheValue existingMetricValue = new TimelineMetricsCacheValue(
-      now - 1000, now + 1000,
-      new HashMap<String, TimelineMetric>() {{
-        put("cpu_user", timelineMetric1);
-        put("cpu_nice", timelineMetric2);
-      }}, null);
+      now - 1000, now + 1000, existingMetrics, null);
 
     // New values
     TimelineMetrics newMetrics = new TimelineMetrics();
@@ -337,12 +340,27 @@ public class TimelineMetricCacheTest {
     newMetrics.getMetrics().add(timelineMetric3);
 
     factory.updateTimelineMetricsInCache(newMetrics, existingMetricValue,
-      now, now + 2000,true);
+      now, now + 2000, false);
+
+    Assert.assertEquals(2, existingMetricValue.getTimelineMetrics().getMetrics().size());
+
+    TimelineMetric newMetric1 = null;
+    TimelineMetric newMetric2 = null;
+
+    for (TimelineMetric metric : existingMetricValue.getTimelineMetrics().getMetrics()) {
+      if (metric.getMetricName().equals("cpu_user")) {
+        newMetric1 = metric;
+      }
+      if (metric.getMetricName().equals("cpu_nice")) {
+        newMetric2 = metric;
+      }
+    }
 
-    Assert.assertEquals(2, existingMetricValue.getTimelineMetrics().size());
-    Assert.assertEquals(3, existingMetricValue.getTimelineMetrics().get("cpu_user").getMetricValues().size());
-    Assert.assertEquals(3, existingMetricValue.getTimelineMetrics().get("cpu_nice").getMetricValues().size());
-    Map<Long, Double> newMetricsMap = existingMetricValue.getTimelineMetrics().get("cpu_user").getMetricValues();
+    Assert.assertNotNull(newMetric1);
+    Assert.assertNotNull(newMetric2);
+    Assert.assertEquals(3, newMetric1.getMetricValues().size());
+    Assert.assertEquals(3, newMetric2.getMetricValues().size());
+    Map<Long, Double> newMetricsMap = newMetric1.getMetricValues();
     Iterator<Long> metricKeyIterator = newMetricsMap.keySet().iterator();
     Assert.assertEquals(now + 1400, metricKeyIterator.next().longValue());
     Assert.assertEquals(now + 1500, metricKeyIterator.next().longValue());