Java 类com.codahale.metrics.ScheduledReporter 实例源码

@Override
public ScheduledReporter build(final MetricRegistry registry) {
  final Sender builtSender = new Sender(sender.build(registry));
  final DropwizardTransformer transformer = new DropwizardTransformer(
    globalTags,
    DropwizardMeasurementParser.withTemplates(metricTemplates),
    groupCounters,
    groupGauges,
    getRateUnit(),
    getDurationUnit()
  );

  return new InfluxDbMeasurementReporter(
    builtSender,
    registry,
    getFilter(),
    getRateUnit(),
    getDurationUnit(),
    Clock.systemUTC(),
    transformer
  );
}

private ScheduledReporter createAndGetConfiguredCSVReporter(String prefix, String csvDir) throws IOException {
    // NOTE:
    // 1) metrics output files are exclusive to a given process
    // 2) the output directory must exist
    // 3) if output files already exist they are not overwritten and there is no metrics output
    File outputDir;
    if (Strings.isNullOrEmpty(prefix)) {
        outputDir = new File(csvDir, prefix);
    } else {
        outputDir = new File(csvDir);
    }
    FileUtils.forceMkdir(outputDir);
    LOG.info("Configuring stats with csv output to directory [{}]", outputDir.getAbsolutePath());
    return CsvReporter.forRegistry(metrics)
            .convertRatesTo(TimeUnit.SECONDS)
            .convertDurationsTo(TimeUnit.MILLISECONDS)
            .build(outputDir);
}

private ScheduledReporter createReporter(String json)
        throws Exception {
    ObjectMapper objectMapper = Jackson.newObjectMapper();
    ReporterFactory reporterFactory = objectMapper.readValue(json, ReporterFactory.class);

    assertTrue(reporterFactory instanceof DatadogExpansionFilteredReporterFactory);
    DatadogExpansionFilteredReporterFactory datadogReporterFactory = (DatadogExpansionFilteredReporterFactory) reporterFactory;

    // Replace the transport with our own mock for testing

    Transport transport = mock(Transport.class);
    when(transport.prepare()).thenReturn(_request);

    AbstractTransportFactory transportFactory = mock(AbstractTransportFactory.class);
    when(transportFactory.build()).thenReturn(transport);

    datadogReporterFactory.setTransport(transportFactory);

    // Build the reporter
    return datadogReporterFactory.build(_metricRegistry);
}

/**
 * Start metric reporting.
 *
 * @param properties configuration properties
 */
public void startMetricReporting(Properties properties) {
  if (this.reportingStarted) {
    LOGGER.warn("Metric reporting has already started");
    return;
  }

  long reportInterval = Long.parseLong(properties.getProperty(ConfigurationKeys.METRICS_REPORT_INTERVAL_KEY,
      ConfigurationKeys.DEFAULT_METRICS_REPORT_INTERVAL));

  buildJmxMetricReporter(properties);
  if (this.jmxReporter.isPresent()) {
    this.jmxReporter.get().start();
  }

  buildFileMetricReporter(properties);
  buildKafkaMetricReporter(properties);
  buildCustomMetricReporters(properties);

  for (ScheduledReporter reporter : this.scheduledReporters) {
    reporter.start(reportInterval, TimeUnit.MILLISECONDS);
  }

  this.reportingStarted = true;
}

@Override
public ScheduledReporter build(MetricRegistry metricRegistry) {
    MetricAdapterFactory metricAdapterFactory = new MetricAdapterFactoryImpl(
            new DefaultMetricDescriptorLookup(),
            new DefaultMetricNameTranslator(getReplacements()));

    MonitorableRegistry monitorableRegistry = MonitorableRegistry.getNamedInstance(getRegistryName());

    PcpMmvWriter pcpMmvWriter = new PcpMmvWriter(getPcpName(), IdentifierSourceSet.DEFAULT_SET);
    pcpMmvWriter.setClusterIdentifier(getClusterIdentifier());

    MonitoringView monitoringView = new PcpMonitorBridge(pcpMmvWriter);

    DynamicMonitoringView dynamicMonitoringView = new DynamicMonitoringView(monitorableRegistry, monitoringView, quietPeriod);

    return new ParfaitReporter(metricRegistry,
            monitorableRegistry,
            dynamicMonitoringView,
            metricAdapterFactory,
            getRateUnit(),
            getDurationUnit(),
            getFilter(),
            getPrefix());
}

