AMBARI-11010. Aggregator function does not work for point in time service component metric query. (swagle)

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

@@ -0,0 +1,107 @@
+/**
+ * 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;
+
+/**
+ * This class prevents creating a TreeMap for every instantiation of a metric
+ * read from the store. The methods are meant to provide interoperability
+ * with @TimelineMetric
+ */
+public class SingleValuedTimelineMetric {
+  private Long timestamp;
+  private Double value;
+  private String metricName;
+  private String appId;
+  private String instanceId;
+  private String hostName;
+  private Long startTime;
+  private String type;
+
+  public void setSingleTimeseriesValue(Long timestamp, Double value) {
+    this.timestamp = timestamp;
+    this.value = value;
+  }
+
+  public SingleValuedTimelineMetric(String metricName, String appId,
+                                    String instanceId, String hostName,
+                                    long timestamp, long startTime, String type) {
+    this.metricName = metricName;
+    this.appId = appId;
+    this.instanceId = instanceId;
+    this.hostName = hostName;
+    this.timestamp = timestamp;
+    this.startTime = startTime;
+    this.type = type;
+  }
+
+  public Long getTimestamp() {
+    return timestamp;
+  }
+
+  public long getStartTime() {
+    return startTime;
+  }
+
+  public String getType() {
+    return type;
+  }
+
+  public Double getValue() {
+    return value;
+  }
+
+  public String getMetricName() {
+    return metricName;
+  }
+
+  public String getAppId() {
+    return appId;
+  }
+
+  public String getInstanceId() {
+    return instanceId;
+  }
+
+  public String getHostName() {
+    return hostName;
+  }
+
+  public boolean equalsExceptTime(TimelineMetric metric) {
+    if (!metricName.equals(metric.getMetricName())) return false;
+    if (hostName != null ? !hostName.equals(metric.getHostName()) : metric.getHostName() != null)
+      return false;
+    if (appId != null ? !appId.equals(metric.getAppId()) : metric.getAppId() != null)
+      return false;
+    if (instanceId != null ? !instanceId.equals(metric.getInstanceId()) : metric.getInstanceId() != null) return false;
+
+    return true;
+  }
+
+  public TimelineMetric getTimelineMetric() {
+    TimelineMetric metric = new TimelineMetric();
+    metric.setMetricName(this.metricName);
+    metric.setAppId(this.appId);
+    metric.setHostName(this.hostName);
+    metric.setType(this.type);
+    metric.setInstanceId(this.instanceId);
+    metric.setStartTime(this.startTime);
+    metric.setTimestamp(this.timestamp);
+    metric.getMetricValues().put(timestamp, value);
+    return metric;
+  }
+}

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

@@ -99,4 +99,30 @@ public class TimelineMetrics {
       allMetrics.add(metric);
     }
   }
+
+  // Optimization that addresses too many TreeMaps from getting created.
+  public void addOrMergeTimelineMetric(SingleValuedTimelineMetric metric) {
+    TimelineMetric metricToMerge = null;
+
+    if (!allMetrics.isEmpty()) {
+      for (TimelineMetric timelineMetric : allMetrics) {
+        if (metric.equalsExceptTime(timelineMetric)) {
+          metricToMerge = timelineMetric;
+          break;
+        }
+      }
+    }
+
+    if (metricToMerge != null) {
+      metricToMerge.getMetricValues().put(metric.getTimestamp(), metric.getValue());
+      if (metricToMerge.getTimestamp() > metric.getTimestamp()) {
+        metricToMerge.setTimestamp(metric.getTimestamp());
+      }
+      if (metricToMerge.getStartTime() > metric.getStartTime()) {
+        metricToMerge.setStartTime(metric.getStartTime());
+      }
+    } else {
+      allMetrics.add(metric.getTimelineMetric());
+    }
+  }
 }

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

@@ -191,8 +191,7 @@ public class HBaseTimelineMetricStore extends AbstractService
     return metricValues;
   }
 
