@Test public void testLiveHostCounterMetrics() 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(); TimelineMetric metric1 = new TimelineMetric("m1", "h1", "a1", null); TimelineMetric metric2 = new TimelineMetric("m2", "h1", "a1", null); TimelineMetric metric3 = new TimelineMetric("m2", "h1", "a2", null); TimelineMetric metric4 = new TimelineMetric("m2", "h2", "a1", null); TimelineMetric metric5 = new TimelineMetric("m2", "h2", "a2", null); TimelineMetric metric6 = new TimelineMetric("m2", "h3", "a2", null); expect(metricMetadataManagerMock.getMetricFromUuid((byte[]) anyObject())). andReturn(metric1).andReturn(metric2).andReturn(metric3). andReturn(metric4).andReturn(metric5).andReturn(metric6); replay(metricMetadataManagerMock); TimelineMetricClusterAggregatorSecond secondAggregator = new TimelineMetricClusterAggregatorSecond( METRIC_AGGREGATE_SECOND, metricMetadataManagerMock, null, configuration, null, aggregatorInterval, 2, "false", "", "", aggregatorInterval, sliceInterval, null); long now = System.currentTimeMillis(); long startTime = now - 120000; long seconds = 1000; List<Long[]> slices = getTimeSlices(startTime, now, sliceInterval); ResultSet rs = createNiceMock(ResultSet.class); TreeMap<Long, Double> metricValues = new TreeMap<>(); 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); expect(rs.next()).andReturn(true).times(6); expect(rs.next()).andReturn(false); expect(rs.getLong("SERVER_TIME")).andReturn(now - 150000).times(6); expect(rs.getLong("START_TIME")).andReturn(now - 150000).times(6); ObjectMapper mapper = new ObjectMapper(); expect(rs.getString("METRICS")).andReturn(mapper.writeValueAsString(metricValues)).times(6); replay(rs); Map<TimelineClusterMetric, MetricClusterAggregate> aggregates = secondAggregator.aggregateMetricsFromResultSet(rs, slices); 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(3d, a2.getSum()); }
Map<TimelineClusterMetric, MetricClusterAggregate> aggregateMetricsFromResultSet(ResultSet rs, List<Long[]> timeSlices) throws SQLException, IOException { Map<TimelineClusterMetric, MetricClusterAggregate> aggregateClusterMetrics = new HashMap<TimelineClusterMetric, MetricClusterAggregate>(); TimelineMetric metric = null; Map<String, MutableInt> hostedAppCounter = new HashMap<>(); if (rs.next()) { metric = timelineMetricReadHelper.getTimelineMetricFromResultSet(rs); while (metric == null && rs.next()) { metric = timelineMetricReadHelper.getTimelineMetricFromResultSet(rs); } while (rs.next()) { TimelineMetric nextMetric = timelineMetricReadHelper.getTimelineMetricFromResultSet(rs); if (nextMetric == null) { continue; } if (metric.equalsExceptTime(nextMetric)) { metric.addMetricValues(nextMetric.getMetricValues()); } else { int numHosts = processAggregateClusterMetrics(aggregateClusterMetrics, metric, timeSlices); if (!hostedAppCounter.containsKey(metric.getAppId())) { hostedAppCounter.put(metric.getAppId(), new MutableInt(numHosts)); } else { int currentHostCount = hostedAppCounter.get(metric.getAppId()).intValue(); if (currentHostCount < numHosts) { hostedAppCounter.put(metric.getAppId(), new MutableInt(numHosts)); } } metric = nextMetric; } } } if (metric != null) { int numHosts = processAggregateClusterMetrics(aggregateClusterMetrics, metric, timeSlices); if (!hostedAppCounter.containsKey(metric.getAppId())) { hostedAppCounter.put(metric.getAppId(), new MutableInt(numHosts)); } else { int currentHostCount = hostedAppCounter.get(metric.getAppId()).intValue(); if (currentHostCount < numHosts) { hostedAppCounter.put(metric.getAppId(), new MutableInt(numHosts)); } } } aggregateClusterMetrics.putAll(appAggregator.getAggregateClusterMetrics()); long timestamp = timeSlices.get(timeSlices.size() - 1)[1]; processLiveAppCountMetrics(aggregateClusterMetrics, hostedAppCounter, timestamp); return aggregateClusterMetrics; }
TimelineMetricClusterAggregatorSecond extends AbstractTimelineAggregator { Map<TimelineClusterMetric, MetricClusterAggregate> aggregateMetricsFromResultSet(ResultSet rs, List<Long[]> timeSlices) throws SQLException, IOException { Map<TimelineClusterMetric, MetricClusterAggregate> aggregateClusterMetrics = new HashMap<TimelineClusterMetric, MetricClusterAggregate>(); TimelineMetric metric = null; Map<String, MutableInt> hostedAppCounter = new HashMap<>(); if (rs.next()) { metric = timelineMetricReadHelper.getTimelineMetricFromResultSet(rs); while (metric == null && rs.next()) { metric = timelineMetricReadHelper.getTimelineMetricFromResultSet(rs); } while (rs.next()) { TimelineMetric nextMetric = timelineMetricReadHelper.getTimelineMetricFromResultSet(rs); if (nextMetric == null) { continue; } if (metric.equalsExceptTime(nextMetric)) { metric.addMetricValues(nextMetric.getMetricValues()); } else { int numHosts = processAggregateClusterMetrics(aggregateClusterMetrics, metric, timeSlices); if (!hostedAppCounter.containsKey(metric.getAppId())) { hostedAppCounter.put(metric.getAppId(), new MutableInt(numHosts)); } else { int currentHostCount = hostedAppCounter.get(metric.getAppId()).intValue(); if (currentHostCount < numHosts) { hostedAppCounter.put(metric.getAppId(), new MutableInt(numHosts)); } } metric = nextMetric; } } } if (metric != null) { int numHosts = processAggregateClusterMetrics(aggregateClusterMetrics, metric, timeSlices); if (!hostedAppCounter.containsKey(metric.getAppId())) { hostedAppCounter.put(metric.getAppId(), new MutableInt(numHosts)); } else { int currentHostCount = hostedAppCounter.get(metric.getAppId()).intValue(); if (currentHostCount < numHosts) { hostedAppCounter.put(metric.getAppId(), new MutableInt(numHosts)); } } } aggregateClusterMetrics.putAll(appAggregator.getAggregateClusterMetrics()); long timestamp = timeSlices.get(timeSlices.size() - 1)[1]; processLiveAppCountMetrics(aggregateClusterMetrics, hostedAppCounter, timestamp); return aggregateClusterMetrics; } }
TimelineMetricClusterAggregatorSecond extends AbstractTimelineAggregator { Map<TimelineClusterMetric, MetricClusterAggregate> aggregateMetricsFromResultSet(ResultSet rs, List<Long[]> timeSlices) throws SQLException, IOException { Map<TimelineClusterMetric, MetricClusterAggregate> aggregateClusterMetrics = new HashMap<TimelineClusterMetric, MetricClusterAggregate>(); TimelineMetric metric = null; Map<String, MutableInt> hostedAppCounter = new HashMap<>(); if (rs.next()) { metric = timelineMetricReadHelper.getTimelineMetricFromResultSet(rs); while (metric == null && rs.next()) { metric = timelineMetricReadHelper.getTimelineMetricFromResultSet(rs); } while (rs.next()) { TimelineMetric nextMetric = timelineMetricReadHelper.getTimelineMetricFromResultSet(rs); if (nextMetric == null) { continue; } if (metric.equalsExceptTime(nextMetric)) { metric.addMetricValues(nextMetric.getMetricValues()); } else { int numHosts = processAggregateClusterMetrics(aggregateClusterMetrics, metric, timeSlices); if (!hostedAppCounter.containsKey(metric.getAppId())) { hostedAppCounter.put(metric.getAppId(), new MutableInt(numHosts)); } else { int currentHostCount = hostedAppCounter.get(metric.getAppId()).intValue(); if (currentHostCount < numHosts) { hostedAppCounter.put(metric.getAppId(), new MutableInt(numHosts)); } } metric = nextMetric; } } } if (metric != null) { int numHosts = processAggregateClusterMetrics(aggregateClusterMetrics, metric, timeSlices); if (!hostedAppCounter.containsKey(metric.getAppId())) { hostedAppCounter.put(metric.getAppId(), new MutableInt(numHosts)); } else { int currentHostCount = hostedAppCounter.get(metric.getAppId()).intValue(); if (currentHostCount < numHosts) { hostedAppCounter.put(metric.getAppId(), new MutableInt(numHosts)); } } } aggregateClusterMetrics.putAll(appAggregator.getAggregateClusterMetrics()); long timestamp = timeSlices.get(timeSlices.size() - 1)[1]; processLiveAppCountMetrics(aggregateClusterMetrics, hostedAppCounter, timestamp); return aggregateClusterMetrics; } 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 { Map<TimelineClusterMetric, MetricClusterAggregate> aggregateMetricsFromResultSet(ResultSet rs, List<Long[]> timeSlices) throws SQLException, IOException { Map<TimelineClusterMetric, MetricClusterAggregate> aggregateClusterMetrics = new HashMap<TimelineClusterMetric, MetricClusterAggregate>(); TimelineMetric metric = null; Map<String, MutableInt> hostedAppCounter = new HashMap<>(); if (rs.next()) { metric = timelineMetricReadHelper.getTimelineMetricFromResultSet(rs); while (metric == null && rs.next()) { metric = timelineMetricReadHelper.getTimelineMetricFromResultSet(rs); } while (rs.next()) { TimelineMetric nextMetric = timelineMetricReadHelper.getTimelineMetricFromResultSet(rs); if (nextMetric == null) { continue; } if (metric.equalsExceptTime(nextMetric)) { metric.addMetricValues(nextMetric.getMetricValues()); } else { int numHosts = processAggregateClusterMetrics(aggregateClusterMetrics, metric, timeSlices); if (!hostedAppCounter.containsKey(metric.getAppId())) { hostedAppCounter.put(metric.getAppId(), new MutableInt(numHosts)); } else { int currentHostCount = hostedAppCounter.get(metric.getAppId()).intValue(); if (currentHostCount < numHosts) { hostedAppCounter.put(metric.getAppId(), new MutableInt(numHosts)); } } metric = nextMetric; } } } if (metric != null) { int numHosts = processAggregateClusterMetrics(aggregateClusterMetrics, metric, timeSlices); if (!hostedAppCounter.containsKey(metric.getAppId())) { hostedAppCounter.put(metric.getAppId(), new MutableInt(numHosts)); } else { int currentHostCount = hostedAppCounter.get(metric.getAppId()).intValue(); if (currentHostCount < numHosts) { hostedAppCounter.put(metric.getAppId(), new MutableInt(numHosts)); } } } aggregateClusterMetrics.putAll(appAggregator.getAggregateClusterMetrics()); long timestamp = timeSlices.get(timeSlices.size() - 1)[1]; processLiveAppCountMetrics(aggregateClusterMetrics, hostedAppCounter, timestamp); return aggregateClusterMetrics; } 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 { Map<TimelineClusterMetric, MetricClusterAggregate> aggregateMetricsFromResultSet(ResultSet rs, List<Long[]> timeSlices) throws SQLException, IOException { Map<TimelineClusterMetric, MetricClusterAggregate> aggregateClusterMetrics = new HashMap<TimelineClusterMetric, MetricClusterAggregate>(); TimelineMetric metric = null; Map<String, MutableInt> hostedAppCounter = new HashMap<>(); if (rs.next()) { metric = timelineMetricReadHelper.getTimelineMetricFromResultSet(rs); while (metric == null && rs.next()) { metric = timelineMetricReadHelper.getTimelineMetricFromResultSet(rs); } while (rs.next()) { TimelineMetric nextMetric = timelineMetricReadHelper.getTimelineMetricFromResultSet(rs); if (nextMetric == null) { continue; } if (metric.equalsExceptTime(nextMetric)) { metric.addMetricValues(nextMetric.getMetricValues()); } else { int numHosts = processAggregateClusterMetrics(aggregateClusterMetrics, metric, timeSlices); if (!hostedAppCounter.containsKey(metric.getAppId())) { hostedAppCounter.put(metric.getAppId(), new MutableInt(numHosts)); } else { int currentHostCount = hostedAppCounter.get(metric.getAppId()).intValue(); if (currentHostCount < numHosts) { hostedAppCounter.put(metric.getAppId(), new MutableInt(numHosts)); } } metric = nextMetric; } } } if (metric != null) { int numHosts = processAggregateClusterMetrics(aggregateClusterMetrics, metric, timeSlices); if (!hostedAppCounter.containsKey(metric.getAppId())) { hostedAppCounter.put(metric.getAppId(), new MutableInt(numHosts)); } else { int currentHostCount = hostedAppCounter.get(metric.getAppId()).intValue(); if (currentHostCount < numHosts) { hostedAppCounter.put(metric.getAppId(), new MutableInt(numHosts)); } } } aggregateClusterMetrics.putAll(appAggregator.getAggregateClusterMetrics()); long timestamp = timeSlices.get(timeSlices.size() - 1)[1]; processLiveAppCountMetrics(aggregateClusterMetrics, hostedAppCounter, timestamp); return aggregateClusterMetrics; } 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; }