@Override
public void report(
    SortedMap<String, Gauge> gauges,
    SortedMap<String, Counter> counters,
    SortedMap<String, Histogram> histograms,
    SortedMap<String, Meter> meters,
    SortedMap<String, Timer> timers) {
  final long now = clock.getTime() / 1000;
  final PayloadBuilder builder = new PayloadBuilder();

  builder.withMetricType(MetricType.gauge);
  gauges.forEach((metricName, metric) -> {
    builder.addGauge(prefix(metricName), now, (long) metric.getValue());
  });

  host.ifPresent(builder::withHost);
  tags.ifPresent(builder::withTags);

  client.send(builder.build());
}
 

private void reportHistogram(String name, Histogram histogram, long now) {
    if (canSkipMetric(name, histogram)) {
        return;
    }
    final Snapshot snapshot = histogram.getSnapshot();
    Map<String, Object> fields = new HashMap<String, Object>();
    fields.put("count", histogram.getCount());
    fields.put("min", snapshot.getMin());
    fields.put("max", snapshot.getMax());
    fields.put("mean", snapshot.getMean());
    fields.put("stddev", snapshot.getStdDev());
    fields.put("p50", snapshot.getMedian());
    fields.put("p75", snapshot.get75thPercentile());
    fields.put("p95", snapshot.get95thPercentile());
    fields.put("p98", snapshot.get98thPercentile());
    fields.put("p99", snapshot.get99thPercentile());
    fields.put("p999", snapshot.get999thPercentile());

    influxDb.appendPoints(
        new InfluxDbPoint(
            getMeasurementName(name),
            getTags(name),
            now,
            fields));
}
 

@Test
public void testResettingHistogram() {
    Histogram resettingHistogram = sfxMetrics.resettingHistogram("resettingHistogram");
    resettingHistogram.update(20);
    resettingHistogram.update(30);
    reporter.report();
    assertEquals(3, dbank.addDataPoints.size());
    assertEquals(2, dbank.lastValueFor(SOURCE_NAME, "resettingHistogram.count").getIntValue());
    assertEquals(20, dbank.lastValueFor(SOURCE_NAME, "resettingHistogram.min").getIntValue());
    assertEquals(30, dbank.lastValueFor(SOURCE_NAME, "resettingHistogram.max").getIntValue());
    dbank.addDataPoints.clear();
    resettingHistogram.update(25);
    reporter.report();
    assertEquals(3, dbank.addDataPoints.size());
    assertEquals(3, dbank.lastValueFor(SOURCE_NAME, "resettingHistogram.count").getIntValue());
    assertEquals(25, dbank.lastValueFor(SOURCE_NAME, "resettingHistogram.min").getIntValue());
    assertEquals(25, dbank.lastValueFor(SOURCE_NAME, "resettingHistogram.max").getIntValue());
}
 

@SuppressWarnings("rawtypes")
@Test
public void metrics2CycleIsNonDestructive() {
    metrics2Reporter.setDropwizardCounters(Collections.unmodifiableSortedMap(new TreeMap<String, Counter>()));
    metrics2Reporter.setDropwizardGauges(Collections.unmodifiableSortedMap(new TreeMap<String, Gauge>()));
    metrics2Reporter.setDropwizardHistograms(Collections.unmodifiableSortedMap(new TreeMap<String, Histogram>()));
    metrics2Reporter.setDropwizardMeters(Collections.unmodifiableSortedMap(new TreeMap<String, Meter>()));
    metrics2Reporter.setDropwizardTimers(Collections.unmodifiableSortedMap(new TreeMap<String, Timer>()));

    MetricsCollector collector = mock(MetricsCollector.class);
    MetricsRecordBuilder recordBuilder = mock(MetricsRecordBuilder.class);

    Mockito.when(collector.addRecord(recordName)).thenReturn(recordBuilder);

    metrics2Reporter.getMetrics(collector, true);
}
 

