[FLINK-31769] feat: quantile to aggregated metrcis

docs/layouts/shortcodes/generated/rest_v1_dispatcher.html

@@ -1495,7 +1495,7 @@
       <td colspan="2">
         <ul>
 <li><code>get</code> (optional): Comma-separated list of string values to select specific metrics.</li>
-<li><code>agg</code> (optional): Comma-separated list of aggregation modes which should be calculated. Available aggregations are: "min, max, sum, avg, skew".</li>
+<li><code>agg</code> (optional): Comma-separated list of aggregation modes which should be calculated. Available aggregations are: "min, max, sum, avg, skew, p50, p90, p99".</li>
 <li><code>jobs</code> (optional): Comma-separated list of 32-character hexadecimal strings to select specific jobs.</li>
         </ul>
       </td>
@@ -5636,7 +5636,7 @@
       <td colspan="2">
         <ul>
 <li><code>get</code> (optional): Comma-separated list of string values to select specific metrics.</li>
-<li><code>agg</code> (optional): Comma-separated list of aggregation modes which should be calculated. Available aggregations are: "min, max, sum, avg, skew".</li>
+<li><code>agg</code> (optional): Comma-separated list of aggregation modes which should be calculated. Available aggregations are: "min, max, sum, avg, skew, p50, p90, p99".</li>
 <li><code>subtasks</code> (optional): Comma-separated list of integer ranges (e.g. "1,3,5-9") to select specific subtasks.</li>
         </ul>
       </td>
@@ -6697,7 +6697,7 @@
       <td colspan="2">
         <ul>
 <li><code>get</code> (optional): Comma-separated list of string values to select specific metrics.</li>
-<li><code>agg</code> (optional): Comma-separated list of aggregation modes which should be calculated. Available aggregations are: "min, max, sum, avg, skew".</li>
+<li><code>agg</code> (optional): Comma-separated list of aggregation modes which should be calculated. Available aggregations are: "min, max, sum, avg, skew, p50, p90, p99".</li>
 <li><code>taskmanagers</code> (optional): Comma-separated list of 32-character hexadecimal strings to select specific task managers.</li>
         </ul>
       </td>

docs/static/generated/rest_v1_dispatcher.yml

@@ -490,7 +490,8 @@ paths:
       - name: agg
         in: query
         description: "Comma-separated list of aggregation modes which should be calculated.\
-          \ Available aggregations are: \"min, max, sum, avg, skew\"."
+          \ Available aggregations are: \"min, max, sum, avg, skew, p50, p90, p99\"\
+          ."
         required: false
         style: form
         schema:
@@ -1458,7 +1459,8 @@ paths:
       - name: agg
         in: query
         description: "Comma-separated list of aggregation modes which should be calculated.\
-          \ Available aggregations are: \"min, max, sum, avg, skew\"."
+          \ Available aggregations are: \"min, max, sum, avg, skew, p50, p90, p99\"\
+          ."
         required: false
         style: form
         schema:
@@ -1771,7 +1773,8 @@ paths:
       - name: agg
         in: query
         description: "Comma-separated list of aggregation modes which should be calculated.\
-          \ Available aggregations are: \"min, max, sum, avg, skew\"."
+          \ Available aggregations are: \"min, max, sum, avg, skew, p50, p90, p99\"\
+          ."
         required: false
         style: form
         schema:
@@ -1899,6 +1902,9 @@ components:
       - SUM
       - AVG
       - SKEW
+      - P50
+      - P90
+      - P99
     ApplicationDetails:
       type: object
       properties:

flink-runtime/pom.xml

@@ -215,6 +215,11 @@ under the License.
 			<artifactId>commons-text</artifactId>
 		</dependency>
 
+		<dependency>
+			<groupId>org.apache.commons</groupId>
+			<artifactId>commons-math3</artifactId>
+		</dependency>
+
 		<dependency>
 			<groupId>commons-cli</groupId>
 			<artifactId>commons-cli</artifactId>

flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/metrics/AbstractAggregatingMetricsHandler.java

