HDFS-16949 Introduce inverse quantiles for metrics where higher numeric value is better

Author: Ravindra Dingankar

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

@@ -227,6 +227,29 @@ public synchronized MutableQuantiles newQuantiles(String name, String desc,
     return ret;
   }
 
+  /**
+   * Create a mutable inverse metric that estimates inverse quantiles of a stream of values
+   * @param name of the metric
+   * @param desc metric description
+   * @param sampleName of the metric (e.g., "Ops")
+   * @param valueName of the metric (e.g., "Rate")
+   * @param interval rollover interval of estimator in seconds
+   * @return a new inverse quantile estimator object
+   * @throws MetricsException if interval is not a positive integer
+   */
+  public synchronized MutableQuantiles newInverseQuantiles(String name, String desc,
+      String sampleName, String valueName, int interval) {
+    checkMetricName(name);
+    if (interval <= 0) {
+      throw new MetricsException("Interval should be positive.  Value passed" +
+          " is: " + interval);
+    }
+    MutableQuantiles ret =
+        new MutableInverseQuantiles(name, desc, sampleName, valueName, interval);
+    metricsMap.put(name, ret);
+    return ret;
+  }
+
   /**
    * Create a mutable metric with stats
    * @param name  of the metric

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

@@ -0,0 +1,102 @@
+/**
+ * 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 org.apache.commons.lang3.StringUtils;
+import org.apache.hadoop.classification.InterfaceAudience;
+import org.apache.hadoop.classification.InterfaceStability;
+import org.apache.hadoop.classification.VisibleForTesting;
+import org.apache.hadoop.metrics2.MetricsInfo;
+import org.apache.hadoop.metrics2.util.Quantile;
+import org.apache.hadoop.metrics2.util.SampleQuantiles;
+import org.apache.hadoop.thirdparty.com.google.common.util.concurrent.ThreadFactoryBuilder;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.TimeUnit;
+import static org.apache.hadoop.metrics2.lib.Interns.info;
+
+/**
+ * Watches a stream of long values, maintaining online estimates of specific
+ * quantiles with provably low error bounds. Inverse quantiles are meant for 
+ * highly accurate low-percentile (e.g. 1st, 5th) latency metrics.
+ * InverseQuantiles are used for metrics where higher the value better it is.
+ * ( eg: data transfer rate ).
+ * The 1st percentile here corresponds to the 99th inverse percentile metric,
+ * 5th percentile to 95th and so on.
+ */
+@InterfaceAudience.Public
+@InterfaceStability.Evolving
+public class MutableInverseQuantiles extends MutableQuantiles{
+
+  @VisibleForTesting
+  public static final Quantile[] inverseQuantiles = { new Quantile(0.50, 0.050),
+      new Quantile(0.25, 0.025), new Quantile(0.10, 0.010),
+      new Quantile(0.05, 0.005), new Quantile(0.01, 0.001) };
+
+  private ScheduledFuture<?> scheduledTask = null;
+    
+  private static final ScheduledExecutorService scheduler = Executors
+      .newScheduledThreadPool(1, new ThreadFactoryBuilder().setDaemon(true)
+          .setNameFormat("MutableInverseQuantiles-%d").build());
+
+  /**
+   * Instantiates a new {@link MutableInverseQuantiles} for a metric that rolls itself
+   * over on the specified time interval.
+   *
+   * @param name        of the metric
+   * @param description long-form textual description of the metric
+   * @param sampleName  type of items in the stream (e.g., "Ops")
+   * @param valueName   type of the values
+   * @param interval
+   */
+  public MutableInverseQuantiles(String name, String description, String sampleName,
+      String valueName, int interval) {
+    String ucName = StringUtils.capitalize(name);
+    String usName = StringUtils.capitalize(sampleName);
+    String uvName = StringUtils.capitalize(valueName);
+    String desc = StringUtils.uncapitalize(description);
+    String lsName = StringUtils.uncapitalize(sampleName);
+    String lvName = StringUtils.uncapitalize(valueName);
+
+    numInfo = info(ucName + "Num" + usName, String.format(
+        "Number of %s for %s with %ds interval", lsName, desc, interval));
+    // Construct the MetricsInfos for the inverse quantiles, converting to inverse percentiles
+    quantileInfos = new MetricsInfo[inverseQuantiles.length];
+    String nameTemplate = ucName + "%dthInversePercentile" + uvName;
+    String descTemplate = "%d inverse percentile " + lvName + " with " + interval
+        + " second interval for " + desc;
+    for (int i = 0; i < inverseQuantiles.length; i++) {
+      int inversePercentile = (int) (100 * (1 - inverseQuantiles[i].quantile));
+      quantileInfos[i] = info(String.format(nameTemplate, inversePercentile),
+          String.format(descTemplate, inversePercentile));
+    }
+
+    estimator = new SampleQuantiles(inverseQuantiles);
+    setInterval(interval);
+    scheduledTask = scheduler.scheduleWithFixedDelay(new RolloverSample(this),
+        interval, interval, TimeUnit.SECONDS);
+  }
+
+  public void stop() {
+    if (scheduledTask != null) {
+      scheduledTask.cancel(false);
+    }
+    scheduledTask = null;
+  }
+}

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

