public int flush ( final MetricSender sender ) throws IOException , HttpException { Preconditions . checkNotNull ( sender ) ; // aggregate metrics that were enqueued before the current minute long currentMinute = ( System . currentTimeMillis ( ) / MS_IN_MIN ) * MS_IN_MIN ; LOGGER . debug ( "Flushing metrics < {}" , currentMinute ) ; LOGGER . debug ( "Metrics queue size {}" , queue . size ( ) ) ; MetricAggregator aggregator = new MetricAggregator ( currentMinute , lastValues ) ; while ( ( ! queue . isEmpty ( ) ) && ( queue . peek ( ) . getOccurredMillis ( ) < currentMinute ) ) { aggregator . add ( queue . remove ( ) ) ; } // handle the auto reports if ( lastFlush < currentMinute ) { aggregator . autoReportZero ( autoReportZeroMetrics ) ; aggregator . autoReportLast ( autoReportLastMetrics ) ; } lastFlush = currentMinute ; // get the aggregates List < MetricAggregate > aggregates = aggregator . getAggregates ( ) ; // Save the values of gauge and average metrics for the next iteration // Gauge last value = aggregate last value // Average last value = aggregate last value / aggregate count if ( ( aggregates != null ) && ( ! aggregates . isEmpty ( ) ) ) { for ( MetricAggregate aggregate : aggregates ) { if ( aggregate . getIdentity ( ) . getType ( ) . equals ( MetricMonitorType . GAUGE ) ) { lastValues . put ( aggregate . getIdentity ( ) , aggregate . getValue ( ) ) ; } else if ( aggregate . getIdentity ( ) . getType ( ) . equals ( MetricMonitorType . AVERAGE ) ) { lastValues . put ( aggregate . getIdentity ( ) , aggregate . getValue ( ) / aggregate . getCount ( ) ) ; } } } // send the aggregates to Stackify int numSent = 0 ; if ( ( aggregates != null ) && ( ! aggregates . isEmpty ( ) ) ) { LOGGER . debug ( "Sending aggregate metrics: {}" , aggregates ) ; sender . send ( aggregates ) ; numSent = aggregates . size ( ) ; } return numSent ; }
Flushes all queued metrics to Stackify