/**
 * Verify that a Ganglia reporter is correctly registered when the configuration option is set.
 *
 * @throws IOException if an I/O error occurs while closing a reporter.
 */
@Test
public void testInitializeReporters_initializeGanglia() throws IOException
{
  CacheConfig.setMetricsReporters(conf, MetricsReporter.GANGLIA.name());

  try (final BookKeeperMetrics bookKeeperMetrics = new BookKeeperMetrics(conf, metrics)) {
    assertTrue(containsReporterType(bookKeeperMetrics.reporters, GangliaReporter.class));
  }
}
 

@Override
protected void configure() {
    bind(GMetric.class).toProvider(GMetricProvider.class);
    bind(GangliaReporter.class).toProvider(GangliaReporterProvider.class);

    addConfigBeans();
    addInitializer(MetricsGangliaReporterService.class);
}
 

@Override
public GangliaReporter get() {
    return GangliaReporter.forRegistry(metricRegistry)
            .prefixedWith(configuration.getPrefix())
            .convertRatesTo(configuration.getUnitRates())
            .convertDurationsTo(configuration.getUnitDurations())
            .withDMax(configuration.getDMax())
            .withTMax(configuration.getTMax())
            .filter(new RegexMetricFilter(configuration.getIncludeMetrics()))
            .build(gMetric);
}
 

@Test
public void get() throws Exception {
    final MetricsGangliaReporterConfiguration configuration = new MetricsGangliaReporterConfiguration();
    final GMetric graphiteSender = new GMetric("127.0.0.1", 12345, null, 23);
    final MetricRegistry metricRegistry = new MetricRegistry();
    final GangliaReporterProvider provider = new GangliaReporterProvider(configuration, graphiteSender, metricRegistry);

    final GangliaReporter reporter = provider.get();
    assertNotNull(reporter);
}
 

/**
 * Builds reporter with the given ganglia metric.
 *
 * @param gMetric ganglia metric
 * @return reporter
 */
private GangliaReporter buildReporter(GMetric gMetric) {
    MetricRegistry mr = metricsService.getMetricRegistry();

    return GangliaReporter.forRegistry(filter(mr))
            .convertRatesTo(TimeUnit.SECONDS)
            .convertDurationsTo(TimeUnit.MILLISECONDS)
            .build(gMetric);
}
 

/**
 * Initialize reporters for reporting metrics to desired services.
 */
protected void initializeReporters()
{
  final Iterable<String> metricsReporterNames = Splitter.on(",").trimResults().omitEmptyStrings().split(CacheConfig.getMetricsReporters(conf));

  final Set<MetricsReporter> metricsReporters = new HashSet<>();
  for (String reporterName : metricsReporterNames) {
    metricsReporters.add(MetricsReporter.valueOf(reporterName.toUpperCase()));
  }

  for (MetricsReporter reporter : metricsReporters) {
    switch (reporter) {
      case JMX:
        final JmxReporter jmxReporter = JmxReporter.forRegistry(metrics)
            .convertRatesTo(TimeUnit.SECONDS)
            .convertDurationsTo(TimeUnit.MILLISECONDS)
            .filter(metricsFilter)
            .build();

        log.debug("Reporting metrics to JMX");
        jmxReporter.start();
        reporters.add(jmxReporter);
        break;
      case STATSD:
        if (!CacheConfig.isOnMaster(conf)) {
          CacheConfig.setStatsDMetricsHost(conf, ClusterUtil.getMasterHostname(conf));
        }
        final StatsDReporter statsDReporter = StatsDReporter.forRegistry(metrics)
            .convertRatesTo(TimeUnit.SECONDS)
            .convertDurationsTo(TimeUnit.MILLISECONDS)
            .filter(metricsFilter)
            .build(CacheConfig.getStatsDMetricsHost(conf), CacheConfig.getStatsDMetricsPort(conf));

        log.debug(String.format("Reporting metrics to StatsD [%s:%d]", CacheConfig.getStatsDMetricsHost(conf), CacheConfig.getStatsDMetricsPort(conf)));
        statsDReporter.start(CacheConfig.getMetricsReportingInterval(conf), TimeUnit.MILLISECONDS);
        reporters.add(statsDReporter);
        break;
      case GANGLIA:
        if (!CacheConfig.isOnMaster(conf)) {
          CacheConfig.setGangliaMetricsHost(conf, ClusterUtil.getMasterHostname(conf));
        }
        log.debug(String.format("Reporting metrics to Ganglia [%s:%s]", CacheConfig.getGangliaMetricsHost(conf), CacheConfig.getGangliaMetricsPort(conf)));
        final GMetric ganglia = new GMetric(CacheConfig.getGangliaMetricsHost(conf), CacheConfig.getGangliaMetricsPort(conf), GMetric.UDPAddressingMode.MULTICAST, 1);
        final GangliaReporter gangliaReporter = GangliaReporter.forRegistry(metrics)
                .convertRatesTo(TimeUnit.SECONDS)
                .convertDurationsTo(TimeUnit.MILLISECONDS)
                .filter(metricsFilter)
                .build(ganglia);
        gangliaReporter.start(CacheConfig.getMetricsReportingInterval(conf), TimeUnit.MILLISECONDS);
        reporters.add(gangliaReporter);
        break;
    }
  }
}
 

@Inject
public MetricsGangliaReporterService(GangliaReporter gangliaReporter,
                                     MetricsGangliaReporterConfiguration configuration) {
    this.gangliaReporter = requireNonNull(gangliaReporter);
    this.configuration = requireNonNull(configuration);
}