TimerClearingGraphiteReporter.java

import com.yammer.metrics.Metrics;
import com.yammer.metrics.core.*;
import com.yammer.metrics.reporting.GraphiteReporter;
import com.yammer.metrics.reporting.SocketProvider;
import com.yammer.metrics.stats.Snapshot;

import java.io.IOException;
import java.util.concurrent.TimeUnit;

/**
 * A Metrics Reporter which sends out application metrics to a Graphite server periodically.
 * After extracting metrics, each Timer's histogram and each Histogram are cleared, so that percentiles
 * and max/min readings do not persist for long time periods (which they do by default in Yammer Metrics).
 *
 * Note: this works with Metrics 2.x; it hasn't yet been ported to 3.x.  (Shouldn't be hard though,
 * the GraphiteReporter hasn't changed very much! In fact 3.x makes some of this code cleaner...)
 */
public class TimerClearingGraphiteReporter extends GraphiteReporter {

    /**
     * Enables and starts the graphite reporter.
     *
     * @param period          the period between successive outputs
     * @param unit            the time unit of {@code period}
     * @param host            the host name of graphite server (carbon-cache agent)
     * @param port            the port number on which the graphite server is listening
     * @param prefix          the string which is prepended to all metric names
     */
    public static void enableClearingReporter(long period, TimeUnit unit,
                                              String host, int port, String prefix) throws IOException {

        final TimerClearingGraphiteReporter reporter = new TimerClearingGraphiteReporter(Metrics.defaultRegistry(),
                prefix, MetricPredicate.ALL, new DefaultSocketProvider(host, port), Clock.defaultClock());

        reporter.start(period, unit);
    }

    private TimerClearingGraphiteReporter(MetricsRegistry metricsRegistry, String prefix, MetricPredicate predicate,
                                          SocketProvider socketProvider, Clock clock) throws IOException {
        super(metricsRegistry, prefix, predicate, socketProvider, clock);
    }

    @Override
    public void processTimer(MetricName name, Timer timer, Long epoch) throws IOException {
        this.processMeter(name, timer, epoch);

        // capture the values to report, in an unlocked fashion. This is still
        // vulnerable to a race, since the values are not captured atomically with the
        // object locked, but then, the underlying Histogram itself isn't atomic (eg. mean() works
        // with sum and count separately without locking), so this is probably good enough.
        // At least we're not making any system calls between capturing and clearing this time!
        SummarizableSnapshot summarizableSnapshot = snapshotSummarizable(timer);
        Snapshot sampleSnapshot = timer.getSnapshot();
        timer.clear();

        sendSnapshots(name, epoch, summarizableSnapshot, sampleSnapshot);
    }

    @Override
    public void processHistogram(MetricName name, Histogram histogram, Long epoch) throws IOException {
        SummarizableSnapshot summarizableSnapshot = snapshotSummarizable(histogram);
        Snapshot sampleSnapshot = histogram.getSnapshot();
        histogram.clear();
        sendSnapshots(name, epoch, summarizableSnapshot, sampleSnapshot);
    }

    private SummarizableSnapshot snapshotSummarizable(Summarizable metric) throws IOException {
        return new SummarizableSnapshot(metric.min(), metric.max(), metric.mean(), metric.stdDev());
    }

    private void sendSnapshots(MetricName name, Long epoch, SummarizableSnapshot summarizableSnapshot, Snapshot sampleSnapshot) throws IOException {
        final String sanitizedName = this.sanitizeName(name);
        sendFloat(epoch, sanitizedName, "min", summarizableSnapshot.min);
        sendFloat(epoch, sanitizedName, "max", summarizableSnapshot.max);
        sendFloat(epoch, sanitizedName, "mean", summarizableSnapshot.mean);
        sendFloat(epoch, sanitizedName, "stddev", summarizableSnapshot.stdDev);
        sendFloat(epoch, sanitizedName, "median", sampleSnapshot.getMedian());
        sendFloat(epoch, sanitizedName, "75percentile", sampleSnapshot.get75thPercentile());
        sendFloat(epoch, sanitizedName, "95percentile", sampleSnapshot.get95thPercentile());
        sendFloat(epoch, sanitizedName, "98percentile", sampleSnapshot.get98thPercentile());
        sendFloat(epoch, sanitizedName, "99percentile", sampleSnapshot.get99thPercentile());
        sendFloat(epoch, sanitizedName, "999percentile", sampleSnapshot.get999thPercentile());
    }

    private class SummarizableSnapshot {
        private final double min, max, mean, stdDev;

        public SummarizableSnapshot(double min, double max, double mean, double stdDev) {
            this.min = min;
            this.max = max;
            this.mean = mean;
            this.stdDev = stdDev;
        }
    }
}