@@ -125,7 +125,11 @@ protected CompletableFuture<AggregatedMetricsResponseBody> handleRequest(
                         DoubleAccumulator.DoubleAverageFactory averageFactory = null;
                         DoubleAccumulator.DoubleSumFactory sumFactory = null;
                         DoubleAccumulator.DoubleDataSkewFactory skewFactory = null;
-                        // by default we return all aggregations
+                        DoubleAccumulator.DoublePercentileFactory p50Factory = null;
+                        DoubleAccumulator.DoublePercentileFactory p90Factory = null;
+                        DoubleAccumulator.DoublePercentileFactory p99Factory = null;
+                        // by default we return all aggregations (percentiles are opt-in: they
+                        // buffer + sort, so they are only computed when explicitly requested)
                         if (requestedAggregations.isEmpty()) {
                             minimumFactory = DoubleAccumulator.DoubleMinimumFactory.get();
                             maximumFactory = DoubleAccumulator.DoubleMaximumFactory.get();
@@ -154,6 +158,18 @@ protected CompletableFuture<AggregatedMetricsResponseBody> handleRequest(
                                     case SKEW:
                                         skewFactory = DoubleAccumulator.DoubleDataSkewFactory.get();
                                         break;
+                                    case P50:
+                                        p50Factory =
+                                                DoubleAccumulator.DoublePercentileFactory.p50();
+                                        break;
+                                    case P90:
+                                        p90Factory =
+                                                DoubleAccumulator.DoublePercentileFactory.p90();
+                                        break;
+                                    case P99:
+                                        p99Factory =
+                                                DoubleAccumulator.DoublePercentileFactory.p99();
+                                        break;
                                     default:
                                         log.warn(
                                                 "Unsupported aggregation specified: {}",
@@ -167,7 +183,10 @@ protected CompletableFuture<AggregatedMetricsResponseBody> handleRequest(
                                         maximumFactory,
                                         averageFactory,
                                         sumFactory,
-                                        skewFactory);
+                                        skewFactory,
+                                        p50Factory,
+                                        p90Factory,
+                                        p99Factory);
 
                         return getAggregatedMetricValues(
                                 stores, requestedMetrics, metricAccumulatorFactory);
@@ -255,18 +274,27 @@ private static class MetricAccumulatorFactory {
 
         @Nullable private final DoubleAccumulator.DoubleSumFactory sumFactory;
         @Nullable private final DoubleAccumulator.DoubleDataSkewFactory dataSkewFactory;
+        @Nullable private final DoubleAccumulator.DoublePercentileFactory p50Factory;
+        @Nullable private final DoubleAccumulator.DoublePercentileFactory p90Factory;
+        @Nullable private final DoubleAccumulator.DoublePercentileFactory p99Factory;
 
         private MetricAccumulatorFactory(
                 @Nullable DoubleAccumulator.DoubleMinimumFactory minimumFactory,
                 @Nullable DoubleAccumulator.DoubleMaximumFactory maximumFactory,
                 @Nullable DoubleAccumulator.DoubleAverageFactory averageFactory,
                 @Nullable DoubleAccumulator.DoubleSumFactory sumFactory,
-                @Nullable DoubleAccumulator.DoubleDataSkewFactory dataSkewFactory) {
+                @Nullable DoubleAccumulator.DoubleDataSkewFactory dataSkewFactory,
+                @Nullable DoubleAccumulator.DoublePercentileFactory p50Factory,
+                @Nullable DoubleAccumulator.DoublePercentileFactory p90Factory,
+                @Nullable DoubleAccumulator.DoublePercentileFactory p99Factory) {
             this.minimumFactory = minimumFactory;
             this.maximumFactory = maximumFactory;
             this.averageFactory = averageFactory;
             this.sumFactory = sumFactory;
             this.dataSkewFactory = dataSkewFactory;
+            this.p50Factory = p50Factory;
+            this.p90Factory = p90Factory;
+            this.p99Factory = p99Factory;
         }
 
         MetricAccumulator get(String metricName, double init) {
@@ -276,7 +304,10 @@ MetricAccumulator get(String metricName, double init) {
                     maximumFactory == null ? null : maximumFactory.get(init),
                     averageFactory == null ? null : averageFactory.get(init),
                     sumFactory == null ? null : sumFactory.get(init),
-                    dataSkewFactory == null ? null : dataSkewFactory.get(init));
+                    dataSkewFactory == null ? null : dataSkewFactory.get(init),
+                    p50Factory == null ? null : p50Factory.get(init),
+                    p90Factory == null ? null : p90Factory.get(init),
+                    p99Factory == null ? null : p99Factory.get(init));
         }
     }
 
@@ -288,20 +319,29 @@ private static class MetricAccumulator {
         @Nullable private final DoubleAccumulator avg;
         @Nullable private final DoubleAccumulator sum;
         @Nullable private final DoubleAccumulator skew;
+        @Nullable private final DoubleAccumulator p50;
+        @Nullable private final DoubleAccumulator p90;
+        @Nullable private final DoubleAccumulator p99;
 
         private MetricAccumulator(
                 String metricName,
                 @Nullable DoubleAccumulator min,
                 @Nullable DoubleAccumulator max,
                 @Nullable DoubleAccumulator avg,
                 @Nullable DoubleAccumulator sum,
-                @Nullable DoubleAccumulator.DoubleDataSkew skew) {
+                @Nullable DoubleAccumulator.DoubleDataSkew skew,
+                @Nullable DoubleAccumulator p50,
+                @Nullable DoubleAccumulator p90,
+                @Nullable DoubleAccumulator p99) {
             this.metricName = Preconditions.checkNotNull(metricName);
             this.min = min;
             this.max = max;
             this.avg = avg;
             this.sum = sum;
             this.skew = skew;
+            this.p50 = p50;
+            this.p90 = p90;
+            this.p99 = p99;
         }
 
         void add(double value) {
@@ -320,6 +360,15 @@ void add(double value) {
             if (skew != null) {
                 skew.add(value);
             }
+            if (p50 != null) {
+                p50.add(value);
+            }
+            if (p90 != null) {
+                p90.add(value);
+            }
+            if (p99 != null) {
+                p99.add(value);
+            }
         }
 
         AggregatedMetric get() {
@@ -329,7 +378,10 @@ AggregatedMetric get() {
                     max == null ? null : max.getValue(),
                     avg == null ? null : avg.getValue(),
                     sum == null ? null : sum.getValue(),
-                    skew == null ? null : skew.getValue());
+                    skew == null ? null : skew.getValue(),
+                    p50 == null ? null : p50.getValue(),
+                    p90 == null ? null : p90.getValue(),
+                    p99 == null ? null : p99.getValue());
         }
     }
 }

flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/metrics/DoubleAccumulator.java

@@ -20,6 +20,9 @@
 
 import org.apache.flink.annotation.VisibleForTesting;
 
+import org.apache.commons.math3.stat.descriptive.rank.Percentile;
+import org.apache.commons.math3.stat.ranking.NaNStrategy;
+
 import java.util.ArrayList;
 import java.util.List;
 
@@ -144,6 +147,36 @@ public static DoubleDataSkewFactory get() {
         }
     }
 
