AMBARI-15476 : [AMS / Grafana] Rate calculation with sum() is wrong (avijayan)

11 files changed, 361 insertions, 16 deletions

diff --git a/ambari-metrics/ambari-metrics-common/src/main/java/org/apache/hadoop/metrics2/sink/timeline/PostProcessingUtil.java b/ambari-metrics/ambari-metrics-common/src/main/java/org/apache/hadoop/metrics2/sink/timeline/PostProcessingUtil.java
new file mode 100644
index 0000000000..ab890ecc02
--- /dev/null
+++ b/ambari-metrics/ambari-metrics-common/src/main/java/org/apache/hadoop/metrics2/sink/timeline/PostProcessingUtil.java
@@ -0,0 +1,81 @@
+/**
+ * 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
+ * <p/>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p/>
+ * 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 java.util.Map;
+import java.util.TreeMap;
+
+public class PostProcessingUtil {
+
+  /*
+    Helper function to interpolate missing data on a series.
+  */
+  public static Map<Long, Double> interpolateMissingData(Map<Long, Double> metricValues, long expectedInterval) {
+
+    if (metricValues == null)
+      return null;
+
+    Long prevTime = null;
+    Double prevVal = null;
+    Map<Long, Double> interpolatedMetricValues = new TreeMap<Long, Double>();
+
+    for (Map.Entry<Long, Double> timeValueEntry : metricValues.entrySet()) {
+      Long currTime = timeValueEntry.getKey();
+      Double currVal = timeValueEntry.getValue();
+
+      if (prevTime != null) {
+        Long stepTime = prevTime;
+        while ((currTime - stepTime) > expectedInterval) {
+          stepTime+=expectedInterval;
+          double interpolatedValue = interpolate(stepTime,
+            prevTime, prevVal,
+            currTime, currVal);
+          interpolatedMetricValues.put(stepTime, interpolatedValue);
+        }
+      }
+
+      interpolatedMetricValues.put(currTime, currVal);
+      prevTime = currTime;
+      prevVal = currVal;
+    }
+    return interpolatedMetricValues;
+  }
+
+  public static Double interpolate(Long t, Long t1, Double m1,
+                                   Long t2, Double m2) {
+
+
+    //Linear Interpolation : y = y0 + (y1 - y0) * ((x - x0) / (x1 - x0))
+    if (m1 == null && m2 == null) {
+      return null;
+    }
+
+    if (m1 == null)
+      return m2;
+
+    if (m2 == null)
+      return m1;
+
+    if (t1 == null || t2 == null)
+      return null;
+
+    double slope = (m2 - m1) / (t2 - t1);
+    return m1 +  slope * (t - t1);
+  }
+
+}
diff --git a/ambari-metrics/ambari-metrics-common/src/test/java/org/apache/hadoop/metrics2/sink/timeline/cache/PostProcessingUtilTest.java b/ambari-metrics/ambari-metrics-common/src/test/java/org/apache/hadoop/metrics2/sink/timeline/cache/PostProcessingUtilTest.java
new file mode 100644
index 0000000000..1ec71d03b0
--- /dev/null
+++ b/ambari-metrics/ambari-metrics-common/src/test/java/org/apache/hadoop/metrics2/sink/timeline/cache/PostProcessingUtilTest.java
@@ -0,0 +1,85 @@
+/**
+ * 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.cache;
+
+import junit.framework.Assert;
+import org.apache.hadoop.metrics2.sink.timeline.PostProcessingUtil;
+import org.apache.hadoop.metrics2.sink.timeline.SingleValuedTimelineMetric;
+import org.junit.Test;
+
+import java.util.Iterator;
+import java.util.Map;
+import java.util.TreeMap;
+
+public class PostProcessingUtilTest {
+
+  @Test
+  public void testInterpolateMissinData() throws Exception {
+
+    Map<Long, Double> metricValues = new TreeMap<Long, Double>();
+    long interval = 60*1000;
+
+    long currentTime = System.currentTimeMillis();
+
+    for(int i = 10; i>=1;i--) {
+      if (i%4 != 0 && i != 5) { //Skip time points 4,5,8
+        metricValues.put(currentTime - i*interval, (double)i);
+      }
+    }
+    metricValues = PostProcessingUtil.interpolateMissingData(metricValues, interval);
+    Assert.assertTrue(metricValues.size() == 10);
+
+    Iterator it = metricValues.entrySet().iterator();
+    double sum = 0;
+    while (it.hasNext()) {
+      Map.Entry entry = (Map.Entry)it.next();
+      sum+= (double)entry.getValue();
+    }
+    Assert.assertEquals(sum, 55.0);
+  }
+
+  @Test
+  public void testInterpolate() throws Exception {
+
+    long t2 = System.currentTimeMillis();
+    long t1 = t2 - 60000;
+    double interpolatedValue;
+
+    //Test Equal Values
+    interpolatedValue = PostProcessingUtil.interpolate((t1 + 30000), t1, 10.0, t2, 10.0);
+    Assert.assertEquals(interpolatedValue, 10.0);
+
+    //Test Linear increase Values
+    interpolatedValue = PostProcessingUtil.interpolate((t1 + 30000), t1, 10.0, t2, 20.0);
+    Assert.assertEquals(interpolatedValue, 15.0);
+
+    //Test Linear decrease Values
+    interpolatedValue = PostProcessingUtil.interpolate((t1 + 30000), t1, 20.0, t2, 10.0);
+    Assert.assertEquals(interpolatedValue, 15.0);
+
+    //Test interpolation with non mid point time
+    interpolatedValue = PostProcessingUtil.interpolate((t1 + 20000), t1, 15.0, t2, 30.0); // 1:2 ratio
+    Assert.assertEquals(interpolatedValue, 20.0);
+
+    //Test interpolation with past time
+    interpolatedValue = PostProcessingUtil.interpolate((t1 - 60000), t1, 20.0, t2, 30.0);
+    Assert.assertEquals(interpolatedValue, 10.0);
+
+  }
+
+  }
diff --git a/ambari-metrics/ambari-metrics-timelineservice/src/main/java/org/apache/hadoop/yarn/server/applicationhistoryservice/metrics/timeline/TimelineMetricConfiguration.java b/ambari-metrics/ambari-metrics-timelineservice/src/main/java/org/apache/hadoop/yarn/server/applicationhistoryservice/metrics/timeline/TimelineMetricConfiguration.java
index 3431ac82c1..e57f02d167 100644
--- a/ambari-metrics/ambari-metrics-timelineservice/src/main/java/org/apache/hadoop/yarn/server/applicationhistoryservice/metrics/timeline/TimelineMetricConfiguration.java
+++ b/ambari-metrics/ambari-metrics-timelineservice/src/main/java/org/apache/hadoop/yarn/server/applicationhistoryservice/metrics/timeline/TimelineMetricConfiguration.java
@@ -216,6 +216,9 @@ public class TimelineMetricConfiguration {
   public static final String METRICS_METADATA_SYNC_SCHEDULE_DELAY =
     "timeline.metrics.service.metadata.sync.delay";
 
+  public static final String TIMELINE_METRICS_CLUSTER_AGGREGATOR_INTERPOLATION_ENABLED =
+    "timeline.metrics.cluster.aggregator.interpolation.enabled";
+
   public static final String HOST_APP_ID = "HOST";
 
   private Configuration hbaseConf;
diff --git a/ambari-metrics/ambari-metrics-timelineservice/src/main/java/org/apache/hadoop/yarn/server/applicationhistoryservice/metrics/timeline/aggregators/TimelineMetricClusterAggregatorSecond.java b/ambari-metrics/ambari-metrics-timelineservice/src/main/java/org/apache/hadoop/yarn/server/applicationhistoryservice/metrics/timeline/aggregators/TimelineMetricClusterAggregatorSecond.java
index de90685ce0..e0e065b5b8 100644
--- a/ambari-metrics/ambari-metrics-timelineservice/src/main/java/org/apache/hadoop/yarn/server/applicationhistoryservice/metrics/timeline/aggregators/TimelineMetricClusterAggregatorSecond.java
+++ b/ambari-metrics/ambari-metrics-timelineservice/src/main/java/org/apache/hadoop/yarn/server/applicationhistoryservice/metrics/timeline/aggregators/TimelineMetricClusterAggregatorSecond.java
@@ -18,9 +18,8 @@
 package org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline.aggregators;
 
 
-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.PostProcessingUtil;
 import org.apache.hadoop.metrics2.sink.timeline.TimelineMetric;
 import org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline.PhoenixHBaseAccessor;
 import org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline.discovery.TimelineMetricMetadataManager;
@@ -31,10 +30,13 @@ import java.io.IOException;
 import java.sql.ResultSet;
 import java.sql.SQLException;
 import java.util.ArrayList;
+import java.util.Date;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+
 import static org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline.TimelineMetricConfiguration.SERVER_SIDE_TIMESIFT_ADJUSTMENT;
+import static org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline.TimelineMetricConfiguration.TIMELINE_METRICS_CLUSTER_AGGREGATOR_INTERPOLATION_ENABLED;
 import static org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline.query.PhoenixTransactSQL.GET_METRIC_SQL;
 import static org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline.query.PhoenixTransactSQL.METRICS_RECORD_TABLE_NAME;
 import static org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline.query.PhoenixTransactSQL.NATIVE_TIME_RANGE_DELTA;
@@ -50,6 +52,7 @@ public class TimelineMetricClusterAggregatorSecond extends AbstractTimelineAggre
   private final TimelineMetricAppAggregator appAggregator;
   // 1 minute client side buffering adjustment
   private final Long serverTimeShiftAdjustment;
+  private final boolean interpolationEnabled;
 
 
   public TimelineMetricClusterAggregatorSecond(String aggregatorName,
@@ -71,13 +74,16 @@ public class TimelineMetricClusterAggregatorSecond extends AbstractTimelineAggre
     appAggregator = new TimelineMetricAppAggregator(metadataManager, metricsConf);
     this.timeSliceIntervalMillis = timeSliceInterval;
     this.serverTimeShiftAdjustment = Long.parseLong(metricsConf.get(SERVER_SIDE_TIMESIFT_ADJUSTMENT, "90000"));
+    this.interpolationEnabled = Boolean.parseBoolean(metricsConf.get(TIMELINE_METRICS_CLUSTER_AGGREGATOR_INTERPOLATION_ENABLED, "true"));
   }
 
   @Override
   protected void aggregate(ResultSet rs, long startTime, long endTime) throws SQLException, IOException {
     // Account for time shift due to client side buffering by shifting the
     // timestamps with the difference between server time and series start time
-    List<Long[]> timeSlices = getTimeSlices(startTime - serverTimeShiftAdjustment, endTime);
+    // Also, we do not want to look at the shift time period from the end as well since we can interpolate those points
+    // that come earlier than the expected, during the next run.
+    List<Long[]> timeSlices = getTimeSlices(startTime - serverTimeShiftAdjustment, endTime - serverTimeShiftAdjustment);
     // Initialize app aggregates for host metrics
     appAggregator.init();
     Map<TimelineClusterMetric, MetricClusterAggregate> aggregateClusterMetrics =
@@ -99,15 +105,15 @@ public class TimelineMetricClusterAggregatorSecond extends AbstractTimelineAggre
     // Retaining order of the row-key avoids client side merge sort.
     condition.addOrderByColumn("METRIC_NAME");
     condition.addOrderByColumn("HOSTNAME");
-    condition.addOrderByColumn("SERVER_TIME");
     condition.addOrderByColumn("APP_ID");
+    condition.addOrderByColumn("SERVER_TIME");
     return condition;
   }
 
   /**
    * Return time slices to normalize the timeseries data.
    */