@Override
public void serialize(Histogram histogram, JsonGenerator json,
                      SerializerProvider provider) throws IOException {
    json.writeStartObject();
    final Snapshot snapshot = histogram.getSnapshot();
    json.writeNumberField("count", histogram.getCount());
    json.writeNumberField("max", snapshot.getMax());
    json.writeNumberField("mean", snapshot.getMean());
    json.writeNumberField("min", snapshot.getMin());
    json.writeNumberField("p50", snapshot.getMedian());
    json.writeNumberField("p75", snapshot.get75thPercentile());
    json.writeNumberField("p95", snapshot.get95thPercentile());
    json.writeNumberField("p98", snapshot.get98thPercentile());
    json.writeNumberField("p99", snapshot.get99thPercentile());
    json.writeNumberField("p999", snapshot.get999thPercentile());

    if (this.showSamples) {
        json.writeObjectField("values", snapshot.getValues());
    }

    json.writeNumberField("stddev", snapshot.getStdDev());
    json.writeEndObject();
}
 

@Test
public void testInit() throws StageException {

  MetricRegistry metrics = metricAggregationProcessor.getMetrics();
  SortedMap<String, Timer> timers = metrics.getTimers();

  Assert.assertEquals(5, timers.size()); // 1 each for 4 stages, 1 for pipeline

  SortedMap<String, Counter> counters = metrics.getCounters();
  Assert.assertEquals(24, counters.size()); // 4 each for 4 stages, 5 for pipeline, one each for 3 output lanes

  SortedMap<String, Meter> meters = metrics.getMeters();
  Assert.assertEquals(24, meters.size()); // 4 each for 4 stages, 5 for pipeline, one each for 3 output lanes

  SortedMap<String, Histogram> histograms = metrics.getHistograms();
  Assert.assertEquals(20, histograms.size()); // 4 each for 4 stages, 4 for pipeline
}
 

public static String printHistogram(String name, Histogram histogram) {
	StringBuilder sb = new StringBuilder();
	sb.append(String.format("              name = %s%n", name));
	sb.append(String.format("             count = %d%n", histogram.getCount()));
	Snapshot snapshot = histogram.getSnapshot();
	sb.append(String.format("               min = %d%n", snapshot.getMin()));
	sb.append(String.format("               max = %d%n", snapshot.getMax()));
	sb.append(String.format("              mean = %2.2f%n", snapshot.getMean()));
	sb.append(String.format("            stddev = %2.2f%n", snapshot.getStdDev()));
	sb.append(String.format("            median = %2.2f%n", snapshot.getMedian()));
	sb.append(String.format("              75%% <= %2.2f%n", snapshot.get75thPercentile()));
	sb.append(String.format("              95%% <= %2.2f%n", snapshot.get95thPercentile()));
	sb.append(String.format("              98%% <= %2.2f%n", snapshot.get98thPercentile()));
	sb.append(String.format("              99%% <= %2.2f%n", snapshot.get99thPercentile()));
	sb.append(String.format("            99.9%% <= %2.2f%n", snapshot.get999thPercentile()));
	return sb.toString();
}
 

@DataProvider(value = {
    "null",
    "0",
    "1",
    "2"
}, splitBy = "\\|")
@Test
public void getNamedHistogram_with_varargs_dimensions_creates_dimensioned_histogram_using_sfx_mechanisms(
    Integer numDimensions
) {
    // given
    String histogramName = UUID.randomUUID().toString();
    Pair<String, String>[] varargDims = generateVarargDimensions(numDimensions);
    List<Pair<String, String>> dimsAsList = (varargDims == null) ? null : Arrays.asList(varargDims);

    // when
    Histogram result = sfxImpl.getNamedHistogram(histogramName, varargDims);

    // then
    verifyMetricCreation(histogramBuilderMock, histogramTaggerMock, histogramName, dimsAsList, histogramMock, result);
}
 

private void notifyListenerOfRemovedMetric(
    final MetricId name, final Metric metric, final SemanticMetricRegistryListener listener
) {
    if (metric instanceof Gauge) {
        listener.onGaugeRemoved(name);
    } else if (metric instanceof Counter) {
        listener.onCounterRemoved(name);
    } else if (metric instanceof Histogram) {
        listener.onHistogramRemoved(name);
    } else if (metric instanceof Meter) {
        listener.onMeterRemoved(name);
    } else if (metric instanceof Timer) {
        listener.onTimerRemoved(name);
    } else if (metric instanceof DerivingMeter) {
        listener.onDerivingMeterRemoved(name);
    } else {
        throw new IllegalArgumentException("Unknown metric type: " + metric.getClass());
    }
}
 

