AMBARI-18154 : Ambari Dashboard, Cluster load widget - Incorrect value in Nodes._avg metric. (avijayan)

ambari-metrics/ambari-metrics-timelineservice/src/main/java/org/apache/hadoop/yarn/server/applicationhistoryservice/metrics/timeline/aggregators/TimelineMetricClusterAggregatorSecond.java

@@ -31,11 +31,15 @@ import java.io.IOException;
 import java.sql.ResultSet;
 import java.sql.SQLException;
 import java.util.ArrayList;
+import java.util.Collections;
 import java.util.Date;
 import java.util.HashMap;
+import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
+import java.util.Set;
 
+import static org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline.TimelineMetricConfiguration.HOST_APP_ID;
 import static org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline.TimelineMetricConfiguration.SERVER_SIDE_TIMESIFT_ADJUSTMENT;
 import static org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline.TimelineMetricConfiguration.TIMELINE_METRICS_CLUSTER_AGGREGATOR_INTERPOLATION_ENABLED;
 import static org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline.query.PhoenixTransactSQL.GET_METRIC_SQL;
@@ -127,6 +131,7 @@ public class TimelineMetricClusterAggregatorSecond extends AbstractTimelineAggre
     throws SQLException, IOException {
     Map<TimelineClusterMetric, MetricClusterAggregate> aggregateClusterMetrics =
       new HashMap<TimelineClusterMetric, MetricClusterAggregate>();
+    int numLiveHosts = 0;
 
     TimelineMetric metric = null;
     if (rs.next()) {
@@ -142,18 +147,25 @@ public class TimelineMetricClusterAggregatorSecond extends AbstractTimelineAggre
           metric.addMetricValues(nextMetric.getMetricValues());
         } else {
           // Process the current metric
-          processAggregateClusterMetrics(aggregateClusterMetrics, metric, timeSlices);
+          int numHosts = processAggregateClusterMetrics(aggregateClusterMetrics, metric, timeSlices);
+          numLiveHosts = Math.max(numHosts, numLiveHosts);
           metric = nextMetric;
         }
       }
     }
     // Process last metric
     if (metric != null) {
-      processAggregateClusterMetrics(aggregateClusterMetrics, metric, timeSlices);
+      int numHosts = processAggregateClusterMetrics(aggregateClusterMetrics, metric, timeSlices);
+      numLiveHosts = Math.max(numHosts, numLiveHosts);
     }
 
     // Add app level aggregates to save
     aggregateClusterMetrics.putAll(appAggregator.getAggregateClusterMetrics());
+
+    // Add liveHosts metric.
+    long timestamp = timeSlices.get(timeSlices.size() - 1)[1];
+    processLiveHostsMetric(aggregateClusterMetrics, numLiveHosts, timestamp);
+
     return aggregateClusterMetrics;
   }
 
@@ -162,10 +174,11 @@ public class TimelineMetricClusterAggregatorSecond extends AbstractTimelineAggre
    * timeline.metrics.cluster.aggregator.minute.timeslice.interval
    * Normalize value by averaging them within the interval
    */
-  protected void processAggregateClusterMetrics(Map<TimelineClusterMetric, MetricClusterAggregate> aggregateClusterMetrics,
+  protected int processAggregateClusterMetrics(Map<TimelineClusterMetric, MetricClusterAggregate> aggregateClusterMetrics,
                                               TimelineMetric metric, List<Long[]> timeSlices) {
     // Create time slices
     Map<TimelineClusterMetric, Double> clusterMetrics = sliceFromTimelineMetric(metric, timeSlices);
+    int numHosts = 0;
 
     if (clusterMetrics != null && !clusterMetrics.isEmpty()) {
       for (Map.Entry<TimelineClusterMetric, Double> clusterMetricEntry :
@@ -185,10 +198,13 @@ public class TimelineMetricClusterAggregatorSecond extends AbstractTimelineAggre
           aggregate.updateMax(avgValue);
           aggregate.updateMin(avgValue);
         }
+
+        numHosts = aggregate.getNumberOfHosts();
         // Update app level aggregates
         appAggregator.processTimelineClusterMetric(clusterMetric, metric.getHostName(), avgValue);
       }
     }
+    return numHosts;
   }
 
   protected Map<TimelineClusterMetric, Double> sliceFromTimelineMetric(
@@ -371,4 +387,16 @@ public class TimelineMetricClusterAggregatorSecond extends AbstractTimelineAggre
     return -1l;
   }
 
+  private void processLiveHostsMetric(Map<TimelineClusterMetric, MetricClusterAggregate> aggregateClusterMetrics,
+                                     int numLiveHosts, long timestamp) {
+
+    TimelineClusterMetric timelineClusterMetric = new TimelineClusterMetric(
+      "live_hosts", HOST_APP_ID, null, timestamp, null);
+
+    MetricClusterAggregate metricClusterAggregate = new MetricClusterAggregate((double) numLiveHosts,
+      1, null, (double) numLiveHosts, (double) numLiveHosts);
+
+    aggregateClusterMetrics.put(timelineClusterMetric, metricClusterAggregate);
+  }
+
 }

ambari-metrics/ambari-metrics-timelineservice/src/test/java/org/apache/hadoop/yarn/server/applicationhistoryservice/metrics/timeline/aggregators/TimelineMetricClusterAggregatorSecondTest.java

@@ -258,8 +258,9 @@ public class TimelineMetricClusterAggregatorSecondTest {
     aggregateClusterMetrics.clear();
 
     timelineMetric.setType("COUNTER");
-    secondAggregator.processAggregateClusterMetrics(aggregateClusterMetrics, timelineMetric, timeslices);
+    int liveHosts = secondAggregator.processAggregateClusterMetrics(aggregateClusterMetrics, timelineMetric, timeslices);
 
+    Assert.assertEquals(liveHosts, 1);
     Assert.assertEquals(aggregateClusterMetrics.size(), 4);
     timelineClusterMetric.setTimestamp(startTime + 30*seconds);
     Assert.assertTrue(aggregateClusterMetrics.containsKey(timelineClusterMetric));
@@ -293,8 +294,9 @@ public class TimelineMetricClusterAggregatorSecondTest {
     aggregateClusterMetrics.clear();
 
     timelineMetric.setType("COUNTER");
-    secondAggregator.processAggregateClusterMetrics(aggregateClusterMetrics, timelineMetric, timeslices);
+    liveHosts = secondAggregator.processAggregateClusterMetrics(aggregateClusterMetrics, timelineMetric, timeslices);
 
+    Assert.assertEquals(liveHosts, 1);
     Assert.assertEquals(aggregateClusterMetrics.size(), 4);
     timelineClusterMetric.setTimestamp(startTime + 60*seconds);
     Assert.assertTrue(aggregateClusterMetrics.containsKey(timelineClusterMetric));

ambari-server/src/main/resources/ganglia_properties.json

@@ -47,7 +47,7 @@
         "metric":"load_report.Nodes",
         "pointInTime":false,
         "temporal":true,
-        "amsId":"cpu_num"
+        "amsId":"live_hosts"
       },
       "metrics/load/Procs":{
         "metric":"load_report.Procs",