hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/metrics2/lib/TestMutableMetrics.java - hadoop - Git at Google

 /**
  * Licensed to the Apache Software Foundation (ASF) under one
  * or more contributor license agreements.  See the NOTICE file
  * distributed with this work for additional information
  * regarding copyright ownership.  The ASF licenses this file
  * to you under the Apache License, Version 2.0 (the
  * "License"); you may not use this file except in compliance
  * with the License.  You may obtain a copy of the License at
  *
  *     http://www.apache.org/licenses/LICENSE-2.0
  *
  * Unless required by applicable law or agreed to in writing, software
  * distributed under the License is distributed on an "AS IS" BASIS,
  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */

 package org.apache.hadoop.metrics2.lib;

 import static org.apache.hadoop.metrics2.lib.Interns.info;
 import static org.apache.hadoop.test.MetricsAsserts.*;
 import static org.mockito.AdditionalMatchers.eq;
 import static org.mockito.AdditionalMatchers.geq;
 import static org.mockito.AdditionalMatchers.leq;
 import static org.mockito.Matchers.anyLong;
 import static org.mockito.Matchers.eq;
 import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
 import static org.junit.Assert.*;

 import java.util.Map;
 import java.util.Map.Entry;
 import java.util.Random;
 import java.util.concurrent.CountDownLatch;

 import org.apache.hadoop.metrics2.MetricsRecordBuilder;
 import org.apache.hadoop.metrics2.util.Quantile;
 import org.junit.Test;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;

 /**
  * Test metrics record builder interface and mutable metrics
  */
 public class TestMutableMetrics {

   private static final Logger LOG =
       LoggerFactory.getLogger(TestMutableMetrics.class);
   private final double EPSILON = 1e-42;

   /**
    * Test the snapshot method
    */
   @Test public void testSnapshot() {
     MetricsRecordBuilder mb = mockMetricsRecordBuilder();

     MetricsRegistry registry = new MetricsRegistry("test");
     registry.newCounter("c1", "int counter", 1);
     registry.newCounter("c2", "long counter", 2L);
     registry.newGauge("g1", "int gauge", 3);
     registry.newGauge("g2", "long gauge", 4L);
     registry.newGauge("g3", "float gauge", 5f);
     registry.newStat("s1", "stat", "Ops", "Time", true).add(0);
     registry.newRate("s2", "stat", false).add(0);

     registry.snapshot(mb, true);

     MutableStat s2 = (MutableStat) registry.get("s2");

     s2.snapshot(mb, true); // should get the same back.
     s2.add(1);
     s2.snapshot(mb, true); // should get new interval values back

     verify(mb).addCounter(info("c1", "int counter"), 1);
     verify(mb).addCounter(info("c2", "long counter"), 2L);
     verify(mb).addGauge(info("g1", "int gauge"), 3);
     verify(mb).addGauge(info("g2", "long gauge"), 4L);
     verify(mb).addGauge(info("g3", "float gauge"), 5f);
     verify(mb).addCounter(info("S1NumOps", "Number of ops for stat"), 1L);
     verify(mb).addGauge(eq(info("S1AvgTime", "Average time for stat")),
                            eq(0.0, EPSILON));
     verify(mb).addGauge(eq(info("S1StdevTime",
                                 "Standard deviation of time for stat")),
                            eq(0.0, EPSILON));
     verify(mb).addGauge(eq(info("S1IMinTime",
                                 "Interval min time for stat")),
                            eq(0.0, EPSILON));
     verify(mb).addGauge(eq(info("S1IMaxTime",
                                 "Interval max time for stat")),
                            eq(0.0, EPSILON));
     verify(mb).addGauge(eq(info("S1MinTime","Min time for stat")),
                            eq(0.0, EPSILON));
     verify(mb).addGauge(eq(info("S1MaxTime","Max time for stat")),
                            eq(0.0, EPSILON));
     verify(mb).addGauge(
         eq(info("S1INumOps", "Interval number of ops for stat")),
         eq(1L));

     verify(mb, times(2))
         .addCounter(info("S2NumOps", "Number of ops for stat"), 1L);
     verify(mb, times(2)).addGauge(eq(info("S2AvgTime",
                                           "Average time for stat")),
                                   eq(0.0, EPSILON));
     verify(mb).addCounter(info("S2NumOps", "Number of ops for stat"), 2L);
     verify(mb).addGauge(eq(info("S2AvgTime", "Average time for stat")),
                            eq(1.0, EPSILON));

     // Add one more sample to s1 and verify that total number of ops
     // has increased to 2, but interval number is 1 for both intervals.
     MutableStat s1 = (MutableStat) registry.get("s1");
     s1.add(0);
     registry.snapshot(mb, true);
     verify(mb).addCounter(info("S1NumOps", "Number of ops for stat"), 2L);
     verify(mb, times(2)).addGauge(
         eq(info("S1INumOps", "Interval number of ops for stat")),
         eq(1L));
   }

   interface TestProtocol {
     void foo();
     void bar();
   }

   @Test public void testMutableRates() {
     MetricsRecordBuilder rb = mockMetricsRecordBuilder();
     MetricsRegistry registry = new MetricsRegistry("test");
     MutableRates rates = new MutableRates(registry);

     rates.init(TestProtocol.class);
     registry.snapshot(rb, false);

     assertCounter("FooNumOps", 0L, rb);
     assertGauge("FooAvgTime", 0.0, rb);
     assertCounter("BarNumOps", 0L, rb);
     assertGauge("BarAvgTime", 0.0, rb);
   }

   @Test public void testMutableRatesWithAggregationInit() {
     MetricsRecordBuilder rb = mockMetricsRecordBuilder();
     MutableRatesWithAggregation rates = new MutableRatesWithAggregation();

     rates.init(TestProtocol.class);
     rates.snapshot(rb, false);

     assertCounter("FooNumOps", 0L, rb);
     assertGauge("FooAvgTime", 0.0, rb);
     assertCounter("BarNumOps", 0L, rb);
     assertGauge("BarAvgTime", 0.0, rb);
   }

   @Test public void testMutableRatesWithAggregationSingleThread() {
     MutableRatesWithAggregation rates = new MutableRatesWithAggregation();

     rates.add("foo", 1);
     rates.add("bar", 5);

     MetricsRecordBuilder rb = mockMetricsRecordBuilder();
     rates.snapshot(rb, false);
     assertCounter("FooNumOps", 1L, rb);
     assertGauge("FooAvgTime", 1.0, rb);
     assertCounter("BarNumOps", 1L, rb);
     assertGauge("BarAvgTime", 5.0, rb);

     rates.add("foo", 1);
     rates.add("foo", 3);
     rates.add("bar", 6);

     rb = mockMetricsRecordBuilder();
     rates.snapshot(rb, false);
     assertCounter("FooNumOps", 3L, rb);
     assertGauge("FooAvgTime", 2.0, rb);
     assertCounter("BarNumOps", 2L, rb);
     assertGauge("BarAvgTime", 6.0, rb);
   }

   @Test public void testMutableRatesWithAggregationManyThreads()
       throws InterruptedException {
     final MutableRatesWithAggregation rates = new MutableRatesWithAggregation();

     final int n = 10;
     long[] opCount = new long[n];
     double[] opTotalTime = new double[n];

     for (int i = 0; i < n; i++) {
       opCount[i] = 0;
       opTotalTime[i] = 0;
       // Initialize so that the getLongCounter() method doesn't complain
       rates.add("metric" + i, 0);
     }

     Thread[] threads = new Thread[n];
     final CountDownLatch firstAddsFinished = new CountDownLatch(threads.length);
     final CountDownLatch firstSnapshotsFinished = new CountDownLatch(1);
     final CountDownLatch secondAddsFinished =
         new CountDownLatch(threads.length);
     final CountDownLatch secondSnapshotsFinished = new CountDownLatch(1);
     long seed = new Random().nextLong();
     LOG.info("Random seed = " + seed);
     final Random sleepRandom = new Random(seed);
     for (int tIdx = 0; tIdx < threads.length; tIdx++) {
       final int threadIdx = tIdx;
       threads[threadIdx] = new Thread() {
         @Override
         public void run() {
           try {
             for (int i = 0; i < 1000; i++) {
               rates.add("metric" + (i % n), (i / n) % 2 == 0 ? 1 : 2);
               // Sleep so additions can be interleaved with snapshots
               Thread.sleep(sleepRandom.nextInt(5));
             }
             firstAddsFinished.countDown();

             // Make sure all threads stay alive long enough for the first
             // snapshot to complete; else their metrics may be lost to GC
             firstSnapshotsFinished.await();

             // Let half the threads continue with more metrics and let half die
             if (threadIdx % 2 == 0) {
               for (int i = 0; i < 1000; i++) {
                 rates.add("metric" + (i % n), (i / n) % 2 == 0 ? 1 : 2);
               }
               secondAddsFinished.countDown();
               secondSnapshotsFinished.await();
             } else {
               secondAddsFinished.countDown();
             }
           } catch (InterruptedException e) {
             // Ignore
           }
         }
       };
     }
     for (Thread t : threads) {
       t.start();
     }
     // Snapshot concurrently with additions but aggregate the totals into
     // opCount / opTotalTime
     for (int i = 0; i < 100; i++) {
       snapshotMutableRatesWithAggregation(rates, opCount, opTotalTime);
       Thread.sleep(sleepRandom.nextInt(20));
     }
     firstAddsFinished.await();
     // Final snapshot to grab any remaining metrics and then verify that
     // the totals are as expected
     snapshotMutableRatesWithAggregation(rates, opCount, opTotalTime);
     for (int i = 0; i < n; i++) {
       assertEquals("metric" + i + " count", 1001, opCount[i]);
       assertEquals("metric" + i + " total", 1500, opTotalTime[i], 1.0);
     }
     firstSnapshotsFinished.countDown();

     // After half of the threads die, ensure that the remaining ones still
     // add metrics correctly and that snapshot occurs correctly
     secondAddsFinished.await();
     snapshotMutableRatesWithAggregation(rates, opCount, opTotalTime);
     for (int i = 0; i < n; i++) {
       assertEquals("metric" + i + " count", 1501, opCount[i]);
       assertEquals("metric" + i + " total", 2250, opTotalTime[i], 1.0);
     }
     secondSnapshotsFinished.countDown();
   }

   private static void snapshotMutableRatesWithAggregation(
       MutableRatesWithAggregation rates, long[] opCount, double[] opTotalTime) {
     MetricsRecordBuilder rb = mockMetricsRecordBuilder();
     rates.snapshot(rb, true);
     for (int i = 0; i < opCount.length; i++) {
       long prevOpCount = opCount[i];
       long newOpCount = getLongCounter("Metric" + i + "NumOps", rb);
       opCount[i] = newOpCount;
       double avgTime = getDoubleGauge("Metric" + i + "AvgTime", rb);
       opTotalTime[i] += avgTime * (newOpCount - prevOpCount);
     }
   }

   /**
    * Tests that when using {@link MutableStat#add(long, long)}, even with a high
    * sample count, the mean does not lose accuracy.
    */
   @Test public void testMutableStatWithBulkAdd() {
     MetricsRecordBuilder rb = mockMetricsRecordBuilder();
     MetricsRegistry registry = new MetricsRegistry("test");
     MutableStat stat = registry.newStat("Test", "Test", "Ops", "Val", false);

     stat.add(1000, 1000);
     stat.add(1000, 2000);
     registry.snapshot(rb, false);

     assertCounter("TestNumOps", 2000L, rb);
     assertGauge("TestAvgVal", 1.5, rb);
   }

   /**
    * Ensure that quantile estimates from {@link MutableQuantiles} are within
    * specified error bounds.
    */
   @Test(timeout = 30000)
   public void testMutableQuantilesError() throws Exception {
     MetricsRecordBuilder mb = mockMetricsRecordBuilder();
     MetricsRegistry registry = new MetricsRegistry("test");
     // Use a 5s rollover period
     MutableQuantiles quantiles = registry.newQuantiles("foo", "stat", "Ops",
         "Latency", 5);
     // Push some values in and wait for it to publish
     long start = System.nanoTime() / 1000000;
     for (long i = 1; i <= 1000; i++) {
       quantiles.add(i);
       quantiles.add(1001 - i);
     }
     long end = System.nanoTime() / 1000000;

     Thread.sleep(6000 - (end - start));

     registry.snapshot(mb, false);

     // Print out the snapshot
     Map<Quantile, Long> previousSnapshot = quantiles.previousSnapshot;
     for (Entry<Quantile, Long> item : previousSnapshot.entrySet()) {
       System.out.println(String.format("Quantile %.2f has value %d",
           item.getKey().quantile, item.getValue()));
     }

     // Verify the results are within our requirements
     verify(mb).addGauge(
         info("FooNumOps", "Number of ops for stat with 5s interval"),
         (long) 2000);
     Quantile[] quants = MutableQuantiles.quantiles;
     String name = "Foo%dthPercentileLatency";
     String desc = "%d percentile latency with 5 second interval for stat";
     for (Quantile q : quants) {
       int percentile = (int) (100 * q.quantile);
       int error = (int) (1000 * q.error);
       String n = String.format(name, percentile);
       String d = String.format(desc, percentile);
       long expected = (long) (q.quantile * 1000);
       verify(mb).addGauge(eq(info(n, d)), leq(expected + error));
       verify(mb).addGauge(eq(info(n, d)), geq(expected - error));
     }
   }

   /**
    * Test that {@link MutableQuantiles} rolls the window over at the specified
    * interval.
    */
   @Test(timeout = 30000)
   public void testMutableQuantilesRollover() throws Exception {
     MetricsRecordBuilder mb = mockMetricsRecordBuilder();
     MetricsRegistry registry = new MetricsRegistry("test");
     // Use a 5s rollover period
     MutableQuantiles quantiles = registry.newQuantiles("foo", "stat", "Ops",
         "Latency", 5);

     Quantile[] quants = MutableQuantiles.quantiles;
     String name = "Foo%dthPercentileLatency";
     String desc = "%d percentile latency with 5 second interval for stat";

     // Push values for three intervals
     long start = System.nanoTime() / 1000000;
     for (int i = 1; i <= 3; i++) {
       // Insert the values
       for (long j = 1; j <= 1000; j++) {
         quantiles.add(i);
       }
       // Sleep until 1s after the next 5s interval, to let the metrics
       // roll over
       long sleep = (start + (5000 * i) + 1000) - (System.nanoTime() / 1000000);
       Thread.sleep(sleep);
       // Verify that the window reset, check it has the values we pushed in
       registry.snapshot(mb, false);
       for (Quantile q : quants) {
         int percentile = (int) (100 * q.quantile);
         String n = String.format(name, percentile);
         String d = String.format(desc, percentile);
         verify(mb).addGauge(info(n, d), (long) i);
       }
     }

     // Verify the metrics were added the right number of times
     verify(mb, times(3)).addGauge(
         info("FooNumOps", "Number of ops for stat with 5s interval"),
         (long) 1000);
     for (Quantile q : quants) {
       int percentile = (int) (100 * q.quantile);
       String n = String.format(name, percentile);
       String d = String.format(desc, percentile);
       verify(mb, times(3)).addGauge(eq(info(n, d)), anyLong());
     }
   }

   /**
    * Test that {@link MutableQuantiles} rolls over correctly even if no items
    * have been added to the window
    */
   @Test(timeout = 30000)
   public void testMutableQuantilesEmptyRollover() throws Exception {
     MetricsRecordBuilder mb = mockMetricsRecordBuilder();
     MetricsRegistry registry = new MetricsRegistry("test");
     // Use a 5s rollover period
     MutableQuantiles quantiles = registry.newQuantiles("foo", "stat", "Ops",
         "Latency", 5);

     // Check it initially
     quantiles.snapshot(mb, true);
     verify(mb).addGauge(
         info("FooNumOps", "Number of ops for stat with 5s interval"), (long) 0);
     Thread.sleep(6000);
     quantiles.snapshot(mb, false);
     verify(mb, times(2)).addGauge(
         info("FooNumOps", "Number of ops for stat with 5s interval"), (long) 0);
   }
 }
	/**
	* Licensed to the Apache Software Foundation (ASF) under one
	* or more contributor license agreements. See the NOTICE file
	* distributed with this work for additional information
	* regarding copyright ownership. The ASF licenses this file
	* to you under the Apache License, Version 2.0 (the
	* "License"); you may not use this file except in compliance
	* with the License. You may obtain a copy of the License at
	*
	* http://www.apache.org/licenses/LICENSE-2.0
	*
	* Unless required by applicable law or agreed to in writing, software
	* distributed under the License is distributed on an "AS IS" BASIS,
	* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
	* See the License for the specific language governing permissions and
	* limitations under the License.
	*/

	package org.apache.hadoop.metrics2.lib;

	import static org.apache.hadoop.metrics2.lib.Interns.info;
	import static org.apache.hadoop.test.MetricsAsserts.*;
	import static org.mockito.AdditionalMatchers.eq;
	import static org.mockito.AdditionalMatchers.geq;
	import static org.mockito.AdditionalMatchers.leq;
	import static org.mockito.Matchers.anyLong;
	import static org.mockito.Matchers.eq;
	import static org.mockito.Mockito.times;
	import static org.mockito.Mockito.verify;
	import static org.junit.Assert.*;

	import java.util.Map;
	import java.util.Map.Entry;
	import java.util.Random;
	import java.util.concurrent.CountDownLatch;

	import org.apache.hadoop.metrics2.MetricsRecordBuilder;
	import org.apache.hadoop.metrics2.util.Quantile;
	import org.junit.Test;
	import org.slf4j.Logger;
	import org.slf4j.LoggerFactory;

	/**
	* Test metrics record builder interface and mutable metrics
	*/
	public class TestMutableMetrics {

	private static final Logger LOG =
	LoggerFactory.getLogger(TestMutableMetrics.class);
	private final double EPSILON = 1e-42;

	/**
	* Test the snapshot method
	*/
	@Test public void testSnapshot() {
	MetricsRecordBuilder mb = mockMetricsRecordBuilder();

	MetricsRegistry registry = new MetricsRegistry("test");
	registry.newCounter("c1", "int counter", 1);
	registry.newCounter("c2", "long counter", 2L);
	registry.newGauge("g1", "int gauge", 3);
	registry.newGauge("g2", "long gauge", 4L);
	registry.newGauge("g3", "float gauge", 5f);
	registry.newStat("s1", "stat", "Ops", "Time", true).add(0);
	registry.newRate("s2", "stat", false).add(0);

	registry.snapshot(mb, true);

	MutableStat s2 = (MutableStat) registry.get("s2");

	s2.snapshot(mb, true); // should get the same back.
	s2.add(1);
	s2.snapshot(mb, true); // should get new interval values back

	verify(mb).addCounter(info("c1", "int counter"), 1);
	verify(mb).addCounter(info("c2", "long counter"), 2L);
	verify(mb).addGauge(info("g1", "int gauge"), 3);
	verify(mb).addGauge(info("g2", "long gauge"), 4L);
	verify(mb).addGauge(info("g3", "float gauge"), 5f);
	verify(mb).addCounter(info("S1NumOps", "Number of ops for stat"), 1L);
	verify(mb).addGauge(eq(info("S1AvgTime", "Average time for stat")),
	eq(0.0, EPSILON));
	verify(mb).addGauge(eq(info("S1StdevTime",
	"Standard deviation of time for stat")),
	eq(0.0, EPSILON));
	verify(mb).addGauge(eq(info("S1IMinTime",
	"Interval min time for stat")),
	eq(0.0, EPSILON));
	verify(mb).addGauge(eq(info("S1IMaxTime",
	"Interval max time for stat")),
	eq(0.0, EPSILON));
	verify(mb).addGauge(eq(info("S1MinTime","Min time for stat")),
	eq(0.0, EPSILON));
	verify(mb).addGauge(eq(info("S1MaxTime","Max time for stat")),
	eq(0.0, EPSILON));
	verify(mb).addGauge(
	eq(info("S1INumOps", "Interval number of ops for stat")),
	eq(1L));

	verify(mb, times(2))
	.addCounter(info("S2NumOps", "Number of ops for stat"), 1L);
	verify(mb, times(2)).addGauge(eq(info("S2AvgTime",
	"Average time for stat")),
	eq(0.0, EPSILON));
	verify(mb).addCounter(info("S2NumOps", "Number of ops for stat"), 2L);
	verify(mb).addGauge(eq(info("S2AvgTime", "Average time for stat")),
	eq(1.0, EPSILON));

	// Add one more sample to s1 and verify that total number of ops
	// has increased to 2, but interval number is 1 for both intervals.
	MutableStat s1 = (MutableStat) registry.get("s1");
	s1.add(0);
	registry.snapshot(mb, true);
	verify(mb).addCounter(info("S1NumOps", "Number of ops for stat"), 2L);
	verify(mb, times(2)).addGauge(
	eq(info("S1INumOps", "Interval number of ops for stat")),
	eq(1L));
	}

	interface TestProtocol {
	void foo();
	void bar();
	}

	@Test public void testMutableRates() {
	MetricsRecordBuilder rb = mockMetricsRecordBuilder();
	MetricsRegistry registry = new MetricsRegistry("test");
	MutableRates rates = new MutableRates(registry);

	rates.init(TestProtocol.class);
	registry.snapshot(rb, false);

	assertCounter("FooNumOps", 0L, rb);
	assertGauge("FooAvgTime", 0.0, rb);
	assertCounter("BarNumOps", 0L, rb);
	assertGauge("BarAvgTime", 0.0, rb);
	}

	@Test public void testMutableRatesWithAggregationInit() {
	MetricsRecordBuilder rb = mockMetricsRecordBuilder();
	MutableRatesWithAggregation rates = new MutableRatesWithAggregation();

	rates.init(TestProtocol.class);
	rates.snapshot(rb, false);

	assertCounter("FooNumOps", 0L, rb);
	assertGauge("FooAvgTime", 0.0, rb);
	assertCounter("BarNumOps", 0L, rb);
	assertGauge("BarAvgTime", 0.0, rb);
	}

	@Test public void testMutableRatesWithAggregationSingleThread() {
	MutableRatesWithAggregation rates = new MutableRatesWithAggregation();

	rates.add("foo", 1);
	rates.add("bar", 5);

	MetricsRecordBuilder rb = mockMetricsRecordBuilder();
	rates.snapshot(rb, false);
	assertCounter("FooNumOps", 1L, rb);
	assertGauge("FooAvgTime", 1.0, rb);
	assertCounter("BarNumOps", 1L, rb);
	assertGauge("BarAvgTime", 5.0, rb);

	rates.add("foo", 1);
	rates.add("foo", 3);
	rates.add("bar", 6);

	rb = mockMetricsRecordBuilder();
	rates.snapshot(rb, false);
	assertCounter("FooNumOps", 3L, rb);
	assertGauge("FooAvgTime", 2.0, rb);
	assertCounter("BarNumOps", 2L, rb);
	assertGauge("BarAvgTime", 6.0, rb);
	}

	@Test public void testMutableRatesWithAggregationManyThreads()
	throws InterruptedException {
	final MutableRatesWithAggregation rates = new MutableRatesWithAggregation();

	final int n = 10;
	long[] opCount = new long[n];
	double[] opTotalTime = new double[n];

	for (int i = 0; i < n; i++) {
	opCount[i] = 0;
	opTotalTime[i] = 0;
	// Initialize so that the getLongCounter() method doesn't complain
	rates.add("metric" + i, 0);
	}

	Thread[] threads = new Thread[n];
	final CountDownLatch firstAddsFinished = new CountDownLatch(threads.length);
	final CountDownLatch firstSnapshotsFinished = new CountDownLatch(1);
	final CountDownLatch secondAddsFinished =
	new CountDownLatch(threads.length);
	final CountDownLatch secondSnapshotsFinished = new CountDownLatch(1);
	long seed = new Random().nextLong();
	LOG.info("Random seed = " + seed);
	final Random sleepRandom = new Random(seed);
	for (int tIdx = 0; tIdx < threads.length; tIdx++) {
	final int threadIdx = tIdx;
	threads[threadIdx] = new Thread() {
	@Override
	public void run() {
	try {
	for (int i = 0; i < 1000; i++) {
	rates.add("metric" + (i % n), (i / n) % 2 == 0 ? 1 : 2);
	// Sleep so additions can be interleaved with snapshots
	Thread.sleep(sleepRandom.nextInt(5));
	}
	firstAddsFinished.countDown();

	// Make sure all threads stay alive long enough for the first
	// snapshot to complete; else their metrics may be lost to GC
	firstSnapshotsFinished.await();

	// Let half the threads continue with more metrics and let half die
	if (threadIdx % 2 == 0) {
	for (int i = 0; i < 1000; i++) {
	rates.add("metric" + (i % n), (i / n) % 2 == 0 ? 1 : 2);
	}
	secondAddsFinished.countDown();
	secondSnapshotsFinished.await();
	} else {
	secondAddsFinished.countDown();
	}
	} catch (InterruptedException e) {
	// Ignore
	}
	}
	};
	}
	for (Thread t : threads) {
	t.start();
	}
	// Snapshot concurrently with additions but aggregate the totals into
	// opCount / opTotalTime
	for (int i = 0; i < 100; i++) {
	snapshotMutableRatesWithAggregation(rates, opCount, opTotalTime);
	Thread.sleep(sleepRandom.nextInt(20));
	}
	firstAddsFinished.await();
	// Final snapshot to grab any remaining metrics and then verify that
	// the totals are as expected
	snapshotMutableRatesWithAggregation(rates, opCount, opTotalTime);
	for (int i = 0; i < n; i++) {
	assertEquals("metric" + i + " count", 1001, opCount[i]);
	assertEquals("metric" + i + " total", 1500, opTotalTime[i], 1.0);
	}
	firstSnapshotsFinished.countDown();

	// After half of the threads die, ensure that the remaining ones still
	// add metrics correctly and that snapshot occurs correctly
	secondAddsFinished.await();
	snapshotMutableRatesWithAggregation(rates, opCount, opTotalTime);
	for (int i = 0; i < n; i++) {
	assertEquals("metric" + i + " count", 1501, opCount[i]);
	assertEquals("metric" + i + " total", 2250, opTotalTime[i], 1.0);
	}
	secondSnapshotsFinished.countDown();
	}

	private static void snapshotMutableRatesWithAggregation(
	MutableRatesWithAggregation rates, long[] opCount, double[] opTotalTime) {
	MetricsRecordBuilder rb = mockMetricsRecordBuilder();
	rates.snapshot(rb, true);
	for (int i = 0; i < opCount.length; i++) {
	long prevOpCount = opCount[i];
	long newOpCount = getLongCounter("Metric" + i + "NumOps", rb);
	opCount[i] = newOpCount;
	double avgTime = getDoubleGauge("Metric" + i + "AvgTime", rb);
	opTotalTime[i] += avgTime * (newOpCount - prevOpCount);
	}
	}

	/**
	* Tests that when using {@link MutableStat#add(long, long)}, even with a high
	* sample count, the mean does not lose accuracy.
	*/
	@Test public void testMutableStatWithBulkAdd() {
	MetricsRecordBuilder rb = mockMetricsRecordBuilder();
	MetricsRegistry registry = new MetricsRegistry("test");
	MutableStat stat = registry.newStat("Test", "Test", "Ops", "Val", false);

	stat.add(1000, 1000);
	stat.add(1000, 2000);
	registry.snapshot(rb, false);

	assertCounter("TestNumOps", 2000L, rb);
	assertGauge("TestAvgVal", 1.5, rb);
	}

	/**
	* Ensure that quantile estimates from {@link MutableQuantiles} are within
	* specified error bounds.
	*/
	@Test(timeout = 30000)
	public void testMutableQuantilesError() throws Exception {
	MetricsRecordBuilder mb = mockMetricsRecordBuilder();
	MetricsRegistry registry = new MetricsRegistry("test");
	// Use a 5s rollover period
	MutableQuantiles quantiles = registry.newQuantiles("foo", "stat", "Ops",
	"Latency", 5);
	// Push some values in and wait for it to publish
	long start = System.nanoTime() / 1000000;
	for (long i = 1; i <= 1000; i++) {
	quantiles.add(i);
	quantiles.add(1001 - i);
	}
	long end = System.nanoTime() / 1000000;

	Thread.sleep(6000 - (end - start));

	registry.snapshot(mb, false);

	// Print out the snapshot
	Map<Quantile, Long> previousSnapshot = quantiles.previousSnapshot;
	for (Entry<Quantile, Long> item : previousSnapshot.entrySet()) {
	System.out.println(String.format("Quantile %.2f has value %d",
	item.getKey().quantile, item.getValue()));
	}

	// Verify the results are within our requirements
	verify(mb).addGauge(
	info("FooNumOps", "Number of ops for stat with 5s interval"),
	(long) 2000);
	Quantile[] quants = MutableQuantiles.quantiles;
	String name = "Foo%dthPercentileLatency";
	String desc = "%d percentile latency with 5 second interval for stat";
	for (Quantile q : quants) {
	int percentile = (int) (100 * q.quantile);
	int error = (int) (1000 * q.error);
	String n = String.format(name, percentile);
	String d = String.format(desc, percentile);
	long expected = (long) (q.quantile * 1000);
	verify(mb).addGauge(eq(info(n, d)), leq(expected + error));
	verify(mb).addGauge(eq(info(n, d)), geq(expected - error));
	}
	}

	/**
	* Test that {@link MutableQuantiles} rolls the window over at the specified
	* interval.
	*/
	@Test(timeout = 30000)
	public void testMutableQuantilesRollover() throws Exception {
	MetricsRecordBuilder mb = mockMetricsRecordBuilder();
	MetricsRegistry registry = new MetricsRegistry("test");
	// Use a 5s rollover period
	MutableQuantiles quantiles = registry.newQuantiles("foo", "stat", "Ops",
	"Latency", 5);

	Quantile[] quants = MutableQuantiles.quantiles;
	String name = "Foo%dthPercentileLatency";
	String desc = "%d percentile latency with 5 second interval for stat";

	// Push values for three intervals
	long start = System.nanoTime() / 1000000;
	for (int i = 1; i <= 3; i++) {
	// Insert the values
	for (long j = 1; j <= 1000; j++) {
	quantiles.add(i);
	}
	// Sleep until 1s after the next 5s interval, to let the metrics
	// roll over
	long sleep = (start + (5000 * i) + 1000) - (System.nanoTime() / 1000000);
	Thread.sleep(sleep);
	// Verify that the window reset, check it has the values we pushed in
	registry.snapshot(mb, false);
	for (Quantile q : quants) {
	int percentile = (int) (100 * q.quantile);
	String n = String.format(name, percentile);
	String d = String.format(desc, percentile);
	verify(mb).addGauge(info(n, d), (long) i);
	}
	}

	// Verify the metrics were added the right number of times
	verify(mb, times(3)).addGauge(
	info("FooNumOps", "Number of ops for stat with 5s interval"),
	(long) 1000);
	for (Quantile q : quants) {
	int percentile = (int) (100 * q.quantile);
	String n = String.format(name, percentile);
	String d = String.format(desc, percentile);
	verify(mb, times(3)).addGauge(eq(info(n, d)), anyLong());
	}
	}

	/**
	* Test that {@link MutableQuantiles} rolls over correctly even if no items
	* have been added to the window
	*/
	@Test(timeout = 30000)
	public void testMutableQuantilesEmptyRollover() throws Exception {
	MetricsRecordBuilder mb = mockMetricsRecordBuilder();
	MetricsRegistry registry = new MetricsRegistry("test");
	// Use a 5s rollover period
	MutableQuantiles quantiles = registry.newQuantiles("foo", "stat", "Ops",
	"Latency", 5);

	// Check it initially
	quantiles.snapshot(mb, true);
	verify(mb).addGauge(
	info("FooNumOps", "Number of ops for stat with 5s interval"), (long) 0);
	Thread.sleep(6000);
	quantiles.snapshot(mb, false);
	verify(mb, times(2)).addGauge(
	info("FooNumOps", "Number of ops for stat with 5s interval"), (long) 0);
	}
	}