@Override
protected void report(SortedMap<String, Gauge> gauges, SortedMap<String, Counter> counters,
    SortedMap<String, Histogram> histograms, SortedMap<String, Meter> meters, SortedMap<String, Timer> timers,
    Map<String, Object> tags) {

  String prefix = getMetricNamePrefix(tags);
  long timestamp = System.currentTimeMillis() / 1000l;

  try {
    for (Map.Entry<String, Gauge> gauge : gauges.entrySet()) {
      reportGauge(prefix, gauge.getKey(), gauge.getValue(), timestamp);
    }

    for (Map.Entry<String, Counter> counter : counters.entrySet()) {
      reportCounter(prefix, counter.getKey(), counter.getValue(), timestamp);
    }

    for (Map.Entry<String, Histogram> histogram : histograms.entrySet()) {
      reportHistogram(prefix, histogram.getKey(), histogram.getValue(), timestamp);
    }

    for (Map.Entry<String, Meter> meter : meters.entrySet()) {
      reportMetered(prefix, meter.getKey(), meter.getValue(), timestamp);
    }

    for (Map.Entry<String, Timer> timer : timers.entrySet()) {
      reportTimer(prefix, timer.getKey(), timer.getValue(), timestamp);
    }

    this.graphitePusher.flush();

  } catch (IOException ioe) {
    LOGGER.error("Error sending metrics to Graphite", ioe);
    try {
      this.graphitePusher.close();
    } catch (IOException innerIoe) {
      LOGGER.error("Error closing the Graphite sender", innerIoe);
    }
  }
}
 

/**
 * Creates a json object for a certain metric.
 *
 * @param metric metric object
 * @return json object
 */
private ObjectNode json(Metric metric) {
    ObjectMapper mapper = new ObjectMapper();
    ObjectNode objectNode = mapper.createObjectNode();
    ObjectNode dataNode = mapper.createObjectNode();

    if (metric instanceof Counter) {
        dataNode.put(COUNTER, ((Counter) metric).getCount());
        objectNode.set(COUNTER, dataNode);
    } else if (metric instanceof Gauge) {
        objectNode.put(VALUE, ((Gauge) metric).getValue().toString());
        objectNode.set(GAUGE, dataNode);
    } else if (metric instanceof Meter) {
        dataNode.put(COUNTER, ((Meter) metric).getCount());
        dataNode.put(MEAN_RATE, ((Meter) metric).getMeanRate());
        dataNode.put(ONE_MIN_RATE, ((Meter) metric).getOneMinuteRate());
        dataNode.put(FIVE_MIN_RATE, ((Meter) metric).getFiveMinuteRate());
        dataNode.put(FIFT_MIN_RATE, ((Meter) metric).getFifteenMinuteRate());
        objectNode.set(METER, dataNode);
    } else if (metric instanceof Histogram) {
        dataNode.put(COUNTER, ((Histogram) metric).getCount());
        dataNode.put(MEAN, ((Histogram) metric).getSnapshot().getMean());
        dataNode.put(MIN, ((Histogram) metric).getSnapshot().getMin());
        dataNode.put(MAX, ((Histogram) metric).getSnapshot().getMax());
        dataNode.put(STDDEV, ((Histogram) metric).getSnapshot().getStdDev());
        objectNode.set(HISTOGRAM, dataNode);
    } else if (metric instanceof Timer) {
        dataNode.put(COUNTER, ((Timer) metric).getCount());
        dataNode.put(MEAN_RATE, ((Timer) metric).getMeanRate());
        dataNode.put(ONE_MIN_RATE, ((Timer) metric).getOneMinuteRate());
        dataNode.put(FIVE_MIN_RATE, ((Timer) metric).getFiveMinuteRate());
        dataNode.put(FIFT_MIN_RATE, ((Timer) metric).getFifteenMinuteRate());
        dataNode.put(MEAN, nanoToMs(((Timer) metric).getSnapshot().getMean()));
        dataNode.put(MIN, nanoToMs(((Timer) metric).getSnapshot().getMin()));
        dataNode.put(MAX, nanoToMs(((Timer) metric).getSnapshot().getMax()));
        dataNode.put(STDDEV, nanoToMs(((Timer) metric).getSnapshot().getStdDev()));
        objectNode.set(TIMER, dataNode);
    }
    return objectNode;
}
 

/**
 * Test the case where all requests time out within the GetOperation.
 * @throws Exception
 */