+    /** Factory for {@link DoublePercentile}. */
+    final class DoublePercentileFactory implements DoubleAccumulatorFactory<DoublePercentile> {
+        private static final DoublePercentileFactory P50 = new DoublePercentileFactory(50);
+        private static final DoublePercentileFactory P90 = new DoublePercentileFactory(90);
+        private static final DoublePercentileFactory P99 = new DoublePercentileFactory(99);
+
+        private final double percentile;
+
+        private DoublePercentileFactory(double percentile) {
+            this.percentile = percentile;
+        }
+
+        @Override
+        public DoublePercentile get(double init) {
+            return new DoublePercentile(percentile, init);
+        }
+
+        public static DoublePercentileFactory p50() {
+            return P50;
+        }
+
+        public static DoublePercentileFactory p90() {
+            return P90;
+        }
+
+        public static DoublePercentileFactory p99() {
+            return P99;
+        }
+    }
+
     /** {@link DoubleAccumulator} that returns the maximum value. */
     final class DoubleMaximum implements DoubleAccumulator {
 
@@ -302,4 +335,40 @@ public String getName() {
             return NAME;
         }
     }
+
+    /**
+     * {@link DoubleAccumulator} that returns a percentile (e.g. p50/p90/p99) over all values. A
+     * percentile needs the whole sample, so this buffers every value (like {@link DoubleDataSkew})
+     * and computes the result lazily in {@link #getValue()}.
+     */
+    final class DoublePercentile implements DoubleAccumulator {
+
+        private final double percentile;
+
+        private final String name;
+
+        private final List<Double> values = new ArrayList<>();
+
+        private DoublePercentile(double percentile, double init) {
+            this.percentile = percentile;
+            this.name = "p" + (int) percentile;
+            values.add(init);
+        }
+
+        @Override
+        public void add(double value) {
+            values.add(value);
+        }
+
+        @Override
+        public double getValue() {
+            double[] arr = values.stream().mapToDouble(Double::doubleValue).toArray();
+            return new Percentile().withNaNStrategy(NaNStrategy.FIXED).evaluate(arr, percentile);
+        }
+
+        @Override
+        public String getName() {
+            return name;
+        }
+    }
 }