/**
 * Create a new ElasticSearch reporter.
 *
 * @param metricRegistry
 *            the registry to report on
 * @param config
 *            the configuration map (see {@link MetricsFactory})
 * @return the reporter instance
 */
public static ScheduledReporter createElasticSearchReporter(MetricRegistry metricRegistry,
        Map<String, Object> config) throws BaleenException {
    try {
        String server = (String) config.getOrDefault("server", "localhost:9200");
        String index = (String) config.getOrDefault("index", "metrics");
        int timeout = (int) config.getOrDefault("timeout", 1000);
        int bulkSize = (int) config.getOrDefault("bulkSize", 2500);

        return ElasticsearchReporter.forRegistry(metricRegistry).convertRatesTo(getRatesUnit(config))
                .convertDurationsTo(getDurationUnit(config)).bulkSize(bulkSize).hosts(server).index(index)
                .timeout(timeout).build();
    } catch (IOException e) {
        throw new BaleenException("Unable to create ElasticSearch reporter", e);
    }
}

@Override
public void onPassivate( Application application )
{
    requestTimers.values().forEach( t -> t.stop() );
    requestTimers = null;
    reporters.forEach(
        r ->
        {
            if( r instanceof ScheduledReporter )
            {
                ( (ScheduledReporter) r ).stop();
            }
            else if( r instanceof JmxReporter )
            {
                ( (JmxReporter) r ).stop();
            }
        }
    );
    reporters = null;
    api = null;
    eventRegistration.unregister();
    eventRegistration = null;
    SharedMetricRegistries.clear();
    SharedHealthCheckRegistries.clear();
}

@Override
public ScheduledReporter build(MetricRegistry registry) {

    final EnumSet<Expansions> expansions = EnumSet.of(COUNT, RATE_1_MINUTE,
            RATE_15_MINUTE, MEDIAN, P95, P99);
    final DatadogReporter reporter;
    try {
        reporter = new DatadogReporter.Builder(registry).withEC2Host()
                .withApiKey(apiKey).withExpansions(expansions).build();
    } catch (IOException e) {
        e.printStackTrace();
        throw new IllegalStateException(
                "Unable to construct DataDog metrics reporter", e);
    }
    return reporter;
}

/**
 * Builds a {@link ScheduledReporter} with the given properties, sending
 * metrics using the given InfluxDB.
 *
 * @return a {@link ScheduledReporter}
 */
public ScheduledReporter build() {
    ScheduledReporter reporter;

    switch (influxdbVersion) {
    case V08:
        Influxdb influxdb = buildInfluxdb();
        reporter = (executor == null)
                ? new ReporterV08(registry, influxdb, clock, prefix, rateUnit, durationUnit, filter, skipIdleMetrics)
                : new ReporterV08(registry, influxdb, clock, prefix, rateUnit, durationUnit, filter, skipIdleMetrics, executor)
                ;
        break;
    default:
        Sender s = buildSender();
        reporter = executor == null
                ? new MeasurementReporter(s, registry, filter, rateUnit, durationUnit, clock, tags, transformer)
                : new MeasurementReporter(s, registry, filter, rateUnit, durationUnit, clock, tags, transformer, executor)
                ;
    }
    return reporter;
}

