@Test public void testLiveHostCounterMetrics() throws Exception { long aggregatorInterval = 120000; long sliceInterval = 30000; Configuration configuration = new Configuration(); TimelineMetricClusterAggregatorSecondWithCacheSource secondAggregator = new TimelineMetricClusterAggregatorSecondWithCacheSource( METRIC_AGGREGATE_SECOND, metricMetadataManagerMock, null, configuration, null, aggregatorInterval, 2, "false", "", "", aggregatorInterval, sliceInterval, null, timelineMetricsIgniteCache); long now = System.currentTimeMillis(); long startTime = now - 120000; long seconds = 1000; Map<TimelineClusterMetric, MetricClusterAggregate> metricsFromCache = new HashMap<>(); metricsFromCache.put(new TimelineClusterMetric("m1", "a1", "i1",startTime + 15 * seconds), new MetricClusterAggregate(1.0, 2, 1.0, 1.0, 1.0)); metricsFromCache.put(new TimelineClusterMetric("m2", "a2", "i1",startTime + 18 * seconds), new MetricClusterAggregate(1.0, 5, 1.0, 1.0, 1.0)); List<Long[]> timeslices = getTimeSlices(startTime, startTime + 120*seconds, 30*seconds); Map<TimelineClusterMetric, MetricClusterAggregate> aggregates = secondAggregator.aggregateMetricsFromMetricClusterAggregates(metricsFromCache, timeslices); Assert.assertNotNull(aggregates); MetricClusterAggregate a1 = null, a2 = null; for (Map.Entry<TimelineClusterMetric, MetricClusterAggregate> m : aggregates.entrySet()) { if (m.getKey().getMetricName().equals("live_hosts") && m.getKey().getAppId().equals("a1")) { a1 = m.getValue(); } if (m.getKey().getMetricName().equals("live_hosts") && m.getKey().getAppId().equals("a2")) { a2 = m.getValue(); } } Assert.assertNotNull(a1); Assert.assertNotNull(a2); Assert.assertEquals(2d, a1.getSum()); Assert.assertEquals(5d, a2.getSum()); }
Map<TimelineClusterMetric, MetricClusterAggregate> aggregateMetricsFromMetricClusterAggregates(Map<TimelineClusterMetric, MetricClusterAggregate> metricsFromCache, List<Long[]> timeSlices) { Map<String, MutableInt> hostedAppCounter = new HashMap<>(); for (Map.Entry<TimelineClusterMetric, MetricClusterAggregate> clusterMetricAggregateEntry : metricsFromCache.entrySet()) { int numHosts = clusterMetricAggregateEntry.getValue().getNumberOfHosts(); String appId = clusterMetricAggregateEntry.getKey().getAppId(); if (!hostedAppCounter.containsKey(appId)) { hostedAppCounter.put(appId, new MutableInt(numHosts)); } else { int currentHostCount = hostedAppCounter.get(appId).intValue(); if (currentHostCount < numHosts) { hostedAppCounter.put(appId, new MutableInt(numHosts)); } } } processLiveAppCountMetrics(metricsFromCache, hostedAppCounter, timeSlices.get(timeSlices.size() - 1)[1]); return metricsFromCache; }
TimelineMetricClusterAggregatorSecondWithCacheSource extends TimelineMetricClusterAggregatorSecond { Map<TimelineClusterMetric, MetricClusterAggregate> aggregateMetricsFromMetricClusterAggregates(Map<TimelineClusterMetric, MetricClusterAggregate> metricsFromCache, List<Long[]> timeSlices) { Map<String, MutableInt> hostedAppCounter = new HashMap<>(); for (Map.Entry<TimelineClusterMetric, MetricClusterAggregate> clusterMetricAggregateEntry : metricsFromCache.entrySet()) { int numHosts = clusterMetricAggregateEntry.getValue().getNumberOfHosts(); String appId = clusterMetricAggregateEntry.getKey().getAppId(); if (!hostedAppCounter.containsKey(appId)) { hostedAppCounter.put(appId, new MutableInt(numHosts)); } else { int currentHostCount = hostedAppCounter.get(appId).intValue(); if (currentHostCount < numHosts) { hostedAppCounter.put(appId, new MutableInt(numHosts)); } } } processLiveAppCountMetrics(metricsFromCache, hostedAppCounter, timeSlices.get(timeSlices.size() - 1)[1]); return metricsFromCache; } }
TimelineMetricClusterAggregatorSecondWithCacheSource extends TimelineMetricClusterAggregatorSecond { Map<TimelineClusterMetric, MetricClusterAggregate> aggregateMetricsFromMetricClusterAggregates(Map<TimelineClusterMetric, MetricClusterAggregate> metricsFromCache, List<Long[]> timeSlices) { Map<String, MutableInt> hostedAppCounter = new HashMap<>(); for (Map.Entry<TimelineClusterMetric, MetricClusterAggregate> clusterMetricAggregateEntry : metricsFromCache.entrySet()) { int numHosts = clusterMetricAggregateEntry.getValue().getNumberOfHosts(); String appId = clusterMetricAggregateEntry.getKey().getAppId(); if (!hostedAppCounter.containsKey(appId)) { hostedAppCounter.put(appId, new MutableInt(numHosts)); } else { int currentHostCount = hostedAppCounter.get(appId).intValue(); if (currentHostCount < numHosts) { hostedAppCounter.put(appId, new MutableInt(numHosts)); } } } processLiveAppCountMetrics(metricsFromCache, hostedAppCounter, timeSlices.get(timeSlices.size() - 1)[1]); return metricsFromCache; } 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); }
TimelineMetricClusterAggregatorSecondWithCacheSource extends TimelineMetricClusterAggregatorSecond { Map<TimelineClusterMetric, MetricClusterAggregate> aggregateMetricsFromMetricClusterAggregates(Map<TimelineClusterMetric, MetricClusterAggregate> metricsFromCache, List<Long[]> timeSlices) { Map<String, MutableInt> hostedAppCounter = new HashMap<>(); for (Map.Entry<TimelineClusterMetric, MetricClusterAggregate> clusterMetricAggregateEntry : metricsFromCache.entrySet()) { int numHosts = clusterMetricAggregateEntry.getValue().getNumberOfHosts(); String appId = clusterMetricAggregateEntry.getKey().getAppId(); if (!hostedAppCounter.containsKey(appId)) { hostedAppCounter.put(appId, new MutableInt(numHosts)); } else { int currentHostCount = hostedAppCounter.get(appId).intValue(); if (currentHostCount < numHosts) { hostedAppCounter.put(appId, new MutableInt(numHosts)); } } } processLiveAppCountMetrics(metricsFromCache, hostedAppCounter, timeSlices.get(timeSlices.size() - 1)[1]); return metricsFromCache; } 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); @Override boolean doWork(long startTime, long endTime); }
TimelineMetricClusterAggregatorSecondWithCacheSource extends TimelineMetricClusterAggregatorSecond { Map<TimelineClusterMetric, MetricClusterAggregate> aggregateMetricsFromMetricClusterAggregates(Map<TimelineClusterMetric, MetricClusterAggregate> metricsFromCache, List<Long[]> timeSlices) { Map<String, MutableInt> hostedAppCounter = new HashMap<>(); for (Map.Entry<TimelineClusterMetric, MetricClusterAggregate> clusterMetricAggregateEntry : metricsFromCache.entrySet()) { int numHosts = clusterMetricAggregateEntry.getValue().getNumberOfHosts(); String appId = clusterMetricAggregateEntry.getKey().getAppId(); if (!hostedAppCounter.containsKey(appId)) { hostedAppCounter.put(appId, new MutableInt(numHosts)); } else { int currentHostCount = hostedAppCounter.get(appId).intValue(); if (currentHostCount < numHosts) { hostedAppCounter.put(appId, new MutableInt(numHosts)); } } } processLiveAppCountMetrics(metricsFromCache, hostedAppCounter, timeSlices.get(timeSlices.size() - 1)[1]); return metricsFromCache; } 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); @Override boolean doWork(long startTime, long endTime);  }