@@ -52,11 +52,11 @@ public class MutableQuantiles extends MutableMetric {
       new Quantile(0.75, 0.025), new Quantile(0.90, 0.010),
       new Quantile(0.95, 0.005), new Quantile(0.99, 0.001) };
 
-  private final MetricsInfo numInfo;
-  private final MetricsInfo[] quantileInfos;
-  private final int interval;
+  MetricsInfo numInfo;
+  MetricsInfo[] quantileInfos;
+  private int interval;
 
-  private QuantileEstimator estimator;
+  QuantileEstimator estimator;
   private long previousCount = 0;
   private ScheduledFuture<?> scheduledTask = null;
 
@@ -106,11 +106,13 @@ public MutableQuantiles(String name, String description, String sampleName,
 
     estimator = new SampleQuantiles(quantiles);
 
-    this.interval = interval;
+    setInterval(interval);
     scheduledTask = scheduler.scheduleWithFixedDelay(new RolloverSample(this),
         interval, interval, TimeUnit.SECONDS);
   }
 
+  public MutableQuantiles() {}
+
   @Override
   public synchronized void snapshot(MetricsRecordBuilder builder, boolean all) {
     if (all || changed()) {
@@ -133,6 +135,10 @@ public synchronized void add(long value) {
     estimator.insert(value);
   }
 
+  synchronized void setInterval(int interval) {
+    this.interval = interval;
+  }
+
   public int getInterval() {
     return interval;
   }
@@ -162,7 +168,7 @@ public synchronized void setEstimator(QuantileEstimator quantileEstimator) {
    * Runnable used to periodically roll over the internal
    * {@link SampleQuantiles} every interval.
    */
-  private static class RolloverSample implements Runnable {
+  static class RolloverSample implements Runnable {
 
     MutableQuantiles parent;
 

hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/metrics2/util/TestSampleQuantiles.java

@@ -24,6 +24,7 @@
 import java.util.Map;
 import java.util.Random;
 
+import org.apache.hadoop.metrics2.lib.MutableInverseQuantiles;
 import org.junit.Before;
 import org.junit.Test;
 
@@ -118,4 +119,40 @@ public void testQuantileError() throws IOException {
       }
     }
   }
+
+  /**
+   * Correctness test that checks that absolute error of the estimate for inverse quantiles
+   * is within specified error bounds for some randomly permuted streams of items.
+   */
+  @Test
+  public void testInverseQuantiles() throws IOException {
+    SampleQuantiles estimator = new SampleQuantiles(MutableInverseQuantiles.inverseQuantiles);
+    final int count = 100000;
+    Random r = new Random(0xDEADDEAD);
+    Long[] values = new Long[count];
+    for (int i = 0; i < count; i++) {
+      values[i] = (long) (i + 1);
+    }
+    // Do 10 shuffle/insert/check cycles
+    for (int i = 0; i < 10; i++) {
+      System.out.println("Starting run " + i);
+      Collections.shuffle(Arrays.asList(values), r);
+      estimator.clear();
+      for (int j = 0; j < count; j++) {
+        estimator.insert(values[j]);
+      }
+      Map<Quantile, Long> snapshot;
+      snapshot = estimator.snapshot();
+      for (Quantile q : MutableInverseQuantiles.inverseQuantiles) {
+        long actual = (long) (q.quantile * count);
+        long error = (long) (q.error * count);
+        long estimate = snapshot.get(q);
+        System.out.println(String.format("For inverse quantile %f " +
+            "Expected %d with error %d, estimated %d",
+                (1 - q.quantile), actual, error, estimate));
+        assertThat(estimate <= actual + error).isTrue();
+        assertThat(estimate >= actual - error).isTrue();
+      }
+    }
+  }
 }

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/metrics/DataNodeMetrics.java

@@ -258,7 +258,7 @@ public DataNodeMetrics(String name, String sessionId, int[] intervals,
           "ramDiskBlocksLazyPersistWindows" + interval + "s",
           "Time between the RamDisk block write and disk persist in ms",
           "ops", "latency", interval);
-      readTransferRateQuantiles[i] = registry.newQuantiles(
+      readTransferRateQuantiles[i] = registry.newInverseQuantiles(
           "readTransferRate" + interval + "s",
           "Rate at which bytes are read from datanode calculated in bytes per second",
           "ops", "rate", interval);