@Test
public void builder_api_with_tags() {
    String tagKey = "tag-name";
    String tagValue = "tag-value";

    Builder builder = InfluxdbReporter
            .forRegistry(registry)
            .tag(tagKey, tagValue)
            .protocol(new HttpInfluxdbProtocol());

    assertThat(builder.tags, notNullValue());
    assertThat(builder.tags, hasEntry(tagKey, tagValue));

    ScheduledReporter reporter = builder.build();
    assertThat(reporter, notNullValue());
}

@Test
public void testPointsSent() throws Exception {
    MetricRegistry registry = new MetricRegistry();
    new BasicJvmMetrics(registry);

    ScheduledReporter reporter = new ScheduledReporter(registry, "test", MetricFilter.ALL,
            TimeUnit.SECONDS, TimeUnit.MILLISECONDS) {

        @Override
        public void report(SortedMap<String, Gauge> gauges, SortedMap<String, Counter> counters,
                           SortedMap<String, Histogram> histograms,
                           SortedMap<String, Meter> meters, SortedMap<String, Timer> timers) {
            Assert.assertFalse(gauges.isEmpty());
            Assert.assertNotNull(gauges.get("jvm.uptime"));
            for (Map.Entry<String, Gauge> entry : gauges.entrySet()) {
                Assert.assertNotNull(entry.getValue().getValue());
            }
        }
    };

    reporter.report();
    reporter.close();
}

@Inject
public NbdStatsReporter(MetricRegistry metrics) {
    ScheduledReporter reporter = Slf4jReporter.forRegistry(metrics)
            .withLoggingLevel(Slf4jReporter.LoggingLevel.DEBUG)
            .outputTo(LOGGER)
            .convertRatesTo(TimeUnit.SECONDS)
            .convertDurationsTo(TimeUnit.MILLISECONDS)
            .build();
    reporter.start(5, TimeUnit.SECONDS);
}

@Override
protected void shutDown() throws Exception {
    for (ScheduledReporter entry : reporter.values()) {
        LOG.debug("Stopping metrics reporter: {}", entry);
        entry.stop();
    }
}

@Override
public ScheduledReporter newInstance(String qualifiedReplicaName) {
  InetSocketAddress address = new InetSocketAddressFactory().newInstance(graphiteHost);
  Graphite graphite = new Graphite(address);
  String prefix = DotJoiner.join(graphitePrefix, qualifiedReplicaName);
  return GraphiteReporter.forRegistry(runningMetricRegistry).prefixedWith(prefix).build(graphite);
}

public ScheduledReporter build(MetricRegistry registry) {
  try {
    return new ApptuitReporter(registry, getFilter(), getRateUnit(), getDurationUnit(),
        globalTags, apiKey, apiUrl != null ? new URL(apiUrl) : null, reportingMode);
  } catch (MalformedURLException e) {
    throw new IllegalArgumentException(e);
  }
}

private ScheduledReporter createReporter(ApptuitReporterFactory factory,
    Map<String, String> globalTags, String apiToken, String apiUrl, ReportingMode reportingMode) {
  factory.setRateUnit(TimeUnit.SECONDS);
  factory.setDurationUnit(TimeUnit.MILLISECONDS);
  globalTags.forEach(factory::addGlobalTag);
  factory.setApiKey(apiToken);
  factory.setApiUrl(apiUrl);
  factory.setReportingMode(reportingMode);

  return factory.build(registry);
}

@Before
public void setUp() throws Exception {
  mockFactory = mock(ApptuitReporterFactory.class);
  when(mockFactory.build(any(MetricRegistry.class))).thenReturn(mock(ScheduledReporter.class));
  mockConfigService = mock(ConfigService.class);
  when(mockConfigService.getGlobalTags()).thenReturn(ConfigService.getInstance().getGlobalTags());
}

/**
 * Will report to ignite that node is overloaded if the latency for requests from the given quantile will become
 * larger than a given threshold.
 *
 * @param ignite Local ignite instance to be used for overload reporting.
 * @param latencyThreshold Hard threshold after exceeding which node will report overload.
 * @param quantile A quantile in {@code [0..1]}.
 */
