@Test public void testShouldAggregateProperly() { long aggregatorInterval = 120000l; long sliceInterval = 30000l; Configuration configuration = new Configuration(); TimelineMetricMetadataManager metricMetadataManagerMock = createNiceMock(TimelineMetricMetadataManager.class); expect(metricMetadataManagerMock.getMetadataCacheValue((TimelineMetricMetadataKey) anyObject())) .andReturn(null).anyTimes(); replay(metricMetadataManagerMock); TimelineMetricClusterAggregatorSecond secondAggregator = new TimelineMetricClusterAggregatorSecond( METRIC_AGGREGATE_SECOND, metricMetadataManagerMock, null, configuration, null, aggregatorInterval, 2, "false", "", "", aggregatorInterval, sliceInterval, null ); long startTime = getRoundedCheckPointTimeMillis(System.currentTimeMillis(),aggregatorInterval); List<Long[]> timeslices = getTimeSlices(startTime, startTime + aggregatorInterval, sliceInterval); Map<TimelineClusterMetric, MetricClusterAggregate> aggregateClusterMetrics = new HashMap<>(); long seconds = 1000; TimelineMetric timelineMetric = new TimelineMetric(); timelineMetric.setMetricName("m1"); timelineMetric.setHostName("h1"); timelineMetric.setAppId("a1"); timelineMetric.setType("GUAGE"); timelineMetric.setStartTime(startTime); TreeMap<Long, Double> metricValues = new TreeMap<Long, Double>(); metricValues.put(startTime + 15*seconds, 1.0); metricValues.put(startTime + 45*seconds, 2.0); metricValues.put(startTime + 75*seconds, 3.0); metricValues.put(startTime + 105*seconds, 4.0); metricValues.put(startTime + 135*seconds, 5.0); metricValues.put(startTime + 165*seconds, 6.0); timelineMetric.setMetricValues(metricValues); secondAggregator.processAggregateClusterMetrics(aggregateClusterMetrics, timelineMetric, timeslices); Assert.assertEquals(aggregateClusterMetrics.size(), 4); TimelineClusterMetric timelineClusterMetric = new TimelineClusterMetric(timelineMetric.getMetricName(), timelineMetric.getAppId(), timelineMetric.getInstanceId(), startTime + 30*seconds); Assert.assertTrue(aggregateClusterMetrics.containsKey(timelineClusterMetric)); Assert.assertEquals(aggregateClusterMetrics.get(timelineClusterMetric).getSum(), 1.0); timelineClusterMetric.setTimestamp(startTime + 4*30*seconds); Assert.assertTrue(aggregateClusterMetrics.containsKey(timelineClusterMetric)); Assert.assertEquals(aggregateClusterMetrics.get(timelineClusterMetric).getSum(),4.0); metricValues.clear(); aggregateClusterMetrics.clear(); metricValues.put(startTime + 15*seconds, 1.0); metricValues.put(startTime + 75*seconds, 3.0); metricValues.put(startTime + 105*seconds, 4.0); metricValues.put(startTime + 135*seconds, 5.0); metricValues.put(startTime + 165*seconds, 6.0); timelineMetric.setMetricValues(metricValues); secondAggregator.processAggregateClusterMetrics(aggregateClusterMetrics, timelineMetric, timeslices); Assert.assertEquals(aggregateClusterMetrics.size(), 4); timelineClusterMetric.setTimestamp(startTime + 60*seconds); Assert.assertTrue(aggregateClusterMetrics.containsKey(timelineClusterMetric)); Assert.assertEquals(aggregateClusterMetrics.get(timelineClusterMetric).getSum(), 2.0); metricValues.clear(); aggregateClusterMetrics.clear(); metricValues.put(startTime + 15*seconds, 1.0); metricValues.put(startTime + 45*seconds, 2.0); metricValues.put(startTime + 75*seconds, 3.0); metricValues.put(startTime + 135*seconds, 5.0); metricValues.put(startTime + 165*seconds, 6.0); timelineMetric.setMetricValues(metricValues); timelineMetric.setType("GUAGE"); secondAggregator.processAggregateClusterMetrics(aggregateClusterMetrics, timelineMetric, timeslices); Assert.assertEquals(aggregateClusterMetrics.size(), 4); timelineClusterMetric.setTimestamp(startTime + 120*seconds); Assert.assertTrue(aggregateClusterMetrics.containsKey(timelineClusterMetric)); Assert.assertEquals(aggregateClusterMetrics.get(timelineClusterMetric).getSum(), 3.0); aggregateClusterMetrics.clear(); timelineMetric.setType("COUNTER"); secondAggregator.processAggregateClusterMetrics(aggregateClusterMetrics, timelineMetric, timeslices); Assert.assertEquals(aggregateClusterMetrics.size(), 4); timelineClusterMetric.setTimestamp(startTime + 120*seconds); Assert.assertTrue(aggregateClusterMetrics.containsKey(timelineClusterMetric)); Assert.assertEquals(aggregateClusterMetrics.get(timelineClusterMetric).getSum(), 4.5); metricValues.clear(); aggregateClusterMetrics.clear(); metricValues.put(startTime + 45*seconds, 2.0); metricValues.put(startTime + 75*seconds, 3.0); metricValues.put(startTime + 105*seconds, 4.0); metricValues.put(startTime + 135*seconds, 5.0); metricValues.put(startTime + 165*seconds, 6.0); timelineMetric.setMetricValues(metricValues); timelineMetric.setType("GUAGE"); secondAggregator.processAggregateClusterMetrics(aggregateClusterMetrics, timelineMetric, timeslices); Assert.assertEquals(aggregateClusterMetrics.size(), 4); timelineClusterMetric.setTimestamp(startTime + 30*seconds); Assert.assertTrue(aggregateClusterMetrics.containsKey(timelineClusterMetric)); Assert.assertEquals(aggregateClusterMetrics.get(timelineClusterMetric).getSum(), 2.0); aggregateClusterMetrics.clear(); timelineMetric.setType("COUNTER"); int liveHosts = secondAggregator.processAggregateClusterMetrics(aggregateClusterMetrics, timelineMetric, timeslices); Assert.assertEquals(liveHosts, 1); Assert.assertEquals(aggregateClusterMetrics.size(), 4); timelineClusterMetric.setTimestamp(startTime + 30*seconds); Assert.assertTrue(aggregateClusterMetrics.containsKey(timelineClusterMetric)); Assert.assertEquals(aggregateClusterMetrics.get(timelineClusterMetric).getSum(), 1.5); metricValues.clear(); aggregateClusterMetrics.clear(); metricValues.put(startTime + 135*seconds, 5.0); metricValues.put(startTime + 165*seconds, 6.0); timelineMetric.setMetricValues(metricValues); timelineMetric.setType("GUAGE"); secondAggregator.processAggregateClusterMetrics(aggregateClusterMetrics, timelineMetric, timeslices); Assert.assertEquals(aggregateClusterMetrics.size(), 4); timelineClusterMetric.setTimestamp(startTime + 30*seconds); Assert.assertTrue(aggregateClusterMetrics.containsKey(timelineClusterMetric)); Assert.assertEquals(aggregateClusterMetrics.get(timelineClusterMetric).getSum(), 5.0); aggregateClusterMetrics.clear(); timelineMetric.setType("COUNTER"); liveHosts = secondAggregator.processAggregateClusterMetrics(aggregateClusterMetrics, timelineMetric, timeslices); Assert.assertEquals(liveHosts, 1); Assert.assertEquals(aggregateClusterMetrics.size(), 4); timelineClusterMetric.setTimestamp(startTime + 60*seconds); Assert.assertTrue(aggregateClusterMetrics.containsKey(timelineClusterMetric)); Assert.assertEquals(aggregateClusterMetrics.get(timelineClusterMetric).getSum(), 2.5); metricValues.clear(); aggregateClusterMetrics.clear(); }
protected int processAggregateClusterMetrics(Map<TimelineClusterMetric, MetricClusterAggregate> aggregateClusterMetrics, TimelineMetric metric, List<Long[]> timeSlices) { TimelineMetricMetadataKey appKey = new TimelineMetricMetadataKey(metric.getMetricName(), metric.getAppId(), metric.getInstanceId()); TimelineMetricMetadata metricMetadata = metadataManagerInstance.getMetadataCacheValue(appKey); if (metricMetadata != null && !metricMetadata.isSupportsAggregates()) { LOG.debug("Skipping cluster aggregation for " + metric.getMetricName()); return 0; } boolean skipInterpolationForMetric = shouldInterpolationBeSkipped(metric.getMetricName()); Map<TimelineClusterMetric, Double> clusterMetrics = sliceFromTimelineMetric(metric, timeSlices, !skipInterpolationForMetric && interpolationEnabled); return aggregateClusterMetricsFromSlices(clusterMetrics, aggregateClusterMetrics, metric.getHostName()); }
TimelineMetricClusterAggregatorSecond extends AbstractTimelineAggregator { protected int processAggregateClusterMetrics(Map<TimelineClusterMetric, MetricClusterAggregate> aggregateClusterMetrics, TimelineMetric metric, List<Long[]> timeSlices) { TimelineMetricMetadataKey appKey = new TimelineMetricMetadataKey(metric.getMetricName(), metric.getAppId(), metric.getInstanceId()); TimelineMetricMetadata metricMetadata = metadataManagerInstance.getMetadataCacheValue(appKey); if (metricMetadata != null && !metricMetadata.isSupportsAggregates()) { LOG.debug("Skipping cluster aggregation for " + metric.getMetricName()); return 0; } boolean skipInterpolationForMetric = shouldInterpolationBeSkipped(metric.getMetricName()); Map<TimelineClusterMetric, Double> clusterMetrics = sliceFromTimelineMetric(metric, timeSlices, !skipInterpolationForMetric && interpolationEnabled); return aggregateClusterMetricsFromSlices(clusterMetrics, aggregateClusterMetrics, metric.getHostName()); } }
TimelineMetricClusterAggregatorSecond extends AbstractTimelineAggregator { protected int processAggregateClusterMetrics(Map<TimelineClusterMetric, MetricClusterAggregate> aggregateClusterMetrics, TimelineMetric metric, List<Long[]> timeSlices) { TimelineMetricMetadataKey appKey = new TimelineMetricMetadataKey(metric.getMetricName(), metric.getAppId(), metric.getInstanceId()); TimelineMetricMetadata metricMetadata = metadataManagerInstance.getMetadataCacheValue(appKey); if (metricMetadata != null && !metricMetadata.isSupportsAggregates()) { LOG.debug("Skipping cluster aggregation for " + metric.getMetricName()); return 0; } boolean skipInterpolationForMetric = shouldInterpolationBeSkipped(metric.getMetricName()); Map<TimelineClusterMetric, Double> clusterMetrics = sliceFromTimelineMetric(metric, timeSlices, !skipInterpolationForMetric && interpolationEnabled); return aggregateClusterMetricsFromSlices(clusterMetrics, aggregateClusterMetrics, metric.getHostName()); } TimelineMetricClusterAggregatorSecond(AGGREGATOR_NAME aggregatorName,
                                               TimelineMetricMetadataManager metadataManager,
                                               PhoenixHBaseAccessor hBaseAccessor,
                                               Configuration metricsConf,
                                               String checkpointLocation,
                                               Long sleepIntervalMillis,
                                               Integer checkpointCutOffMultiplier,
                                               String aggregatorDisabledParam,
                                               String tableName,
                                               String outputTableName,
                                               Long nativeTimeRangeDelay,
                                               Long timeSliceInterval,
                                               MetricCollectorHAController haController); }