-  public static HashMap<String, List<Function>>
-  parseMetricNamesToAggregationFunctions(List<String> metricNames) {
+  public static HashMap<String, List<Function>> parseMetricNamesToAggregationFunctions(List<String> metricNames) {
     HashMap<String, List<Function>> metricsFunctions = new HashMap<String,
       List<Function>>();
 

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

@@ -22,6 +22,7 @@ import org.apache.commons.logging.LogFactory;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.hbase.util.RetryCounter;
 import org.apache.hadoop.hbase.util.RetryCounterFactory;
+import org.apache.hadoop.metrics2.sink.timeline.SingleValuedTimelineMetric;
 import org.apache.hadoop.metrics2.sink.timeline.TimelineMetric;
 import org.apache.hadoop.metrics2.sink.timeline.TimelineMetrics;
 import org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline.aggregators.AggregatorUtils;
@@ -39,7 +40,6 @@ import org.apache.hadoop.yarn.util.timeline.TimelineUtils;
 import org.apache.phoenix.exception.SQLExceptionCode;
 import org.codehaus.jackson.map.ObjectMapper;
 import org.codehaus.jackson.type.TypeReference;
-
 import java.io.IOException;
 import java.sql.Connection;
 import java.sql.PreparedStatement;
@@ -50,9 +50,7 @@ import java.util.Collections;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
-import java.util.TreeMap;
 import java.util.concurrent.TimeUnit;
-
 import static java.util.concurrent.TimeUnit.SECONDS;
 import static org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline.TimelineMetricConfiguration.CLUSTER_DAILY_TABLE_TTL;
 import static org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline.TimelineMetricConfiguration.CLUSTER_HOUR_TABLE_TTL;
@@ -132,23 +130,23 @@ public class PhoenixHBaseAccessor {
 
   private static TimelineMetric getLastTimelineMetricFromResultSet(ResultSet rs)
     throws SQLException, IOException {
-    TimelineMetric metric = TIMELINE_METRIC_READ_HELPER
-      .getTimelineMetricCommonsFromResultSet(rs);
+    TimelineMetric metric = TIMELINE_METRIC_READ_HELPER.getTimelineMetricCommonsFromResultSet(rs);
     metric.setMetricValues(readLastMetricValueFromJSON(rs.getString("METRICS")));
-
     return metric;
   }
 
-  public static TimelineMetric getAggregatedTimelineMetricFromResultSet(
-    ResultSet rs, Function f) throws SQLException, IOException {
+  public static SingleValuedTimelineMetric getAggregatedTimelineMetricFromResultSet(
+      ResultSet rs, Function f) throws SQLException, IOException {
 
-    TimelineMetric metric = new TimelineMetric();
-    metric.setHostName(rs.getString("HOSTNAME"));
-    metric.setAppId(rs.getString("APP_ID"));
-    metric.setInstanceId(rs.getString("INSTANCE_ID"));
-    metric.setTimestamp(rs.getLong("SERVER_TIME"));
-    metric.setStartTime(rs.getLong("SERVER_TIME"));
-    metric.setType(rs.getString("UNITS"));
+    SingleValuedTimelineMetric metric = new SingleValuedTimelineMetric(
+      rs.getString("METRIC_NAME") + f.getSuffix(),
+      rs.getString("APP_ID"),
+      rs.getString("INSTANCE_ID"),
+      rs.getString("HOSTNAME"),
+      rs.getLong("SERVER_TIME"),
+      rs.getLong("SERVER_TIME"),
+      rs.getString("UNITS")
+    );
 
     // get functions for metricnames
 
@@ -171,11 +169,8 @@ public class PhoenixHBaseAccessor {
         break;
     }
 
-    metric.setMetricName(rs.getString("METRIC_NAME") + f.getSuffix());
+    metric.setSingleTimeseriesValue(rs.getLong("SERVER_TIME"), value);
 
-    Map<Long, Double> valueMap = new TreeMap<Long, Double>();
-    valueMap.put(rs.getLong("SERVER_TIME"), value);
-    metric.setMetricValues(valueMap);
     return metric;
   }
 
@@ -194,30 +189,6 @@ public class PhoenixHBaseAccessor {
     return (Map<Long, Double>) mapper.readValue(json, metricValuesTypeRef);
   }
 
-  public static TimelineMetric getTimelineMetricKeyFromResultSet(ResultSet rs)
-    throws SQLException, IOException {
-    TimelineMetric metric = new TimelineMetric();
-    metric.setMetricName(rs.getString("METRIC_NAME"));
-    metric.setAppId(rs.getString("APP_ID"));
-    metric.setInstanceId(rs.getString("INSTANCE_ID"));
-    metric.setHostName(rs.getString("HOSTNAME"));
-    metric.setTimestamp(rs.getLong("SERVER_TIME"));
-    metric.setType(rs.getString("UNITS"));
-    return metric;
-  }
-
-  public static MetricHostAggregate getMetricHostAggregateFromResultSet(ResultSet rs)
-    throws SQLException {
-    MetricHostAggregate metricHostAggregate = new MetricHostAggregate();
-    metricHostAggregate.setSum(rs.getDouble("METRIC_SUM"));
-    metricHostAggregate.setMax(rs.getDouble("METRIC_MAX"));
-    metricHostAggregate.setMin(rs.getDouble("METRIC_MIN"));
-    metricHostAggregate.setNumberOfSamples(rs.getLong("METRIC_COUNT"));
-
-    metricHostAggregate.setDeviation(0.0);
-    return metricHostAggregate;
-  }
-
   private Connection getConnectionRetryingOnException()
     throws SQLException, InterruptedException {
     RetryCounter retryCounter = retryCounterFactory.create();
@@ -465,9 +436,9 @@ public class PhoenixHBaseAccessor {
   }
 
   private void appendMetricFromResultSet(
-    TimelineMetrics metrics, Condition condition, Map<String,
-    List<Function>> metricFunctions, ResultSet rs)
-    throws SQLException, IOException {
+      TimelineMetrics metrics, Condition condition, Map<String,
+      List<Function>> metricFunctions, ResultSet rs)
+      throws SQLException, IOException {
     if (condition.getPrecision() == Precision.HOURS
       || condition.getPrecision() == Precision.MINUTES) {
 
@@ -475,14 +446,12 @@ public class PhoenixHBaseAccessor {
       List<Function> functions = metricFunctions.get(metricName);
 
       for (Function f : functions) {
-        TimelineMetric metric;
-
-        metric = getAggregatedTimelineMetricFromResultSet(rs, f);
+        SingleValuedTimelineMetric metric = getAggregatedTimelineMetricFromResultSet(rs, f);
 
         if (condition.isGrouped()) {
           metrics.addOrMergeTimelineMetric(metric);
         } else {
-          metrics.getMetrics().add(metric);
+          metrics.getMetrics().add(metric.getTimelineMetric());
         }
       }
     }
@@ -540,10 +509,8 @@ public class PhoenixHBaseAccessor {
    * @return @TimelineMetrics
    * @throws SQLException
    */
-  public TimelineMetrics getAggregateMetricRecords(
-    final Condition condition,
-    Map<String, List<Function>> metricFunctions)
-    throws SQLException {
+  public TimelineMetrics getAggregateMetricRecords(final Condition condition,
+      Map<String, List<Function>> metricFunctions) throws SQLException {
 
     validateConditionIsNotEmpty(condition);
 
@@ -555,14 +522,13 @@ public class PhoenixHBaseAccessor {
     try {
       //get latest
       if(condition.isPointInTime()) {
-        stmt = getLatestAggregateMetricRecords(condition, conn, metrics);
+        stmt = getLatestAggregateMetricRecords(condition, conn, metrics, metricFunctions);
       } else {
         stmt = PhoenixTransactSQL.prepareGetAggregateSqlStmt(conn, condition);
 
         rs = stmt.executeQuery();
         while (rs.next()) {
-          appendAggregateMetricFromResultSet(metrics, condition,
-            metricFunctions, rs);
+          appendAggregateMetricFromResultSet(metrics, condition, metricFunctions, rs);
         }
       }
     } finally {
@@ -593,34 +559,34 @@ public class PhoenixHBaseAccessor {
     return metrics;
   }
 
-  private void appendAggregateMetricFromResultSet(
-    TimelineMetrics metrics, Condition condition,
-    Map<String, List<Function>> metricFunctions, ResultSet rs)
-    throws SQLException {
+  private void appendAggregateMetricFromResultSet(TimelineMetrics metrics,
+      Condition condition, Map<String, List<Function>> metricFunctions,
+      ResultSet rs) throws SQLException {
 
     String metricName = rs.getString("METRIC_NAME");
     List<Function> functions = metricFunctions.get(metricName);
 
     for (Function aggregateFunction : functions) {
-      TimelineMetric metric;
+      SingleValuedTimelineMetric metric;
 
-      if (condition.getPrecision() == Precision.HOURS) {
-        metric = getAggregateHoursTimelineMetricFromResultSet(rs, aggregateFunction);
+      if (condition.getPrecision() == Precision.HOURS
+          || condition.getPrecision() == Precision.DAYS) {
+        metric = getAggregateTimelineMetricFromResultSet(rs, aggregateFunction, false);
       } else {
-        metric = getAggregateTimelineMetricFromResultSet(rs, aggregateFunction);
+        metric = getAggregateTimelineMetricFromResultSet(rs, aggregateFunction, true);
       }
 
       if (condition.isGrouped()) {
         metrics.addOrMergeTimelineMetric(metric);
       } else {
-        metrics.getMetrics().add(metric);
+        metrics.getMetrics().add(metric.getTimelineMetric());
       }
     }
   }
 
-  private PreparedStatement getLatestAggregateMetricRecords(
-    Condition condition, Connection conn, TimelineMetrics metrics)
-    throws SQLException {
+  private PreparedStatement getLatestAggregateMetricRecords(Condition condition,
+      Connection conn, TimelineMetrics metrics,
+      Map<String, List<Function>> metricFunctions) throws SQLException {
 
     PreparedStatement stmt = null;
     SplitByMetricNamesCondition splitCondition =
@@ -629,15 +595,28 @@ public class PhoenixHBaseAccessor {
     for (String metricName: splitCondition.getOriginalMetricNames()) {
 
       splitCondition.setCurrentMetric(metricName);
-      stmt = PhoenixTransactSQL.prepareGetLatestAggregateMetricSqlStmt(conn,
-        splitCondition);
+      stmt = PhoenixTransactSQL.prepareGetLatestAggregateMetricSqlStmt(conn, splitCondition);
       ResultSet rs = null;
       try {
         rs = stmt.executeQuery();
         while (rs.next()) {
-          TimelineMetric metric = getAggregateTimelineMetricFromResultSet(rs,
-            new Function());
-          metrics.getMetrics().add(metric);
+          List<Function> functions = metricFunctions.get(metricName);
+          if (functions != null) {
+            for (Function f : functions) {
+              SingleValuedTimelineMetric metric =
+                getAggregateTimelineMetricFromResultSet(rs, f, true);
+
+              if (condition.isGrouped()) {
+                metrics.addOrMergeTimelineMetric(metric);
+              } else {
+                metrics.getMetrics().add(metric.getTimelineMetric());
+              }
+            }
+          } else {
+            SingleValuedTimelineMetric metric =
+              getAggregateTimelineMetricFromResultSet(rs, new Function(), true);
+            metrics.getMetrics().add(metric.getTimelineMetric());
+          }
         }
       } finally {
         if (rs != null) {
@@ -653,54 +632,28 @@ public class PhoenixHBaseAccessor {
     return stmt;
   }
 
-  private TimelineMetric getAggregateTimelineMetricFromResultSet(
-    ResultSet rs, Function f) throws SQLException {
-    TimelineMetric metric = new TimelineMetric();
-    metric.setAppId(rs.getString("APP_ID"));
-    metric.setInstanceId(rs.getString("INSTANCE_ID"));
-    metric.setTimestamp(rs.getLong("SERVER_TIME"));
-    metric.setStartTime(rs.getLong("SERVER_TIME"));
+  private SingleValuedTimelineMetric getAggregateTimelineMetricFromResultSet(ResultSet rs,
+      Function f, boolean useHostCount) throws SQLException {
 
-    double value;
-    switch(f.getReadFunction()){
-      case AVG:
-        value = rs.getDouble("METRIC_SUM") / rs.getInt("HOSTS_COUNT");
-        break;
-      case MIN:
-        value = rs.getDouble("METRIC_MIN");
-        break;
-      case MAX:
-        value = rs.getDouble("METRIC_MAX");
-        break;
-      case SUM:
-        value = rs.getDouble("METRIC_SUM");
-        break;
-      default:
-        value = rs.getDouble("METRIC_SUM") / rs.getInt("HOSTS_COUNT");
-        break;
+    String countColumnName = "METRIC_COUNT";
+    if (useHostCount) {
+      countColumnName = "HOSTS_COUNT";
     }
 
-    metric.setMetricName(rs.getString("METRIC_NAME") + f.getSuffix());
-
-    Map<Long, Double> valueMap = new TreeMap<Long, Double>();
-    valueMap.put(rs.getLong("SERVER_TIME"), value);
-    metric.setMetricValues(valueMap);
-
-    return metric;
-  }
-
-  private TimelineMetric getAggregateHoursTimelineMetricFromResultSet(
-    ResultSet rs, Function f) throws SQLException {
-    TimelineMetric metric = new TimelineMetric();
-    metric.setAppId(rs.getString("APP_ID"));
-    metric.setInstanceId(rs.getString("INSTANCE_ID"));
-    metric.setTimestamp(rs.getLong("SERVER_TIME"));
-    metric.setStartTime(rs.getLong("SERVER_TIME"));
+    SingleValuedTimelineMetric metric = new SingleValuedTimelineMetric(
+      rs.getString("METRIC_NAME") + f.getSuffix(),
+      rs.getString("APP_ID"),
+      rs.getString("INSTANCE_ID"),
+      null,
+      rs.getLong("SERVER_TIME"),
+      rs.getLong("SERVER_TIME"),
+      rs.getString("UNITS")
+    );
 
     double value;
     switch(f.getReadFunction()){
       case AVG:
-        value = rs.getDouble("METRIC_SUM") / rs.getInt("METRIC_COUNT");
+        value = rs.getDouble("METRIC_SUM") / rs.getInt(countColumnName);
         break;
       case MIN:
         value = rs.getDouble("METRIC_MIN");
@@ -712,15 +665,11 @@ public class PhoenixHBaseAccessor {
         value = rs.getDouble("METRIC_SUM");
         break;
       default:
-        value = rs.getDouble("METRIC_SUM") / rs.getInt("METRIC_COUNT");
+        value = rs.getDouble("METRIC_SUM") / rs.getInt(countColumnName);
         break;
     }
 
-    metric.setMetricName(rs.getString("METRIC_NAME") + f.getSuffix());
-
-    Map<Long, Double> valueMap = new TreeMap<Long, Double>();
-    valueMap.put(rs.getLong("SERVER_TIME"), value);
-    metric.setMetricValues(valueMap);
+    metric.setSingleTimeseriesValue(rs.getLong("SERVER_TIME"), value);
 
     return metric;
   }

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

@@ -34,6 +34,7 @@ import static org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.ti
 
 public class TimelineMetricHostAggregator extends AbstractTimelineAggregator {
   private static final Log LOG = LogFactory.getLog(TimelineMetricHostAggregator.class);
+  TimelineMetricReadHelper readHelper = new TimelineMetricReadHelper(false);
 
   public TimelineMetricHostAggregator(PhoenixHBaseAccessor hBaseAccessor,
                                       Configuration metricsConf,
@@ -79,14 +80,14 @@ public class TimelineMetricHostAggregator extends AbstractTimelineAggregator {
       throws IOException, SQLException {
     TimelineMetric existingMetric = null;
     MetricHostAggregate hostAggregate = null;
-    Map<TimelineMetric, MetricHostAggregate> hostAggregateMap =
-      new HashMap<TimelineMetric, MetricHostAggregate>();
+    Map<TimelineMetric, MetricHostAggregate> hostAggregateMap =  new HashMap<TimelineMetric, MetricHostAggregate>();
+
 
     while (rs.next()) {
       TimelineMetric currentMetric =
-        PhoenixHBaseAccessor.getTimelineMetricKeyFromResultSet(rs);
+        readHelper.getTimelineMetricKeyFromResultSet(rs);
       MetricHostAggregate currentHostAggregate =
-        PhoenixHBaseAccessor.getMetricHostAggregateFromResultSet(rs);
+        readHelper.getMetricHostAggregateFromResultSet(rs);
 
       if (existingMetric == null) {
         // First row

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

@@ -38,7 +38,7 @@ public class TimelineMetricReadHelper {
   }
 
   public TimelineMetric getTimelineMetricFromResultSet(ResultSet rs)
-    throws SQLException, IOException {
+      throws SQLException, IOException {
     TimelineMetric metric = getTimelineMetricCommonsFromResultSet(rs);
     Map<Long, Double> sortedByTimeMetrics = new TreeMap<Long, Double>(
         PhoenixHBaseAccessor.readMetricFromJSON(rs.getString("METRICS")));
@@ -50,7 +50,7 @@ public class TimelineMetricReadHelper {
    * Returns common part of timeline metrics record without the values.
    */
   public TimelineMetric getTimelineMetricCommonsFromResultSet(ResultSet rs)
-    throws SQLException {
+      throws SQLException {
     TimelineMetric metric = new TimelineMetric();
     metric.setMetricName(rs.getString("METRIC_NAME"));
     metric.setAppId(rs.getString("APP_ID"));
@@ -76,7 +76,7 @@ public class TimelineMetricReadHelper {
   }
 
   public MetricClusterAggregate getMetricClusterTimeAggregateFromResultSet(ResultSet rs)
-    throws SQLException {
+      throws SQLException {
     MetricClusterAggregate agg = new MetricClusterAggregate();
     agg.setSum(rs.getDouble("METRIC_SUM"));
     agg.setMax(rs.getDouble("METRIC_MAX"));
@@ -98,5 +98,28 @@ public class TimelineMetricReadHelper {
       rs.getString("UNITS"));
   }
 
+  public MetricHostAggregate getMetricHostAggregateFromResultSet(ResultSet rs)
+      throws SQLException {
+    MetricHostAggregate metricHostAggregate = new MetricHostAggregate();
+    metricHostAggregate.setSum(rs.getDouble("METRIC_SUM"));
+    metricHostAggregate.setMax(rs.getDouble("METRIC_MAX"));
+    metricHostAggregate.setMin(rs.getDouble("METRIC_MIN"));
+    metricHostAggregate.setNumberOfSamples(rs.getLong("METRIC_COUNT"));
+
+    metricHostAggregate.setDeviation(0.0);
+    return metricHostAggregate;
+  }
+
+  public TimelineMetric getTimelineMetricKeyFromResultSet(ResultSet rs)
+      throws SQLException, IOException {
+    TimelineMetric metric = new TimelineMetric();
+    metric.setMetricName(rs.getString("METRIC_NAME"));
+    metric.setAppId(rs.getString("APP_ID"));
+    metric.setInstanceId(rs.getString("INSTANCE_ID"));
+    metric.setHostName(rs.getString("HOSTNAME"));
+    metric.setTimestamp(rs.getLong("SERVER_TIME"));
+    metric.setType(rs.getString("UNITS"));
+    return metric;
+  }
 }
 

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

@@ -135,7 +135,6 @@ public class ITClusterAggregator extends AbstractMiniHBaseClusterTest {
     }
   }
 
-
   @Test
   public void testShouldAggregateClusterIgnoringInstance() throws Exception {
     // GIVEN

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

@@ -23,6 +23,7 @@ import org.apache.hadoop.metrics2.sink.timeline.TimelineMetrics;
 import org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline.aggregators.MetricHostAggregate;
 import org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline.aggregators.TimelineMetricAggregator;
 import org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline.aggregators.TimelineMetricAggregatorFactory;
+import org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline.aggregators.TimelineMetricReadHelper;
 import org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline.query.Condition;
 import org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline.query.DefaultCondition;
 import org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline.query.PhoenixTransactSQL;
@@ -107,6 +108,7 @@ public class ITMetricAggregator extends AbstractMiniHBaseClusterTest {
     // GIVEN
     TimelineMetricAggregator aggregatorMinute =
       TimelineMetricAggregatorFactory.createTimelineMetricAggregatorMinute(hdb, new Configuration());
+    TimelineMetricReadHelper readHelper = new TimelineMetricReadHelper(false);
 
     long startTime = System.currentTimeMillis();
     long ctime = startTime;
@@ -137,9 +139,9 @@ public class ITMetricAggregator extends AbstractMiniHBaseClusterTest {
     int count = 0;
     while (rs.next()) {
       TimelineMetric currentMetric =
-        PhoenixHBaseAccessor.getTimelineMetricKeyFromResultSet(rs);
+        readHelper.getTimelineMetricKeyFromResultSet(rs);
       MetricHostAggregate currentHostAggregate =
-        PhoenixHBaseAccessor.getMetricHostAggregateFromResultSet(rs);
+        readHelper.getMetricHostAggregateFromResultSet(rs);
 
       if ("disk_free".equals(currentMetric.getMetricName())) {
         assertEquals(2.0, currentHostAggregate.getMax());
@@ -167,6 +169,7 @@ public class ITMetricAggregator extends AbstractMiniHBaseClusterTest {
     // GIVEN
     TimelineMetricAggregator aggregator =
       TimelineMetricAggregatorFactory.createTimelineMetricAggregatorHourly(hdb, new Configuration());
+    TimelineMetricReadHelper readHelper = new TimelineMetricReadHelper(false);
     long startTime = System.currentTimeMillis();
 
     MetricHostAggregate expectedAggregate =
@@ -210,9 +213,9 @@ public class ITMetricAggregator extends AbstractMiniHBaseClusterTest {
 
     while (rs.next()) {
       TimelineMetric currentMetric =
-        PhoenixHBaseAccessor.getTimelineMetricKeyFromResultSet(rs);
+        readHelper.getTimelineMetricKeyFromResultSet(rs);
       MetricHostAggregate currentHostAggregate =
-        PhoenixHBaseAccessor.getMetricHostAggregateFromResultSet(rs);
+        readHelper.getMetricHostAggregateFromResultSet(rs);
 
       if ("disk_used".equals(currentMetric.getMetricName())) {
         assertEquals(2.0, currentHostAggregate.getMax());
@@ -229,6 +232,7 @@ public class ITMetricAggregator extends AbstractMiniHBaseClusterTest {
     // GIVEN
     TimelineMetricAggregator aggregator =
       TimelineMetricAggregatorFactory.createTimelineMetricAggregatorDaily(hdb, new Configuration());
+    TimelineMetricReadHelper readHelper = new TimelineMetricReadHelper(false);
     long startTime = System.currentTimeMillis();
 
     MetricHostAggregate expectedAggregate =
@@ -270,9 +274,9 @@ public class ITMetricAggregator extends AbstractMiniHBaseClusterTest {
 
     while (rs.next()) {
       TimelineMetric currentMetric =
-        PhoenixHBaseAccessor.getTimelineMetricKeyFromResultSet(rs);
+        readHelper.getTimelineMetricKeyFromResultSet(rs);
       MetricHostAggregate currentHostAggregate =
-        PhoenixHBaseAccessor.getMetricHostAggregateFromResultSet(rs);
+        readHelper.getMetricHostAggregateFromResultSet(rs);
 
       if ("disk_used".equals(currentMetric.getMetricName())) {
         assertEquals(2.0, currentHostAggregate.getMax());
@@ -301,9 +305,7 @@ public class ITMetricAggregator extends AbstractMiniHBaseClusterTest {
     return metrics;
   }
 
-  private TimelineMetric createMetric(long startTime,
-                                      String metricName,
-                                      String host) {
+  private TimelineMetric createMetric(long startTime, String metricName, String host) {
     TimelineMetric m = new TimelineMetric();
     m.setAppId("host");
     m.setHostName(host);

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

@@ -86,20 +86,20 @@ public class ITPhoenixHBaseAccessor extends AbstractMiniHBaseClusterTest {
     long ctime = startTime;
     long minute = 60 * 1000;
     hdb.insertMetricRecords(prepareSingleTimelineMetric(ctime, "local1",
-        "disk_free", 1));
+      "disk_free", 1));
     hdb.insertMetricRecords(prepareSingleTimelineMetric(ctime, "local2",
-        "disk_free", 2));
+      "disk_free", 2));
     ctime += minute;
     hdb.insertMetricRecords(prepareSingleTimelineMetric(ctime, "local1",
-        "disk_free", 2));
+      "disk_free", 2));
     hdb.insertMetricRecords(prepareSingleTimelineMetric(ctime, "local2",
-        "disk_free", 1));
+      "disk_free", 1));
 
     // WHEN
     long endTime = ctime + minute;
     Condition condition = new DefaultCondition(
-        Collections.singletonList("disk_free"), "local1", null, null, startTime,
-        endTime, Precision.SECONDS, null, true);
+      Collections.singletonList("disk_free"), "local1", null, null, startTime,
+      endTime, Precision.SECONDS, null, true);
     TimelineMetrics timelineMetrics = hdb.getMetricRecords(condition,
       singletonValueFunctionMap("disk_free"));
 
@@ -239,6 +239,46 @@ public class ITPhoenixHBaseAccessor extends AbstractMiniHBaseClusterTest {
     assertEquals(1.5, metric.getMetricValues().values().iterator().next(), 0.00001);
   }
 
+  @Test
+  public void testGetClusterMetricRecordLatestWithFunction() throws Exception {
+    // GIVEN
+    TimelineMetricAggregator agg =
+      TimelineMetricAggregatorFactory.createTimelineClusterAggregatorMinute(hdb, new Configuration());
+
+    long startTime = System.currentTimeMillis();
+    long ctime = startTime + 1;
+    long minute = 60 * 1000;
+    hdb.insertMetricRecords(prepareSingleTimelineMetric(ctime, "local1",
+      "disk_free", 1));
+    hdb.insertMetricRecords(prepareSingleTimelineMetric(ctime, "local2",
+      "disk_free", 2));
+    ctime += minute;
+    hdb.insertMetricRecords(prepareSingleTimelineMetric(ctime, "local1",
+      "disk_free", 2));
+    hdb.insertMetricRecords(prepareSingleTimelineMetric(ctime, "local2",
+      "disk_free", 1));
+
+    long endTime = ctime + minute + 1;
+    boolean success = agg.doWork(startTime, endTime);
+    assertTrue(success);
+
+    // WHEN
+    Condition condition = new DefaultCondition(
+      Collections.singletonList("disk_free"), null, null, null,
+      null, null, Precision.SECONDS, null, true);
+    TimelineMetrics timelineMetrics = hdb.getAggregateMetricRecords(condition,
+      Collections.singletonMap("disk_free",
+        Collections.singletonList(new Function(Function.ReadFunction.SUM, null))));
+
+    //THEN
+    assertEquals(1, timelineMetrics.getMetrics().size());
+    TimelineMetric metric = timelineMetrics.getMetrics().get(0);
+
+    assertEquals("disk_free._sum", metric.getMetricName());
+    assertEquals(1, metric.getMetricValues().size());
+    assertEquals(3, metric.getMetricValues().values().iterator().next().intValue());
+  }
+
   @Test
   public void testGetClusterMetricRecordsHours() throws Exception {
     // GIVEN