public void enableIgniteNodeOverloadStop(Ignite ignite, long latencyThreshold, double quantile) {
    ScheduledReporter reporter = IgniteNodeOverloadReporter.forRegistry(registry)
        .setIgnite(ignite)
        .setLatencyThreshold(latencyThreshold)
        .setQuantile(quantile)
        .build();
    reporter.start(reportFrequency, TimeUnit.MILLISECONDS);
}

@Override public ScheduledReporter getIfEnabled() {
    if (!config.isNodeOverloadStopEnabled()) {
        return null;
    }
    return new IgniteNodeOverloadReporter(
        registry,
        config.getWarmupDuration(),
        config.getLatencyThreshold(),
        config.getQuantile(),
        ignite
    );
}

@Override public ScheduledReporter getIfEnabled() {
    if (!config.isGangliaReportingEnabled()) {
        return null;
    }
    try {
        InetSocketAddress gangliaAddress = config.getGangliaAddress();
        GMetric ganglia = new GMetric(gangliaAddress.getHostString(), gangliaAddress.getPort(),
            GMetric.UDPAddressingMode.UNICAST, 1);
        return GangliaReporter.forRegistry(registry)
            .prefixedWith(StatisticsCollector.GANGLIA_METRICS_PREFIX)
            .build(ganglia);
    } catch (IOException e) {
        throw new RuntimeException(e);
    }
}

@Override public ScheduledReporter getIfEnabled() {
    if (!config.isDebugReportingEnabled()) {
        return null;
    }
    return ConsoleReporter.forRegistry(registry)
        .build();
}

@Override public ScheduledReporter getIfEnabled() {
    if (!config.isCsvReportingEnabled()) {
        return null;
    }
    return new HumanReadableCsvReporter(
        registry,
        config.getWarmupDuration(),
        new File(config.getCsvReportDirectory())
    );
}

public void startReporters() {
    for (ReporterProvider provider : reporterProviders) {
        ScheduledReporter reporter = provider.getIfEnabled();

        if (reporter != null) {
            reporters.add(reporter);
            reporter.start(config.getReportFrequency(), TimeUnit.MILLISECONDS);
        }
    }
}

@Inject
public NbdStatsReporter(MetricRegistry metrics) {
    ScheduledReporter reporter = Slf4jReporter.forRegistry(metrics)
            .withLoggingLevel(Slf4jReporter.LoggingLevel.DEBUG)
            .outputTo(LOGGER)
            .convertRatesTo(TimeUnit.SECONDS)
            .convertDurationsTo(TimeUnit.MILLISECONDS)
            .build();
    reporter.start(5, TimeUnit.SECONDS);
}

/**
 * Configures and builds a {@link ScheduledReporter} instance for the given registry.
 *
 * @param registry the metrics registry to report metrics from.
 * @return a reporter configured for the given metrics registry.
 */
@Override
public ScheduledReporter build(MetricRegistry registry) {

    StatsDReporter.Builder builder
            = StatsDReporter.forRegistry(registry)
                            .convertDurationsTo(getDurationUnit())
                            .convertRatesTo(getRateUnit())
                            .filter(getFilter());

    LOG.info("StatsDReporterFactory built with host: {}, port: {}", getHost(), getPort());

    return builder.build(getHost(), getPort());
}

public CodahaleMetricsProvider(CodahaleMetricsConfig conf) throws IOException {
    metricsOutputFrequencyInSecs = conf.getOutputFreqInSecs();
    int reporterCount = 0;
    for (Reporter reporter : conf.getReporters()) {
        ScheduledReporter codahaleReporter = null;
        switch (reporter) {
            case CONSOLE:
                codahaleReporter = createAndGetConfiguredConsoleReporter();
                break;
            case GRAPHITE:
                codahaleReporter = createAndGetConfiguredGraphiteReporter(conf.getPrefix(),
                                                                          conf.getGraphiteHostConfig());
                break;
            case CSV:
                codahaleReporter = createAndGetConfiguredCSVReporter(conf.getPrefix(),
                                                                     conf.getCsvDir());
                break;
            case SLF4J:
                codahaleReporter = createAndGetConfiguredSlf4jReporter(conf.getSlf4jLogger());
                break;
        }
        if (codahaleReporter != null) {
            reporters.add(codahaleReporter);
            reporterCount++;
        }
    }
    if (reporterCount == 0) {
        LOG.warn("No metric reporters found, so metrics won't be available");
    }
    startMetrics();
}