-  private List<Long[]> getTimeSlices(long startTime, long endTime) {
+  protected List<Long[]> getTimeSlices(long startTime, long endTime) {
     List<Long[]> timeSlices = new ArrayList<Long[]>();
     long sliceStartTime = startTime;
     while (sliceStartTime < endTime) {
@@ -118,7 +124,7 @@ public class TimelineMetricClusterAggregatorSecond extends AbstractTimelineAggre
   }
 
   private Map<TimelineClusterMetric, MetricClusterAggregate> aggregateMetricsFromResultSet(ResultSet rs, List<Long[]> timeSlices)
-      throws SQLException, IOException {
+    throws SQLException, IOException {
     Map<TimelineClusterMetric, MetricClusterAggregate> aggregateClusterMetrics =
       new HashMap<TimelineClusterMetric, MetricClusterAggregate>();
 
@@ -185,8 +191,8 @@ public class TimelineMetricClusterAggregatorSecond extends AbstractTimelineAggre
     }
   }
 
-  private Map<TimelineClusterMetric, Double> sliceFromTimelineMetric(
-      TimelineMetric timelineMetric, List<Long[]> timeSlices) {
+  protected Map<TimelineClusterMetric, Double> sliceFromTimelineMetric(
+    TimelineMetric timelineMetric, List<Long[]> timeSlices) {
 
     if (timelineMetric.getMetricValues().isEmpty()) {
       return null;
@@ -202,6 +208,7 @@ public class TimelineMetricClusterAggregatorSecond extends AbstractTimelineAggre
       timeShift = 0l;
     }
 
+    Map<Long,Double> timeSliceValueMap = new HashMap<>();
     for (Map.Entry<Long, Double> metric : timelineMetric.getMetricValues().entrySet()) {
       // TODO: investigate null values - pre filter
       if (metric.getValue() == null) {
@@ -228,13 +235,68 @@ public class TimelineMetricClusterAggregatorSecond extends AbstractTimelineAggre
           Double oldValue = timelineClusterMetricMap.get(clusterMetric);
           sum = oldValue + metric.getValue();
         }
-        timelineClusterMetricMap.put(clusterMetric, (sum / count));
+        double metricValue = sum / count;
+        timelineClusterMetricMap.put(clusterMetric, metricValue);
+        timeSliceValueMap.put(timestamp, metricValue);
       }
     }
+    if (interpolationEnabled) {
+      interpolateMissingPeriods(timelineClusterMetricMap, timelineMetric, timeSlices, timeSliceValueMap);
+    }
 
     return timelineClusterMetricMap;
   }
 
+  private void interpolateMissingPeriods(Map<TimelineClusterMetric, Double> timelineClusterMetricMap,
+                                         TimelineMetric timelineMetric,
+                                         List<Long[]> timeSlices,
+                                         Map<Long, Double> timeSliceValueMap) {
+
+
+    for (int sliceNum = 0; sliceNum < timeSlices.size(); sliceNum++) {
+      Long[] timeSlice = timeSlices.get(sliceNum);
+
+      if (!timeSliceValueMap.containsKey(timeSlice[1])) {
+        LOG.debug("Found an empty slice : " + new Date(timeSlice[0]) + ", " + new Date(timeSlice[1]));
+
+        Double lastSeenValue = null;
+        int index = sliceNum - 1;
+        Long[] prevTimeSlice = null;
+        while (lastSeenValue == null && index >= 0) {
+          prevTimeSlice = timeSlices.get(index--);
+          lastSeenValue = timeSliceValueMap.get(prevTimeSlice[1]);
+        }
+
+        Double nextSeenValue = null;
+        index = sliceNum + 1;
+        Long[] nextTimeSlice = null;
+        while ( nextSeenValue == null && index < timeSlices.size()) {
+          nextTimeSlice = timeSlices.get(index++);
+          nextSeenValue = timeSliceValueMap.get(nextTimeSlice[1]);
+        }
+
+        Double interpolatedValue = PostProcessingUtil.interpolate(timeSlice[1],
+          (prevTimeSlice != null ? prevTimeSlice[1] : null), lastSeenValue,
+          (nextTimeSlice != null ? nextTimeSlice[1] : null), nextSeenValue);
+
+        if (interpolatedValue != null) {
+          TimelineClusterMetric clusterMetric = new TimelineClusterMetric(
+            timelineMetric.getMetricName(),
+            timelineMetric.getAppId(),
+            timelineMetric.getInstanceId(),
+            timeSlice[1],
+            timelineMetric.getType());
+
+          LOG.debug("Interpolated value : " + interpolatedValue);
+          timelineClusterMetricMap.put(clusterMetric, interpolatedValue);
+        } else {
+          LOG.debug("Cannot compute interpolated value, hence skipping.");
+        }
+
+      }
+    }
+  }
+
   /**
    * Return end of the time slice into which the metric fits.
    */
diff --git a/ambari-metrics/ambari-metrics-timelineservice/src/test/java/org/apache/hadoop/yarn/server/applicationhistoryservice/metrics/timeline/aggregators/TimelineMetricClusterAggregatorSecondTest.java b/ambari-metrics/ambari-metrics-timelineservice/src/test/java/org/apache/hadoop/yarn/server/applicationhistoryservice/metrics/timeline/aggregators/TimelineMetricClusterAggregatorSecondTest.java
new file mode 100644
index 0000000000..d2d478c0b9
--- /dev/null
+++ b/ambari-metrics/ambari-metrics-timelineservice/src/test/java/org/apache/hadoop/yarn/server/applicationhistoryservice/metrics/timeline/aggregators/TimelineMetricClusterAggregatorSecondTest.java
@@ -0,0 +1,87 @@
+/**
+ * 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.yarn.server.applicationhistoryservice.metrics.timeline.aggregators;
+
+import junit.framework.Assert;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.metrics2.sink.timeline.TimelineMetric;
+import org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline.discovery.TimelineMetricMetadataManager;
+import org.easymock.EasyMock;
+import org.junit.Test;
+
+import java.util.Date;
+import java.util.List;
+import java.util.Map;
+import java.util.TreeMap;
+
+public class TimelineMetricClusterAggregatorSecondTest {
+
+  @Test
+  public void testClusterSecondAggregatorWithInterpolation() {
+
+    long aggregatorInterval = 120000l;
+    long sliceInterval = 30000l;
+    long metricInterval = 10000l;
+
+    Configuration configuration = new Configuration();
+    TimelineMetricMetadataManager metricMetadataManagerMock = EasyMock.createNiceMock(TimelineMetricMetadataManager.class);
+
+    TimelineMetricClusterAggregatorSecond secondAggregator = new TimelineMetricClusterAggregatorSecond(
+      "TimelineClusterAggregatorSecond", metricMetadataManagerMock, null, configuration, null,
+      aggregatorInterval, 2, "false", "", "", aggregatorInterval, sliceInterval
+    );
+
+    secondAggregator.timeSliceIntervalMillis = sliceInterval;
+    long roundedEndTime = AbstractTimelineAggregator.getRoundedAggregateTimeMillis(aggregatorInterval);
+    long roundedStartTime = roundedEndTime - aggregatorInterval;
+    List<Long[]> timeSlices = secondAggregator.getTimeSlices(roundedStartTime ,
+      roundedEndTime);
+
+    TreeMap<Long, Double> metricValues = new TreeMap<Long, Double>();
+
+    long startTime = roundedEndTime - aggregatorInterval;
+
+    for (int i=1; startTime < roundedEndTime; i++) {
+      startTime += metricInterval;
+      if (i%6 == 1 || i%6 == 2) {
+        metricValues.put(startTime, (double)i);
+      }
+    }
+
+    TimelineMetric metric = new TimelineMetric();
+    metric.setMetricName("TestMetric");
+    metric.setHostName("TestHost");
+    metric.setAppId("TestAppId");
+    metric.setMetricValues(metricValues);
+
+    Map<TimelineClusterMetric, Double> timelineClusterMetricMap = secondAggregator.sliceFromTimelineMetric(metric, timeSlices);
+
+    TimelineClusterMetric timelineClusterMetric = new TimelineClusterMetric(metric.getMetricName(), metric.getAppId(),
+      metric.getInstanceId(), 0l, null);
+
+    timelineClusterMetric.setTimestamp(roundedStartTime + 2*sliceInterval);
+    Assert.assertTrue(timelineClusterMetricMap.containsKey(timelineClusterMetric));
+    Assert.assertEquals(timelineClusterMetricMap.get(timelineClusterMetric), 4.5);
+
+    timelineClusterMetric.setTimestamp(roundedStartTime + 4*sliceInterval);
+    Assert.assertTrue(timelineClusterMetricMap.containsKey(timelineClusterMetric));
+    Assert.assertEquals(timelineClusterMetricMap.get(timelineClusterMetric), 7.5);
+
+  }
+
+}
diff --git a/ambari-server/src/main/java/org/apache/ambari/server/upgrade/UpgradeCatalog222.java b/ambari-server/src/main/java/org/apache/ambari/server/upgrade/UpgradeCatalog222.java
index 8cb5e5a8bf..598861be75 100644
--- a/ambari-server/src/main/java/org/apache/ambari/server/upgrade/UpgradeCatalog222.java
+++ b/ambari-server/src/main/java/org/apache/ambari/server/upgrade/UpgradeCatalog222.java
@@ -97,6 +97,8 @@ public class UpgradeCatalog222 extends AbstractUpgradeCatalog {
   public static final String HBASE_RPC_TIMEOUT_PROPERTY = "hbase.rpc.timeout";
   public static final String PHOENIX_QUERY_TIMEOUT_PROPERTY = "phoenix.query.timeoutMs";
   public static final String PHOENIX_QUERY_KEEPALIVE_PROPERTY = "phoenix.query.keepAliveMs";
+  public static final String TIMELINE_METRICS_CLUSTER_AGGREGATOR_INTERPOLATION_ENABLED
+    = "timeline.metrics.cluster.aggregator.interpolation.enabled";
 
   public static final String AMS_SERVICE_NAME = "AMBARI_METRICS";
   public static final String AMS_COLLECTOR_COMPONENT_NAME = "METRICS_COLLECTOR";
@@ -249,7 +251,7 @@ public class UpgradeCatalog222 extends AbstractUpgradeCatalog {
 
   }
 
-  protected void updateHostRoleCommands() throws SQLException{
+  protected void updateHostRoleCommands() throws SQLException {
     dbAccessor.createIndex("idx_hrc_status", "host_role_command", "status", "role");
   }
 
@@ -312,7 +314,7 @@ public class UpgradeCatalog222 extends AbstractUpgradeCatalog {
               String newTtl = oldTtl;
               if (isDistributed) {
                 if ("86400".equals(oldTtl)) {
-                  newTtl = String.valueOf(7 * 86400); // 7 days
+                  newTtl = String.valueOf(3 * 86400); // 3 days
                 }
               }
               newProperties.put(PRECISION_TABLE_TTL_PROPERTY, newTtl);
@@ -343,6 +345,11 @@ public class UpgradeCatalog222 extends AbstractUpgradeCatalog {
               LOG.info("Setting value of " + CLUSTER_MINUTE_TABLE_TTL_PROPERTY + " : " + newTtl);
             }
 
+            if (!amsSiteProperties.containsKey(TIMELINE_METRICS_CLUSTER_AGGREGATOR_INTERPOLATION_ENABLED)) {
+              LOG.info("Add config  " + TIMELINE_METRICS_CLUSTER_AGGREGATOR_INTERPOLATION_ENABLED + " = true");
+              newProperties.put(TIMELINE_METRICS_CLUSTER_AGGREGATOR_INTERPOLATION_ENABLED, String.valueOf(true));
+            }
+
             updateConfigurationPropertiesForCluster(cluster, AMS_SITE, newProperties, true, true);
           }
 
@@ -515,7 +522,7 @@ public class UpgradeCatalog222 extends AbstractUpgradeCatalog {
    */
   protected void updateUpgradeTable() throws AmbariException, SQLException {
     dbAccessor.addColumn(UPGRADE_TABLE,
-        new DBAccessor.DBColumnInfo(UPGRADE_SUSPENDED_COLUMN, Short.class, 1, 0, false));
+      new DBAccessor.DBColumnInfo(UPGRADE_SUSPENDED_COLUMN, Short.class, 1, 0, false));
   }
 
 }