@Test
public void testRouterRequestTimeoutAllFailure() throws Exception {
  doPut();
  GetBlobOperation op = createOperation(routerConfig, null);
  op.poll(requestRegistrationCallback);
  while (!op.isOperationComplete()) {
    time.sleep(routerConfig.routerRequestTimeoutMs + 1);
    op.poll(requestRegistrationCallback);
  }
  // At this time requests would have been created for all replicas, as none of them were delivered,
  // and cross-colo proxying is enabled by default.
  Assert.assertEquals("Must have attempted sending requests to all replicas", replicasCount,
      correlationIdToGetOperation.size());
  assertFailureAndCheckErrorCode(op, RouterErrorCode.OperationTimedOut);

  // test that timed out response won't update latency histogram if exclude timeout is enabled.
  assumeTrue(operationTrackerType.equals(AdaptiveOperationTracker.class.getSimpleName()));
  AdaptiveOperationTracker tracker = (AdaptiveOperationTracker) op.getFirstChunkOperationTrackerInUse();
  Histogram localColoTracker =
      tracker.getLatencyHistogram(RouterTestHelpers.getAnyReplica(blobId, true, localDcName));
  Histogram crossColoTracker =
      tracker.getLatencyHistogram(RouterTestHelpers.getAnyReplica(blobId, false, localDcName));
  Assert.assertEquals("Timed-out response shouldn't be counted into local colo latency histogram", 0,
      localColoTracker.getCount());
  Assert.assertEquals("Timed-out response shouldn't be counted into cross colo latency histogram", 0,
      crossColoTracker.getCount());
}
 

@Override
public void updateHistogram(String name, long count) {
    try {
        histogramLock.lock();
        Histogram histogram = histograms.get(name);
        histogram.update(count);
    } catch (ExecutionException e) {
        throw new IllegalStateException("Error retrieving meter " + name + " from the metric registry ", e);
    } finally {
        histogramLock.unlock();
    }
}
 

@SuppressWarnings("rawtypes")
@Test
public void cachedMetricsAreClearedAfterCycle() {
    // After we perform a metrics2 reporting cycle, the maps should be reset to avoid double-reporting
    TreeMap<String, Counter> counters = new TreeMap<>();
    TreeMap<String, Gauge> gauges = new TreeMap<>();
    TreeMap<String, Histogram> histograms = new TreeMap<>();
    TreeMap<String, Meter> meters = new TreeMap<>();
    TreeMap<String, Timer> timers = new TreeMap<>();

    metrics2Reporter.setDropwizardCounters(counters);
    metrics2Reporter.setDropwizardGauges(gauges);
    metrics2Reporter.setDropwizardHistograms(histograms);
    metrics2Reporter.setDropwizardMeters(meters);
    metrics2Reporter.setDropwizardTimers(timers);

    MetricsCollector collector = mock(MetricsCollector.class);
    MetricsRecordBuilder recordBuilder = mock(MetricsRecordBuilder.class);

    Mockito.when(collector.addRecord(recordName)).thenReturn(recordBuilder);

    metrics2Reporter.getMetrics(collector, true);

    assertTrue(counters != metrics2Reporter.getDropwizardCounters());
    assertEquals(0, metrics2Reporter.getDropwizardCounters().size());
    assertTrue(gauges != metrics2Reporter.getDropwizardGauges());
    assertEquals(0, metrics2Reporter.getDropwizardGauges().size());
    assertTrue(histograms != metrics2Reporter.getDropwizardHistograms());
    assertEquals(0, metrics2Reporter.getDropwizardHistograms().size());
    assertTrue(meters != metrics2Reporter.getDropwizardMeters());
    assertEquals(0, metrics2Reporter.getDropwizardMeters().size());
    assertTrue(timers != metrics2Reporter.getDropwizardTimers());
    assertEquals(0, metrics2Reporter.getDropwizardTimers().size());
}
 

