gistfile1.txt

/**
 * @author Denis Gabaydulin
 * @since 12/08/2016
 */
public class InfluxDbReporter extends ScheduledReporter {
    private static final Logger log = LoggerFactory.getLogger(InfluxDbReporter.class);

    private static final String REPORTER_NAME = "influxdb_reporter";

    private final InfluxDB influxDB;
    private final ConfigDatabaseConfiguration configuration;
    private final InfluxDbClientPropertiesConfiguration clientPropertiesConfiguration;

    /**
     * Creates a new {@link ScheduledReporter} instance.
     *
     * @param registry     the {@link MetricRegistry} containing the metrics this
     *                     reporter will report
     * @param filter       the filter for which metrics to report
     * @param rateUnit
     * @param durationUnit
     */
    protected InfluxDbReporter(
            InfluxDB influxDB,
            InfluxDbClientPropertiesConfiguration clientPropertiesConfiguration,
            ConfigDatabaseConfiguration configuration,
            MetricRegistry registry,
            MetricFilter filter,
            TimeUnit rateUnit,
            TimeUnit durationUnit
    ) {
        super(registry, REPORTER_NAME, filter, rateUnit, durationUnit);
        this.influxDB = influxDB;
        this.configuration = configuration;
        this.clientPropertiesConfiguration = clientPropertiesConfiguration;
    }

    @Override
    public void report(
            SortedMap<String, Gauge> gauges,
            SortedMap<String, Counter> counters,
            SortedMap<String, Histogram> histograms,
            SortedMap<String, Meter> meters,
            SortedMap<String, Timer> timers
    ) {
        long reportTime = now(UTC).getMillis();

        gauges.entrySet().forEach(entry -> logGauge(reportTime, entry.getKey(), entry.getValue()));

        counters.entrySet().forEach(entry -> logCounter(reportTime, entry.getKey(), entry.getValue()));

        histograms.entrySet().forEach(entry -> logHistogram(reportTime, entry.getKey(), entry.getValue()));

        meters.entrySet().forEach(entry -> logMeter(reportTime, entry.getKey(), entry.getValue()));

        timers.entrySet().forEach(entry -> logTimer(reportTime, entry.getKey(), entry.getValue()));
    }

    private void logTimer(long reportTime, String name, Timer timer) {
        Snapshot snapshot = timer.getSnapshot();

        Point point = Point.measurement(name.replace(".", "_"))
                .time(reportTime, TimeUnit.MILLISECONDS)
                .tag("host", configuration.getHost())
                .tag("app", configuration.getModule())
                .tag("machine", valueOf(configuration.getMachineId()))
                .addField("count", timer.getCount())
                .addField("min", convertDuration(snapshot.getMin()))
                .addField("max", convertDuration(snapshot.getMax()))
                .addField("mean", convertDuration(snapshot.getMean()))
                .addField("stdDev", convertDuration(snapshot.getStdDev()))
                .addField("median", convertDuration(snapshot.getMedian()))
                .addField("75thPercentile", convertDuration(snapshot.get75thPercentile()))
                .addField("95thPercentile", convertDuration(snapshot.get95thPercentile()))
                .addField("99thPercentile", convertDuration(snapshot.get99thPercentile()))
                .addField("meanRate", timer.getMeanRate())
                .build();

        try {
            influxDB.write(clientPropertiesConfiguration.getDatabase(), "default", point);
        } catch (Exception ignored) {
        }
    }

    private void logMeter(long reportTime, String name, Meter meter) {
        Point point = Point.measurement(name.replace(".", "_"))
                .time(reportTime, TimeUnit.MILLISECONDS)
                .tag("host", configuration.getHost())
                .tag("app", configuration.getModule())
                .tag("machine", valueOf(configuration.getMachineId()))
                .addField("count", meter.getCount())
                .addField("meanRate", meter.getMeanRate())
                .build();

        try {
            influxDB.write(clientPropertiesConfiguration.getDatabase(), "default", point);
        } catch (Exception ignored) {
        }
    }