TimelineMetricClusterAggregatorSecond extends AbstractTimelineAggregator { protected int processAggregateClusterMetrics(Map<TimelineClusterMetric, MetricClusterAggregate> aggregateClusterMetrics, TimelineMetric metric, List<Long[]> timeSlices) { TimelineMetricMetadataKey appKey = new TimelineMetricMetadataKey(metric.getMetricName(), metric.getAppId(), metric.getInstanceId()); TimelineMetricMetadata metricMetadata = metadataManagerInstance.getMetadataCacheValue(appKey); if (metricMetadata != null && !metricMetadata.isSupportsAggregates()) { LOG.debug("Skipping cluster aggregation for " + metric.getMetricName()); return 0; } boolean skipInterpolationForMetric = shouldInterpolationBeSkipped(metric.getMetricName()); Map<TimelineClusterMetric, Double> clusterMetrics = sliceFromTimelineMetric(metric, timeSlices, !skipInterpolationForMetric && interpolationEnabled); return aggregateClusterMetricsFromSlices(clusterMetrics, aggregateClusterMetrics, metric.getHostName()); } TimelineMetricClusterAggregatorSecond(AGGREGATOR_NAME aggregatorName,
                                               TimelineMetricMetadataManager metadataManager,
                                               PhoenixHBaseAccessor hBaseAccessor,
                                               Configuration metricsConf,
                                               String checkpointLocation,
                                               Long sleepIntervalMillis,
                                               Integer checkpointCutOffMultiplier,
                                               String aggregatorDisabledParam,
                                               String tableName,
                                               String outputTableName,
                                               Long nativeTimeRangeDelay,
                                               Long timeSliceInterval,
                                               MetricCollectorHAController haController);  }