public static Object getMetricValue(
  MetricElement metricElement,
  MetricType metricType,
  Metric metric
) throws ObserverException {
  Object value;
  switch (metricType) {
    case HISTOGRAM:
      value = MetricRuleEvaluatorHelper.getHistogramValue((Histogram) metric, metricElement);
      break;
    case METER:
      value = MetricRuleEvaluatorHelper.getMeterValue((ExtendedMeter) metric, metricElement);
      break;
    case COUNTER:
      value = MetricRuleEvaluatorHelper.getCounterValue((Counter) metric, metricElement);
      break;
    case TIMER:
      value = MetricRuleEvaluatorHelper.getTimerValue((Timer) metric, metricElement);
      break;
    case GAUGE:
      value = MetricRuleEvaluatorHelper.getGaugeValue((Gauge) metric, metricElement);
      break;
    default :
      throw new IllegalArgumentException(Utils.format("Unknown metric type '{}'", metricType));
  }
  return value;
}
 

/**
 * Instantiates a new Histogram dynamic metric.
 *
 * @param metricName   the metric name
 * @param metricObject the metric object
 */
public HistogramDynamicMetric(String metricName, Histogram metricObject) {
  super(metricName, metricObject);

  _attributeNameSet = new HashSet<>();
  Iterator<MBeanAttributeInfo> iter = getAttributeInfos().iterator();
  while (iter.hasNext()) {
    MBeanAttributeInfo attributeInfo = iter.next();
    _attributeNameSet.add(attributeInfo.getName());
  }
}
 

private void initMetrics() {
    //Main DB Metrics
    metrics.register(metricName("write"), new Timer(new UniformReservoir()));
    metrics.register(metricName("write.rate"), new Meter());
    metrics.register(metricName("read"), new Timer(new UniformReservoir()));
    metrics.register(metricName("read.rate"), new Meter());
    metrics.register(metricName("scan"), new Timer(new UniformReservoir()));
    metrics.register(metricName("scan.rate"), new Meter());

    //Write
    metrics.register(metricName("write.concurrentMemoryTableSerializers"), new Histogram(new UniformReservoir()));
    metrics.register(metricName("write.memoryTableSerialize"), new Timer(new UniformReservoir()));

    //Read
    metrics.register(metricName("read.tablesConsulted"), new Histogram(new UniformReservoir()));
    metrics.register(metricName("read.bloomFilterFalsePositiveRate"), new CacheHitGauge());
    metrics.register(metricName("read.recordNotFoundRate"), new CacheHitGauge());

    //FileTable
    metrics.register(metricName("table.cacheHitRate"), new CacheHitGauge());

    //Index
    metrics.register(metricName("index.searchLevels"), new Histogram(new UniformReservoir()));
    metrics.register(metricName("index.cacheHitRate"), new CacheHitGauge());

    //Compactor
    metrics.register(metricName("compactor.concurrentTasks"), new Histogram(new UniformReservoir()));
    metrics.register(metricName("compactor.taskExecution"), new Timer(new UniformReservoir()));
}
 

public static JsonObject convertMetric(Metric metric, TimeUnit rateUnit, TimeUnit durationUnit) {
  if (metric instanceof Timer) {
    return toJson((Timer) metric, rateUnit, durationUnit);
  } else if (metric instanceof Gauge) {
    return toJson((Gauge) metric);
  } else if (metric instanceof Counter) {
    return toJson((Counter) metric);
  } else if (metric instanceof Histogram) {
    return toJson((Histogram) metric);
  } else if (metric instanceof Meter) {
    return toJson((Meter) metric, rateUnit);
  } else {
    throw new IllegalArgumentException("Unknown metric " + metric);
  }
}
 

/**
 * Updates the {@code tracker} to mimic {@code numRequests} each taking {@code latency} ms.
 * @param tracker the {@link Histogram} to update
 * @param numRequests the number of requests (data points)
 * @param latencyRange the range of latencies (in ms) to generate and record.
 */
private void primeTracker(Histogram tracker, long numRequests, Pair<Long, Long> latencyRange) {
  for (long i = 0; i < numRequests; i++) {
    // Given latencyRange specifies boundaries of latency: low = latencyRange.getFirst(), high = latencyRange.getSecond().
    // Any randomly generated latency should fall in the range [low, high).
    long latency = Utils.getRandomLong(TestUtils.RANDOM, latencyRange.getSecond() - latencyRange.getFirst())
        + latencyRange.getFirst();
    tracker.update(latency);
  }
}
 

@SafeVarargs
public final @NotNull Histogram getNamedHistogram(
    @NotNull String histogramName,
    @Nullable Pair<String, String>... dimensions
) {
    return getNamedHistogram(histogramName, convertDimensionsToList(dimensions));
}
 

