AMBARI-22215 Refine cluster second aggregator by aligning sink publish times to 1 minute boundaries. (dsen)

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

@@ -19,6 +19,8 @@ package org.apache.hadoop.metrics2.sink.timeline;
 
 import com.google.common.base.Supplier;
 import com.google.common.base.Suppliers;
+import com.google.common.cache.Cache;
+import com.google.common.cache.CacheBuilder;
 import com.google.common.reflect.TypeToken;
 import com.google.gson.Gson;
 import com.google.gson.JsonSyntaxException;
@@ -58,6 +60,7 @@ import java.util.List;
 import java.util.Random;
 import java.util.Set;
 import java.util.SortedSet;
+import java.util.TreeMap;
 import java.util.TreeSet;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
@@ -130,6 +133,13 @@ public abstract class AbstractTimelineMetricsSink {
   private static final int COLLECTOR_HOST_CACHE_MAX_EXPIRATION_MINUTES = 75;
   private static final int COLLECTOR_HOST_CACHE_MIN_EXPIRATION_MINUTES = 60;
 
+  //10 seconds
+  protected int collectionPeriodMillis = 10000;
+
+  private int cacheExpireTimeMinutesDefault = 10;
+
+  private volatile Cache<String, TimelineMetric> metricsPostCache = CacheBuilder.newBuilder().expireAfterAccess(cacheExpireTimeMinutesDefault, TimeUnit.MINUTES).build();
+
   static {
     mapper = new ObjectMapper();
     AnnotationIntrospector introspector = new JaxbAnnotationIntrospector();
@@ -289,7 +299,21 @@ public abstract class AbstractTimelineMetricsSink {
     return collectorHost;
   }
 
+  /**
+   * @param metrics metrics to post, metric values will be aligned by minute mark,
+   *                last uncompleted minute will be cached to post in future iteration
+   */
   protected boolean emitMetrics(TimelineMetrics metrics) {
+    return emitMetrics(metrics, false);
+  }
+
+  /**
+   * @param metrics metrics to post, if postAllCachedMetrics is false metric values will be aligned by minute mark,
+   *                last uncompleted minute will be cached to post in future iteration
+   * @param postAllCachedMetrics if set to true all cached metrics will be posted, ignoring the minute aligning
+   * @return
+   */
+  protected boolean emitMetrics(TimelineMetrics metrics, boolean postAllCachedMetrics) {
     String connectUrl;
     boolean validCollectorHost = true;
 
@@ -307,11 +331,20 @@ public abstract class AbstractTimelineMetricsSink {
       connectUrl = getCollectorUri(collectorHost);
     }
 
+    TimelineMetrics metricsToEmit = alignMetricsByMinuteMark(metrics);
+
+    if (postAllCachedMetrics) {
+      for (TimelineMetric timelineMetric : metricsPostCache.asMap().values()) {
+        metricsToEmit.addOrMergeTimelineMetric(timelineMetric);
+      }
+      metricsPostCache.invalidateAll();
+    }
+
     if (validCollectorHost) {
       String jsonData = null;
       LOG.debug("EmitMetrics connectUrl = "  + connectUrl);
       try {
-        jsonData = mapper.writeValueAsString(metrics);
+        jsonData = mapper.writeValueAsString(metricsToEmit);
       } catch (IOException e) {
         LOG.error("Unable to parse metrics", e);
       }
@@ -334,6 +367,61 @@ public abstract class AbstractTimelineMetricsSink {
     return appCookieManager;
   }
 
+  /**
+   * Align metrics by the minutes so that only complete minutes are send.
+   * Not completed minutes data points will be cached and posted when the minute will be completed.
+   * Cached metrics are merged with currently posting metrics
+   * e.g:
+   * first iteration if metrics from 00m15s to 01m15s are processed,
+   *               then metrics from 00m15s to 00m59s will be posted
+   *                        and from 01m00s to 01m15s will be cached
+   * second iteration   metrics from 01m25s to 02m55s are processed,
+   *     cached metrics from previous call will be merged with current,
+   *                    metrics from 01m00s to 02m55s will be posted, cache will be empty
+   * @param metrics
+   * @return
+   */
+  protected TimelineMetrics alignMetricsByMinuteMark(TimelineMetrics metrics) {
+    TimelineMetrics allMetricsToPost = new TimelineMetrics();
+
+    for (TimelineMetric metric : metrics.getMetrics()) {
+      TimelineMetric cachedMetric = metricsPostCache.getIfPresent(metric.getMetricName());
+      if (cachedMetric != null) {
+        metric.addMetricValues(cachedMetric.getMetricValues());
+        metricsPostCache.invalidate(metric.getMetricName());
+      }
+    }
+
+    for (TimelineMetric metric : metrics.getMetrics()) {
+      TreeMap<Long, Double> valuesToCache = new TreeMap<>();
+      TreeMap<Long, Double> valuesToPost = metric.getMetricValues();
+
+      // in case there can't be any more datapoints in last minute just post the metrics,
+      // otherwise need to cut off and cache the last uncompleted minute
+      if (!(valuesToPost.lastKey() % 60000 > 60000 - collectionPeriodMillis)) {
+        Long lastMinute = valuesToPost.lastKey() / 60000;
+        while (!valuesToPost.isEmpty() && valuesToPost.lastKey() / 60000 == lastMinute) {
+          valuesToCache.put(valuesToPost.lastKey(), valuesToPost.get(valuesToPost.lastKey()));
+          valuesToPost.remove(valuesToPost.lastKey());
+        }
+      }
+
+      if (!valuesToCache.isEmpty()) {
+        TimelineMetric metricToCache = new TimelineMetric(metric);
+        metricToCache.setMetricValues(valuesToCache);
+        metricsPostCache.put(metricToCache.getMetricName(), metricToCache);
+      }
+
+      if (!valuesToPost.isEmpty()) {
+        TimelineMetric metricToPost = new TimelineMetric(metric);
+        metricToPost.setMetricValues(valuesToPost);
+        allMetricsToPost.addOrMergeTimelineMetric(metricToPost);
+      }
+    }
+
+    return allMetricsToPost;
+  }
+
   /**
    * Cleans up and closes an input stream
    * see http://docs.oracle.com/javase/6/docs/technotes/guides/net/http-keepalive.html
@@ -609,6 +697,11 @@ public abstract class AbstractTimelineMetricsSink {
       rand.nextInt(zookeeperMaxBackoffTimeMins - zookeeperMinBackoffTimeMins + 1)) * 60*1000l;
   }
 
+  //for now it's used only for testing
+  protected Cache<String, TimelineMetric> getMetricsPostCache() {
+    return metricsPostCache;
+  }
+
   /**
    * Get a pre-formatted URI for the collector
    */

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

@@ -146,6 +146,9 @@ public class TimelineMetric implements Comparable<TimelineMetric>, Serializable
 
   public void addMetricValues(Map<Long, Double> metricValues) {
     this.metricValues.putAll(metricValues);
+    if (!this.metricValues.isEmpty()) {
+      this.setStartTime(this.metricValues.firstKey());
+    }
   }
 
   @XmlElement(name = "metadata")

ambari-metrics/ambari-metrics-common/src/test/java/org/apache/hadoop/metrics2/sink/timeline/AbstractTimelineMetricSinkTest.java

@@ -0,0 +1,240 @@
+/**
+ * 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 junit.framework.Assert;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.powermock.api.easymock.PowerMock;
+import org.powermock.core.classloader.annotations.PrepareForTest;
+import org.powermock.modules.junit4.PowerMockRunner;
+
+import java.io.OutputStream;
+import java.net.HttpURLConnection;
+import java.net.URL;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.TreeMap;
+
+import static org.easymock.EasyMock.anyString;
+import static org.easymock.EasyMock.expect;
+import static org.powermock.api.easymock.PowerMock.expectNew;
+import static org.powermock.api.easymock.PowerMock.replayAll;
+
+@RunWith(PowerMockRunner.class)
+@PrepareForTest({AbstractTimelineMetricsSink.class, HttpURLConnection.class})
+public class AbstractTimelineMetricSinkTest {
+
+  @Test
+  public void testParseHostsStringIntoCollection() {
+    AbstractTimelineMetricsSink sink = new TestTimelineMetricsSink();
+    Collection<String> hosts;
+
+    hosts = sink.parseHostsStringIntoCollection("");
+    Assert.assertTrue(hosts.isEmpty());
+
+    hosts = sink.parseHostsStringIntoCollection("test1.123.abc.def.local");
+    Assert.assertTrue(hosts.size() == 1);
+    Assert.assertTrue(hosts.contains("test1.123.abc.def.local"));
+
+    hosts = sink.parseHostsStringIntoCollection("test1.123.abc.def.local ");
+    Assert.assertTrue(hosts.size() == 1);
+    Assert.assertTrue(hosts.contains("test1.123.abc.def.local"));
+
+    hosts = sink.parseHostsStringIntoCollection("test1.123.abc.def.local,test1.456.abc.def.local");
+    Assert.assertTrue(hosts.size() == 2);
+
+    hosts = sink.parseHostsStringIntoCollection("test1.123.abc.def.local, test1.456.abc.def.local");
+    Assert.assertTrue(hosts.size() == 2);
+    Assert.assertTrue(hosts.contains("test1.123.abc.def.local"));
+    Assert.assertTrue(hosts.contains("test1.456.abc.def.local"));
+  }
+
+  @Test
+  @PrepareForTest({URL.class, OutputStream.class, AbstractTimelineMetricsSink.class, HttpURLConnection.class, TimelineMetric.class})
+  public void testEmitMetrics() throws Exception {
+    HttpURLConnection connection = PowerMock.createNiceMock(HttpURLConnection.class);
+    URL url = PowerMock.createNiceMock(URL.class);
+    expectNew(URL.class, anyString()).andReturn(url).anyTimes();
+    expect(url.openConnection()).andReturn(connection).anyTimes();
+    expect(connection.getResponseCode()).andReturn(200).anyTimes();
+    OutputStream os = PowerMock.createNiceMock(OutputStream.class);
+    expect(connection.getOutputStream()).andReturn(os).anyTimes();
+
+
+    TestTimelineMetricsSink sink = new TestTimelineMetricsSink();
+    TimelineMetrics timelineMetrics = new TimelineMetrics();
+    long startTime = System.currentTimeMillis() / 60000 * 60000;
+
+    long seconds = 1000;
+    TreeMap<Long, Double> metricValues = new TreeMap<>();
+    /*
+
+    0        +30s      +60s
+    |         |         |
+      (1)(2)(3) (4)(5)   (6)  m1
+
+    */
+    // (6) should be cached, the rest - posted
+
+    metricValues.put(startTime + 4*seconds, 1.0);
+    metricValues.put(startTime + 14*seconds, 2.0);
+    metricValues.put(startTime + 24*seconds, 3.0);
+    metricValues.put(startTime + 34*seconds, 4.0);
+    metricValues.put(startTime + 44*seconds, 5.0);
+    metricValues.put(startTime + 64*seconds, 6.0);
+
+    TimelineMetric timelineMetric = new TimelineMetric("metric1", "host1", "app1", "instance1");
+    timelineMetric.setStartTime(metricValues.firstKey());
+    timelineMetric.addMetricValues(metricValues);
+
+    timelineMetrics.addOrMergeTimelineMetric(timelineMetric);
+
+    replayAll();
+    sink.emitMetrics(timelineMetrics);
+    Assert.assertEquals(1, sink.getMetricsPostCache().size());
+    metricValues = new TreeMap<>();
+    metricValues.put(startTime + 64*seconds, 6.0);
+    Assert.assertEquals(metricValues, sink.getMetricsPostCache().getIfPresent("metric1").getMetricValues());
+
+    timelineMetrics = new TimelineMetrics();
+    metricValues = new TreeMap<>();
+    /*
+
+    +60      +90s     +120s     +150s     +180s
+    |         |         |         |         |
+       (7)      (8)       (9)           (10)   (11)   m1
+
+    */
+    // (6) from previous post should be merged with current data
+    // (6),(7),(8),(9),(10) - should be posted, (11) - cached
+    metricValues.put(startTime + 74*seconds, 7.0);
+    metricValues.put(startTime + 94*seconds, 8.0);
+    metricValues.put(startTime + 124*seconds, 9.0);
+    metricValues.put(startTime + 154*seconds, 10.0);
+    metricValues.put(startTime + 184*seconds, 11.0);
+
+    timelineMetric = new TimelineMetric("metric1", "host1", "app1", "instance1");
+    timelineMetric.setStartTime(metricValues.firstKey());
+    timelineMetric.addMetricValues(metricValues);
+
+    timelineMetrics.addOrMergeTimelineMetric(timelineMetric);
+    sink.emitMetrics(timelineMetrics);
+
+    Assert.assertEquals(1, sink.getMetricsPostCache().size());
+    metricValues = new TreeMap<>();
+    metricValues.put(startTime + 184*seconds, 11.0);
+    Assert.assertEquals(metricValues, sink.getMetricsPostCache().getIfPresent("metric1").getMetricValues());timelineMetrics = new TimelineMetrics();
+
+    metricValues = new TreeMap<>();
+    /*
+
+    +180s   +210s   +240s
+    |         |       |
+       (12)        (13)
+
+    */
+    // (11) from previous post should be merged with current data
+    // (11),(12),(13) - should be posted, cache should be empty
+    metricValues.put(startTime + 194*seconds, 12.0);
+    metricValues.put(startTime + 239*seconds, 13.0);
+
+    timelineMetric = new TimelineMetric("metric1", "host1", "app1", "instance1");
+    timelineMetric.setStartTime(metricValues.firstKey());
+    timelineMetric.addMetricValues(metricValues);
+
+    timelineMetrics.addOrMergeTimelineMetric(timelineMetric);
+    sink.emitMetrics(timelineMetrics);
+
+    Assert.assertEquals(0, sink.getMetricsPostCache().size());
+
+    metricValues = new TreeMap<>();
+    /*
+
+    +240s   +270s   +300s   +330s
+    |         |       |       |
+       (14)        (15)   (16)
+
+    */
+    // since postAllCachedMetrics in emitMetrics call is true (14),(15),(16) - should be posted, cache should be empty
+    metricValues.put(startTime + 245*seconds, 14.0);
+    metricValues.put(startTime + 294*seconds, 15.0);
+    metricValues.put(startTime + 315*seconds, 16.0);
+
+    timelineMetric = new TimelineMetric("metric1", "host1", "app1", "instance1");
+    timelineMetric.setStartTime(metricValues.firstKey());
+    timelineMetric.addMetricValues(metricValues);
+
+    timelineMetrics.addOrMergeTimelineMetric(timelineMetric);
+    sink.emitMetrics(timelineMetrics, true);
+
+    Assert.assertEquals(0, sink.getMetricsPostCache().size());
+  }
+
+  private class TestTimelineMetricsSink extends AbstractTimelineMetricsSink {
+    @Override
+    protected String getCollectorUri(String host) {
+      return "";
+    }
+
+    @Override
+    protected String getCollectorProtocol() {
+      return "http";
+    }
+
+    @Override
+    protected String getCollectorPort() {
+      return "2181";
+    }
+
+    @Override
+    protected int getTimeoutSeconds() {
+      return 10;
+    }
+
+    @Override
+    protected String getZookeeperQuorum() {
+      return "localhost:2181";
+    }
+
+    @Override
+    protected Collection<String> getConfiguredCollectorHosts() {
+      return Arrays.asList("localhost");
+    }
+
+    @Override
+    protected String getHostname() {
+      return "h1";
+    }
+
+    @Override
+    protected boolean isHostInMemoryAggregationEnabled() {
+      return true;
+    }
+
+    @Override
+    protected int getHostInMemoryAggregationPort() {
+      return 61888;
+    }
+
+    @Override
+    protected String getHostInMemoryAggregationProtocol() {
+      return "http";
+    }
+  }
+}

ambari-metrics/ambari-metrics-common/src/test/java/org/apache/hadoop/metrics2/sink/timeline/availability/AbstractTimelineMetricSinkTest.java

@@ -1,113 +0,0 @@
-/**
- * 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.availability;
-
-import junit.framework.Assert;
-import org.apache.hadoop.metrics2.sink.timeline.AbstractTimelineMetricsSink;
-import org.apache.hadoop.metrics2.sink.timeline.TimelineMetrics;
-import org.junit.Test;
-
-import java.util.Arrays;
-import java.util.Collection;
-
-public class AbstractTimelineMetricSinkTest {
-
-  @Test
-  public void testParseHostsStringIntoCollection() {
-    AbstractTimelineMetricsSink sink = new TestTimelineMetricsSink();
-    Collection<String> hosts;
-
-    hosts = sink.parseHostsStringIntoCollection("");
-    Assert.assertTrue(hosts.isEmpty());
-
-    hosts = sink.parseHostsStringIntoCollection("test1.123.abc.def.local");
-    Assert.assertTrue(hosts.size() == 1);
-    Assert.assertTrue(hosts.contains("test1.123.abc.def.local"));
-
-    hosts = sink.parseHostsStringIntoCollection("test1.123.abc.def.local ");
-    Assert.assertTrue(hosts.size() == 1);
-    Assert.assertTrue(hosts.contains("test1.123.abc.def.local"));
-
-    hosts = sink.parseHostsStringIntoCollection("test1.123.abc.def.local,test1.456.abc.def.local");
-    Assert.assertTrue(hosts.size() == 2);
-
-    hosts = sink.parseHostsStringIntoCollection("test1.123.abc.def.local, test1.456.abc.def.local");
-    Assert.assertTrue(hosts.size() == 2);
-    Assert.assertTrue(hosts.contains("test1.123.abc.def.local"));
-    Assert.assertTrue(hosts.contains("test1.456.abc.def.local"));
-
-  }
-
-  private class TestTimelineMetricsSink extends AbstractTimelineMetricsSink {
-    @Override
-    protected String getCollectorUri(String host) {
-      return "";
-    }
-
-    @Override
-    protected String getCollectorProtocol() {
-      return "http";
-    }
-
-    @Override
-    protected String getCollectorPort() {
-      return "2181";
-    }
-
-    @Override
-    protected int getTimeoutSeconds() {
-      return 10;
-    }
-
-    @Override
-    protected String getZookeeperQuorum() {
-      return "localhost:2181";
-    }
-
-    @Override
-    protected Collection<String> getConfiguredCollectorHosts() {
-      return Arrays.asList("localhost");
-    }
-
-    @Override
-    protected String getHostname() {
-      return "h1";
-    }
-
-    @Override
-    protected boolean isHostInMemoryAggregationEnabled() {
-      return true;
-    }
-
-    @Override
-    protected int getHostInMemoryAggregationPort() {
-      return 61888;
-    }
-
-    @Override
-    protected String getHostInMemoryAggregationProtocol() {
-      return "http";
-    }
-
-    @Override
-    public boolean emitMetrics(TimelineMetrics metrics) {
-      super.init();
-      return super.emitMetrics(metrics);
-    }
-  }
-}

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

@@ -509,7 +509,7 @@ public class HadoopTimelineMetricsSink extends AbstractTimelineMetricsSink imple
         LOG.debug("Closing HadoopTimelineMetricSink. Flushing metrics to collector...");
         TimelineMetrics metrics = metricsCache.getAllMetrics();
         if (metrics != null) {
-          emitMetrics(metrics);
+          emitMetrics(metrics, true);
         }
       }
     });

ambari-metrics/ambari-metrics-hadoop-sink/src/test/java/org/apache/hadoop/metrics2/sink/timeline/HadoopTimelineMetricsSinkTest.java

@@ -180,7 +180,7 @@ public class HadoopTimelineMetricsSinkTest {
       createMockBuilder(HadoopTimelineMetricsSink.class)
         .withConstructor().addMockedMethod("appendPrefix")
         .addMockedMethod("findLiveCollectorHostsFromKnownCollector")
-        .addMockedMethod("emitMetrics").createNiceMock();
+        .addMockedMethod("emitMetrics", TimelineMetrics.class).createNiceMock();
 
     SubsetConfiguration conf = PowerMock.createNiceMock(SubsetConfiguration.class);
     expect(conf.getString("slave.host.name")).andReturn("localhost").anyTimes();
@@ -311,7 +311,7 @@ public class HadoopTimelineMetricsSinkTest {
       createMockBuilder(HadoopTimelineMetricsSink.class)
         .withConstructor().addMockedMethod("appendPrefix")
         .addMockedMethod("findLiveCollectorHostsFromKnownCollector")
-        .addMockedMethod("emitMetrics").createNiceMock();
+        .addMockedMethod("emitMetrics", TimelineMetrics.class).createNiceMock();
 
     SubsetConfiguration conf = PowerMock.createNiceMock(SubsetConfiguration.class);
     expect(conf.getString("slave.host.name")).andReturn("localhost").anyTimes();

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

@@ -41,6 +41,7 @@ class ApplicationMetricMap:
     self.ip_address = ip_address
     self.lock = RLock()
     self.app_metric_map = {}
+    self.cached_metric_map = {}
   pass
 
   def put_metric(self, application_id, metric_id_to_value_map, timestamp):
@@ -98,7 +99,7 @@ class ApplicationMetricMap:
             "appid" : "HOST",
             "instanceid" : result_instanceid,
             "starttime" : self.get_start_time(appId, metricId),
-            "metrics" : metricData
+            "metrics" : self.align_values_by_minute_mark(appId, metricId, metricData) if clear_once_flattened else metricData
           }
           timeline_metrics[ "metrics" ].append( timeline_metric )
         pass
@@ -114,6 +115,10 @@ class ApplicationMetricMap:
 
   def get_start_time(self, app_id, metric_id):
     with self.lock:
+      if self.cached_metric_map.has_key(app_id):
+        if self.cached_metric_map.get(app_id).has_key(metric_id):
+          metrics = self.cached_metric_map.get(app_id).get(metric_id)
+          return min(metrics.iterkeys())
       if self.app_metric_map.has_key(app_id):
         if self.app_metric_map.get(app_id).has_key(metric_id):
           metrics = self.app_metric_map.get(app_id).get(metric_id)
@@ -137,3 +142,48 @@ class ApplicationMetricMap:
     with self.lock:
       self.app_metric_map.clear()
   pass
+
+  # Align metrics by the minutes so that only complete minutes are send.
+  # Not completed minutes data points will be cached and posted when the minute will be completed.
+  # Cached metrics are merged with currently posting metrics
+  # e.g:
+  # first iteration if metrics from 00m15s to 01m15s are processed,
+  #               then metrics from 00m15s to 00m59s will be posted
+  #                        and from 01m00s to 01m15s will be cached
+  # second iteration   metrics from 01m25s to 02m55s are processed,
+  #     cached metrics from previous call will be merged with current,
+  #                    metrics from 01m00s to 02m55s will be posted, cache will be empty
+  def align_values_by_minute_mark(self, appId, metricId, metricData):
+    with self.lock:
+      # append with cached values
+      if self.cached_metric_map.get(appId) and self.cached_metric_map.get(appId).get(metricId):
+        metricData.update(self.cached_metric_map[appId][metricId])
+        self.cached_metric_map[appId].pop(metricId)
+
+      # check if needs to be cached
+      # in case there can't be any more datapoints in last minute just post the metrics,
+      # otherwise need to cut off and cache the last uncompleted minute
+      max_time = max(metricData.iterkeys())
+      if max_time % 60000 <= 60000 - 10000:
+        max_minute = max_time / 60000
+        metric_data_copy = metricData.copy()
+        for time,value in metric_data_copy.iteritems():
+          if time / 60000 == max_minute:
+            cached_metric_map = self.cached_metric_map.get(appId)
+            if not cached_metric_map:
+              cached_metric_map = { metricId : { time : value } }
+              self.cached_metric_map[ appId ] = cached_metric_map
+            else:
+              cached_metric_id_map = cached_metric_map.get(metricId)
+              if not cached_metric_id_map:
+                cached_metric_id_map = { time : value }
+                cached_metric_map[ metricId ] = cached_metric_id_map
+              else:
+                cached_metric_map[ metricId ].update( { time : value } )
+              pass
+            pass
+            metricData.pop(time)
+          pass
+        pass
+
+      return metricData

ambari-metrics/ambari-metrics-host-monitoring/src/test/python/core/TestApplicationMetricMap.py

@@ -50,7 +50,7 @@ class TestApplicationMetricMap(TestCase):
     self.assertEqual(p['metrics'][0]['metrics'][str(timestamp)], 'bv')
     
     self.assertEqual(application_metric_map.get_start_time(application_id, "b"), timestamp)
-    
+
     metrics = {}
     metrics.update({"b" : 'bv'})
     metrics.update({"a" : 'av'})
@@ -71,4 +71,38 @@ class TestApplicationMetricMap(TestCase):
     json_data = json.loads(application_metric_map.flatten('A1', True))
     self.assertEqual(len(json_data['metrics']), 1)
     self.assertTrue(json_data['metrics'][0]['metricname'] == 'a')
-    self.assertFalse(application_metric_map.app_metric_map)
+    self.assertFalse(application_metric_map.app_metric_map)
+
+  def test_flatten_and_align_values_by_minute_mark(self):
+    application_metric_map = ApplicationMetricMap("host", "10.10.10.10")
+    second = 1000
+    timestamp = int(round(1415390640.3806491 * second))
+    application_id = application_metric_map.format_app_id("A","1")
+    metrics = {}
+    metrics.update({"b" : 'bv'})
+
+    #   0s    60s   120s
+    #   (0) (1)   (2)    (3)
+    # (3) should be cut off and cached
+    application_metric_map.put_metric(application_id, metrics, timestamp)
+    application_metric_map.put_metric(application_id, metrics, timestamp + second*24)
+    application_metric_map.put_metric(application_id, metrics, timestamp + second*84)
+    application_metric_map.put_metric(application_id, metrics, timestamp + second*124)
+
+    json_data = json.loads(application_metric_map.flatten(application_id, True))
+    self.assertEqual(len(json_data['metrics'][0]['metrics']), 3)
+    self.assertEqual(len(application_metric_map.cached_metric_map.get(application_id).get("b")), 1)
+    self.assertEqual(application_metric_map.cached_metric_map.get(application_id).get("b"), {timestamp + second*124 : 'bv'})
+
+    #   120s    180s
+    #      (3)  (4)
+    # cached (3) should be added to the post;
+    # (4) should be posted as well because there can't be more data points in the minute
+    application_metric_map.put_metric(application_id, metrics, timestamp + second * 176)
+
+    json_data = json.loads(application_metric_map.flatten(application_id, True))
+    self.assertEqual(len(json_data['metrics'][0]['metrics']), 2)
+
+    # starttime should be set to (3)
+    self.assertEqual(json_data['metrics'][0]['starttime'], timestamp + second*124)
+    self.assertEqual(len(application_metric_map.cached_metric_map.get(application_id)), 0)

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

@@ -121,9 +121,6 @@ public class TimelineMetricConfiguration {
   public static final String CLUSTER_AGGREGATOR_TIMESLICE_INTERVAL =
     "timeline.metrics.cluster.aggregator.second.timeslice.interval";
 
-  public static final String CLUSTER_CACHE_AGGREGATOR_TIMESLICE_INTERVAL =
-    "timeline.metrics.cluster.cache.aggregator.second.timeslice.interval";
-
   public static final String AGGREGATOR_CHECKPOINT_DELAY =
     "timeline.metrics.service.checkpointDelay";
 

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

@@ -50,7 +50,6 @@ import java.net.URISyntaxException;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collection;
-import java.util.Date;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
@@ -60,12 +59,11 @@ import java.util.concurrent.locks.Lock;
 
 import static java.util.concurrent.TimeUnit.SECONDS;
 import static org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline.TimelineMetricConfiguration.CLUSTER_AGGREGATOR_SECOND_SLEEP_INTERVAL;
-import static org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline.TimelineMetricConfiguration.CLUSTER_CACHE_AGGREGATOR_TIMESLICE_INTERVAL;
+import static org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline.TimelineMetricConfiguration.CLUSTER_AGGREGATOR_TIMESLICE_INTERVAL;
 import static org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline.TimelineMetricConfiguration.HOST_APP_ID;
 import static org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline.TimelineMetricConfiguration.TIMELINE_METRICS_CLUSTER_AGGREGATOR_INTERPOLATION_ENABLED;
 import static org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline.TimelineMetricConfiguration.TIMELINE_METRICS_COLLECTOR_IGNITE_BACKUPS;
 import static org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline.TimelineMetricConfiguration.TIMELINE_METRICS_COLLECTOR_IGNITE_NODES;
-import static org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline.TimelineMetricConfiguration.TIMELINE_METRICS_SINK_COLLECTION_PERIOD;
 import static org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline.TimelineMetricConfiguration.TIMELINE_METRIC_AGGREGATION_SQL_FILTERS;
 import static org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline.TimelineMetricConfiguration.TIMELINE_SERVICE_HTTP_POLICY;
 import static org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline.aggregators.AggregatorUtils.getRoundedCheckPointTimeMillis;
@@ -77,7 +75,6 @@ public class TimelineMetricsIgniteCache implements TimelineMetricDistributedCach
       LogFactory.getLog(TimelineMetricsIgniteCache.class);
   private IgniteCache<TimelineClusterMetric, MetricClusterAggregate> igniteCache;
   private long cacheSliceIntervalMillis;
-  private int collectionPeriodMillis;
   private boolean interpolationEnabled;
   private List<String> skipAggrPatternStrings = new ArrayList<>();
   private List<String> appIdsToAggregate;
@@ -110,8 +107,7 @@ public class TimelineMetricsIgniteCache implements TimelineMetricDistributedCach
     //aggregation parameters
     appIdsToAggregate = timelineMetricConfiguration.getAppIdsForHostAggregation();
     interpolationEnabled = Boolean.parseBoolean(metricConf.get(TIMELINE_METRICS_CLUSTER_AGGREGATOR_INTERPOLATION_ENABLED, "true"));
-    collectionPeriodMillis = (int) SECONDS.toMillis(metricConf.getInt(TIMELINE_METRICS_SINK_COLLECTION_PERIOD, 10));
-    cacheSliceIntervalMillis = SECONDS.toMillis(metricConf.getInt(CLUSTER_CACHE_AGGREGATOR_TIMESLICE_INTERVAL, 30));
+    cacheSliceIntervalMillis = SECONDS.toMillis(metricConf.getInt(CLUSTER_AGGREGATOR_TIMESLICE_INTERVAL, 30));
     Long aggregationInterval = metricConf.getLong(CLUSTER_AGGREGATOR_SECOND_SLEEP_INTERVAL, 120L);
 
     String filteredMetricPatterns = metricConf.get(TIMELINE_METRIC_AGGREGATION_SQL_FILTERS);
@@ -215,12 +211,6 @@ public class TimelineMetricsIgniteCache implements TimelineMetricDistributedCach
 
       if (slicedClusterMetrics != null) {
         for (Map.Entry<TimelineClusterMetric, Double> metricDoubleEntry : slicedClusterMetrics.entrySet()) {
-          if (metricDoubleEntry.getKey().getTimestamp() == timeSlices.get(timeSlices.size()-1)[1] && metricDoubleEntry.getKey().getTimestamp() - metric.getMetricValues().lastKey() > collectionPeriodMillis) {
-            if(LOG.isDebugEnabled()) {
-              LOG.debug("Last skipped timestamp @ " + new Date(metric.getMetricValues().lastKey()) + " slice timestamp @ " + new Date(metricDoubleEntry.getKey().getTimestamp()));
-            }
-            continue;
-          }
           MetricClusterAggregate newMetricClusterAggregate  = new MetricClusterAggregate(
               metricDoubleEntry.getValue(), 1, null, metricDoubleEntry.getValue(), metricDoubleEntry.getValue());
           //put app metric into cache

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

@@ -223,7 +223,7 @@ public class AggregatorUtils {
    */
   public static Long getSliceTimeForMetric(List<Long[]> timeSlices, Long timestamp) {
     for (Long[] timeSlice : timeSlices) {
-      if (timestamp > timeSlice[0] && timestamp <= timeSlice[1]) {
+      if (timestamp >= timeSlice[0] && timestamp < timeSlice[1]) {
         return timeSlice[1];
       }
     }

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

@@ -41,7 +41,6 @@ import static org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.ti
 import static org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline.TimelineMetricConfiguration.CLUSTER_AGGREGATOR_SECOND_DISABLED;
 import static org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline.TimelineMetricConfiguration.CLUSTER_AGGREGATOR_SECOND_SLEEP_INTERVAL;
 import static org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline.TimelineMetricConfiguration.CLUSTER_AGGREGATOR_TIMESLICE_INTERVAL;
-import static org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline.TimelineMetricConfiguration.CLUSTER_CACHE_AGGREGATOR_TIMESLICE_INTERVAL;
 import static org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline.TimelineMetricConfiguration.DEFAULT_CHECKPOINT_LOCATION;
 import static org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline.TimelineMetricConfiguration.HOST_AGGREGATOR_DAILY_CHECKPOINT_CUTOFF_MULTIPLIER;
 import static org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline.TimelineMetricConfiguration.HOST_AGGREGATOR_DAILY_DISABLED;
@@ -273,9 +272,6 @@ public class TimelineMetricAggregatorFactory {
     long timeSliceIntervalMillis = SECONDS.toMillis(metricsConf.getInt
       (CLUSTER_AGGREGATOR_TIMESLICE_INTERVAL, 30));
 
-    long cacheTimeSliceIntervalMillis = SECONDS.toMillis(metricsConf.getInt
-      (CLUSTER_CACHE_AGGREGATOR_TIMESLICE_INTERVAL, 30));
-
     int checkpointCutOffMultiplier =
       metricsConf.getInt(CLUSTER_AGGREGATOR_SECOND_CHECKPOINT_CUTOFF_MULTIPLIER, 2);
 
@@ -297,8 +293,7 @@ public class TimelineMetricAggregatorFactory {
         120000l,
         timeSliceIntervalMillis,
         haController,
-        distributedCache,
-        cacheTimeSliceIntervalMillis
+        distributedCache
       );
     }
 

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

@@ -31,19 +31,16 @@ import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 
-import static org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline.aggregators.AggregatorUtils.getSliceTimeForMetric;
 import static org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline.aggregators.AggregatorUtils.getTimeSlices;
 
 public class TimelineMetricClusterAggregatorSecondWithCacheSource extends TimelineMetricClusterAggregatorSecond {
   private TimelineMetricDistributedCache distributedCache;
-  private Long cacheTimeSliceIntervalMillis;
   public TimelineMetricClusterAggregatorSecondWithCacheSource(AggregationTaskRunner.AGGREGATOR_NAME metricAggregateSecond, TimelineMetricMetadataManager metricMetadataManager, PhoenixHBaseAccessor hBaseAccessor, Configuration metricsConf, String checkpointLocation, long sleepIntervalMillis, int checkpointCutOffMultiplier, String aggregatorDisabledParam, String inputTableName, String outputTableName,
                                                               Long nativeTimeRangeDelay,
                                                               Long timeSliceInterval,
-                                                              MetricCollectorHAController haController, TimelineMetricDistributedCache distributedCache, Long cacheTimeSliceIntervalMillis) {
+                                                              MetricCollectorHAController haController, TimelineMetricDistributedCache distributedCache) {
     super(metricAggregateSecond, metricMetadataManager, hBaseAccessor, metricsConf, checkpointLocation, sleepIntervalMillis, checkpointCutOffMultiplier, aggregatorDisabledParam, inputTableName, outputTableName, nativeTimeRangeDelay, timeSliceInterval, haController);
     this.distributedCache = distributedCache;
-    this.cacheTimeSliceIntervalMillis = cacheTimeSliceIntervalMillis;
   }
 
   @Override
@@ -81,36 +78,11 @@ public class TimelineMetricClusterAggregatorSecondWithCacheSource extends Timeli
 
   //Slices in cache could be different from aggregate slices, so need to recalculate. Counts hosted apps
   Map<TimelineClusterMetric, MetricClusterAggregate> aggregateMetricsFromMetricClusterAggregates(Map<TimelineClusterMetric, MetricClusterAggregate> metricsFromCache, List<Long[]> timeSlices) {
-    Map<TimelineClusterMetric, MetricClusterAggregate> result = new HashMap<>();
-
-    //normalize if slices in cache are different from the aggregation slices
-    //TODO add basic interpolation, current implementation assumes that cacheTimeSliceIntervalMillis <= timeSliceIntervalMillis
-    if (cacheTimeSliceIntervalMillis.equals(timeSliceIntervalMillis)) {
-      result = metricsFromCache;
-    } else {
-      for (Map.Entry<TimelineClusterMetric, MetricClusterAggregate> clusterMetricAggregateEntry : metricsFromCache.entrySet()) {
-        Long timestamp = getSliceTimeForMetric(timeSlices, clusterMetricAggregateEntry.getKey().getTimestamp());
-        if (timestamp <= 0) {
-          LOG.warn("Entry doesn't match any slice. Slices : " + timeSlices + " metric timestamp : " + clusterMetricAggregateEntry.getKey().getTimestamp());
-          continue;
-        }
-        TimelineClusterMetric timelineClusterMetric = new TimelineClusterMetric(clusterMetricAggregateEntry.getKey().getMetricName(), clusterMetricAggregateEntry.getKey().getAppId(), clusterMetricAggregateEntry.getKey().getInstanceId(), timestamp);
-        if (result.containsKey(timelineClusterMetric)) {
-          MetricClusterAggregate metricClusterAggregate = result.get(timelineClusterMetric);
-          metricClusterAggregate.updateMax(clusterMetricAggregateEntry.getValue().getMax());
-          metricClusterAggregate.updateMin(clusterMetricAggregateEntry.getValue().getMin());
-          metricClusterAggregate.setSum((metricClusterAggregate.getSum() + clusterMetricAggregateEntry.getValue().getSum()) / 2D);
-          metricClusterAggregate.setNumberOfHosts(Math.max(metricClusterAggregate.getNumberOfHosts(), clusterMetricAggregateEntry.getValue().getNumberOfHosts()));
-        } else {
-          result.put(timelineClusterMetric, clusterMetricAggregateEntry.getValue());
-        }
-      }
-    }
-
+    //TODO add basic interpolation
     //TODO investigate if needed, maybe add config to disable/enable
     //count hosted apps
     Map<String, MutableInt> hostedAppCounter = new HashMap<>();
-    for (Map.Entry<TimelineClusterMetric, MetricClusterAggregate> clusterMetricAggregateEntry : result.entrySet()) {
+    for (Map.Entry<TimelineClusterMetric, MetricClusterAggregate> clusterMetricAggregateEntry : metricsFromCache.entrySet()) {
       int numHosts = clusterMetricAggregateEntry.getValue().getNumberOfHosts();
       String appId = clusterMetricAggregateEntry.getKey().getAppId();
       if (!hostedAppCounter.containsKey(appId)) {
@@ -124,9 +96,9 @@ public class TimelineMetricClusterAggregatorSecondWithCacheSource extends Timeli
     }
 
     // Add liveHosts per AppId metrics.
-    processLiveAppCountMetrics(result, hostedAppCounter, timeSlices.get(timeSlices.size() - 1)[1]);
+    processLiveAppCountMetrics(metricsFromCache, hostedAppCounter, timeSlices.get(timeSlices.size() - 1)[1]);
 
-    return result;
+    return metricsFromCache;
   }
 
 }

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

@@ -167,62 +167,6 @@ public class TimelineMetricsIgniteCacheTest {
     metricValues.clear();
     timelineMetrics.clear();
 
-    /*
-
-    0      +30s    +60s    +90s
-    |       |       |       |
-     (1)      (2)                h1
-                (3)       (4)    h2
-                 (5)      (6)    h1
-
-    */
-    // Case 3 : merging host data points, ignore (2) for h1 as it will conflict with (5), two hosts.
-    metricValues = new TreeMap<>();
-    metricValues.put(startTime + 15*seconds, 1.0);
-    metricValues.put(startTime + 45*seconds, 2.0);
-    timelineMetric = new TimelineMetric("metric1", "host1", "app1", "instance1");
-    timelineMetric.setMetricValues(metricValues);
-    timelineMetrics.add(timelineMetric);
-
-    metricValues = new TreeMap<>();
-    metricValues.put(startTime + 45*seconds, 3.0);
-    metricValues.put(startTime + 85*seconds, 4.0);
-    timelineMetric = new TimelineMetric("metric1", "host2", "app1", "instance1");
-    timelineMetric.setMetricValues(metricValues);
-    timelineMetrics.add(timelineMetric);
-
-    metricValues = new TreeMap<>();
-    metricValues.put(startTime + 55*seconds, 5.0);
-    metricValues.put(startTime + 85*seconds, 6.0);
-    timelineMetric = new TimelineMetric("metric1", "host1", "app1", "instance1");
-    timelineMetric.setMetricValues(metricValues);
-    timelineMetrics.add(timelineMetric);
-
-    timelineMetricsIgniteCache.putMetrics(timelineMetrics, metricMetadataManagerMock);
-
-    aggregateMap = timelineMetricsIgniteCache.evictMetricAggregates(startTime, startTime + 120*seconds);
-
-    Assert.assertEquals(aggregateMap.size(), 3);
-    timelineClusterMetric = new TimelineClusterMetric(timelineMetric.getMetricName(),
-      timelineMetric.getAppId(), timelineMetric.getInstanceId(), startTime + 30*seconds);
-
-    Assert.assertTrue(aggregateMap.containsKey(timelineClusterMetric));
-    Assert.assertEquals(1.0, aggregateMap.get(timelineClusterMetric).getSum());
-    Assert.assertEquals(1, aggregateMap.get(timelineClusterMetric).getNumberOfHosts());
-
-    timelineClusterMetric.setTimestamp(startTime + 2*30*seconds);
-    Assert.assertTrue(aggregateMap.containsKey(timelineClusterMetric));
-    Assert.assertEquals(8.0, aggregateMap.get(timelineClusterMetric).getSum());
-    Assert.assertEquals(2, aggregateMap.get(timelineClusterMetric).getNumberOfHosts());
-
-    timelineClusterMetric.setTimestamp(startTime + 3*30*seconds);
-    Assert.assertTrue(aggregateMap.containsKey(timelineClusterMetric));
-    Assert.assertEquals(10.0, aggregateMap.get(timelineClusterMetric).getSum());
-    Assert.assertEquals(2, aggregateMap.get(timelineClusterMetric).getNumberOfHosts());
-
-    metricValues.clear();
-    timelineMetrics.clear();
-
     Assert.assertEquals(0d, timelineMetricsIgniteCache.getPointInTimeCacheMetrics().get("Cluster_KeySize"));
   }
 

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

@@ -79,7 +79,7 @@ public class TimelineMetricClusterAggregatorSecondWithCacheSourceTest {
     TimelineMetricClusterAggregatorSecondWithCacheSource secondAggregator = new TimelineMetricClusterAggregatorSecondWithCacheSource(
         METRIC_AGGREGATE_SECOND, metricMetadataManagerMock, null, configuration, null,
         aggregatorInterval, 2, "false", "", "", aggregatorInterval,
-        sliceInterval, null, timelineMetricsIgniteCache, 30L);
+        sliceInterval, null, timelineMetricsIgniteCache);
 
     long now = System.currentTimeMillis();
     long startTime = now - 120000;
@@ -112,67 +112,4 @@ public class TimelineMetricClusterAggregatorSecondWithCacheSourceTest {
     Assert.assertEquals(2d, a1.getSum());
     Assert.assertEquals(5d, a2.getSum());
   }
-
-  @Test
-  public void testSlicesRecalculation() throws Exception {
-    long aggregatorInterval = 120000;
-    long sliceInterval = 30000;
-
-    Configuration configuration = new Configuration();
-
-    TimelineMetricMetadataManager metricMetadataManagerMock = createNiceMock(TimelineMetricMetadataManager.class);
-    expect(metricMetadataManagerMock.getMetadataCacheValue((TimelineMetricMetadataKey) anyObject())).andReturn(null).anyTimes();
-    replay(metricMetadataManagerMock);
-
-    TimelineMetricClusterAggregatorSecondWithCacheSource secondAggregator = new TimelineMetricClusterAggregatorSecondWithCacheSource(
-        METRIC_AGGREGATE_SECOND, metricMetadataManagerMock, null, configuration, null,
-        aggregatorInterval, 2, "false", "", "", aggregatorInterval,
-        sliceInterval, null, timelineMetricsIgniteCache, 30L);
-
-    long seconds = 1000;
-    long now = getRoundedCheckPointTimeMillis(System.currentTimeMillis(), 120*seconds);
-    long startTime = now - 120*seconds;
-
-    Map<TimelineClusterMetric, MetricClusterAggregate> metricsFromCache = new HashMap<>();
-    metricsFromCache.put(new TimelineClusterMetric("m1", "a1", "i1",startTime + 5 * seconds),
-        new MetricClusterAggregate(1.0, 2, 1.0, 1.0, 1.0));
-    metricsFromCache.put(new TimelineClusterMetric("m1", "a1", "i1",startTime + 25 * seconds),
-        new MetricClusterAggregate(2.0, 2, 1.0, 2.0, 2.0));
-    metricsFromCache.put(new TimelineClusterMetric("m1", "a1", "i1",startTime + 45 * seconds),
-        new MetricClusterAggregate(3.0, 2, 1.0, 1.0, 1.0));
-    metricsFromCache.put(new TimelineClusterMetric("m1", "a1", "i1",startTime + 65 * seconds),
-        new MetricClusterAggregate(4.0, 2, 1.0, 4.0, 4.0));
-    metricsFromCache.put(new TimelineClusterMetric("m1", "a1", "i1",startTime + 85 * seconds),
-        new MetricClusterAggregate(5.0, 2, 1.0, 5.0, 5.0));
-
-    List<Long[]> timeslices = getTimeSlices(startTime, startTime + 120*seconds, 30*seconds);
-
-    Map<TimelineClusterMetric, MetricClusterAggregate> aggregates = secondAggregator.aggregateMetricsFromMetricClusterAggregates(metricsFromCache, timeslices);
-
-    Assert.assertNotNull(aggregates);
-    Assert.assertEquals(4, aggregates.size());
-
-    TimelineClusterMetric timelineClusterMetric = new TimelineClusterMetric("m1", "a1", "i1", startTime + 30*seconds);
-    MetricClusterAggregate metricClusterAggregate = aggregates.get(timelineClusterMetric);
-    Assert.assertNotNull(metricClusterAggregate);
-    Assert.assertEquals(1.5, metricClusterAggregate.getSum());
-    Assert.assertEquals(1d, metricClusterAggregate.getMin());
-    Assert.assertEquals(2d, metricClusterAggregate.getMax());
-    Assert.assertEquals(2, metricClusterAggregate.getNumberOfHosts());
-
-    timelineClusterMetric.setTimestamp(startTime + 60*seconds);
-    metricClusterAggregate = aggregates.get(timelineClusterMetric);
-    Assert.assertNotNull(metricClusterAggregate);
-    Assert.assertEquals(3d, metricClusterAggregate.getSum());
-
-    timelineClusterMetric.setTimestamp(startTime + 90*seconds);
-    metricClusterAggregate = aggregates.get(timelineClusterMetric);
-    Assert.assertNotNull(metricClusterAggregate);
-    Assert.assertEquals(4.5d, metricClusterAggregate.getSum());
-
-    timelineClusterMetric = new TimelineClusterMetric("live_hosts", "a1", null, startTime + 120*seconds);
-    metricClusterAggregate = aggregates.get(timelineClusterMetric);
-    Assert.assertNotNull(metricClusterAggregate);
-    Assert.assertEquals(2d, metricClusterAggregate.getSum());
-  }
 }