AMBARI-15952 Add container metrics in AMS (jianhe via avijayan)

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

@@ -75,17 +75,15 @@ public abstract class AbstractTimelineMetricsSink {
     LOG = LogFactory.getLog(this.getClass());
   }
 
-  protected void emitMetrics(TimelineMetrics metrics) {
-    String connectUrl = getCollectorUri();
+  protected void emitMetricsJson(String connectUrl, String jsonData) {
     int timeout = getTimeoutSeconds() * 1000;
     HttpURLConnection connection = null;
     try {
       if (connectUrl == null) {
         throw new IOException("Unknown URL. Unable to connect to metrics collector.");
       }
-      String jsonData = mapper.writeValueAsString(metrics);
       connection = connectUrl.startsWith("https") ?
-        getSSLConnection(connectUrl) : getConnection(connectUrl);
+          getSSLConnection(connectUrl) : getConnection(connectUrl);
 
       connection.setRequestMethod("POST");
       connection.setRequestProperty("Content-Type", "application/json");
@@ -104,7 +102,7 @@ public abstract class AbstractTimelineMetricsSink {
 
       if (statusCode != 200) {
         LOG.info("Unable to POST metrics to collector, " + connectUrl + ", " +
-          "statusCode = " + statusCode);
+            "statusCode = " + statusCode);
       } else {
         if (LOG.isDebugEnabled()) {
           LOG.debug("Metrics posted to Collector " + connectUrl);
@@ -113,7 +111,7 @@ public abstract class AbstractTimelineMetricsSink {
       cleanupInputStream(connection.getInputStream());
     } catch (IOException ioe) {
       StringBuilder errorMessage =
-        new StringBuilder("Unable to connect to collector, " + connectUrl + "\n");
+          new StringBuilder("Unable to connect to collector, " + connectUrl + "\n");
       try {
         if ((connection != null)) {
           errorMessage.append(cleanupInputStream(connection.getErrorStream()));
@@ -130,6 +128,19 @@ public abstract class AbstractTimelineMetricsSink {
     }
   }
 
+  protected void emitMetrics(TimelineMetrics metrics) {
+    String connectUrl = getCollectorUri();
+    String jsonData = null;
+    try {
+      jsonData = mapper.writeValueAsString(metrics);
+    } catch (IOException e) {
+      LOG.error("Unable to parse metrics", e);
+    }
+    if (jsonData != null) {
+      emitMetricsJson(connectUrl, jsonData);
+    }
+  }
+
   /**
    * Cleans up and closes an input stream
    * see http://docs.oracle.com/javase/6/docs/technotes/guides/net/http-keepalive.html
@@ -137,7 +148,7 @@ public abstract class AbstractTimelineMetricsSink {
    * @return string read from the InputStream
    * @throws IOException
    */
-  private String cleanupInputStream(InputStream is) throws IOException {
+  protected String cleanupInputStream(InputStream is) throws IOException {
     StringBuilder sb = new StringBuilder();
     if (is != null) {
       try (

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

@@ -0,0 +1,218 @@
+/**
+ * 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;
+
+import org.apache.hadoop.classification.InterfaceAudience;
+import org.apache.hadoop.classification.InterfaceStability;
+
+import javax.xml.bind.annotation.XmlAccessType;
+import javax.xml.bind.annotation.XmlAccessorType;
+import javax.xml.bind.annotation.XmlRootElement;
+
+/**
+ * This differs from TimelineMetric in that this class contains all the fields
+ * for a single metric.
+ */
+@XmlRootElement(name = "containermetric")
+@XmlAccessorType(XmlAccessType.FIELD)
+@InterfaceAudience.Public
+@InterfaceStability.Unstable
+public class ContainerMetric {
+  private String hostName;
+  private String containerId;
+  private int pmemLimit;
+  private int vmemLimit;
+  private int pmemUsedAvg;
+  private int pmemUsedMin;
+  private int pmemUsedMax;
+  private int pmem50Pct;
+  private int pmem75Pct;
+  private int pmem90Pct;
+  private int pmem95Pct;
+  private int pmem99Pct;
+  private long launchDuration;
+  private long localizationDuration;
+  private long startTime;
+  private long finishTime;
+  private int exitCode;
+
+
+  public ContainerMetric() {
+
+  }
+
+  public String getHostName() {
+    return hostName;
+  }
+
+  public void setHostName(String hostName) {
+    this.hostName = hostName;
+  }
+
+  public String getContainerId() {
+    return containerId;
+  }
+
+  public void setContainerId(String containerId) {
+    this.containerId = containerId;
+  }
+
+  public int getPmemLimit() {
+    return pmemLimit;
+  }
+
+  public void setPmemLimit(int pmemLimit) {
+    this.pmemLimit = pmemLimit;
+  }
+
+  public int getVmemLimit() {
+    return vmemLimit;
+  }
+
+  public void setVmemLimit(int vmemLimit) {
+    this.vmemLimit = vmemLimit;
+  }
+
+  public int getPmemUsedAvg() {
+    return pmemUsedAvg;
+  }
+
+  public void setPmemUsedAvg(int pmemUsedAvg) {
+    this.pmemUsedAvg = pmemUsedAvg;
+  }
+
+  public int getPmemUsedMin() {
+    return pmemUsedMin;
+  }
+
+  public void setPmemUsedMin(int pmemUsedMin) {
+    this.pmemUsedMin = pmemUsedMin;
+  }
+
+  public int getPmemUsedMax() {
+    return pmemUsedMax;
+  }
+
+  public void setPmemUsedMax(int pmemUsedMax) {
+    this.pmemUsedMax = pmemUsedMax;
+  }
+
+  public int getPmem50Pct() {
+    return pmem50Pct;
+  }
+
+  public void setPmem50Pct(int pmem50Pct) {
+    this.pmem50Pct = pmem50Pct;
+  }
+
+  public int getPmem75Pct() {
+    return pmem75Pct;
+  }
+
+  public void setPmem75Pct(int pmem75Pct) {
+    this.pmem75Pct = pmem75Pct;
+  }
+
+  public int getPmem90Pct() {
+    return pmem90Pct;
+  }
+
+  public void setPmem90Pct(int pmem90Pct) {
+    this.pmem90Pct = pmem90Pct;
+  }
+
+  public int getPmem95Pct() {
+    return pmem95Pct;
+  }
+
+  public void setPmem95Pct(int pmem95Pct) {
+    this.pmem95Pct = pmem95Pct;
+  }
+
+  public int getPmem99Pct() {
+    return pmem99Pct;
+  }
+
+  public void setPmem99Pct(int pmem99Pct) {
+    this.pmem99Pct = pmem99Pct;
+  }
+
+  public long getLaunchDuration() {
+    return launchDuration;
+  }
+
+  public void setLaunchDuration(long launchDuration) {
+    this.launchDuration = launchDuration;
+  }
+
+  public long getLocalizationDuration() {
+    return localizationDuration;
+  }
+
+  public void setLocalizationDuration(long localizationDuration) {
+    this.localizationDuration = localizationDuration;
+  }
+
+  public long getStartTime() {
+    return startTime;
+  }
+
+  public void setStartTime(long startTime) {
+    this.startTime = startTime;
+  }
+
+  public long getFinishTime() {
+    return finishTime;
+  }
+
+  public void setFinishTime(long finishTime) {
+    this.finishTime = finishTime;
+  }
+
+  public int getExitCode() {
+    return exitCode;
+  }
+
+  public void setExitCode(int exitCode) {
+    this.exitCode = exitCode;
+  }
+
+  @Override
+  public String toString() {
+    return "ContainerMetric{" +
+        "hostName='" + hostName + '\'' +
+        ", containerId='" + containerId + '\'' +
+        ", pmemLimit=" + pmemLimit +
+        ", vmemLimit=" + vmemLimit +
+        ", pmemUsedAvg=" + pmemUsedAvg +
+        ", pmemUsedMin=" + pmemUsedMin +
+        ", pmemUsedMax=" + pmemUsedMax +
+        ", pmem50Pct=" + pmem50Pct +
+        ", pmem75Pct=" + pmem75Pct +
+        ", pmem90Pct=" + pmem90Pct +
+        ", pmem95Pct=" + pmem95Pct +
+        ", pmem99Pct=" + pmem99Pct +
+        ", launchDuration=" + launchDuration +
+        ", localizationDuration=" + localizationDuration +
+        ", startTime=" + startTime +
+        ", finishTime=" + finishTime +
+        ", exitCode=" + exitCode +
+        '}';
+  }
+}

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

@@ -31,6 +31,9 @@ import org.apache.hadoop.metrics2.sink.timeline.cache.TimelineMetricsCache;
 import org.apache.hadoop.metrics2.util.Servers;
 import org.apache.hadoop.net.DNS;
 
+import java.io.IOException;
+import java.io.OutputStream;
+import java.net.HttpURLConnection;
 import java.net.SocketAddress;
 import java.net.UnknownHostException;
 import java.util.ArrayList;
@@ -51,6 +54,9 @@ public class HadoopTimelineMetricsSink extends AbstractTimelineMetricsSink imple
   private String serviceName = "";
   private List<? extends SocketAddress> metricsServers;
   private String collectorUri;
+  private String containerMetricsUri;
+  public static final String WS_V1_CONTAINER_METRICS = "/ws/v1/timeline/containermetrics";
+
   private static final String SERVICE_NAME_PREFIX = "serviceName-prefix";
   private static final String SERVICE_NAME = "serviceName";
   private int timeoutSeconds = 10;
@@ -88,6 +94,7 @@ public class HadoopTimelineMetricsSink extends AbstractTimelineMetricsSink imple
       LOG.error("No Metric collector configured.");
     } else {
       collectorUri = conf.getString(COLLECTOR_PROPERTY).trim() + WS_V1_TIMELINE_METRICS;
+      containerMetricsUri = conf.getString(COLLECTOR_PROPERTY).trim() + WS_V1_CONTAINER_METRICS;
       if (collectorUri.toLowerCase().startsWith("https://")) {
         String trustStorePath = conf.getString(SSL_KEYSTORE_PATH_PROPERTY).trim();
         String trustStoreType = conf.getString(SSL_KEYSTORE_TYPE_PROPERTY).trim();
@@ -220,6 +227,11 @@ public class HadoopTimelineMetricsSink extends AbstractTimelineMetricsSink imple
         }
       }
 
+      if (record.context().equals("container")) {
+        emitContainerMetrics(record);
+        return;
+      }
+
       int sbBaseLen = sb.length();
 
       Collection<AbstractMetric> metrics = (Collection<AbstractMetric>) record.metrics();
@@ -263,6 +275,89 @@ public class HadoopTimelineMetricsSink extends AbstractTimelineMetricsSink imple
     }
   }
 
+  private void parseContainerMetrics(MetricsRecord record,
+      ContainerMetric containerMetric) {
+    for (AbstractMetric metric : record.metrics() ) {
+      switch (metric.name()) {
+      case "PMemUsageMBsAvgMBs":
+        containerMetric.setPmemUsedAvg(metric.value().intValue());
+        break;
+      case "PMemUsageMBsMinMBs":
+        containerMetric.setPmemUsedMin(metric.value().intValue());
+        break;
+      case "PMemUsageMBsMaxMBs":
+        containerMetric.setPmemUsedMax(metric.value().intValue());
+        break;
+      case "PMemUsageMBHistogram50thPercentileMBs":
+        containerMetric.setPmem50Pct(metric.value().intValue());
+        break;
+      case "PMemUsageMBHistogram75thPercentileMBs":
+        containerMetric.setPmem75Pct(metric.value().intValue());
+        break;
+      case "PMemUsageMBHistogram90thPercentileMBs":
+        containerMetric.setPmem90Pct(metric.value().intValue());
+        break;
+      case "PMemUsageMBHistogram95thPercentileMBs":
+        containerMetric.setPmem95Pct(metric.value().intValue());
+        break;
+      case "PMemUsageMBHistogram99thPercentileMBs":
+        containerMetric.setPmem99Pct(metric.value().intValue());
+        break;
+      case "pMemLimitMBs":
+        containerMetric.setPmemLimit(metric.value().intValue());
+        break;
+      case "vMemLimitMBs":
+        containerMetric.setVmemLimit(metric.value().intValue());
+        break;
+      case "launchDurationMs":
+        containerMetric.setLaunchDuration(metric.value().longValue());
+        break;
+      case "localizationDurationMs":
+        containerMetric.setLocalizationDuration(metric.value().longValue());
+        break;
+      case "StartTime":
+        containerMetric.setStartTime(metric.value().longValue());
+        break;
+      case "FinishTime":
+        containerMetric.setFinishTime(metric.value().longValue());
+        break;
+      case "ExitCode":
+        containerMetric.setExitCode((metric.value().intValue()));
+        break;
+      default:
+        break;
+      }
+    }
+    if (LOG.isDebugEnabled()) {
+      LOG.debug(containerMetric);
+    }
+  }
+
+  private void emitContainerMetrics(MetricsRecord record) {
+
+    ContainerMetric containerMetric = new ContainerMetric();
+    containerMetric.setHostName(hostName);
+
+    for (MetricsTag tag : record.tags()) {
+      if (tag.name().equals("ContainerResource")) {
+        containerMetric.setContainerId(tag.value());
+      }
+    }
+
+    parseContainerMetrics(record, containerMetric);
+    List<ContainerMetric> list = new ArrayList<>();
+    list.add(containerMetric);
+    String jsonData = null;
+    try {
+      jsonData = mapper.writeValueAsString(list);
+    } catch (IOException e) {
+      LOG.error("Unable to parse container metrics ", e);
+    }
+    if (jsonData != null) {
+      emitMetricsJson(containerMetricsUri, jsonData);
+    }
+  }
+
   // Taken as is from Ganglia30 implementation
   @InterfaceAudience.Private
   public void appendPrefix(MetricsRecord record, StringBuilder sb) {

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

@@ -21,6 +21,7 @@ package org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.metrics2.sink.timeline.ContainerMetric;
 import org.apache.hadoop.metrics2.sink.timeline.Precision;
 import org.apache.hadoop.metrics2.sink.timeline.TimelineMetric;
 import org.apache.hadoop.metrics2.sink.timeline.TimelineMetricMetadata;
@@ -335,6 +336,13 @@ public class HBaseTimelineMetricStore extends AbstractService implements Timelin
     return response;
   }
 
+  @Override
+  public TimelinePutResponse putContainerMetrics(List<ContainerMetric> metrics)
+      throws SQLException, IOException {
+    hBaseAccessor.insertContainerMetrics(metrics);
+    return new TimelinePutResponse();
+  }
+
   @Override
   public Map<String, List<TimelineMetricMetadata>> getTimelineMetricMetadata() throws SQLException, IOException {
     Map<TimelineMetricMetadataKey, TimelineMetricMetadata> metadata =

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

@@ -28,11 +28,13 @@ import org.apache.hadoop.hbase.client.Durability;
 import org.apache.hadoop.hbase.client.HBaseAdmin;
 import org.apache.hadoop.hbase.util.RetryCounter;
 import org.apache.hadoop.hbase.util.RetryCounterFactory;
+import org.apache.hadoop.metrics2.sink.timeline.ContainerMetric;
 import org.apache.hadoop.metrics2.sink.timeline.Precision;
 import org.apache.hadoop.metrics2.sink.timeline.SingleValuedTimelineMetric;
 import org.apache.hadoop.metrics2.sink.timeline.TimelineMetric;
 import org.apache.hadoop.metrics2.sink.timeline.TimelineMetricMetadata;
 import org.apache.hadoop.metrics2.sink.timeline.TimelineMetrics;
+import org.apache.hadoop.yarn.api.records.ContainerId;
 import org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline.aggregators.AggregatorUtils;
 import org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline.aggregators.Function;
 import org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline.aggregators.MetricClusterAggregate;
@@ -57,6 +59,7 @@ import java.sql.PreparedStatement;
 import java.sql.ResultSet;
 import java.sql.SQLException;
 import java.sql.Statement;
+import java.sql.Timestamp;
 import java.util.Arrays;
 import java.util.Collection;
 import java.util.HashMap;
@@ -86,6 +89,9 @@ import static org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.ti
 import static org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline.TimelineMetricConfiguration.OUT_OFF_BAND_DATA_TIME_ALLOWANCE;
 import static org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline.TimelineMetricConfiguration.PRECISION_TABLE_SPLIT_POINTS;
 import static org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline.TimelineMetricConfiguration.PRECISION_TABLE_TTL;
+import static org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline.TimelineMetricConfiguration.CONTAINER_METRICS_TTL;
+import static org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline.query.PhoenixTransactSQL.CONTAINER_METRICS_TABLE_NAME;
+import static org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline.query.PhoenixTransactSQL.CREATE_CONTAINER_METRICS_TABLE_SQL;
 import static org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline.query.PhoenixTransactSQL.CREATE_HOSTED_APPS_METADATA_TABLE_SQL;
 import static org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline.query.PhoenixTransactSQL.CREATE_METRICS_AGGREGATE_TABLE_SQL;
 import static org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline.query.PhoenixTransactSQL.CREATE_METRICS_CLUSTER_AGGREGATE_GROUPED_TABLE_SQL;
@@ -111,6 +117,8 @@ import static org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.ti
 import static org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline.query.PhoenixTransactSQL.UPSERT_HOSTED_APPS_METADATA_SQL;
 import static org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline.query.PhoenixTransactSQL.UPSERT_METADATA_SQL;
 import static org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline.query.PhoenixTransactSQL.UPSERT_METRICS_SQL;
+import static org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline.query.PhoenixTransactSQL.UPSERT_CONTAINER_METRICS_SQL;
+
 
 /**
  * Provides a facade over the Phoenix API to access HBase schema
@@ -180,6 +188,7 @@ public class PhoenixHBaseAccessor {
     this.timelineMetricsTablesDurability = metricsConf.get(TIMELINE_METRICS_TABLES_DURABILITY, "");
 
     tableTTL.put(METRICS_RECORD_TABLE_NAME, metricsConf.get(PRECISION_TABLE_TTL, String.valueOf(1 * 86400)));  // 1 day
+    tableTTL.put(CONTAINER_METRICS_TABLE_NAME, metricsConf.get(CONTAINER_METRICS_TTL, String.valueOf(30 * 86400)));  // 30 days
     tableTTL.put(METRICS_AGGREGATE_MINUTE_TABLE_NAME, metricsConf.get(HOST_MINUTE_TABLE_TTL, String.valueOf(7 * 86400))); //7 days
     tableTTL.put(METRICS_AGGREGATE_HOURLY_TABLE_NAME, metricsConf.get(HOST_HOUR_TABLE_TTL, String.valueOf(30 * 86400))); //30 days
     tableTTL.put(METRICS_AGGREGATE_DAILY_TABLE_NAME, metricsConf.get(HOST_DAILY_TABLE_TTL, String.valueOf(365 * 86400))); //1 year
@@ -271,6 +280,10 @@ public class PhoenixHBaseAccessor {
         encoding, compression);
       stmt.executeUpdate(hostedAppSql);
 
+      // Container Metrics
+      stmt.executeUpdate( String.format(CREATE_CONTAINER_METRICS_TABLE_SQL,
+        encoding, tableTTL.get(CONTAINER_METRICS_TABLE_NAME), compression));
+
       // Host level
       String precisionSql = String.format(CREATE_METRICS_TABLE_SQL,
         encoding, tableTTL.get(METRICS_RECORD_TABLE_NAME), compression);
@@ -462,6 +475,71 @@ public class PhoenixHBaseAccessor {
     return "";
   }
 
+  public void insertContainerMetrics(List<ContainerMetric> metrics)
+      throws SQLException, IOException {
+    Connection conn = getConnection();
+    PreparedStatement metricRecordStmt = null;
+
+    try {
+      metricRecordStmt = conn.prepareStatement(
+          String.format(UPSERT_CONTAINER_METRICS_SQL, CONTAINER_METRICS_TABLE_NAME));
+      for (ContainerMetric metric : metrics) {
+        metricRecordStmt.clearParameters();
+        metricRecordStmt.setString(1, ContainerId.fromString(metric.getContainerId())
+            .getApplicationAttemptId().getApplicationId().toString());
+        metricRecordStmt.setString(2, metric.getContainerId());
+        metricRecordStmt.setTimestamp(3, new Timestamp(metric.getStartTime()));
+        metricRecordStmt.setTimestamp(4, new Timestamp(metric.getFinishTime()));
+        metricRecordStmt.setLong(5, metric.getFinishTime() - metric.getStartTime());
+        metricRecordStmt.setString(6, metric.getHostName());
+        metricRecordStmt.setInt(7, metric.getExitCode());
+        metricRecordStmt.setLong(8, metric.getLocalizationDuration());
+        metricRecordStmt.setLong(9, metric.getLaunchDuration());
+        metricRecordStmt.setDouble(10, (double) metric.getPmemLimit() / 1024);
+        metricRecordStmt.setDouble(11,
+            ((double) metric.getPmemLimit() / 1024) * (metric.getFinishTime()
+                - metric.getStartTime()));
+        metricRecordStmt.setDouble(12, (double) metric.getVmemLimit() / 1024);
+        metricRecordStmt.setDouble(13, (double) metric.getPmemUsedMin() / 1024);
+        metricRecordStmt.setDouble(14, (double) metric.getPmemUsedMax() / 1024);
+        metricRecordStmt.setDouble(15, (double) metric.getPmemUsedAvg() / 1024);
+        metricRecordStmt.setDouble(16, (double) metric.getPmem50Pct() / 1024);
+        metricRecordStmt.setDouble(17, (double) metric.getPmem75Pct() / 1024);
+        metricRecordStmt.setDouble(18, (double) metric.getPmem90Pct() / 1024);
+        metricRecordStmt.setDouble(19, (double) metric.getPmem95Pct()/ 1024);
+        metricRecordStmt.setDouble(20, (double) metric.getPmem99Pct() / 1024);
+        metricRecordStmt.setDouble(21, (double) metric.getPmemLimit() / 1024
+            - (double) metric.getPmemUsedMax() / 1024);
+        metricRecordStmt.setDouble(22, ((double) metric.getPmemLimit() / 1024
+            - (double) metric.getPmemUsedMax() / 1024) * (metric.getFinishTime()
+            - metric.getStartTime()));
+
+        try {
+          metricRecordStmt.executeUpdate();
+        } catch (SQLException sql) {
+          LOG.error("Failed on insert records to store.", sql);
+        }
+      }
+
+      conn.commit();
+    } finally {
+      if (metricRecordStmt != null) {
+        try {
+          metricRecordStmt.close();
+        } catch (SQLException e) {
+          // Ignore
+        }
+      }
+      if (conn != null) {
+        try {
+          conn.close();
+        } catch (SQLException sql) {
+          // Ignore
+        }
+      }
+    }
+  }
+
   public void insertMetricRecordsWithMetadata(TimelineMetricMetadataManager metadataManager,
                                               TimelineMetrics metrics) throws SQLException, IOException {
     List<TimelineMetric> timelineMetrics = metrics.getMetrics();
@@ -860,7 +938,7 @@ public class PhoenixHBaseAccessor {
         }
         if (stmt != null) {
           stmt.close();
-        }        
+        }
       }
     }
   }

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

@@ -54,6 +54,9 @@ public class TimelineMetricConfiguration {
   public static final String HBASE_COMPRESSION_SCHEME =
     "timeline.metrics.hbase.compression.scheme";
 
+  public static final String CONTAINER_METRICS_TTL =
+    "timeline.container-metrics.ttl";
+
   public static final String PRECISION_TABLE_TTL =
     "timeline.metrics.host.aggregator.ttl";
 

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

@@ -17,6 +17,7 @@
  */
 package org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline;
 
+import org.apache.hadoop.metrics2.sink.timeline.ContainerMetric;
 import org.apache.hadoop.metrics2.sink.timeline.Precision;
 import org.apache.hadoop.metrics2.sink.timeline.TimelineMetric;
 import org.apache.hadoop.metrics2.sink.timeline.TimelineMetricMetadata;
@@ -72,6 +73,12 @@ public interface TimelineMetricStore {
    */
   TimelinePutResponse putMetrics(TimelineMetrics metrics) throws SQLException, IOException;
 
+  /**
+   * Store container metric into the timeliens tore
+   */
+  TimelinePutResponse putContainerMetrics(List<ContainerMetric> metrics)
+      throws SQLException, IOException;
+
   /**
    * Return all metrics metadata that have been written to the store.
    * @return { appId : [ @TimelineMetricMetadata ] }

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

@@ -59,6 +59,33 @@ public class PhoenixTransactSQL {
     "INSTANCE_ID)) DATA_BLOCK_ENCODING='%s', IMMUTABLE_ROWS=true, " +
     "TTL=%s, COMPRESSION='%s'";
 
+  public static final String CREATE_CONTAINER_METRICS_TABLE_SQL =
+      "CREATE TABLE IF NOT EXISTS CONTAINER_METRICS "
+      + "(APP_ID VARCHAR, "
+      + " CONTAINER_ID VARCHAR,"
+      + " START_TIME TIMESTAMP,"
+      + " FINISH_TIME TIMESTAMP, "
+      + " DURATION BIGINT,"
+      + " HOSTNAME VARCHAR,"
+      + " EXIT_CODE INTEGER,"
+      + " LOCALIZATION_DURATION BIGINT,"
+      + " LAUNCH_DURATION BIGINT,"
+      + " MEM_REQUESTED_GB DOUBLE,"
+      + " MEM_REQUESTED_GB_MILLIS DOUBLE,"
+      + " MEM_VIRTUAL_GB DOUBLE,"
+      + " MEM_USED_GB_MIN DOUBLE,"
+      + " MEM_USED_GB_MAX DOUBLE,"
+      + " MEM_USED_GB_AVG DOUBLE,"
+      + " MEM_USED_GB_50_PCT DOUBLE,"
+      + " MEM_USED_GB_75_PCT DOUBLE,"
+      + " MEM_USED_GB_90_PCT DOUBLE,"
+      + " MEM_USED_GB_95_PCT DOUBLE,"
+      + " MEM_USED_GB_99_PCT DOUBLE,"
+      + " MEM_UNUSED_GB DOUBLE,"
+      + " MEM_UNUSED_GB_MILLIS DOUBLE "
+      + " CONSTRAINT pk PRIMARY KEY(APP_ID, CONTAINER_ID)) DATA_BLOCK_ENCODING='%s',"
+      + " IMMUTABLE_ROWS=true, TTL=%s, COMPRESSION='%s'";
+
   public static final String CREATE_METRICS_AGGREGATE_TABLE_SQL =
     "CREATE TABLE IF NOT EXISTS %s " +
       "(METRIC_NAME VARCHAR, " +
@@ -141,6 +168,31 @@ public class PhoenixTransactSQL {
     "METRICS) VALUES " +
     "(?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)";
 
+  public static final String UPSERT_CONTAINER_METRICS_SQL = "UPSERT INTO %s " +
+      "(APP_ID,"
+      + " CONTAINER_ID,"
+      + " START_TIME,"
+      + " FINISH_TIME,"
+      + " DURATION,"
+      + " HOSTNAME,"
+      + " EXIT_CODE,"
+      + " LOCALIZATION_DURATION,"
+      + " LAUNCH_DURATION,"
+      + " MEM_REQUESTED_GB,"
+      + " MEM_REQUESTED_GB_MILLIS,"
+      + " MEM_VIRTUAL_GB,"
+      + " MEM_USED_GB_MIN,"
+      + " MEM_USED_GB_MAX,"
+      + " MEM_USED_GB_AVG,"
+      + " MEM_USED_GB_50_PCT,"
+      + " MEM_USED_GB_75_PCT,"
+      + " MEM_USED_GB_90_PCT,"
+      + " MEM_USED_GB_95_PCT,"
+      + " MEM_USED_GB_99_PCT,"
+      + " MEM_UNUSED_GB,"
+      + " MEM_UNUSED_GB_MILLIS) VALUES " +
+      "(?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)";
+
   public static final String UPSERT_CLUSTER_AGGREGATE_SQL = "UPSERT INTO " +
     "%s (METRIC_NAME, APP_ID, INSTANCE_ID, SERVER_TIME, " +
     "UNITS, " +
@@ -277,6 +329,9 @@ public class PhoenixTransactSQL {
     "SERVER_TIME <= %s GROUP BY METRIC_NAME, APP_ID, INSTANCE_ID, UNITS";
 
   public static final String METRICS_RECORD_TABLE_NAME = "METRIC_RECORD";
+
+  public static final String CONTAINER_METRICS_TABLE_NAME = "CONTAINER_METRICS";
+
   public static final String METRICS_AGGREGATE_MINUTE_TABLE_NAME =
     "METRIC_RECORD_MINUTE";
   public static final String METRICS_AGGREGATE_HOURLY_TABLE_NAME =

ambari-metrics/ambari-metrics-timelineservice/src/main/java/org/apache/hadoop/yarn/server/applicationhistoryservice/webapp/TimelineWebServices.java

@@ -24,6 +24,8 @@ import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.apache.hadoop.classification.InterfaceAudience.Public;
 import org.apache.hadoop.classification.InterfaceStability.Unstable;
+import org.apache.hadoop.metrics2.annotation.Metric;
+import org.apache.hadoop.metrics2.sink.timeline.ContainerMetric;
 import org.apache.hadoop.metrics2.sink.timeline.PrecisionLimitExceededException;
 import org.apache.hadoop.metrics2.sink.timeline.TimelineMetricMetadata;
 import org.apache.hadoop.yarn.api.records.timeline.TimelineEntities;
@@ -282,6 +284,32 @@ public class TimelineWebServices {
     }
   }
 
+  @Path("/containermetrics")
+  @POST
+  @Consumes({ MediaType.APPLICATION_JSON /* , MediaType.APPLICATION_XML */})
+  public TimelinePutResponse postContainerMetrics(
+      @Context HttpServletRequest req,
+      @Context HttpServletResponse res,
+      List<ContainerMetric> metrics) {
+    init(res);
+    if (metrics == null || metrics.isEmpty()) {
+      return new TimelinePutResponse();
+    }
+
+    try {
+      if (LOG.isDebugEnabled()) {
+        LOG.debug("Storing container metrics: " + TimelineUtils
+            .dumpTimelineRecordtoJSON(metrics, true));
+      }
+
+      return timelineMetricStore.putContainerMetrics(metrics);
+
+    } catch (Exception e) {
+      LOG.error("Error saving metrics.", e);
+      throw new WebApplicationException(e, Response.Status.INTERNAL_SERVER_ERROR);
+    }
+  }
+
   /**
    * Query for a particular metric satisfying the filter criteria.
    * @return {@link TimelineMetric}

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

@@ -23,6 +23,7 @@ import org.apache.hadoop.hbase.HColumnDescriptor;
 import org.apache.hadoop.hbase.HTableDescriptor;
 import org.apache.hadoop.hbase.client.Durability;
 import org.apache.hadoop.hbase.client.HBaseAdmin;
+import org.apache.hadoop.metrics2.sink.timeline.ContainerMetric;
 import org.apache.hadoop.metrics2.sink.timeline.Precision;
 import org.apache.hadoop.metrics2.sink.timeline.TimelineMetric;
 import org.apache.hadoop.metrics2.sink.timeline.TimelineMetrics;
@@ -38,8 +39,15 @@ import org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline.
 import org.junit.Test;
 
 import java.io.IOException;
+import java.lang.reflect.Array;
 import java.lang.reflect.Field;
 import java.sql.SQLException;
+import java.sql.Connection;
+import java.sql.PreparedStatement;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.sql.Statement;
+import java.util.Arrays;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.Iterator;
@@ -375,4 +383,50 @@ public class ITPhoenixHBaseAccessor extends AbstractMiniHBaseClusterTest {
   private Map<String, List<Function>> singletonValueFunctionMap(String metricName) {
     return Collections.singletonMap(metricName, Collections.singletonList(new Function()));
   }
+
+  @Test
+  public void testInsertContainerMetrics() throws Exception {
+    ContainerMetric metric = new ContainerMetric();
+    metric.setContainerId("container_1450744875949_0001_01_000001");
+    metric.setHostName("host1");
+    metric.setPmemLimit(2048);
+    metric.setVmemLimit(2048);
+    metric.setPmemUsedAvg(1024);
+    metric.setPmemUsedMin(1024);
+    metric.setPmemUsedMax(1024);
+    metric.setLaunchDuration(2000);
+    metric.setLocalizationDuration(3000);
+    long startTime = System.currentTimeMillis();
+    long finishTime = startTime + 5000;
+    metric.setStartTime(startTime);
+    metric.setFinishTime(finishTime);
+    metric.setExitCode(0);
+    List<ContainerMetric> list = Arrays.asList(metric);
+    hdb.insertContainerMetrics(list);
+    PreparedStatement stmt = conn.prepareStatement("SELECT * FROM CONTAINER_METRICS");
+    ResultSet set = stmt.executeQuery();
+    // check each filed is set properly when read back.
+    boolean foundRecord = false;
+    while (set.next()) {
+      assertEquals("application_1450744875949_0001", set.getString("APP_ID"));
+      assertEquals("container_1450744875949_0001_01_000001", set.getString("CONTAINER_ID"));
+      assertEquals(new java.sql.Timestamp(startTime), set.getTimestamp("START_TIME"));
+      assertEquals(new java.sql.Timestamp(finishTime), set.getTimestamp("FINISH_TIME"));
+      assertEquals(5000, set.getLong("DURATION"));
+      assertEquals("host1", set.getString("HOSTNAME"));
+      assertEquals(0, set.getInt("EXIT_CODE"));
+      assertEquals(3000, set.getLong("LOCALIZATION_DURATION"));
+      assertEquals(2000, set.getLong("LAUNCH_DURATION"));
+      assertEquals((double)2, set.getDouble("MEM_REQUESTED_GB"));
+      assertEquals((double)2 * 5000, set.getDouble("MEM_REQUESTED_GB_MILLIS"));
+      assertEquals((double)2, set.getDouble("MEM_VIRTUAL_GB"));
+      assertEquals((double)1, set.getDouble("MEM_USED_GB_MIN"));
+      assertEquals((double)1, set.getDouble("MEM_USED_GB_MAX"));
+      assertEquals((double)1, set.getDouble("MEM_USED_GB_AVG"));
+      assertEquals((double)(2 - 1), set.getDouble("MEM_UNUSED_GB"));
+      assertEquals((double)(2-1) * 5000, set.getDouble("MEM_UNUSED_GB_MILLIS"));
+      foundRecord = true;
+    }
+    assertTrue(foundRecord);
+  }
 }

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

@@ -17,6 +17,7 @@
  */
 package org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline;
 
+import org.apache.hadoop.metrics2.sink.timeline.ContainerMetric;
 import org.apache.hadoop.metrics2.sink.timeline.Precision;
 import org.apache.hadoop.metrics2.sink.timeline.TimelineMetric;
 import org.apache.hadoop.metrics2.sink.timeline.TimelineMetricMetadata;
@@ -85,6 +86,12 @@ public class TestTimelineMetricStore implements TimelineMetricStore {
     return new TimelinePutResponse();
   }
 
+  @Override
+  public TimelinePutResponse putContainerMetrics(List<ContainerMetric> metrics)
+      throws SQLException, IOException {
+    return new TimelinePutResponse();
+  }
+
   @Override
   public Map<String, List<TimelineMetricMetadata>> getTimelineMetricMetadata() throws SQLException, IOException {
     return null;