Generalise exponential bucket histogram, add percentile calculation

server/src/main/java/org/elasticsearch/common/metrics/ExponentialBucketHistogram.java

@@ -0,0 +1,131 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the "Elastic License
+ * 2.0", the "GNU Affero General Public License v3.0 only", and the "Server Side
+ * Public License v 1"; you may not use this file except in compliance with, at
+ * your election, the "Elastic License 2.0", the "GNU Affero General Public
+ * License v3.0 only", or the "Server Side Public License, v 1".
+ */
+
+package org.elasticsearch.common.metrics;
+
+import java.util.Arrays;
+import java.util.concurrent.atomic.LongAdder;
+
+/**
+ * A histogram with a fixed number of buckets of exponentially increasing width.
+ * <p>
+ * The bucket boundaries are defined by increasing powers of two, e.g.
+ * <code>
+ *     (-&infin;, 1), [1, 2), [2, 4), [4, 8), ..., [2^({@link #bucketCount}-2), &infin;)
+ * </code>
+ * There are {@link #bucketCount} buckets.
+ */
+public class ExponentialBucketHistogram {
+
+    private final int bucketCount;
+    private final long lastBucketLowerBound;
+
+    public static int[] getBucketUpperBounds(int bucketCount) {
+        int[] bounds = new int[bucketCount - 1];
+        for (int i = 0; i < bounds.length; i++) {
+            bounds[i] = 1 << i;
+        }
+        return bounds;
+    }
+
+    private int getBucket(long observedValue) {
+        if (observedValue <= 0) {
+            return 0;
+        } else if (lastBucketLowerBound <= observedValue) {
+            return bucketCount - 1;
+        } else {
+            return Long.SIZE - Long.numberOfLeadingZeros(observedValue);
+        }
+    }
+
+    private final LongAdder[] buckets;
+
+    public ExponentialBucketHistogram(int bucketCount) {
+        if (bucketCount < 2 || bucketCount > Integer.SIZE) {
+            throw new IllegalArgumentException("Bucket count must be in [2, " + Integer.SIZE + "], got " + bucketCount);
+        }
+        this.bucketCount = bucketCount;
+        this.lastBucketLowerBound = getBucketUpperBounds(bucketCount)[bucketCount - 2];
+        buckets = new LongAdder[bucketCount];
+        for (int i = 0; i < bucketCount; i++) {
+            buckets[i] = new LongAdder();
+        }
+    }
+
+    public int[] calculateBucketUpperBounds() {
+        return getBucketUpperBounds(bucketCount);
+    }
+
+    public void addObservation(long observedValue) {
+        buckets[getBucket(observedValue)].increment();
+    }
+
+    /**
+     * @return An array of frequencies of handling times in buckets with upper bounds as returned by {@link #calculateBucketUpperBounds()},
+     *         plus an extra bucket for handling times longer than the longest upper bound.
+     */
+    public long[] getSnapshot() {
+        final long[] histogram = new long[bucketCount];
+        for (int i = 0; i < bucketCount; i++) {
+            histogram[i] = buckets[i].longValue();
+        }
+        return histogram;
+    }
+
+    /**
+     * Calculate the Nth percentile value
+     *
+     * @param percentile The percentile as a fraction (in [0, 1.0])
+     * @return A value greater than the specified fraction of values in the histogram
+     * @throws IllegalArgumentException if the requested percentile is invalid
+     */
+    public long getPercentile(float percentile) {
+        return getPercentile(percentile, getSnapshot(), calculateBucketUpperBounds());
+    }
+
+    /**
+     * Calculate the Nth percentile value
+     *
+     * @param percentile The percentile as a fraction (in [0, 1.0])
+     * @param snapshot An array of frequencies of handling times in buckets with upper bounds as per {@link #calculateBucketUpperBounds()}
+     * @param bucketUpperBounds The upper bounds of the buckets in the histogram, as per {@link #calculateBucketUpperBounds()}
+     * @return A value greater than the specified fraction of values in the histogram
+     * @throws IllegalArgumentException if the requested percentile is invalid
+     */
+    public long getPercentile(float percentile, long[] snapshot, int[] bucketUpperBounds) {
+        assert snapshot.length == bucketCount && bucketUpperBounds.length == bucketCount - 1;
+        if (percentile < 0 || percentile > 1) {
+            throw new IllegalArgumentException("Requested percentile must be in [0, 1.0], percentile=" + percentile);
+        }
+        final long totalCount = Arrays.stream(snapshot).sum();
+        long percentileIndex = (long) Math.ceil(totalCount * percentile);
+        // Find which bucket has the Nth percentile value and return the upper bound value.
+        for (int i = 0; i < bucketCount; i++) {
+            percentileIndex -= snapshot[i];
+            if (percentileIndex <= 0) {
+                if (i == snapshot.length - 1) {
+                    return Long.MAX_VALUE;
+                } else {
+                    return bucketUpperBounds[i];
+                }
+            }
+        }
+        assert false : "We shouldn't ever get here";
+        return Long.MAX_VALUE;
+    }
+
+    /**
+     * Clear all values in the histogram (non-atomic)
+     */
+    public void clear() {
+        for (int i = 0; i < bucketCount; i++) {
+            buckets[i].reset();
+        }
+    }
+}

server/src/main/java/org/elasticsearch/common/network/HandlingTimeTracker.java

@@ -9,58 +9,20 @@
 
 package org.elasticsearch.common.network;
 
