HDFS-16949 Introduce inverse quantiles for metrics where higher numer…

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

@@ -227,6 +227,24 @@ public synchronized MutableQuantiles newQuantiles(String name, String desc,
     return ret;
   }
 
+  /**
+   * Create a mutable metric that estimates 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., "Time" or "Latency")
+   * @param interval rollover interval of estimator in seconds
+   * @param inverseQuantiles inverse the quantiles ( e.g. P99 will give the 1st quantile )      
+   * @return a new quantile estimator object
+   * @throws MetricsException if interval is not a positive integer
+   */
+  public synchronized MutableQuantiles newQuantiles(String name, String desc, String sampleName, 
+                                                    String valueName, int interval, boolean inverseQuantiles) {
+      MutableQuantiles ret = newQuantiles(name, desc, sampleName, valueName, interval);
+      ret.inverseQuantiles = inverseQuantiles;
+      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/MutableQuantiles.java

@@ -52,6 +52,7 @@ 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) };
 
+  protected boolean inverseQuantiles = false;
   private final MetricsInfo numInfo;
   private final MetricsInfo[] quantileInfos;
   private final int interval;
@@ -104,7 +105,7 @@ public MutableQuantiles(String name, String description, String sampleName,
           String.format(descTemplate, percentile));
     }
 
-    estimator = new SampleQuantiles(quantiles);
+    estimator = new SampleQuantiles(quantiles, inverseQuantiles);
 
     this.interval = interval;
     scheduledTask = scheduler.scheduleWithFixedDelay(new RolloverSample(this),

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/metrics2/util/SampleQuantiles.java

@@ -71,9 +71,11 @@ public class SampleQuantiles implements QuantileEstimator {
    * Array of Quantiles that we care about, along with desired error.
    */
   private final Quantile quantiles[];
+  private boolean inverseQuantiles;
 
-  public SampleQuantiles(Quantile[] quantiles) {
+  public SampleQuantiles(Quantile[] quantiles, boolean inverseQuantiles) {
     this.quantiles = quantiles;
+    this. inverseQuantiles = inverseQuantiles;
     this.samples = new LinkedList<SampleItem>();
   }
 
@@ -200,7 +202,7 @@ private void compress() {
   /**
    * Get the estimated value at the specified quantile.
    * 
-   * @param quantile Queried quantile, e.g. 0.50 or 0.99.
+   * @param quantile Queried quantile, e.g. 0.01, 0.50 or 0.99.
    * @return Estimated value at that quantile.
    */
   private long query(double quantile) {
@@ -243,7 +245,12 @@ synchronized public Map<Quantile, Long> snapshot() {
 
     Map<Quantile, Long> values = new TreeMap<Quantile, Long>();
     for (int i = 0; i < quantiles.length; i++) {
-      values.put(quantiles[i], query(quantiles[i].quantile));
+      /* eg : effectiveQuantile for 0.99 with inverseQuantiles will be 0.01. 
+      For inverse quantiles higher numeric value is better and hence we want 
+      to query from the opposite end of the sorted sample
+       */
+      double effectiveQuantile = inverseQuantiles ? 1 - quantiles[i].quantile : quantiles[i].quantile;
+      values.put(quantiles[i], query(effectiveQuantile));
     }
 
     return values;

hadoop-common-project/hadoop-common/src/site/markdown/Metrics.md

@@ -372,7 +372,7 @@ Each metrics record contains tags such as SessionId and Hostname as additional i
 | `BytesRead` | Total number of bytes read from DataNode |
 | `ReadTransferRateNumOps` | Total number of data read transfers |
 | `ReadTransferRateAvgTime` | Average transfer rate of bytes read from DataNode, measured in bytes per second. |
-| `ReadTransferRate`*num*`s(50/75/90/95/99)thPercentileRate` | The 50/75/90/95/99th percentile of the transfer rate of bytes read from DataNode, measured in bytes per second. |
+| `ReadTransferRate`*num*`s(50/75/90/95/99)thPercentileRate` | The 50/75/90/95/99th inverse percentile of the transfer rate of bytes read from DataNode, measured in bytes per second. |
 | `BlocksWritten` | Total number of blocks written to DataNode |
 | `BlocksRead` | Total number of blocks read from DataNode |
 | `BlocksReplicated` | Total number of blocks replicated |

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

@@ -39,7 +39,7 @@ public class TestSampleQuantiles {
 
   @Before
   public void init() {
-    estimator = new SampleQuantiles(quantiles);
+    estimator = new SampleQuantiles(quantiles, false);
   }
 
   /**
@@ -118,4 +118,36 @@ public void testQuantileError() throws IOException {
       }
     }
   }
+
+  @Test
+  public void testInverseQuantiles() {
+      SampleQuantiles estimatorWithInverseQuantiles = new SampleQuantiles(quantiles, true);
+      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);
+          estimatorWithInverseQuantiles.clear();
+          for (int j = 0; j < count; j++) {
+              estimatorWithInverseQuantiles.insert(values[j]);
+          }
+          Map<Quantile, Long> snapshot;
+          snapshot = estimatorWithInverseQuantiles.snapshot();
+          for (Quantile q : quantiles) {
+              long actual = (long) ((1 - q.quantile) * count);
+              long error = (long) ((0.1 - q.error) * count);
+              long estimate = snapshot.get(q);
+              System.out
+                      .println(String.format("For quantile %f Expected %d with error %d, estimated %d",
+                              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

@@ -260,8 +260,8 @@ public DataNodeMetrics(String name, String sessionId, int[] intervals,
           "ops", "latency", interval);
       readTransferRateQuantiles[i] = registry.newQuantiles(
           "readTransferRate" + interval + "s",
-          "Rate at which bytes are read from datanode calculated in bytes per second",
-          "ops", "rate", interval);
+          "Rate at which bytes are read from datanode calculated in bytes per second with inverse quantiles",
+          "ops", "rate", interval, true);
     }
   }
 

hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/TestMultiThreadedHflush.java

@@ -53,7 +53,7 @@ public class TestMultiThreadedHflush {
       new Quantile[] {
         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) }, false);
 
   /*
    * creates a file but does not close it