HDFS-16982 Use the right Quantiles Array for Inverse Quantiles snapshot (#5556)

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/metrics2/lib/MutableInverseQuantiles.java

@@ -21,7 +21,6 @@ import org.apache.hadoop.classification.InterfaceAudience;
 import org.apache.hadoop.classification.InterfaceStability;
 import org.apache.hadoop.classification.VisibleForTesting;
 import org.apache.hadoop.metrics2.util.Quantile;
-import org.apache.hadoop.metrics2.util.SampleQuantiles;
 import java.text.DecimalFormat;
 import static org.apache.hadoop.metrics2.lib.Interns.info;
 
@@ -65,7 +64,7 @@ public class MutableInverseQuantiles extends MutableQuantiles{
   }
 
   /**
-   * Sets quantileInfo and estimator.
+   * Sets quantileInfo.
    *
    * @param ucName capitalized name of the metric
    * @param uvName capitalized type of the values
@@ -74,8 +73,6 @@ public class MutableInverseQuantiles extends MutableQuantiles{
    * @param df Number formatter for inverse percentile value
    */
   void setQuantiles(String ucName, String uvName, String desc, String lvName, DecimalFormat df) {
-    // Construct the MetricsInfos for inverse quantiles, converting to inverse percentiles
-    setQuantileInfos(INVERSE_QUANTILES.length);
     for (int i = 0; i < INVERSE_QUANTILES.length; i++) {
       double inversePercentile = 100 * (1 - INVERSE_QUANTILES[i].quantile);
       String nameTemplate = ucName + df.format(inversePercentile) + "thInversePercentile" + uvName;
@@ -83,7 +80,14 @@ public class MutableInverseQuantiles extends MutableQuantiles{
           + " with " + getInterval() + " second interval for " + desc;
       addQuantileInfo(i, info(nameTemplate, descTemplate));
     }
+  }
 
-    setEstimator(new SampleQuantiles(INVERSE_QUANTILES));
+  /**
+   * Returns the array of Inverse Quantiles declared in MutableInverseQuantiles.
+   *
+   * @return array of Inverse Quantiles
+   */
+  public synchronized Quantile[] getQuantiles() {
+    return INVERSE_QUANTILES;
   }
 }

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/metrics2/lib/MutableQuantiles.java

@@ -49,9 +49,9 @@ import org.apache.hadoop.thirdparty.com.google.common.util.concurrent.ThreadFact
 public class MutableQuantiles extends MutableMetric {
 
   @VisibleForTesting
-  public static final Quantile[] quantiles = { new Quantile(0.50, 0.050),
+  public static final Quantile[] QUANTILES = {new Quantile(0.50, 0.050),
       new Quantile(0.75, 0.025), new Quantile(0.90, 0.010),
-      new Quantile(0.95, 0.005), new Quantile(0.99, 0.001) };
+      new Quantile(0.95, 0.005), new Quantile(0.99, 0.001)};
 
   private MetricsInfo numInfo;
   private MetricsInfo[] quantileInfos;
@@ -98,11 +98,15 @@ public class MutableQuantiles extends MutableMetric {
         "Number of %s for %s with %ds interval", lsName, desc, interval)));
     scheduledTask = scheduler.scheduleWithFixedDelay(new RolloverSample(this),
         interval, interval, TimeUnit.SECONDS);
+    // Construct the MetricsInfos for the quantiles, converting to percentiles
+    Quantile[] quantilesArray = getQuantiles();
+    setQuantileInfos(quantilesArray.length);
     setQuantiles(ucName, uvName, desc, lvName, decimalFormat);