diff --git a/ambari-server/src/main/resources/common-services/AMBARI_METRICS/0.1.0/configuration/ams-hbase-env.xml b/ambari-server/src/main/resources/common-services/AMBARI_METRICS/0.1.0/configuration/ams-hbase-env.xml
index d6e68d116e..378138b475 100644
--- a/ambari-server/src/main/resources/common-services/AMBARI_METRICS/0.1.0/configuration/ams-hbase-env.xml
+++ b/ambari-server/src/main/resources/common-services/AMBARI_METRICS/0.1.0/configuration/ams-hbase-env.xml
@@ -25,6 +25,12 @@
     <name>hbase_log_dir</name>
     <value>/var/log/ambari-metrics-collector</value>
     <description>Log Directories for HBase.</description>
+    <depends-on>
+      <property>
+        <type>ams-env</type>
+        <name>metrics_collector_log_dir</name>
+      </property>
+    </depends-on>
     <value-attributes>
       <type>directory</type>
     </value-attributes>
diff --git a/ambari-server/src/main/resources/common-services/AMBARI_METRICS/0.1.0/configuration/ams-site.xml b/ambari-server/src/main/resources/common-services/AMBARI_METRICS/0.1.0/configuration/ams-site.xml
index 49934128a5..e5758bff90 100644
--- a/ambari-server/src/main/resources/common-services/AMBARI_METRICS/0.1.0/configuration/ams-site.xml
+++ b/ambari-server/src/main/resources/common-services/AMBARI_METRICS/0.1.0/configuration/ams-site.xml
@@ -334,7 +334,7 @@
     <name>timeline.metrics.host.aggregator.ttl</name>
     <value>86400</value>
     <description>