    private void logHistogram(long reportTime, String name, Histogram histogram) {
        Snapshot snapshot = histogram.getSnapshot();

        Point point = Point.measurement(name.replace(".", "_"))
                .time(reportTime, TimeUnit.MILLISECONDS)
                .tag("host", configuration.getHost())
                .tag("app", configuration.getModule())
                .tag("machine", valueOf(configuration.getMachineId()))
                .addField("count", histogram.getCount())
                .addField("min", convertDuration(snapshot.getMin()))
                .addField("max", convertDuration(snapshot.getMax()))
                .addField("mean", convertDuration(snapshot.getMean()))
                .addField("stdDev", convertDuration(snapshot.getStdDev()))
                .addField("median", convertDuration(snapshot.getMedian()))
                .addField("75thPercentile", convertDuration(snapshot.get75thPercentile()))
                .addField("95thPercentile", convertDuration(snapshot.get95thPercentile()))
                .addField("99thPercentile", convertDuration(snapshot.get99thPercentile()))
                .build();

        try {
            influxDB.write(clientPropertiesConfiguration.getDatabase(), "default", point);
        } catch (Exception ignored) {
        }
    }

    private void logCounter(long reportTime, String name, Counter counter) {
        Point point = Point.measurement(name.replace(".", "_"))
                .time(reportTime, TimeUnit.MILLISECONDS)
                .tag("host", configuration.getHost())
                .tag("app", configuration.getModule())
                .tag("machine", valueOf(configuration.getMachineId()))
                .addField("count", counter.getCount())
                .build();

        try {
            influxDB.write(clientPropertiesConfiguration.getDatabase(), "default", point);
        } catch (Exception ignored) {
        }
    }

    private void logGauge(long reportTime, String name, Gauge gauge) {
        Point point = Point.measurement(name.replace(".", "_"))
                .time(reportTime, TimeUnit.MILLISECONDS)
                .tag("host", configuration.getHost())
                .tag("app", configuration.getModule())
                .tag("machine", valueOf(configuration.getMachineId()))
                .addField("value", valueOf(gauge.getValue()))
                .build();

        try {
            influxDB.write(clientPropertiesConfiguration.getDatabase(), "default", point);
        } catch (Exception ignored) {
        }
    }

    public static class Builder {
        private final MetricRegistry registry;
        private TimeUnit rateUnit;
        private TimeUnit durationUnit;
        private MetricFilter filter;
        private InfluxDB influxDB;
        private InfluxDbClientPropertiesConfiguration clientPropertiesConfiguration;
        private ConfigDatabaseConfiguration configuration;

        public Builder(
                MetricRegistry registry,
                InfluxDB influxDB,
                InfluxDbClientPropertiesConfiguration clientPropertiesConfiguration,
                ConfigDatabaseConfiguration configuration
        ) {
            this.registry = registry;
            this.influxDB = influxDB;
            this.clientPropertiesConfiguration = clientPropertiesConfiguration;
            this.configuration = configuration;
            this.rateUnit = TimeUnit.SECONDS;
            this.durationUnit = TimeUnit.MILLISECONDS;
            this.filter = MetricFilter.ALL;
        }

        public Builder convertRatesTo(TimeUnit rateUnit) {
            this.rateUnit = rateUnit;
            return this;
        }

        public Builder convertDurationsTo(TimeUnit durationUnit) {
            this.durationUnit = durationUnit;
            return this;
        }

        public Builder filter(MetricFilter filter) {
            this.filter = filter;
            return this;
        }

        public InfluxDbReporter build() {
            return new InfluxDbReporter(
                    influxDB,
                    clientPropertiesConfiguration,
                    configuration,
                    registry,
                    filter,
                    rateUnit,
                    durationUnit
            );
        }
    }
}