@Override
public void startMetrics() {
    for (ScheduledReporter r : reporters) {
        LOG.info("Starting metrics reporter {} reporting every {} Secs",
                 r.getClass().getCanonicalName(), metricsOutputFrequencyInSecs);
        r.start(metricsOutputFrequencyInSecs, TimeUnit.SECONDS);
    }
}

@Override
public void stopMetrics() {
    for (ScheduledReporter r : reporters) {
        r.report();
        LOG.info("Stopping reporter {}", r.toString());
        r.stop();
    }
}

private ScheduledReporter createAndGetConfiguredGraphiteReporter(String prefix, String graphiteHost) {
    LOG.info("Configuring Graphite reporter. Sendig data to host:port {}", graphiteHost);
    HostAndPort addr = HostAndPort.fromString(graphiteHost);

    final Graphite graphite = new Graphite(
            new InetSocketAddress(addr.getHostText(), addr.getPort()));

    return GraphiteReporter.forRegistry(metrics)
            .prefixedWith(prefix)
            .convertRatesTo(TimeUnit.SECONDS)
            .convertDurationsTo(TimeUnit.MILLISECONDS)
            .filter(MetricFilter.ALL)
            .build(graphite);
}

private ScheduledReporter createAndGetConfiguredSlf4jReporter(String slf4jLogger) {
    LOG.info("Configuring stats with SLF4J with logger {}", slf4jLogger);
    return Slf4jReporter.forRegistry(metrics)
            .outputTo(LoggerFactory.getLogger(slf4jLogger))
            .convertRatesTo(TimeUnit.SECONDS)
            .convertDurationsTo(TimeUnit.MILLISECONDS)
            .build();
}

@ProvidesIntoMap
@StringMapKey("statsdReporter")
public ScheduledReporter statsdReporter() {
    String endpoint = reporterProperties.statsdEndpoint == null ? "localhost" : reporterProperties.statsdEndpoint;

    final StatsdClient statsdClient = new StatsdClient(endpoint, reporterProperties.statsdPort);

    return StatsdReporter.forRegistry(metricRegistry)
            .convertRatesTo(TimeUnit.SECONDS)
            .convertDurationsTo(TimeUnit.SECONDS)
            .build(statsdClient);
}

@ProvidesIntoMap
@StringMapKey("consoleReporter")
public ScheduledReporter consoleReporter() {

    return ConsoleReporter
            .forRegistry(metricRegistry)
            .convertRatesTo(TimeUnit.SECONDS)
            .convertDurationsTo(TimeUnit.SECONDS)
            .build();
}

@Test
public void testExpansionFilterExclusion() throws Exception {
    String json =
            "{" +
                    "\"type\": \"datadogExpansionFiltered\"," +
                    "\"host\": \"test-host\"," +
                    "\"excludeExpansions\": [\"min\", \"max\", \"p75\", \"p95\", \"p98\", \"p99\", \"p999\"]," +
                    "\"transport\": {" +
                    "\"type\": \"http\"," +
                    "\"apiKey\": \"12345\"" +
                    "}" +
                    "}";


    ScheduledReporter reporter = createReporter(json);

    // Create a representative type.
    Histogram histogram = _metricRegistry.histogram("test.histogram");
    histogram.update(1);
    histogram.update(2);
    histogram.update(3);

    reporter.report();

    // Verify only the desired metrics were sent.  Notably min, max, and the nth percentiles should be absent.
    verify(_request).addCounter(argThat(hasCounter("test.histogram.count", 3)));
    verify(_request).addGauge(argThat(hasGauge("test.histogram.mean", 2)));
    verify(_request).addGauge(argThat(hasGauge("test.histogram.median", 2)));
    verify(_request).addGauge(argThat(hasGauge("test.histogram.stddev", 1.0)));

    // Send was called exactly once
    verify(_request).send();

    verifyNoMoreInteractions(_request);
}