/** @return a summary of {@code hist}. */
public static String getHistogramReport(final Histogram hist) {
  Snapshot sn = hist.getSnapshot();
  return "mean=" + DOUBLE_FORMAT.format(sn.getMean()) +
      ", min=" + DOUBLE_FORMAT.format(sn.getMin()) +
      ", max=" + DOUBLE_FORMAT.format(sn.getMax()) +
      ", stdDev=" + DOUBLE_FORMAT.format(sn.getStdDev()) +
      ", 50th=" + DOUBLE_FORMAT.format(sn.getMedian()) +
      ", 75th=" + DOUBLE_FORMAT.format(sn.get75thPercentile()) +
      ", 95th=" + DOUBLE_FORMAT.format(sn.get95thPercentile()) +
      ", 99th=" + DOUBLE_FORMAT.format(sn.get99thPercentile()) +
      ", 99.9th=" + DOUBLE_FORMAT.format(sn.get999thPercentile()) +
      ", 99.99th=" + DOUBLE_FORMAT.format(sn.getValue(0.9999)) +
      ", 99.999th=" + DOUBLE_FORMAT.format(sn.getValue(0.99999));
}
 

@Setup(Level.Iteration)
public void setup() {
    registry = new MetricRegistry();
    histogram = registry.histogram("histogram");
    histogramSlidingTimeWindow =
            registry.register("slidingTimeWindowHistogram",
                    new Histogram(new SlidingTimeWindowReservoir(10, TimeUnit.SECONDS)));
    histogramUniform =
            registry.register("uniformHistogram",
                    new Histogram(new UniformReservoir()));
}
 

@Test
public void testHistogram() throws Exception {
  final Histogram h = metricRegistry.histogram("histo.gram");
  IntStream.range(1, 10).forEach(h::update);

  reporter.reportOnce();

  verifyHistogramStats("histo.gram", "histogram");
}
 

private void reportHistogram(MetricId key, Histogram value) {
    key = MetricId.join(prefix, key);

    final Metric m = FastForward
        .metric(key.getKey())
        .attributes(key.getTags())
        .attribute(METRIC_TYPE, "histogram");

    reportHistogram(m, value.getSnapshot());
}
 

public Histogram getHistogram(String name) {
    try {
        histogramLock.lock();
        return histograms.get(name);
    } catch (ExecutionException e) {
        throw new IllegalStateException("Error retrieving histogram " + name + " from the metric registry ", e);
    } finally {
        histogramLock.unlock();
    }
}
 

/**
 * Update the histogram (or creates it if it does not exist) for the specified key/metricName pair by the given value.
 * @param classSimpleName the simple name of the underlying class
 * @param key the key (i.e. topic or partition) for the metric
 * @param metricName the metric name
 * @param value the value to update on the histogram
 */
public void createOrUpdateHistogram(String classSimpleName, String key, String metricName, long value) {
  validateArguments(classSimpleName, metricName);
  // create and register the metric if it does not exist
  Histogram histogram = (Histogram) checkCache(classSimpleName, key, metricName).orElseGet(() -> {
    Histogram newHistogram = _metricRegistry.histogram(MetricRegistry.name(classSimpleName, key, metricName));
    updateCache(classSimpleName, key, metricName, newHistogram);
    return newHistogram;
  });
  histogram.update(value);
}
 

private static JsonObject toJson(Histogram histogram) {
  Snapshot snapshot = histogram.getSnapshot();
  JsonObject json = new JsonObject();
  json.put("type", "histogram");
  json.put("count", histogram.getCount());

  // Snapshot
  populateSnapshot(json, snapshot, 1);

  return json;
}
 