-      1 minute resolution data purge interval in seconds. Default is 1 day for embedded mode and 7 days for Distributed mode.
+      1 minute resolution data purge interval in seconds. Default is 1 day for embedded mode and 3 days for Distributed mode.
     </description>
     <depends-on>
       <property>
@@ -570,5 +570,12 @@
       Enable Initialization check for HBase tables during Metrics service startup.
     </description>
   </property>
+  <property>
+    <name>timeline.metrics.cluster.aggregator.interpolation.enabled</name>
+    <value>true</value>
+    <description>
+      Enable Linear interpolation for missing slices of data, while aggregating.
+    </description>
+  </property>
 
 </configuration>
diff --git a/ambari-server/src/main/resources/stacks/HDP/2.0.6/services/stack_advisor.py b/ambari-server/src/main/resources/stacks/HDP/2.0.6/services/stack_advisor.py
index 24d6dd215e..4b99081e37 100644
--- a/ambari-server/src/main/resources/stacks/HDP/2.0.6/services/stack_advisor.py
+++ b/ambari-server/src/main/resources/stacks/HDP/2.0.6/services/stack_advisor.py
@@ -522,6 +522,12 @@ class HDP206StackAdvisor(DefaultStackAdvisor):
 
     putAmsSiteProperty("timeline.metrics.service.webapp.address", str(metric_collector_host) + ":6188")
 
