/**
 * Verify that a StatsD 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_initializeStatsD() throws IOException
{
  CacheConfig.setMetricsReporters(conf, MetricsReporter.STATSD.name());

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

/**
 * Verify that both JMX and StatsD reporters are correctly registered when the configuration option is set.
 *
 * @throws IOException if an I/O error occurs while closing a reporter.
 */
@Test
public void testInitializeReporters_initializeJMXAndStatsD() throws IOException
{
  CacheConfig.setMetricsReporters(conf, Joiner.on(",").join(MetricsReporter.STATSD.name(), MetricsReporter.JMX.name()));

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

/**
 * Verify that no reporters are registered when the configuration option is set.
 *
 * @throws IOException if an I/O error occurs while closing a reporter.
 */
@Test
public void testInitializeReporters_noneInitialized() throws IOException
{
  CacheConfig.setMetricsReporters(conf, "");

  try (final BookKeeperMetrics bookKeeperMetrics = new BookKeeperMetrics(conf, metrics)) {
    assertFalse(containsReporterType(bookKeeperMetrics.reporters, StatsDReporter.class));
    assertFalse(containsReporterType(bookKeeperMetrics.reporters, JmxReporter.class));
  }
}
 

/**
 * Starts logging the metrics for cassandra into statsd. 
 */
public static void apply(ActorSystem system) {
    MetricRegistry registry = CassandraMetricsRegistry.get(system).getRegistry();
    
    String statsdhost = system.settings().config().getString("codahale.statsd.hostname");
    int statsdport = system.settings().config().getInt("codahale.statsd.port");
    long interval = system.settings().config().getDuration("codahale.statsd.tick-interval", TimeUnit.MILLISECONDS);
    String prefix = system.settings().config().getString("codahale.statsd.prefix");
    StatsDReporter.forRegistry(registry)
        .prefixedWith(prefix)
        .build(statsdhost, statsdport)           
        .start(interval, TimeUnit.MILLISECONDS);
}
 

public ManagedStatsdReporter(final String endpoint, final MetricRegistry registry) {
  Preconditions.checkArgument(!Strings.isNullOrEmpty(endpoint));

  final List<String> parts = Splitter.on(":").splitToList(endpoint);
  checkArgument(parts.size() == 2, "Specification of statsd host port has wrong number of "
                                   + "parts. Should be host:port");
  final String host = parts.get(0);
  final int port = Integer.valueOf(parts.get(1));
  statsdReporter = StatsDReporter.forRegistry(registry).build(host, port);
}
 

/**
 * 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;
    }
  }
}
 

/**
 * Configures the metrics service to emit StatsD-formatted metrics to the configured UDP host/port with the
 * specified interval.
 */
public static void configureStatsd(SchedulerConfig schedulerConfig) {
  StatsDReporter.forRegistry(METRICS)
      .build(schedulerConfig.getStatsdHost(), schedulerConfig.getStatsdPort())
      .start(schedulerConfig.getStatsDPollIntervalS(), TimeUnit.SECONDS);
}