public ScheduledReporter build(MetricRegistry registry) {
    return DatadogReporter.forRegistry(registry)
            .withTransport(_transport.build())
            .withHost(_host)
            .withTags(_tags)
            .filter(getFilter())
            .withExpansions(getExpansions())
            .convertDurationsTo(getDurationUnit())
            .convertRatesTo(getRateUnit())
            .build();
}

@Override
public ScheduledReporter build(MetricRegistry registry) {
    return HawkularReporter.builder(registry, tenant)
            .withNullableConfig(this)
            .filter(this.getFilter())
            .convertRatesTo(this.getRateUnit())
            .convertDurationsTo(this.getDurationUnit())
            .build();
}

@Synchronized
@Override
public void close() {
    for (ScheduledReporter r : reporters) {
        try {
            r.report();
            r.stop();
        } catch (Exception e) {
            log.error("Exception report or stop reporter", e);
        }
    }

    metrics.removeMatching(MetricFilter.ALL);
}

@Override
public ScheduledReporter getReporter(MetricConfig config) {
    scheduledReporter = new TestingScheduledReporter(
        registry,
        getClass().getName(),
        null,
        TimeUnit.MILLISECONDS,
        TimeUnit.MILLISECONDS);

    return scheduledReporter;
}

@Test
public void testInvalidCharacterReplacement() throws NoSuchMethodException, InvocationTargetException, IllegalAccessException {
    ScheduledDropwizardReporter reporter = new ScheduledDropwizardReporter() {
        @Override
        public ScheduledReporter getReporter(MetricConfig config) {
            return null;
        }
    };

    assertEquals("abc", reporter.filterCharacters("abc"));
    assertEquals("a--b-c-", reporter.filterCharacters("a..b.c."));
    assertEquals("ab-c", reporter.filterCharacters("a\"b.c"));
}

public ScheduledReporter build(MetricRegistry registry) {
  CirconusReporter reporter = CirconusReporter.forRegistry(registry)
      .withTransport(transport.build())
      .withHost(host)
      .withTags(tags)
      .filter(getFilter())
      .convertDurationsTo(getDurationUnit())
      .convertRatesTo(getRateUnit())
      .onlyCirconusAnalytics(circonus_analytics)
      .build();
  return reporter;
}

public static void main(String[] args) {
    InfluxDbReporter influxDbReporter = null;
    ScheduledReporter consoleReporter = null;
    Timer.Context context = null;
    try {
        final MetricRegistry registry = new MetricRegistry();
        consoleReporter = startConsoleReporter(registry);
        influxDbReporter = startInfluxDbReporter(registry, GetHttpSender());

        final Meter myMeter = registry.meter(MetricRegistry.name(SendToLocalInfluxDB.class, "testMetric"));

        final Timer myTimer = registry.timer("testTimer");
        context = myTimer.time();
        for (int i = 0; i < 5000; i++) {
            myMeter.mark();
            myMeter.mark(Math.round(Math.random() * 100.0));
            Thread.sleep(2000);
        }
    } catch (Exception exc) {
        exc.printStackTrace();
        System.exit(1);
    } finally {
        if (context != null) {
            context.stop();
        }
        if (influxDbReporter != null) {
            influxDbReporter.report();
            influxDbReporter.stop();
        }
        if (consoleReporter != null) {
            consoleReporter.report();
            consoleReporter.stop();
        }
        System.out.println("Finished");
    }
}