-import java.util.concurrent.atomic.LongAdder;
+import org.elasticsearch.common.metrics.ExponentialBucketHistogram;
 
 /**
  * Tracks how long message handling takes on a transport thread as a histogram with fixed buckets.
  */
-public class HandlingTimeTracker {
+public class HandlingTimeTracker extends ExponentialBucketHistogram {
 
-    public static int[] getBucketUpperBounds() {
-        int[] bounds = new int[17];
-        for (int i = 0; i < bounds.length; i++) {
-            bounds[i] = 1 << i;
-        }
-        return bounds;
-    }
+    public static final int BUCKET_COUNT = 18;
 
-    private static int getBucket(long handlingTimeMillis) {
-        if (handlingTimeMillis <= 0) {
-            return 0;
-        } else if (LAST_BUCKET_LOWER_BOUND <= handlingTimeMillis) {
-            return BUCKET_COUNT - 1;
-        } else {
-            return Long.SIZE - Long.numberOfLeadingZeros(handlingTimeMillis);
-        }
+    public static int[] getBucketUpperBounds() {
+        return ExponentialBucketHistogram.getBucketUpperBounds(BUCKET_COUNT);
     }
 
-    public static final int BUCKET_COUNT = getBucketUpperBounds().length + 1;
-
-    private static final long LAST_BUCKET_LOWER_BOUND = getBucketUpperBounds()[BUCKET_COUNT - 2];
-
-    private final LongAdder[] buckets;
-
     public HandlingTimeTracker() {
-        buckets = new LongAdder[BUCKET_COUNT];
-        for (int i = 0; i < BUCKET_COUNT; i++) {
-            buckets[i] = new LongAdder();
-        }
+        super(BUCKET_COUNT);
     }
-
-    public void addHandlingTime(long handlingTimeMillis) {
-        buckets[getBucket(handlingTimeMillis)].increment();
-    }
-
-    /**
-     * @return An array of frequencies of handling times in buckets with upper bounds as returned by {@link #getBucketUpperBounds()}, plus
-     *         an extra bucket for handling times longer than the longest upper bound.
-     */
-    public long[] getHistogram() {
-        final long[] histogram = new long[BUCKET_COUNT];
-        for (int i = 0; i < BUCKET_COUNT; i++) {
-            histogram[i] = buckets[i].longValue();
-        }
-        return histogram;
-    }
-
 }

server/src/main/java/org/elasticsearch/http/AbstractHttpServerTransport.java

@@ -460,7 +460,7 @@ public void incomingRequest(final HttpRequest httpRequest, final HttpChannel htt
             handleIncomingRequest(httpRequest, trackingChannel, httpRequest.getInboundException());
         } finally {
             final long took = threadPool.rawRelativeTimeInMillis() - startTime;
-            networkService.getHandlingTimeTracker().addHandlingTime(took);
+            networkService.getHandlingTimeTracker().addObservation(took);
             final long logThreshold = slowLogThresholdMs;
             if (logThreshold > 0 && took > logThreshold) {
                 logger.warn(

server/src/main/java/org/elasticsearch/http/HttpRouteStatsTracker.java

@@ -91,7 +91,7 @@ public void addResponseStats(long contentLength) {
     }
 
     public void addResponseTime(long timeMillis) {
-        responseTimeTracker.addHandlingTime(timeMillis);
+        responseTimeTracker.addObservation(timeMillis);
     }
 
     public HttpRouteStats getStats() {
@@ -102,7 +102,7 @@ public HttpRouteStats getStats() {
             responseStats.count().longValue(),
             responseStats.totalSize().longValue(),
             responseStats.getHistogram(),
-            responseTimeTracker.getHistogram()
+            responseTimeTracker.getSnapshot()
         );
     }
 }

server/src/main/java/org/elasticsearch/transport/InboundHandler.java

@@ -138,7 +138,7 @@ private void messageReceived(TcpChannel channel, InboundMessage message, long st
             }
         } finally {
             final long took = threadPool.rawRelativeTimeInMillis() - startTime;
-            handlingTimeTracker.addHandlingTime(took);
+            handlingTimeTracker.addObservation(took);
             final long logThreshold = slowLogThresholdMs;
             if (logThreshold > 0 && took > logThreshold) {
                 logSlowMessage(message, took, logThreshold, responseHandler);

server/src/main/java/org/elasticsearch/transport/OutboundHandler.java

@@ -414,7 +414,7 @@ private void maybeLogSlowMessage(boolean success) {
                     final long logThreshold = slowLogThresholdMs;
                     if (logThreshold > 0) {
                         final long took = threadPool.rawRelativeTimeInMillis() - startTime;
-                        handlingTimeTracker.addHandlingTime(took);
+                        handlingTimeTracker.addObservation(took);
                         if (took > logThreshold) {
                             logger.warn(
                                 "sending transport message [{}] of size [{}] on [{}] took [{}ms] which is above the warn "

server/src/main/java/org/elasticsearch/transport/TcpTransport.java

@@ -1001,8 +1001,8 @@ public final TransportStats getStats() {
             bytesRead,
             messagesSent,
             bytesWritten,
-            networkService.getHandlingTimeTracker().getHistogram(),
-            outboundHandlingTimeTracker.getHistogram(),
+            networkService.getHandlingTimeTracker().getSnapshot(),
+            outboundHandlingTimeTracker.getSnapshot(),
             requestHandlers.getStats()
         );
     }