+    log_dir = "/var/log/ambari-metrics-collector"
+    if "ams-env" in services["configurations"]:
+      if "metrics_collector_log_dir" in services["configurations"]["ams-env"]["properties"]:
+        log_dir = services["configurations"]["ams-env"]["properties"]["metrics_collector_log_dir"]
+      putHbaseEnvProperty("hbase_log_dir", log_dir)
+
     defaultFs = 'file:///'
     if "core-site" in services["configurations"] and \
       "fs.defaultFS" in services["configurations"]["core-site"]["properties"]:
@@ -534,7 +540,7 @@ class HDP206StackAdvisor(DefaultStackAdvisor):
 
     if operatingMode == "distributed":
       putAmsSiteProperty("timeline.metrics.service.watcher.disabled", 'true')
-      putAmsSiteProperty("timeline.metrics.host.aggregator.ttl", 604800)
+      putAmsSiteProperty("timeline.metrics.host.aggregator.ttl", 259200)
       putAmsHbaseSiteProperty("hbase.cluster.distributed", 'true')
     else:
       putAmsSiteProperty("timeline.metrics.service.watcher.disabled", 'false')
diff --git a/ambari-server/src/test/java/org/apache/ambari/server/upgrade/UpgradeCatalog222Test.java b/ambari-server/src/test/java/org/apache/ambari/server/upgrade/UpgradeCatalog222Test.java
index cbb2797211..a9eccdc4a2 100644
--- a/ambari-server/src/test/java/org/apache/ambari/server/upgrade/UpgradeCatalog222Test.java
+++ b/ambari-server/src/test/java/org/apache/ambari/server/upgrade/UpgradeCatalog222Test.java
@@ -301,7 +301,7 @@ public class UpgradeCatalog222Test {
         put("timeline.metrics.host.aggregator.daily.checkpointCutOffMultiplier", String.valueOf(2));
         put("timeline.metrics.cluster.aggregator.daily.checkpointCutOffMultiplier", String.valueOf(2));
         put("timeline.metrics.service.watcher.disabled", String.valueOf(false));
-        put("timeline.metrics.host.aggregator.ttl", String.valueOf(7 * 86400));
+        put("timeline.metrics.host.aggregator.ttl", String.valueOf(3 * 86400));
         put("timeline.metrics.host.aggregator.minute.ttl", String.valueOf(7 * 86400));
         put("timeline.metrics.host.aggregator.hourly.ttl", String.valueOf(30 * 86400));
         put("timeline.metrics.host.aggregator.daily.ttl", String.valueOf(365 * 86400));
@@ -311,6 +311,7 @@ public class UpgradeCatalog222Test {
         put("timeline.metrics.cluster.aggregator.daily.ttl", String.valueOf(730 * 86400));
         put("timeline.metrics.service.operation.mode", "distributed");
         put("timeline.metrics.service.webapp.address", "host1:6188");
+        put("timeline.metrics.cluster.aggregator.interpolation.enabled", String.valueOf(true));
       }
     };
     EasyMockSupport easyMockSupport = new EasyMockSupport();
diff --git a/ambari-server/src/test/python/stacks/2.2/common/test_stack_advisor.py b/ambari-server/src/test/python/stacks/2.2/common/test_stack_advisor.py
index 1d3789ef4a..06fb9f3586 100644
--- a/ambari-server/src/test/python/stacks/2.2/common/test_stack_advisor.py
+++ b/ambari-server/src/test/python/stacks/2.2/common/test_stack_advisor.py
@@ -2249,7 +2249,7 @@ class TestHDP22StackAdvisor(TestCase):
     expected['ams-hbase-env']['properties']['hbase_regionserver_heapsize'] = '512'
     expected["ams-hbase-env"]['properties']['hbase_master_xmn_size'] = '102'
     expected['ams-hbase-env']['properties']['regionserver_xmn_size'] = '384'
-    expected['ams-site']['properties']['timeline.metrics.host.aggregator.ttl'] = '604800'
+    expected['ams-site']['properties']['timeline.metrics.host.aggregator.ttl'] = '259200'
     expected['ams-site']['properties']['timeline.metrics.service.watcher.disabled'] = 'true'
     self.stackAdvisor.recommendAmsConfigurations(configurations, clusterData, services, hosts)
     self.assertEquals(configurations, expected)