+    setEstimator(new SampleQuantiles(quantilesArray));
   }
 
   /**
-   * Sets quantileInfo and estimator.
+   * Sets quantileInfo.
    *
    * @param ucName capitalized name of the metric
    * @param uvName capitalized type of the values
@@ -111,30 +115,27 @@ public class MutableQuantiles extends MutableMetric {
    * @param pDecimalFormat Number formatter for percentile value
    */
   void setQuantiles(String ucName, String uvName, String desc, String lvName, DecimalFormat pDecimalFormat) {
-    // Construct the MetricsInfos for the quantiles, converting to percentiles
+    for (int i = 0; i < QUANTILES.length; i++) {
-    setQuantileInfos(quantiles.length);
+      double percentile = 100 * QUANTILES[i].quantile;
-    for (int i = 0; i < quantiles.length; i++) {
-      double percentile = 100 * quantiles[i].quantile;
       String nameTemplate = ucName + pDecimalFormat.format(percentile) + "thPercentile" + uvName;
       String descTemplate = pDecimalFormat.format(percentile) + " percentile " + lvName
           + " with " + getInterval() + " second interval for " + desc;
       addQuantileInfo(i, info(nameTemplate, descTemplate));
     }
-
-    setEstimator(new SampleQuantiles(quantiles));
   }
 
   public MutableQuantiles() {}
 
   @Override
   public synchronized void snapshot(MetricsRecordBuilder builder, boolean all) {
+    Quantile[] quantilesArray = getQuantiles();
     if (all || changed()) {
       builder.addGauge(numInfo, previousCount);
-      for (int i = 0; i < quantiles.length; i++) {
+      for (int i = 0; i < quantilesArray.length; i++) {
         long newValue = 0;
         // If snapshot is null, we failed to update since the window was empty
         if (previousSnapshot != null) {
-          newValue = previousSnapshot.get(quantiles[i]);
+          newValue = previousSnapshot.get(quantilesArray[i]);
         }
         builder.addGauge(quantileInfos[i], newValue);
       }
@@ -148,6 +149,15 @@ public class MutableQuantiles extends MutableMetric {
     estimator.insert(value);
   }
 
+  /**
+   * Returns the array of Quantiles declared in MutableQuantiles.
+   *
+   * @return array of Quantiles
+   */
+  public synchronized Quantile[] getQuantiles() {
+    return QUANTILES;
+  }
+
   /**
    * Set info about the metrics.
    *

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

@@ -52,6 +52,8 @@ public class TestMutableMetrics {
   private static final Logger LOG =
       LoggerFactory.getLogger(TestMutableMetrics.class);
   private static final double EPSILON = 1e-42;
+  private static final int SLEEP_TIME_MS = 6 * 1000; // 6 seconds.
+  private static final int SAMPLE_COUNT = 1000;
 
   /**
    * Test the snapshot method
@@ -395,14 +397,14 @@ public class TestMutableMetrics {
     MutableQuantiles quantiles = registry.newQuantiles("foo", "stat", "Ops",
         "Latency", 5);
     // Push some values in and wait for it to publish
-    long start = System.nanoTime() / 1000000;
+    long startTimeMS = System.currentTimeMillis();
-    for (long i = 1; i <= 1000; i++) {
+    for (long i = 1; i <= SAMPLE_COUNT; i++) {
       quantiles.add(i);
       quantiles.add(1001 - i);
     }
-    long end = System.nanoTime() / 1000000;
+    long endTimeMS = System.currentTimeMillis();
 
-    Thread.sleep(6000 - (end - start));
+    Thread.sleep(SLEEP_TIME_MS - (endTimeMS - startTimeMS));
 
     registry.snapshot(mb, false);
 
@@ -414,10 +416,8 @@ public class TestMutableMetrics {
     }
 
     // Verify the results are within our requirements
-    verify(mb).addGauge(
+    verify(mb).addGauge(info("FooNumOps", "Number of ops for stat with 5s interval"), 2000L);
-        info("FooNumOps", "Number of ops for stat with 5s interval"),
+    Quantile[] quants = MutableQuantiles.QUANTILES;
-        (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) {
@@ -431,6 +431,46 @@ public class TestMutableMetrics {
     }
   }
 
+  /**
+   * Ensure that quantile estimates from {@link MutableInverseQuantiles} are within
+   * specified error bounds.
+   */
+  @Test(timeout = 30000)
+  public void testMutableInverseQuantilesError() throws Exception {
+    MetricsRecordBuilder mb = mockMetricsRecordBuilder();
+    MetricsRegistry registry = new MetricsRegistry("test");
+    // Use a 5s rollover period
+    MutableQuantiles inverseQuantiles = registry.newInverseQuantiles("foo", "stat", "Ops",
+        "Latency", 5);
+    // Push some values in and wait for it to publish
+    long startTimeMS = System.currentTimeMillis();
+    for (long i = 1; i <= SAMPLE_COUNT; i++) {
+      inverseQuantiles.add(i);
+      inverseQuantiles.add(1001 - i);
+    }
+    long endTimeMS = System.currentTimeMillis();
+
+    Thread.sleep(SLEEP_TIME_MS - (endTimeMS - startTimeMS));
+
+    registry.snapshot(mb, false);
+
+    // Verify the results are within our requirements
+    verify(mb).addGauge(
+        info("FooNumOps", "Number of ops for stat with 5s interval"), 2000L);
+    Quantile[] inverseQuants = MutableInverseQuantiles.INVERSE_QUANTILES;
+    String name = "Foo%dthInversePercentileLatency";
+    String desc = "%d inverse percentile latency with 5 second interval for stat";
+    for (Quantile q : inverseQuants) {
+      int inversePercentile = (int) (100 * (1 - q.quantile));
+      int error = (int) (1000 * q.error);
+      String n = String.format(name, inversePercentile);
+      String d = String.format(desc, inversePercentile);
+      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.
@@ -443,21 +483,21 @@ public class TestMutableMetrics {
     MutableQuantiles quantiles = registry.newQuantiles("foo", "stat", "Ops",
         "Latency", 5);
 
-    Quantile[] quants = MutableQuantiles.quantiles;
+    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;
+    long startTimeMS = System.currentTimeMillis();
     for (int i = 1; i <= 3; i++) {
       // Insert the values
-      for (long j = 1; j <= 1000; j++) {
+      for (long j = 1; j <= SAMPLE_COUNT; 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);
+      long sleepTimeMS = startTimeMS + (5000L * i) + 1000 - System.currentTimeMillis();
-      Thread.sleep(sleep);
+      Thread.sleep(sleepTimeMS);
       // Verify that the window reset, check it has the values we pushed in
       registry.snapshot(mb, false);
       for (Quantile q : quants) {
@@ -470,8 +510,7 @@ public class TestMutableMetrics {
 
     // 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"),
+        info("FooNumOps", "Number of ops for stat with 5s interval"), 1000L);
-        (long) 1000);
     for (Quantile q : quants) {
       int percentile = (int) (100 * q.quantile);
       String n = String.format(name, percentile);
@@ -481,7 +520,56 @@ public class TestMutableMetrics {
   }
 
   /**
-   * Test that {@link MutableQuantiles} rolls over correctly even if no items
+   * Test that {@link MutableInverseQuantiles} rolls the window over at the specified
+   * interval.
+   */
+  @Test(timeout = 30000)
+  public void testMutableInverseQuantilesRollover() throws Exception {
+    MetricsRecordBuilder mb = mockMetricsRecordBuilder();
+    MetricsRegistry registry = new MetricsRegistry("test");
+    // Use a 5s rollover period
+    MutableQuantiles inverseQuantiles = registry.newInverseQuantiles("foo", "stat", "Ops",
+        "Latency", 5);
+
+    Quantile[] quants = MutableInverseQuantiles.INVERSE_QUANTILES;
+    String name = "Foo%dthInversePercentileLatency";
+    String desc = "%d inverse percentile latency with 5 second interval for stat";
+
+    // Push values for three intervals
+    long startTimeMS = System.currentTimeMillis();
+    for (int i = 1; i <= 3; i++) {
+      // Insert the values
+      for (long j = 1; j <= SAMPLE_COUNT; j++) {
+        inverseQuantiles.add(i);
+      }
+      // Sleep until 1s after the next 5s interval, to let the metrics
+      // roll over
+      long sleepTimeMS = startTimeMS + (5000L * i) + 1000 - System.currentTimeMillis();
+      Thread.sleep(sleepTimeMS);
+      // Verify that the window reset, check it has the values we pushed in
+      registry.snapshot(mb, false);
+      for (Quantile q : quants) {
+        int inversePercentile = (int) (100 * (1 - q.quantile));
+        String n = String.format(name, inversePercentile);
+        String d = String.format(desc, inversePercentile);
+        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"), 1000L);
+
+    for (Quantile q : quants) {
+      int inversePercentile = (int) (100 * (1 - q.quantile));
+      String n = String.format(name, inversePercentile);
+      String d = String.format(desc, inversePercentile);
+      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)
@@ -495,11 +583,33 @@ public class TestMutableMetrics {
     // Check it initially
     quantiles.snapshot(mb, true);
     verify(mb).addGauge(
-        info("FooNumOps", "Number of ops for stat with 5s interval"), (long) 0);
+        info("FooNumOps", "Number of ops for stat with 5s interval"), 0L);
-    Thread.sleep(6000);
+    Thread.sleep(SLEEP_TIME_MS);
     quantiles.snapshot(mb, false);
     verify(mb, times(2)).addGauge(
-        info("FooNumOps", "Number of ops for stat with 5s interval"), (long) 0);
+        info("FooNumOps", "Number of ops for stat with 5s interval"), 0L);
+  }
+
+  /**
+   * Test that {@link MutableInverseQuantiles} rolls over correctly even if no items
+   * have been added to the window
+   */
+  @Test(timeout = 30000)
+  public void testMutableInverseQuantilesEmptyRollover() throws Exception {
+    MetricsRecordBuilder mb = mockMetricsRecordBuilder();
+    MetricsRegistry registry = new MetricsRegistry("test");
+    // Use a 5s rollover period
+    MutableQuantiles inverseQuantiles = registry.newInverseQuantiles("foo", "stat", "Ops",
+        "Latency", 5);
+
+    // Check it initially
+    inverseQuantiles.snapshot(mb, true);
+    verify(mb).addGauge(
+        info("FooNumOps", "Number of ops for stat with 5s interval"), 0L);
+    Thread.sleep(SLEEP_TIME_MS);
+    inverseQuantiles.snapshot(mb, false);
+    verify(mb, times(2)).addGauge(
+        info("FooNumOps", "Number of ops for stat with 5s interval"), 0L);
   }
 
   /**

hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/test/MetricsAsserts.java

@@ -393,7 +393,7 @@ public class MetricsAsserts {
   public static void assertQuantileGauges(String prefix,
       MetricsRecordBuilder rb, String valueName) {
     verify(rb).addGauge(eqName(info(prefix + "NumOps", "")), geq(0L));
-    for (Quantile q : MutableQuantiles.quantiles) {
+    for (Quantile q : MutableQuantiles.QUANTILES) {
       String nameTemplate = prefix + "%dthPercentile" + valueName;
       int percentile = (int) (100 * q.quantile);
       verify(rb).addGauge(
@@ -414,7 +414,7 @@ public class MetricsAsserts {
   public static void assertInverseQuantileGauges(String prefix,
       MetricsRecordBuilder rb, String valueName) {
     verify(rb).addGauge(eqName(info(prefix + "NumOps", "")), geq(0L));
-    for (Quantile q : MutableQuantiles.quantiles) {
+    for (Quantile q : MutableQuantiles.QUANTILES) {
       String nameTemplate = prefix + "%dthInversePercentile" + valueName;
       int percentile = (int) (100 * q.quantile);
       verify(rb).addGauge(

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/monitor/ContainerMetrics.java

@@ -155,7 +155,7 @@ public class ContainerMetrics implements MetricsSource {
         .newQuantiles(PMEM_USAGE_QUANTILES_NAME, "Physical memory quantiles",
             "Usage", "MBs", 1);
     ContainerMetricsQuantiles memEstimator =
-        new ContainerMetricsQuantiles(MutableQuantiles.quantiles);
+        new ContainerMetricsQuantiles(MutableQuantiles.QUANTILES);
     pMemMBQuantiles.setEstimator(memEstimator);
 
     this.cpuCoreUsagePercent = registry.newStat(
@@ -166,7 +166,7 @@ public class ContainerMetrics implements MetricsSource {
             "Physical Cpu core percent usage quantiles", "Usage", "Percents",
             1);
     ContainerMetricsQuantiles cpuEstimator =
-        new ContainerMetricsQuantiles(MutableQuantiles.quantiles);
+        new ContainerMetricsQuantiles(MutableQuantiles.QUANTILES);
     cpuCoreUsagePercentQuantiles.setEstimator(cpuEstimator);
     this.milliVcoresUsed = registry.newStat(
         VCORE_USAGE_METRIC_NAME, "1000 times Vcore usage", "Usage",