TimelineMetricClusterAggregatorSecond extends AbstractTimelineAggregator { protected int processAggregateClusterMetrics(Map<TimelineClusterMetric, MetricClusterAggregate> aggregateClusterMetrics, TimelineMetric metric, List<Long[]> timeSlices) { TimelineMetricMetadataKey appKey = new TimelineMetricMetadataKey(metric.getMetricName(), metric.getAppId(), metric.getInstanceId()); TimelineMetricMetadata metricMetadata = metadataManagerInstance.getMetadataCacheValue(appKey); if (metricMetadata != null && !metricMetadata.isSupportsAggregates()) { LOG.debug("Skipping cluster aggregation for " + metric.getMetricName()); return 0; } boolean skipInterpolationForMetric = shouldInterpolationBeSkipped(metric.getMetricName()); Map<TimelineClusterMetric, Double> clusterMetrics = sliceFromTimelineMetric(metric, timeSlices, !skipInterpolationForMetric && interpolationEnabled); return aggregateClusterMetricsFromSlices(clusterMetrics, aggregateClusterMetrics, metric.getHostName()); } TimelineMetricClusterAggregatorSecond(AGGREGATOR_NAME aggregatorName,
                                               TimelineMetricMetadataManager metadataManager,
                                               PhoenixHBaseAccessor hBaseAccessor,
                                               Configuration metricsConf,
                                               String checkpointLocation,
                                               Long sleepIntervalMillis,
                                               Integer checkpointCutOffMultiplier,
                                               String aggregatorDisabledParam,
                                               String tableName,
                                               String outputTableName,
                                               Long nativeTimeRangeDelay,
                                               Long timeSliceInterval,
                                               MetricCollectorHAController haController);  public Long timeSliceIntervalMillis; }