AMBARI-8742. Implement Report graphs capability in AMS. (swagle)

ambari-metrics/ambari-metrics-host-monitoring/src/main/python/core/host_info.py

@@ -38,9 +38,10 @@ pass
 
 class HostInfo():
   def __init__(self):
-    self.__last_network_io_time = 0;
+    self.__last_network_io_time = 0
     self.__last_network_data = {}
     self.__last_network_lock = threading.Lock()
+    self.__host_static_info = self.get_host_static_info()
 
   def get_cpu_times(self):
     """
@@ -48,10 +49,13 @@ class HostInfo():
     """
     cpu_times = psutil.cpu_times_percent()
     load_avg = os.getloadavg()
+    cpu_count = self.__host_static_info.get('cpu_num', 1)
 
-    number2percents = lambda x: x * 100
+    # Divide by number of cpu's on the system
+    number2percents = lambda x: ((x / int(cpu_count)) * 100)
 
     return {
+      'cpu_num': int(cpu_count),
       'cpu_user': number2percents(cpu_times.user) if hasattr(cpu_times, 'user') else '',
       'cpu_system': number2percents(cpu_times.system) if hasattr(cpu_times, 'system') else '',
       'cpu_idle': number2percents(cpu_times.idle) if hasattr(cpu_times, 'idle') else '',
@@ -96,10 +100,13 @@ class HostInfo():
     mem_stats = psutil.virtual_memory()
     swap_stats = psutil.swap_memory()
     disk_usage = self.get_combined_disk_usage()
+    mem_total = self.__host_static_info.get('mem_total')
 
     bytes2kilobytes = lambda x: x / 1024
 
     return {
+      'mem_total': bytes2kilobytes(mem_total) if mem_total else '',
+      'mem_used': bytes2kilobytes(mem_stats.used) if hasattr(mem_stats, 'used') else '',
       'mem_free': bytes2kilobytes(mem_stats.free) if hasattr(mem_stats, 'free') else '',
       'mem_shared': bytes2kilobytes(mem_stats.shared) if hasattr(mem_stats, 'shared') else '',
       'mem_buffered': bytes2kilobytes(mem_stats.buffers) if hasattr(mem_stats, 'buffers') else '',

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

@@ -23,19 +23,19 @@ package org.apache.hadoop.yarn.server.applicationhistoryservice.metrics
  * since this is a not recoverable situation, and should be handled by main or
  * service method followed by shutdown.
  */
-public class MetricsInitializationException extends RuntimeException {
-  public MetricsInitializationException() {
+public class MetricsSystemInitializationException extends RuntimeException {
+  public MetricsSystemInitializationException() {
   }
 
-  public MetricsInitializationException(String msg) {
+  public MetricsSystemInitializationException(String msg) {
     super(msg);
   }
 
-  public MetricsInitializationException(Throwable t) {
+  public MetricsSystemInitializationException(Throwable t) {
     super(t);
   }
 
-  public MetricsInitializationException(String msg, Throwable t) {
+  public MetricsSystemInitializationException(String msg, Throwable t) {
     super(msg, t);
   }
 

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

@@ -272,12 +272,12 @@ public class PhoenixHBaseAccessor {
         LOG.warn("Cannot update TTL on tables. " + sql.getMessage());
       } else {
         LOG.error("Error creating Metrics Schema in HBase using Phoenix.", sql);
-        throw new MetricsInitializationException(
+        throw new MetricsSystemInitializationException(
           "Error creating Metrics Schema in HBase using Phoenix.", sql);
       }
     } catch (InterruptedException e) {
       LOG.error("Error creating Metrics Schema in HBase using Phoenix.", e);
-      throw new MetricsInitializationException(
+      throw new MetricsSystemInitializationException(
         "Error creating Metrics Schema in HBase using Phoenix.", e);
     } finally {
       if (stmt != null) {
@@ -651,7 +651,13 @@ public class PhoenixHBaseAccessor {
     }
   }
 
-
+  /**
+   * Get metrics aggregated across hosts.
+   *
+   * @param condition @Condition
+   * @return @TimelineMetrics
+   * @throws SQLException
+   */
   public TimelineMetrics getAggregateMetricRecords(final Condition condition)
     throws SQLException {
 

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

@@ -256,15 +256,9 @@ public class PhoenixTransactSQL {
       LOG.debug("Setting pos: " + pos + ", value: " + condition.getHostname());
       stmt.setString(pos++, condition.getHostname());
     }
-    // TODO: Upper case all strings on POST
     if (condition.getAppId() != null) {
-      // TODO: fix case of appId coming from host metrics
-      String appId = condition.getAppId();
-      if (!condition.getAppId().equals("HOST")) {
-        appId = appId.toLowerCase();
-      }
-      LOG.debug("Setting pos: " + pos + ", value: " + appId);
-      stmt.setString(pos++, appId);
+      LOG.debug("Setting pos: " + pos + ", value: " + condition.getAppId());
+      stmt.setString(pos++, condition.getAppId());
     }
     if (condition.getInstanceId() != null) {
       LOG.debug("Setting pos: " + pos + ", value: " + condition.getInstanceId());
@@ -301,11 +295,12 @@ public class PhoenixTransactSQL {
       sb.append(" LIMIT ").append(condition.getLimit());
     }
 
-    LOG.debug("SQL => " + sb.toString() + ", condition => " + condition);
     String query = String.format(sb.toString(),
       PhoenixTransactSQL.getNaiveTimeRangeHint(condition.getStartTime(),
         NATIVE_TIME_RANGE_DELTA));
-    PreparedStatement stmt = connection.prepareStatement(query);    int pos = 1;
+    LOG.debug("SQL => " + query + ", condition => " + condition);
+    PreparedStatement stmt = connection.prepareStatement(query);
+    int pos = 1;
     if (condition.getMetricNames() != null) {
       for (; pos <= condition.getMetricNames().size(); pos++) {
         stmt.setString(pos, condition.getMetricNames().get(pos - 1));
@@ -313,7 +308,7 @@ public class PhoenixTransactSQL {
     }
     // TODO: Upper case all strings on POST
     if (condition.getAppId() != null) {
-      stmt.setString(pos++, condition.getAppId().toLowerCase());
+      stmt.setString(pos++, condition.getAppId());
     }
     if (condition.getInstanceId() != null) {
       stmt.setString(pos++, condition.getInstanceId());
@@ -438,7 +433,14 @@ public class PhoenixTransactSQL {
     }
 
     String getAppId() {
-      return appId == null || appId.isEmpty() ? null : appId;
+      if (appId != null && !appId.isEmpty()) {
+        if (!appId.equals("HOST")) {
+          return appId.toLowerCase();
+        } else {
+          return appId;
+        }
+      }
+      return null;
     }
 
     String getInstanceId() {

ambari-metrics/ambari-metrics-timelineservice/src/test/conf/hbase-site.xml

@@ -227,4 +227,19 @@
       hbase.server.thread.wakefrequency.
     </description>
   </property>
+  <property>
+    <name>hbase.snapshot.enabled</name>
+    <value>false</value>
+    <description>Enable/Disable HBase snapshots.</description>
+  </property>
+  <property>
+    <name>hbase.replication</name>
+    <value>false</value>
+    <description>Enable/Disable HBase replication.</description>
+  </property>
+  <property>
+    <name>zookeeper.session.timeout</name>
+    <value>120000</value>
+    <description>ZooKeeper session timeout in milliseconds.</description>
+  </property>
 </configuration>

ambari-server/src/main/java/org/apache/ambari/server/controller/internal/AbstractPropertyProvider.java

@@ -20,7 +20,10 @@ package org.apache.ambari.server.controller.internal;
 
 import java.lang.reflect.InvocationTargetException;
 import java.lang.reflect.Method;
+import java.text.DecimalFormat;
+import java.util.Collections;
 import java.util.HashMap;
+import java.util.HashSet;
 import java.util.LinkedList;
 import java.util.List;
 import java.util.Map;
@@ -30,6 +33,8 @@ import java.util.regex.Pattern;
 
 import org.apache.ambari.server.controller.spi.PropertyProvider;
 import org.apache.ambari.server.controller.utilities.PropertyHelper;
+import org.apache.hadoop.metrics2.sink.timeline.TimelineMetric;
+import org.apache.http.client.utils.URIBuilder;
 
 /**
  *  Abstract property provider implementation.
@@ -64,6 +69,20 @@ public abstract class AbstractPropertyProvider extends BaseProvider implements P
    */
   private static final String FIND_REGEX_IN_METRIC_REGEX = "\\([^)]+\\)";
 
+  private static final Set<String> PERCENTAGE_METRIC;
+  private static final DecimalFormat decimalFormat = new DecimalFormat("#.00");
+
+  static {
+    Set<String> temp = new HashSet<String>();
+    temp.add("cpu_wio");
+    temp.add("cpu_idle");
+    temp.add("cpu_nice");
+    temp.add("cpu_aidle");
+    temp.add("cpu_system");
+    temp.add("cpu_user");
+    PERCENTAGE_METRIC = Collections.unmodifiableSet(temp);
+  }
+
   // ----- Constructors ------------------------------------------------------
 
   /**
@@ -347,4 +366,49 @@ public abstract class AbstractPropertyProvider extends BaseProvider implements P
     }
     return false;
   }
+
+  // Normalize percent values: Copied over from Ganglia Metric
+  private static Number[][] getGangliaLikeDatapoints(TimelineMetric metric) {
+    Number[][] datapointsArray = new Number[metric.getMetricValues().size()][2];
+    int cnt = 0;
+
+    for (Map.Entry<Long, Double> metricEntry : metric.getMetricValues().entrySet()) {
+      Double value = metricEntry.getValue();
+      Long time = metricEntry.getKey();
+      if (time > 9999999999l) {
+        time = time / 1000;
+      }
+
+      if (PERCENTAGE_METRIC.contains(metric.getMetricName())) {
+        value = new Double(decimalFormat.format(value / 100));
+      }
+
+      datapointsArray[cnt][0] = value;
+      datapointsArray[cnt][1] = time;
+      cnt++;
+    }
+
+    return datapointsArray;
+  }
+
+  /**
+   * Get value from the given metric.
+   *
+   * @param metric      the metric
+   * @param isTemporal  indicates whether or not this a temporal metric
+   *
+   * @return a range of temporal data or a point in time value if not temporal
+   */
+  protected static Object getValue(TimelineMetric metric, boolean isTemporal) {
+    Number[][] dataPoints = getGangliaLikeDatapoints(metric);
+
+    int length = dataPoints.length;
+    if (isTemporal) {
+      return length > 0 ? dataPoints : null;
+    } else {
+      // return the value of the last data point
+      return length > 0 ? dataPoints[length - 1][0] : 0;
+    }
+  }
+
 }

ambari-server/src/main/java/org/apache/ambari/server/controller/internal/PropertyInfo.java

@@ -25,6 +25,7 @@ public class PropertyInfo {
   private final String propertyId;
   private final boolean temporal;
   private final boolean pointInTime;
+  private String amsId;
 
   public PropertyInfo(String propertyId, boolean temporal, boolean pointInTime) {
     this.propertyId = propertyId;
@@ -43,4 +44,12 @@ public class PropertyInfo {
   public boolean isPointInTime() {
     return pointInTime;
   }
+
+  public String getAmsId() {
+    return amsId;
+  }
+
+  public void setAmsId(String amsId) {
+    this.amsId = amsId;
+  }
 }

ambari-server/src/main/java/org/apache/ambari/server/controller/metrics/MetricsPropertyProvider.java

@@ -25,6 +25,7 @@ import org.apache.ambari.server.controller.spi.Request;
 import org.apache.ambari.server.controller.spi.Resource;
 import org.apache.ambari.server.controller.spi.SystemException;
 import org.apache.ambari.server.controller.utilities.StreamProvider;
+import org.apache.http.client.utils.URIBuilder;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import java.util.Map;
@@ -148,7 +149,7 @@ public abstract class MetricsPropertyProvider extends AbstractPropertyProvider {
    *
    * @return a comma delimited string of strings
    */
-  protected static String getSetString(Set<String> set, int limit) {
+  public static String getSetString(Set<String> set, int limit) {
     StringBuilder sb = new StringBuilder();
 
     if (limit == -1 || set.size() <= limit) {
@@ -161,4 +162,5 @@ public abstract class MetricsPropertyProvider extends AbstractPropertyProvider {
     }
     return sb.toString();
   }
+
 }

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

@@ -53,11 +53,7 @@ import static org.codehaus.jackson.map.annotate.JsonSerialize.Inclusion;
 public abstract class AMSPropertyProvider extends MetricsPropertyProvider {
   static final Map<String, String> TIMELINE_APPID_MAP = new HashMap<String, String>();
   private static ObjectMapper mapper;
-  //private final HttpClient httpClient = new HttpClient();
   private final static ObjectReader timelineObjectReader;
-  private static final DecimalFormat decimalFormat = new DecimalFormat("#.00");
-
-  private static final Set<String> PERCENTAGE_METRIC;
   private static final String METRIC_REGEXP_PATTERN = "\\([^)]*\\)";
 
   static {
@@ -71,15 +67,6 @@ public abstract class AMSPropertyProvider extends MetricsPropertyProvider {
     //noinspection deprecation
     mapper.getSerializationConfig().setSerializationInclusion(Inclusion.NON_NULL);
     timelineObjectReader = mapper.reader(TimelineMetrics.class);
-
-    Set<String> temp = new HashSet<String>();
-    temp.add("cpu_wio");
-    temp.add("cpu_idle");
-    temp.add("cpu_nice");
-    temp.add("cpu_aidle");
-    temp.add("cpu_system");
-    temp.add("cpu_user");
-    PERCENTAGE_METRIC = Collections.unmodifiableSet(temp);
   }
 
   public AMSPropertyProvider(Map<String, Map<String, PropertyInfo>> componentPropertyInfoMap,
@@ -324,59 +311,6 @@ public abstract class AMSPropertyProvider extends MetricsPropertyProvider {
     }
   }
 
-  // Normalize percent values: Copied over from Ganglia Metric
-  private static Number[][] getGangliaLikeDatapoints(TimelineMetric metric) {
-    Number[][] datapointsArray = new Number[metric.getMetricValues().size()][2];
-    int cnt = 0;
-
-    for (Map.Entry<Long, Double> metricEntry : metric.getMetricValues().entrySet()) {
-      Double value = metricEntry.getValue();
-      Long time = metricEntry.getKey();
-      if (time > 9999999999l) {
-        time = time / 1000;
-      }
-
-      if (PERCENTAGE_METRIC.contains(metric.getMetricName())) {
-        value = new Double(decimalFormat.format(value / 100));
-      }
-
-      datapointsArray[cnt][0] = value;
-      datapointsArray[cnt][1] = time;
-      cnt++;
-    }
-
-    return datapointsArray;
-  }
-
-  /**
-   * Get value from the given metric.
-   *
-   * @param metric      the metric
-   * @param isTemporal  indicates whether or not this a temporal metric
-   *
-   * @return a range of temporal data or a point in time value if not temporal
-   */
-  private static Object getValue(TimelineMetric metric, boolean isTemporal) {
-    Number[][] dataPoints = getGangliaLikeDatapoints(metric);
-
-    int length = dataPoints.length;
-    if (isTemporal) {
-      return length > 0 ? dataPoints : null;
-    } else {
-      // return the value of the last data point
-      return length > 0 ? dataPoints[length - 1][0] : 0;
-    }
-  }
-
-  protected static URIBuilder getUriBuilder(String hostname, int port) {
-    URIBuilder uriBuilder = new URIBuilder();
-    uriBuilder.setScheme("http");
-    uriBuilder.setHost(hostname);
-    uriBuilder.setPort(port);
-    uriBuilder.setPath("/ws/v1/timeline/metrics");
-    return uriBuilder;
-  }
-
   @Override
   public Set<Resource> populateResourcesWithProperties(Set<Resource> resources,
                Request request, Set<String> propertyIds) throws SystemException {
@@ -447,7 +381,7 @@ public abstract class AMSPropertyProvider extends MetricsPropertyProvider {
             MetricsRequest metricsRequest = requests.get(temporalInfo);
             if (metricsRequest == null) {
               metricsRequest = new MetricsRequest(temporalInfo,
-                getUriBuilder(collectorHostName,
+                getAMSUriBuilder(collectorHostName,
                   collectorPort != null ? Integer.parseInt(collectorPort) : 8188));
               requests.put(temporalInfo, metricsRequest);
             }
@@ -460,4 +394,13 @@ public abstract class AMSPropertyProvider extends MetricsPropertyProvider {
 
     return requestMap;
   }
+
+  static URIBuilder getAMSUriBuilder(String hostname, int port) {
+    URIBuilder uriBuilder = new URIBuilder();
+    uriBuilder.setScheme("http");
+    uriBuilder.setHost(hostname);
+    uriBuilder.setPort(port);
+    uriBuilder.setPath("/ws/v1/timeline/metrics");
+    return uriBuilder;
+  }
 }

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

@@ -20,16 +20,46 @@ package org.apache.ambari.server.controller.metrics.timeline;
 import org.apache.ambari.server.configuration.ComponentSSLConfiguration;
 import org.apache.ambari.server.controller.internal.PropertyInfo;
 import org.apache.ambari.server.controller.metrics.MetricHostProvider;
+import org.apache.ambari.server.controller.metrics.MetricsPropertyProvider;
 import org.apache.ambari.server.controller.metrics.MetricsReportPropertyProvider;
 import org.apache.ambari.server.controller.spi.Predicate;
 import org.apache.ambari.server.controller.spi.Request;
 import org.apache.ambari.server.controller.spi.Resource;
 import org.apache.ambari.server.controller.spi.SystemException;
+import org.apache.ambari.server.controller.spi.TemporalInfo;
 import org.apache.ambari.server.controller.utilities.StreamProvider;
+import org.apache.hadoop.metrics2.sink.timeline.TimelineMetric;
+import org.apache.hadoop.metrics2.sink.timeline.TimelineMetrics;
+import org.apache.http.client.utils.URIBuilder;
+import org.codehaus.jackson.map.AnnotationIntrospector;
+import org.codehaus.jackson.map.ObjectMapper;
+import org.codehaus.jackson.map.ObjectReader;
+import org.codehaus.jackson.map.annotate.JsonSerialize;
+import org.codehaus.jackson.xc.JaxbAnnotationIntrospector;
+
+import java.io.BufferedReader;
+import java.io.IOException;
+import java.io.InputStreamReader;
+import java.util.HashMap;
+import java.util.HashSet;
 import java.util.Map;
 import java.util.Set;
 
+import static org.apache.ambari.server.controller.metrics.MetricsServiceProvider.MetricsService.GANGLIA;
+import static org.apache.ambari.server.controller.metrics.MetricsServiceProvider.MetricsService.TIMELINE_METRICS;
+
 public class AMSReportPropertyProvider extends MetricsReportPropertyProvider {
+  private static ObjectMapper mapper;
+  private final static ObjectReader timelineObjectReader;
+
+  static {
+    mapper = new ObjectMapper();
+    AnnotationIntrospector introspector = new JaxbAnnotationIntrospector();
+    mapper.setAnnotationIntrospector(introspector);
+    //noinspection deprecation
+    mapper.getSerializationConfig().setSerializationInclusion(JsonSerialize.Inclusion.NON_NULL);
+    timelineObjectReader = mapper.reader(TimelineMetrics.class);
+  }
 
   public AMSReportPropertyProvider(Map<String, Map<String, PropertyInfo>> componentPropertyInfoMap,
                                  StreamProvider streamProvider,
@@ -44,6 +74,174 @@ public class AMSReportPropertyProvider extends MetricsReportPropertyProvider {
   @Override
   public Set<Resource> populateResources(Set<Resource> resources,
                Request request, Predicate predicate) throws SystemException {
-    return resources;
+
+    Set<Resource> keepers = new HashSet<Resource>();
+    for (Resource resource : resources) {
+      if (populateResource(resource, request, predicate)) {
+        keepers.add(resource);
+      }
+    }
+    return keepers;
+  }
+
+  /**
+   * Populate a resource by obtaining the requested Ganglia RESOURCE_METRICS.
+   *
+   * @param resource  the resource to be populated
+   * @param request   the request
+   * @param predicate the predicate
+   *
+   * @return true if the resource was successfully populated with the requested properties
+   *
+   * @throws SystemException if unable to populate the resource
+   */
+  private boolean populateResource(Resource resource, Request request, Predicate predicate)
+    throws SystemException {
+
+    Set<String> propertyIds = getPropertyIds();
+
+    if (propertyIds.isEmpty()) {
+      return true;
+    }
+    String clusterName = (String) resource.getPropertyValue(clusterNamePropertyId);
+
+    // Check liveliness of host
+    if (!hostProvider.isCollectorHostLive(clusterName, TIMELINE_METRICS)) {
+      LOG.info("METRIC_COLLECTOR host is not live. Skip populating " +
+        "resources with metrics.");
+      return true;
+    }
+
+    // Check liveliness of Collector
+    if (!hostProvider.isCollectorComponentLive(clusterName, TIMELINE_METRICS)) {
+      LOG.info("METRIC_COLLECTOR is not live. Skip populating resources" +
+        " with metrics.");
+      return true;
+    }
+
+    setProperties(resource, clusterName, request, getRequestPropertyIds(request, predicate));
+
+    return true;
+  }
+
+  private void setProperties(Resource resource, String clusterName,
+                            Request request, Set<String> ids) throws SystemException {
+
+    Map<String, MetricReportRequest> reportRequestMap = getPropertyIdMaps(request, ids);
+    String host = hostProvider.getCollectorHostName(clusterName, TIMELINE_METRICS);
+    String port = hostProvider.getCollectorPortName(clusterName, TIMELINE_METRICS);
+    URIBuilder uriBuilder = AMSPropertyProvider.getAMSUriBuilder(host,
+      port != null ? Integer.parseInt(port) : 8188);
+
+    for (Map.Entry<String, MetricReportRequest> entry : reportRequestMap.entrySet()) {
+      MetricReportRequest reportRequest = entry.getValue();
+      TemporalInfo temporalInfo = reportRequest.getTemporalInfo();
+      Map<String, String> propertyIdMap = reportRequest.getPropertyIdMap();
+
+      uriBuilder.removeQuery();
+      // Call with hostname = null
+      uriBuilder.addParameter("metricNames",
+        MetricsPropertyProvider.getSetString(propertyIdMap.keySet(), -1));
+
+      uriBuilder.setParameter("appId", "HOST");
+      long startTime = temporalInfo.getStartTime();
+      if (startTime != -1) {
+        uriBuilder.setParameter("startTime", String.valueOf(startTime));
+      }
+
+      long endTime = temporalInfo.getEndTime();
+      if (endTime != -1) {
+        uriBuilder.setParameter("endTime", String.valueOf(endTime));
+      }
+
+      BufferedReader reader = null;
+      String spec = uriBuilder.toString();
+      try {
+        LOG.debug("Metrics request url =" + spec);
+        reader = new BufferedReader(new InputStreamReader(streamProvider.readFrom(spec)));
+
+        TimelineMetrics timelineMetrics = timelineObjectReader.readValue(reader);
+        LOG.debug("Timeline metrics response => " + timelineMetrics);
+
+        for (TimelineMetric metric : timelineMetrics.getMetrics()) {
+          if (metric.getMetricName() != null && metric.getMetricValues() != null) {
+            String propertyId = propertyIdMap.get(metric.getMetricName());
+            if (propertyId != null) {
+              resource.setProperty(propertyId, getValue(metric, true));
+            }
+          }
+        }
+
+      } catch (IOException io) {
+        LOG.warn("Error getting timeline metrics.", io);
+      } finally {
+        if (reader != null) {
+          try {
+            reader.close();
+          } catch (IOException e) {
+            if (LOG.isWarnEnabled()) {
+              LOG.warn("Unable to close http input steam : spec=" + spec, e);
+            }
+          }
+        }
+      }
+    }
+  }
+
+  private Map<String, MetricReportRequest> getPropertyIdMaps(Request request, Set<String> ids) {
+    Map<String, MetricReportRequest> propertyMap = new HashMap<String, MetricReportRequest>();
+
+    for (String id : ids) {
+      Map<String, PropertyInfo> propertyInfoMap = getPropertyInfoMap("*", id);
+
+      for (Map.Entry<String, PropertyInfo> entry : propertyInfoMap.entrySet()) {
+        PropertyInfo propertyInfo = entry.getValue();
+        String propertyId = entry.getKey();
+        String amsId = propertyInfo.getAmsId();
+
+        TemporalInfo temporalInfo = request.getTemporalInfo(id);
+
+        if (temporalInfo != null && propertyInfo.isTemporal()) {
+          String propertyName = propertyInfo.getPropertyId();
+          String report = null;
+          // format : report_name.metric_name
+          int dotIndex = propertyName.lastIndexOf('.');
+          if (dotIndex != -1){
+            report = propertyName.substring(0, dotIndex);
+          }
+          if (report !=  null) {
+            MetricReportRequest reportRequest = propertyMap.get(report);
+            if (reportRequest == null) {
+              reportRequest = new MetricReportRequest();
+              propertyMap.put(report, reportRequest);
+              reportRequest.setTemporalInfo(temporalInfo);
+            }
+            reportRequest.addPropertyId(amsId, propertyId);
+          }
+        }
+      }
+    }
+    return propertyMap;
+  }
+
+  class MetricReportRequest {
+    private TemporalInfo temporalInfo;
+    private Map<String, String> propertyIdMap = new HashMap<String, String>();
+
+    public TemporalInfo getTemporalInfo() {
+      return temporalInfo;
+    }
+
+    public void setTemporalInfo(TemporalInfo temporalInfo) {
+      this.temporalInfo = temporalInfo;
+    }
+
+    public Map<String, String> getPropertyIdMap() {
+      return propertyIdMap;
+    }
+
+    public void addPropertyId(String propertyName, String propertyId) {
+      propertyIdMap.put(propertyName, propertyId);
+    }
   }
 }

ambari-server/src/main/java/org/apache/ambari/server/controller/utilities/PropertyHelper.java

@@ -430,8 +430,10 @@ public class PropertyHelper {
           for (Map.Entry<String, Metric> metricEntry : componentEntry.getValue().entrySet()) {
             String property = metricEntry.getKey();
             Metric metric   = metricEntry.getValue();
-
-            metrics.put(property, new PropertyInfo(metric.getMetric(), metric.isTemporal(), metric.isPointInTime()));
+            PropertyInfo propertyInfo = new PropertyInfo(metric.getMetric(),
+              metric.isTemporal(), metric.isPointInTime());
+            propertyInfo.setAmsId(metric.getAmsId());
+            metrics.put(property, propertyInfo);
           }
           componentMetrics.put(componentEntry.getKey(), metrics);
         }
@@ -484,6 +486,7 @@ public class PropertyHelper {
     private String metric;
     private boolean pointInTime;
     private boolean temporal;
+    private String amsId;
 
     private Metric() {
     }
@@ -517,5 +520,13 @@ public class PropertyHelper {
     public void setTemporal(boolean temporal) {
       this.temporal = temporal;
     }
+
+    public String getAmsId() {
+      return amsId;
+    }
+
+    public void setAmsId(String amsId) {
+      this.amsId = amsId;
+    }
   }
 }

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

@@ -4,87 +4,104 @@
       "metrics/cpu/Idle":{
         "metric":"cpu_report.Idle\\g",
         "pointInTime":false,
-        "temporal":true
+        "temporal":true,
+        "amsId":"cpu_idle"
       },
       "metrics/cpu/Nice":{
         "metric":"cpu_report.Nice\\g",
         "pointInTime":false,
-        "temporal":true
+        "temporal":true,
+        "amsId":"cpu_nice"
       },
       "metrics/cpu/System":{
         "metric":"cpu_report.System\\g",
         "pointInTime":false,
-        "temporal":true
+        "temporal":true,
+        "amsId":"cpu_system"
       },
       "metrics/cpu/User":{
         "metric":"cpu_report.User\\g",
         "pointInTime":false,
-        "temporal":true
+        "temporal":true,
+        "amsId":"cpu_user"
       },
       "metrics/cpu/Wait":{
         "metric":"cpu_report.Wait\\g",
         "pointInTime":false,
-        "temporal":true
+        "temporal":true,
+        "amsId":"cpu_wait"
       },
       "metrics/load/1-min":{
         "metric":"load_report.1-min",
         "pointInTime":false,
-        "temporal":true
+        "temporal":true,
+        "amsId":"load_one"
       },
       "metrics/load/CPUs":{
         "metric":"load_report.CPUs ",
         "pointInTime":false,
-        "temporal":true
+        "temporal":true,
+        "amsId":"cpu_num"
       },
       "metrics/load/Nodes":{
         "metric":"load_report.Nodes",
         "pointInTime":false,
-        "temporal":true
+        "temporal":true,
+        "amsId":"cpu_num"
       },
       "metrics/load/Procs":{
         "metric":"load_report.Procs",
         "pointInTime":false,
-        "temporal":true
+        "temporal":true,
+        "amsId":"proc_run"
       },
       "metrics/memory/Buffer":{
         "metric":"mem_report.Buffer\\g",
         "pointInTime":false,
-        "temporal":true
+        "temporal":true,
+        "amsId":"mem_buffered"
       },
       "metrics/memory/Cache":{
         "metric":"mem_report.Cache\\g",
         "pointInTime":false,
-        "temporal":true
+        "temporal":true,
+        "amsId":"mem_cached"
       },
       "metrics/memory/Share":{
         "metric":"mem_report.Share\\g",
         "pointInTime":false,
-        "temporal":true
+        "temporal":true,
+        "amsId":"mem_shared"
       },
       "metrics/memory/Swap":{
         "metric":"mem_report.Swap\\g",
         "pointInTime":false,
-        "temporal":true
+        "temporal":true,
+        "amsId":"swap_free"
       },
       "metrics/memory/Total":{
         "metric":"mem_report.Total\\g",
         "pointInTime":false,
-        "temporal":true
+        "temporal":true,
+        "amsId":"mem_total"
       },
       "metrics/memory/Use":{
         "metric":"mem_report.Use\\g",
         "pointInTime":false,
-        "temporal":true
+        "temporal":true,
+        "amsId":"mem_used"
       },
       "metrics/network/In":{
         "metric":"network_report.In ",
         "pointInTime":false,
-        "temporal":true
+        "temporal":true,
+        "amsId":"bytes_in"
       },
       "metrics/network/Out":{
         "metric":"network_report.Out",
         "pointInTime":false,
-        "temporal":true
+        "temporal":true,
+        "amsId":"bytes_out"
       }
     }
   },

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

@@ -17,27 +17,17 @@
  */
 package org.apache.ambari.server.controller.metrics.timeline;
 
-import com.google.inject.Injector;
 import org.apache.ambari.server.configuration.ComponentSSLConfiguration;
-import org.apache.ambari.server.controller.AmbariManagementController;
-import org.apache.ambari.server.controller.MaintenanceStateHelper;
-import org.apache.ambari.server.controller.internal.AbstractProviderModule;
-import org.apache.ambari.server.controller.internal.ClusterResourceProvider;
-import org.apache.ambari.server.controller.internal.ConfigurationResourceProvider;
-import org.apache.ambari.server.controller.internal.HostComponentResourceProvider;
 import org.apache.ambari.server.controller.internal.PropertyInfo;
 import org.apache.ambari.server.controller.internal.ResourceImpl;
-import org.apache.ambari.server.controller.internal.ServiceResourceProvider;
 import org.apache.ambari.server.controller.internal.TemporalInfoImpl;
 import org.apache.ambari.server.controller.metrics.MetricHostProvider;
 import org.apache.ambari.server.controller.metrics.ganglia.TestStreamProvider;
 import org.apache.ambari.server.controller.spi.Request;
 import org.apache.ambari.server.controller.spi.Resource;
-import org.apache.ambari.server.controller.spi.ResourceProvider;
 import org.apache.ambari.server.controller.spi.SystemException;
 import org.apache.ambari.server.controller.spi.TemporalInfo;
 import org.apache.ambari.server.controller.utilities.PropertyHelper;
-import org.apache.ambari.server.state.Clusters;
 import org.apache.http.client.utils.URIBuilder;
 import org.junit.Assert;
 import org.junit.Test;
@@ -51,8 +41,6 @@ import java.util.Set;
 
 import static org.apache.ambari.server.controller.metrics.MetricsServiceProvider.MetricsService;
 import static org.easymock.EasyMock.createNiceMock;
-import static org.easymock.EasyMock.expect;
-import static org.easymock.EasyMock.replay;
 import static org.mockito.Mockito.mock;
 
 public class AMSPropertyProviderTest {
@@ -97,7 +85,7 @@ public class AMSPropertyProviderTest {
     Resource res = resources.iterator().next();
     Map<String, Object> properties = PropertyHelper.getProperties(resources.iterator().next());
     Assert.assertNotNull(properties);
-    URIBuilder uriBuilder = AMSPropertyProvider.getUriBuilder("localhost", 8188);
+    URIBuilder uriBuilder = AMSPropertyProvider.getAMSUriBuilder("localhost", 8188);
     uriBuilder.addParameter("metricNames", "cpu_user");
     uriBuilder.addParameter("hostname", "h1");
     uriBuilder.addParameter("appId", "HOST");
@@ -137,14 +125,14 @@ public class AMSPropertyProviderTest {
     Resource res = resources.iterator().next();
     Map<String, Object> properties = PropertyHelper.getProperties(resources.iterator().next());
     Assert.assertNotNull(properties);
-    URIBuilder uriBuilder = AMSPropertyProvider.getUriBuilder("localhost", 8188);
+    URIBuilder uriBuilder = AMSPropertyProvider.getAMSUriBuilder("localhost", 8188);
     uriBuilder.addParameter("metricNames", "cpu_user,mem_free");
     uriBuilder.addParameter("hostname", "h1");
     uriBuilder.addParameter("appId", "HOST");
     uriBuilder.addParameter("startTime", "1416445244701");
     uriBuilder.addParameter("endTime", "1416445244901");
 
-    URIBuilder uriBuilder2 = AMSPropertyProvider.getUriBuilder("localhost", 8188);
+    URIBuilder uriBuilder2 = AMSPropertyProvider.getAMSUriBuilder("localhost", 8188);
     uriBuilder2.addParameter("metricNames", "mem_free,cpu_user");
     uriBuilder2.addParameter("hostname", "h1");
     uriBuilder2.addParameter("appId", "HOST");
@@ -196,7 +184,7 @@ public class AMSPropertyProviderTest {
     Resource res = resources.iterator().next();
     Map<String, Object> properties = PropertyHelper.getProperties(resources.iterator().next());
     Assert.assertNotNull(properties);
-    URIBuilder uriBuilder = AMSPropertyProvider.getUriBuilder("localhost", 8188);
+    URIBuilder uriBuilder = AMSPropertyProvider.getAMSUriBuilder("localhost", 8188);
     uriBuilder.addParameter("metricNames", "yarn.QueueMetrics.%.AvailableMB");
     uriBuilder.addParameter("appId", "RESOURCEMANAGER");
     uriBuilder.addParameter("startTime", "1416528819369");
@@ -238,7 +226,7 @@ public class AMSPropertyProviderTest {
     Resource res = resources.iterator().next();
     Map<String, Object> properties = PropertyHelper.getProperties(resources.iterator().next());
     Assert.assertNotNull(properties);
-    URIBuilder uriBuilder = AMSPropertyProvider.getUriBuilder("localhost", 8188);
+    URIBuilder uriBuilder = AMSPropertyProvider.getAMSUriBuilder("localhost", 8188);
     uriBuilder.addParameter("metricNames", "rpc.rpc.RpcQueueTimeAvgTime");
     uriBuilder.addParameter("appId", "NAMENODE");
     uriBuilder.addParameter("startTime", "1416528819369");
@@ -248,7 +236,7 @@ public class AMSPropertyProviderTest {
     Assert.assertEquals(238, val.length);
   }
 
-  public class TestMetricHostProvider implements MetricHostProvider {
+  public static class TestMetricHostProvider implements MetricHostProvider {
 
     @Override
     public String getCollectorHostName(String clusterName, MetricsService service)

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

@@ -0,0 +1,86 @@
+/**
+ * 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.ambari.server.controller.metrics.timeline;
+
+import org.apache.ambari.server.configuration.ComponentSSLConfiguration;
+import org.apache.ambari.server.controller.internal.PropertyInfo;
+import org.apache.ambari.server.controller.internal.ResourceImpl;
+import org.apache.ambari.server.controller.internal.TemporalInfoImpl;
+import org.apache.ambari.server.controller.metrics.ganglia.TestStreamProvider;
+import org.apache.ambari.server.controller.spi.Request;
+import org.apache.ambari.server.controller.spi.Resource;
+import org.apache.ambari.server.controller.spi.TemporalInfo;
+import org.apache.ambari.server.controller.utilities.PropertyHelper;
+import org.apache.http.client.utils.URIBuilder;
+import org.junit.Assert;
+import org.junit.Test;
+import java.io.File;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Set;
+
+import static org.apache.ambari.server.controller.metrics.timeline.AMSPropertyProviderTest.TestMetricHostProvider;
+import static org.mockito.Mockito.mock;
+
+public class AMSReportPropertyProviderTest {
+  private static final String CLUSTER_NAME_PROPERTY_ID = PropertyHelper.getPropertyId("Clusters", "cluster_name");
+  private static final String FILE_PATH_PREFIX = "ams" + File.separator;
+  private static final String SINGLE_HOST_METRICS_FILE_PATH = FILE_PATH_PREFIX + "single_host_metric.json";
+
+  @Test
+  public void testPopulateResources() throws Exception {
+    TestStreamProvider streamProvider = new TestStreamProvider(SINGLE_HOST_METRICS_FILE_PATH);
+    TestMetricHostProvider metricHostProvider = new TestMetricHostProvider();
+    ComponentSSLConfiguration sslConfiguration = mock(ComponentSSLConfiguration.class);
+
+    Map<String, Map<String, PropertyInfo>> propertyIds =
+      PropertyHelper.getMetricPropertyIds(Resource.Type.Cluster);
+
+    AMSReportPropertyProvider propertyProvider =
+      new AMSReportPropertyProvider(
+        propertyIds,
+        streamProvider,
+        sslConfiguration,
+        metricHostProvider,
+        CLUSTER_NAME_PROPERTY_ID
+      );
+
+    String propertyId = PropertyHelper.getPropertyId("metrics/cpu", "User");
+    Resource resource = new ResourceImpl(Resource.Type.Cluster);
+    resource.setProperty(CLUSTER_NAME_PROPERTY_ID, "c1");
+    Map<String, TemporalInfo> temporalInfoMap = new HashMap<String, TemporalInfo>();
+    temporalInfoMap.put(propertyId, new TemporalInfoImpl(1416528819369L, 1416528819569L, 1L));
+    Request request = PropertyHelper.getReadRequest(
+      Collections.singleton(propertyId), temporalInfoMap);
+    Set<Resource> resources =
+      propertyProvider.populateResources(Collections.singleton(resource), request, null);
+    Assert.assertEquals(1, resources.size());
+    Resource res = resources.iterator().next();
+    Map<String, Object> properties = PropertyHelper.getProperties(resources.iterator().next());
+    Assert.assertNotNull(properties);
+    URIBuilder uriBuilder = AMSPropertyProvider.getAMSUriBuilder("localhost", 8188);
+    uriBuilder.addParameter("metricNames", "cpu_user");
+    uriBuilder.addParameter("appId", "HOST");
+    uriBuilder.addParameter("startTime", "1416528819369");
+    uriBuilder.addParameter("endTime", "1416528819569");
+    Assert.assertEquals(uriBuilder.toString(), streamProvider.getLastSpec());
+    Number[][] val = (Number[][]) res.getPropertyValue("metrics/cpu/User");
+    Assert.assertEquals(111, val.length);
+  }
+}