@Test(groups = "slow") public void testAggregation() throws Exception { hostId = timelineDAO.getOrAddHost(HOST_NAME); Assert.assertNotNull(hostId); Assert.assertEquals(timelineDAO.getHosts().values().size(), 1); minHeapUsedKindId = timelineDAO.getOrAddSampleKind(hostId, EVENT_TYPE_ID, MIN_HEAPUSED_KIND); Assert.assertNotNull(minHeapUsedKindId); maxHeapUsedKindId = timelineDAO.getOrAddSampleKind(hostId, EVENT_TYPE_ID, MAX_HEAPUSED_KIND); Assert.assertNotNull(maxHeapUsedKindId); Assert.assertEquals(timelineDAO.getSampleKinds().values().size(), 2); createAOneHourTimelineChunk(125); createAOneHourTimelineChunk(60); checkSamplesForATimeline(185, 126, 0); checkSamplesForATimeline(185, 125, 2); checkSamplesForATimeline(64, 61, 0); checkSamplesForATimeline(125, 65, 2); checkSamplesForATimeline(60, 0, 2); checkSamplesForATimeline(125, 0, 4); checkSamplesForATimeline(124, 0, 4); checkSamplesForATimeline(124, 66, 2); aggregator.getAndProcessTimelineAggregationCandidates(); Assert.assertEquals(timelineDAO.getHosts().values().size(), 1); Assert.assertEquals(timelineDAO.getSampleKinds().values().size(), 2); checkSamplesForATimeline(185, 126, 0); checkSamplesForATimeline(185, 125, 2); checkSamplesForATimeline(64, 61, 2); checkSamplesForATimeline(125, 65, 2); checkSamplesForATimeline(60, 0, 2); checkSamplesForATimeline(125, 0, 2); checkSamplesForATimeline(124, 0, 2); checkSamplesForATimeline(124, 66, 2); }
public void getAndProcessTimelineAggregationCandidates() { if (!isAggregating.compareAndSet(false, true)) { log.info("Asked to aggregate, but we're already aggregating!"); return; } else { log.debug("Starting aggregating"); } aggregationRuns.incrementAndGet(); final String[] chunkCountsToAggregate = config.getChunksToAggregate().split(","); for (int aggregationLevel=0; aggregationLevel<config.getMaxAggregationLevel(); aggregationLevel++) { final long startingAggregatesCreated = aggregatesCreated.get(); final Map<String, Long> initialCounters = captureAggregatorCounters(); final int chunkCountIndex = aggregationLevel >= chunkCountsToAggregate.length ? chunkCountsToAggregate.length - 1 : aggregationLevel; final int chunksToAggregate = Integer.parseInt(chunkCountsToAggregate[chunkCountIndex]); streamingAggregateLevel(aggregationLevel, chunksToAggregate); final Map<String, Long> counterDeltas = subtractFromAggregatorCounters(initialCounters); final long netAggregatesCreated = aggregatesCreated.get() - startingAggregatesCreated; if (netAggregatesCreated == 0) { if (aggregationLevel == 0) { foundNothingRuns.incrementAndGet(); } log.debug("Created no new aggregates, so skipping higher-level aggregations"); break; } else { final StringBuilder builder = new StringBuilder(); builder .append("For aggregation level ") .append(aggregationLevel) .append(", runs ") .append(aggregationRuns.get()) .append(", foundNothingRuns ") .append(foundNothingRuns.get()); for (Map.Entry<String, Long> entry : counterDeltas.entrySet()) { builder.append(", ").append(entry.getKey()).append(": ").append(entry.getValue()); } log.info(builder.toString()); } } log.debug("Aggregation done"); isAggregating.set(false); }
TimelineAggregator { public void getAndProcessTimelineAggregationCandidates() { if (!isAggregating.compareAndSet(false, true)) { log.info("Asked to aggregate, but we're already aggregating!"); return; } else { log.debug("Starting aggregating"); } aggregationRuns.incrementAndGet(); final String[] chunkCountsToAggregate = config.getChunksToAggregate().split(","); for (int aggregationLevel=0; aggregationLevel<config.getMaxAggregationLevel(); aggregationLevel++) { final long startingAggregatesCreated = aggregatesCreated.get(); final Map<String, Long> initialCounters = captureAggregatorCounters(); final int chunkCountIndex = aggregationLevel >= chunkCountsToAggregate.length ? chunkCountsToAggregate.length - 1 : aggregationLevel; final int chunksToAggregate = Integer.parseInt(chunkCountsToAggregate[chunkCountIndex]); streamingAggregateLevel(aggregationLevel, chunksToAggregate); final Map<String, Long> counterDeltas = subtractFromAggregatorCounters(initialCounters); final long netAggregatesCreated = aggregatesCreated.get() - startingAggregatesCreated; if (netAggregatesCreated == 0) { if (aggregationLevel == 0) { foundNothingRuns.incrementAndGet(); } log.debug("Created no new aggregates, so skipping higher-level aggregations"); break; } else { final StringBuilder builder = new StringBuilder(); builder .append("For aggregation level ") .append(aggregationLevel) .append(", runs ") .append(aggregationRuns.get()) .append(", foundNothingRuns ") .append(foundNothingRuns.get()); for (Map.Entry<String, Long> entry : counterDeltas.entrySet()) { builder.append(", ").append(entry.getKey()).append(": ").append(entry.getValue()); } log.info(builder.toString()); } } log.debug("Aggregation done"); isAggregating.set(false); } }
TimelineAggregator { public void getAndProcessTimelineAggregationCandidates() { if (!isAggregating.compareAndSet(false, true)) { log.info("Asked to aggregate, but we're already aggregating!"); return; } else { log.debug("Starting aggregating"); } aggregationRuns.incrementAndGet(); final String[] chunkCountsToAggregate = config.getChunksToAggregate().split(","); for (int aggregationLevel=0; aggregationLevel<config.getMaxAggregationLevel(); aggregationLevel++) { final long startingAggregatesCreated = aggregatesCreated.get(); final Map<String, Long> initialCounters = captureAggregatorCounters(); final int chunkCountIndex = aggregationLevel >= chunkCountsToAggregate.length ? chunkCountsToAggregate.length - 1 : aggregationLevel; final int chunksToAggregate = Integer.parseInt(chunkCountsToAggregate[chunkCountIndex]); streamingAggregateLevel(aggregationLevel, chunksToAggregate); final Map<String, Long> counterDeltas = subtractFromAggregatorCounters(initialCounters); final long netAggregatesCreated = aggregatesCreated.get() - startingAggregatesCreated; if (netAggregatesCreated == 0) { if (aggregationLevel == 0) { foundNothingRuns.incrementAndGet(); } log.debug("Created no new aggregates, so skipping higher-level aggregations"); break; } else { final StringBuilder builder = new StringBuilder(); builder .append("For aggregation level ") .append(aggregationLevel) .append(", runs ") .append(aggregationRuns.get()) .append(", foundNothingRuns ") .append(foundNothingRuns.get()); for (Map.Entry<String, Long> entry : counterDeltas.entrySet()) { builder.append(", ").append(entry.getKey()).append(": ").append(entry.getValue()); } log.info(builder.toString()); } } log.debug("Aggregation done"); isAggregating.set(false); } @Inject  TimelineAggregator(final IDBI dbi, final DefaultTimelineDAO timelineDao, final TimelineCoder timelineCoder, final SampleCoder sampleCoder, final CollectorConfig config); }
TimelineAggregator { public void getAndProcessTimelineAggregationCandidates() { if (!isAggregating.compareAndSet(false, true)) { log.info("Asked to aggregate, but we're already aggregating!"); return; } else { log.debug("Starting aggregating"); } aggregationRuns.incrementAndGet(); final String[] chunkCountsToAggregate = config.getChunksToAggregate().split(","); for (int aggregationLevel=0; aggregationLevel<config.getMaxAggregationLevel(); aggregationLevel++) { final long startingAggregatesCreated = aggregatesCreated.get(); final Map<String, Long> initialCounters = captureAggregatorCounters(); final int chunkCountIndex = aggregationLevel >= chunkCountsToAggregate.length ? chunkCountsToAggregate.length - 1 : aggregationLevel; final int chunksToAggregate = Integer.parseInt(chunkCountsToAggregate[chunkCountIndex]); streamingAggregateLevel(aggregationLevel, chunksToAggregate); final Map<String, Long> counterDeltas = subtractFromAggregatorCounters(initialCounters); final long netAggregatesCreated = aggregatesCreated.get() - startingAggregatesCreated; if (netAggregatesCreated == 0) { if (aggregationLevel == 0) { foundNothingRuns.incrementAndGet(); } log.debug("Created no new aggregates, so skipping higher-level aggregations"); break; } else { final StringBuilder builder = new StringBuilder(); builder .append("For aggregation level ") .append(aggregationLevel) .append(", runs ") .append(aggregationRuns.get()) .append(", foundNothingRuns ") .append(foundNothingRuns.get()); for (Map.Entry<String, Long> entry : counterDeltas.entrySet()) { builder.append(", ").append(entry.getKey()).append(": ").append(entry.getValue()); } log.info(builder.toString()); } } log.debug("Aggregation done"); isAggregating.set(false); } @Inject  TimelineAggregator(final IDBI dbi, final DefaultTimelineDAO timelineDao, final TimelineCoder timelineCoder, final SampleCoder sampleCoder, final CollectorConfig config); void getAndProcessTimelineAggregationCandidates(); void runAggregationThread(); void stopAggregationThread(); @Managed long getAggregationRuns(); @Managed long getFoundNothingRuns(); @Managed long getTimelineChunksConsidered(); @Managed long getTimelineChunkBatchesProcessed(); @Managed long getTimelineChunksCombined(); @Managed long getTimelineChunksQueuedForCreation(); @Managed long getTimelineChunksWritten(); @Managed long getTimelineChunksInvalidatedOrDeleted(); @Managed long getTimelineChunksBytesCreated(); @Managed long getMsSpentAggregating(); @Managed long getMsSpentSleeping(); @Managed long getMsWritingDb(); @Managed(description = "Aggregate candidate timelines") void initiateAggregation(); }
TimelineAggregator { public void getAndProcessTimelineAggregationCandidates() { if (!isAggregating.compareAndSet(false, true)) { log.info("Asked to aggregate, but we're already aggregating!"); return; } else { log.debug("Starting aggregating"); } aggregationRuns.incrementAndGet(); final String[] chunkCountsToAggregate = config.getChunksToAggregate().split(","); for (int aggregationLevel=0; aggregationLevel<config.getMaxAggregationLevel(); aggregationLevel++) { final long startingAggregatesCreated = aggregatesCreated.get(); final Map<String, Long> initialCounters = captureAggregatorCounters(); final int chunkCountIndex = aggregationLevel >= chunkCountsToAggregate.length ? chunkCountsToAggregate.length - 1 : aggregationLevel; final int chunksToAggregate = Integer.parseInt(chunkCountsToAggregate[chunkCountIndex]); streamingAggregateLevel(aggregationLevel, chunksToAggregate); final Map<String, Long> counterDeltas = subtractFromAggregatorCounters(initialCounters); final long netAggregatesCreated = aggregatesCreated.get() - startingAggregatesCreated; if (netAggregatesCreated == 0) { if (aggregationLevel == 0) { foundNothingRuns.incrementAndGet(); } log.debug("Created no new aggregates, so skipping higher-level aggregations"); break; } else { final StringBuilder builder = new StringBuilder(); builder .append("For aggregation level ") .append(aggregationLevel) .append(", runs ") .append(aggregationRuns.get()) .append(", foundNothingRuns ") .append(foundNothingRuns.get()); for (Map.Entry<String, Long> entry : counterDeltas.entrySet()) { builder.append(", ").append(entry.getKey()).append(": ").append(entry.getValue()); } log.info(builder.toString()); } } log.debug("Aggregation done"); isAggregating.set(false); } @Inject  TimelineAggregator(final IDBI dbi, final DefaultTimelineDAO timelineDao, final TimelineCoder timelineCoder, final SampleCoder sampleCoder, final CollectorConfig config); void getAndProcessTimelineAggregationCandidates(); void runAggregationThread(); void stopAggregationThread(); @Managed long getAggregationRuns(); @Managed long getFoundNothingRuns(); @Managed long getTimelineChunksConsidered(); @Managed long getTimelineChunkBatchesProcessed(); @Managed long getTimelineChunksCombined(); @Managed long getTimelineChunksQueuedForCreation(); @Managed long getTimelineChunksWritten(); @Managed long getTimelineChunksInvalidatedOrDeleted(); @Managed long getTimelineChunksBytesCreated(); @Managed long getMsSpentAggregating(); @Managed long getMsSpentSleeping(); @Managed long getMsWritingDb(); @Managed(description = "Aggregate candidate timelines") void initiateAggregation();  }