ElasticsearchReporter extends ScheduledReporter { @Override public void report(SortedMap<String, Gauge> gauges, SortedMap<String, Counter> counters, SortedMap<String, Histogram> histograms, SortedMap<String, Meter> meters, SortedMap<String, Timer> timers) { if (gauges.isEmpty() && counters.isEmpty() && histograms.isEmpty() && meters.isEmpty() && timers.isEmpty()) { LOGGER.info("All metrics empty, nothing to report"); return; } if (!checkedForIndexTemplate) { checkForIndexTemplate(); } final long timestamp = clock.getTime() / 1000; currentIndexName = index; if (indexDateFormat != null) { currentIndexName += "-" + indexDateFormat.format(new Date(timestamp * 1000)); } try { HttpURLConnection connection; try { connection = openConnection("/_bulk", "POST"); } catch (ElasticsearchConnectionException e) { LOGGER.error("Could not connect to any configured elasticsearch instances: {}", Arrays.asList(hosts), e); return; } List<JsonMetrics.JsonMetric> percolationMetrics = new ArrayList<>(); AtomicInteger entriesWritten = new AtomicInteger(0); for (Map.Entry<String, Gauge> entry : gauges.entrySet()) { if (entry.getValue().getValue() != null) { JsonMetrics.JsonMetric jsonMetric = new JsonMetrics.JsonGauge(name(prefix, entry.getKey()), timestamp, entry.getValue()); connection = writeJsonMetricAndRecreateConnectionIfNeeded(jsonMetric, connection, entriesWritten); addJsonMetricToPercolationIfMatching(jsonMetric, percolationMetrics); } } for (Map.Entry<String, Counter> entry : counters.entrySet()) { JsonCounter jsonMetric = new JsonCounter(name(prefix, entry.getKey()), timestamp, entry.getValue()); connection = writeJsonMetricAndRecreateConnectionIfNeeded(jsonMetric, connection, entriesWritten); addJsonMetricToPercolationIfMatching(jsonMetric, percolationMetrics); } for (Map.Entry<String, Histogram> entry : histograms.entrySet()) { JsonHistogram jsonMetric = new JsonHistogram(name(prefix, entry.getKey()), timestamp, entry.getValue()); connection = writeJsonMetricAndRecreateConnectionIfNeeded(jsonMetric, connection, entriesWritten); addJsonMetricToPercolationIfMatching(jsonMetric, percolationMetrics); } for (Map.Entry<String, Meter> entry : meters.entrySet()) { JsonMeter jsonMetric = new JsonMeter(name(prefix, entry.getKey()), timestamp, entry.getValue()); connection = writeJsonMetricAndRecreateConnectionIfNeeded(jsonMetric, connection, entriesWritten); addJsonMetricToPercolationIfMatching(jsonMetric, percolationMetrics); } for (Map.Entry<String, Timer> entry : timers.entrySet()) { JsonTimer jsonMetric = new JsonTimer(name(prefix, entry.getKey()), timestamp, entry.getValue()); connection = writeJsonMetricAndRecreateConnectionIfNeeded(jsonMetric, connection, entriesWritten); addJsonMetricToPercolationIfMatching(jsonMetric, percolationMetrics); } closeConnection(connection); if (percolationMetrics.size() > 0 && notifier != null) { for (JsonMetric jsonMetric : percolationMetrics) { List<String> matches = getPercolationMatches(jsonMetric); for (String match : matches) { notifier.notify(jsonMetric, match); } } } } catch (ElasticsearchConnectionException e) { LOGGER.error("Couldnt report to elasticsearch server", e); } catch (IOException e) { LOGGER.error("Couldnt report to elasticsearch server", e); } } ElasticsearchReporter(MetricRegistry registry, String[] hosts, int timeout,
                                 String index, String indexDateFormat, int bulkSize, Clock clock, String prefix, TimeUnit rateUnit, TimeUnit durationUnit,
                                 MetricFilter filter, MetricFilter percolationFilter, Notifier percolationNotifier, String timestampFieldname, Map<String, Object> additionalFields); static Builder forRegistry(MetricRegistry registry); @Override void report(SortedMap<String, Gauge> gauges,
                       SortedMap<String, Counter> counters,
                       SortedMap<String, Histogram> histograms,
                       SortedMap<String, Meter> meters,
                       SortedMap<String, Timer> timers);  }
@Test public void testThatEmptyMetricsDoNotResultInBrokenBulkRequest() throws Exception { long connectionsBeforeReporting = getTotalHttpConnections(); elasticsearchReporter.report(); long connectionsAfterReporting = getTotalHttpConnections(); org.assertj.core.api.Assertions.assertThat(connectionsAfterReporting).isEqualTo(connectionsBeforeReporting); }