@Test
public void reportsHistograms() throws Exception {
    Histogram histogram = mock(Histogram.class);
    when(histogram.getCount()).thenReturn(1L);

    Snapshot snapshot = mock(Snapshot.class);
    when(snapshot.getMax()).thenReturn(2L);
    when(snapshot.getMean()).thenReturn(3.0);
    when(snapshot.getMin()).thenReturn(4L);
    when(snapshot.getStdDev()).thenReturn(5.0);
    when(snapshot.getMedian()).thenReturn(6.0);
    when(snapshot.get75thPercentile()).thenReturn(7.0);
    when(snapshot.get95thPercentile()).thenReturn(8.0);
    when(snapshot.get98thPercentile()).thenReturn(9.0);
    when(snapshot.get99thPercentile()).thenReturn(10.0);
    when(snapshot.get999thPercentile()).thenReturn(11.0);

    when(histogram.getSnapshot()).thenReturn(snapshot);

    reporter.report(emptyMap(), emptyMap(), map("histogram", histogram), emptyMap(), emptyMap());

    InOrder inOrder = inOrder(graphite);
    inOrder.verify(graphite).connect();
    inOrder.verify(graphite).send("prefix.histogram.count", "1", TIMESTAMP);
    inOrder.verify(graphite).send("prefix.histogram.max", "2", TIMESTAMP);
    inOrder.verify(graphite).send("prefix.histogram.mean", "3.00", TIMESTAMP);
    inOrder.verify(graphite).send("prefix.histogram.min", "4", TIMESTAMP);
    inOrder.verify(graphite).send("prefix.histogram.stddev", "5.00", TIMESTAMP);
    inOrder.verify(graphite).send("prefix.histogram.p50", "6.00", TIMESTAMP);
    inOrder.verify(graphite).send("prefix.histogram.p75", "7.00", TIMESTAMP);
    inOrder.verify(graphite).send("prefix.histogram.p95", "8.00", TIMESTAMP);
    inOrder.verify(graphite).send("prefix.histogram.p98", "9.00", TIMESTAMP);
    inOrder.verify(graphite).send("prefix.histogram.p99", "10.00", TIMESTAMP);
    inOrder.verify(graphite).send("prefix.histogram.p999", "11.00", TIMESTAMP);
    inOrder.verify(graphite).flush();
    inOrder.verify(graphite).close();

    verifyNoMoreInteractions(graphite);
}
 

@SuppressWarnings("rawtypes")
@Override
public void report(SortedMap<String, Gauge> gauges,
		SortedMap<String, Counter> counters,
		SortedMap<String, Histogram> histograms,
		SortedMap<String, Meter> meters, SortedMap<String, Timer> timers) {
	
	final Map<String, Object> result = new HashMap<String, Object>(16);
	
	result.put("hostName", hostName);
	result.put("ip", ip);
	result.put("rateUnit", getRateUnit());
	result.put("durationUnit", getDurationUnit());
	
	result.put("gauges", addPrefix(gauges));
	result.put("counters", addPrefix(counters));
	result.put("histograms", addPrefix(histograms));
	result.put("meters", addPrefix(meters));
	result.put("timers", addPrefix(timers));
	
	result.put("clock", System.currentTimeMillis());
	
	kafkaExecutor.execute(new Runnable() {
		@Override
		public void run() {
			try {
			KeyedMessage<String, String> message = new KeyedMessage<String, String>(
					topic, "" + count++, mapper.writeValueAsString(result));
				producer.send(message);
			} catch (Exception e) {
				logger.error("send metrics to kafka error!", e);
			}
		}
	});
}
 

@Test(dependsOnMethods = "testGetMetrics")
@SuppressWarnings("unchecked")
public void testGetMetricsWithFilter() {
  MetricFilter filter = new MetricFilter() {
    @Override public boolean matches(String name, Metric metric) {
      return !name.equals(MetricContext.GOBBLIN_METRICS_NOTIFICATIONS_TIMER_NAME);
    }
  };

  Map<String, Counter> counters = this.context.getCounters(filter);
  Assert.assertEquals(counters.size(), 1);
  Assert.assertTrue(
      counters.containsKey(RECORDS_PROCESSED));

  Map<String, Meter> meters = this.context.getMeters(filter);
  Assert.assertEquals(meters.size(), 1);
  Assert.assertTrue(
      meters.containsKey(RECORD_PROCESS_RATE));

  Map<String, Histogram> histograms = this.context.getHistograms(filter);
  Assert.assertEquals(histograms.size(), 1);
  Assert.assertTrue(
      histograms.containsKey(RECORD_SIZE_DISTRIBUTION));

  Map<String, Timer> timers = this.context.getTimers(filter);
  Assert.assertEquals(timers.size(), 1);
  Assert.assertTrue(timers.containsKey(TOTAL_DURATION));

  Map<String, Gauge> gauges = this.context.getGauges(filter);
  Assert.assertEquals(gauges.size(), 1);
  Assert.assertTrue(gauges.containsKey(QUEUE_SIZE));
}