diff --git a/docs/layouts/shortcodes/generated/rest_v1_dispatcher.html b/docs/layouts/shortcodes/generated/rest_v1_dispatcher.html index 26749b00f691b..fc4f3aef2b679 100644 --- a/docs/layouts/shortcodes/generated/rest_v1_dispatcher.html +++ b/docs/layouts/shortcodes/generated/rest_v1_dispatcher.html @@ -4988,6 +4988,117 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + +
/jobs/:jobid/vertices/subtasks/metrics/names
Verb: GETResponse code: 200 OK
Provides batch access to aggregated subtask metric names.
Path parameters
+
    +
  • jobid - 32-character hexadecimal string value that identifies a job.
    • +
+
Query parameters
+
    +
  • regex (optional): Comma-separated list of Java regular expressions to filter metric names by full match.
    • +
  • vertices (mandatory): Comma-separated list of 32-character hexadecimal strings to select specific job vertices.
    • +
+
+
+
+ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + +
/jobs/:jobid/vertices/subtasks/metrics/values
Verb: GETResponse code: 200 OK
Provides batch access to aggregated subtask metric values.
Path parameters
+
    +
  • jobid - 32-character hexadecimal string value that identifies a job.
    • +
+
Query parameters
+
    +
  • get (optional): Comma-separated list of string values to select specific metrics.
    • +
  • agg (optional): Comma-separated list of aggregation modes which should be calculated. Available aggregations are: "min, max, sum, avg, skew".
    • +
  • vertices (mandatory): Comma-separated list of 32-character hexadecimal strings to select specific job vertices.
    • +
+
+
+
diff --git a/docs/static/generated/rest_v1_dispatcher.yml b/docs/static/generated/rest_v1_dispatcher.yml index 0c94fafc052e5..c2c8e6d0b3c02 100644 --- a/docs/static/generated/rest_v1_dispatcher.yml +++ b/docs/static/generated/rest_v1_dispatcher.yml @@ -1237,6 +1237,81 @@ paths: application/json: schema: $ref: "#/components/schemas/LogUrlResponse" + /jobs/{jobid}/vertices/subtasks/metrics/names: + get: + description: Provides batch access to aggregated subtask metric names. + operationId: getAggregatedSubtaskMetricsNames + parameters: + - name: jobid + in: path + description: 32-character hexadecimal string value that identifies a job. + required: true + schema: + $ref: "#/components/schemas/JobID" + - name: vertices + in: query + description: Comma-separated list of 32-character hexadecimal strings to select + specific job vertices. + required: true + style: form + schema: + $ref: "#/components/schemas/JobVertexID" + - name: regex + in: query + description: Comma-separated list of Java regular expressions to filter metric + names by full match. + required: false + style: form + schema: + type: string + responses: + "200": + description: The request was successful. + content: + application/json: + schema: + $ref: "#/components/schemas/AggregatedSubtaskMetricsBatchResponseBody" + /jobs/{jobid}/vertices/subtasks/metrics/values: + get: + description: Provides batch access to aggregated subtask metric values. + operationId: getAggregatedSubtaskMetricsValues + parameters: + - name: jobid + in: path + description: 32-character hexadecimal string value that identifies a job. + required: true + schema: + $ref: "#/components/schemas/JobID" + - name: vertices + in: query + description: Comma-separated list of 32-character hexadecimal strings to select + specific job vertices. + required: true + style: form + schema: + $ref: "#/components/schemas/JobVertexID" + - name: get + in: query + description: Comma-separated list of string values to select specific metrics. + required: false + style: form + schema: + type: string + - name: agg + in: query + description: "Comma-separated list of aggregation modes which should be calculated.\ + \ Available aggregations are: \"min, max, sum, avg, skew\"." + required: false + style: form + schema: + $ref: "#/components/schemas/AggregationMode" + responses: + "200": + description: The request was successful. + content: + application/json: + schema: + $ref: "#/components/schemas/AggregatedSubtaskMetricsBatchResponseBody" /jobs/{jobid}/vertices/{vertexid}: get: description: "Returns details for a task, with a summary for each of its subtasks." @@ -1891,6 +1966,8 @@ components: schemas: AggregatedMetricsResponseBody: type: object + AggregatedSubtaskMetricsBatchResponseBody: + type: object AggregatedTaskDetailsInfo: type: object properties: diff --git a/flink-runtime-web/src/test/resources/rest_api_v1.snapshot b/flink-runtime-web/src/test/resources/rest_api_v1.snapshot index a1dd691a7d29d..ee02d6d3aa5d8 100644 --- a/flink-runtime-web/src/test/resources/rest_api_v1.snapshot +++ b/flink-runtime-web/src/test/resources/rest_api_v1.snapshot @@ -3846,6 +3846,61 @@ } } } + }, { + "url" : "/jobs/:jobid/vertices/subtasks/metrics/names", + "method" : "GET", + "status-code" : "200 OK", + "file-upload" : false, + "path-parameters" : { + "pathParameters" : [ { + "key" : "jobid" + } ] + }, + "query-parameters" : { + "queryParameters" : [ { + "key" : "vertices", + "mandatory" : true + }, { + "key" : "regex", + "mandatory" : false + } ] + }, + "request" : { + "type" : "object", + "id" : "urn:jsonschema:org:apache:flink:runtime:rest:messages:EmptyRequestBody" + }, + "response" : { + "type" : "any" + } + }, { + "url" : "/jobs/:jobid/vertices/subtasks/metrics/values", + "method" : "GET", + "status-code" : "200 OK", + "file-upload" : false, + "path-parameters" : { + "pathParameters" : [ { + "key" : "jobid" + } ] + }, + "query-parameters" : { + "queryParameters" : [ { + "key" : "vertices", + "mandatory" : true + }, { + "key" : "get", + "mandatory" : false + }, { + "key" : "agg", + "mandatory" : false + } ] + }, + "request" : { + "type" : "object", + "id" : "urn:jsonschema:org:apache:flink:runtime:rest:messages:EmptyRequestBody" + }, + "response" : { + "type" : "any" + } }, { "url" : "/jobs/:jobid/vertices/:vertexid", "method" : "GET", diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/metrics/AbstractAggregatingMetricsHandler.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/metrics/AbstractAggregatingMetricsHandler.java index 79fa9b944ab6d..6ee99dcb5876b 100644 --- a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/metrics/AbstractAggregatingMetricsHandler.java +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/metrics/AbstractAggregatingMetricsHandler.java @@ -32,22 +32,16 @@ import org.apache.flink.runtime.rest.messages.job.metrics.MetricsFilterParameter; import org.apache.flink.runtime.webmonitor.RestfulGateway; import org.apache.flink.runtime.webmonitor.retriever.GatewayRetriever; -import org.apache.flink.util.CollectionUtil; import org.apache.flink.util.Preconditions; import org.apache.flink.shaded.netty4.io.netty.handler.codec.http.HttpResponseStatus; import javax.annotation.Nonnull; -import javax.annotation.Nullable; import java.time.Duration; -import java.util.ArrayList; import java.util.Collection; -import java.util.Collections; -import java.util.Iterator; import java.util.List; import java.util.Map; -import java.util.Set; import java.util.concurrent.CompletableFuture; import java.util.concurrent.CompletionException; import java.util.concurrent.Executor; @@ -113,64 +107,16 @@ protected CompletableFuture handleRequest( getStores(store, request); if (requestedMetrics.isEmpty()) { - Collection list = getAvailableMetrics(stores); + Collection list = + AggregatedMetricsStoreHelper.getAvailableMetrics(stores); return new AggregatedMetricsResponseBody( list.stream() .map(AggregatedMetric::new) .collect(Collectors.toList())); } - DoubleAccumulator.DoubleMinimumFactory minimumFactory = null; - DoubleAccumulator.DoubleMaximumFactory maximumFactory = null; - DoubleAccumulator.DoubleAverageFactory averageFactory = null; - DoubleAccumulator.DoubleSumFactory sumFactory = null; - DoubleAccumulator.DoubleDataSkewFactory skewFactory = null; - // by default we return all aggregations - if (requestedAggregations.isEmpty()) { - minimumFactory = DoubleAccumulator.DoubleMinimumFactory.get(); - maximumFactory = DoubleAccumulator.DoubleMaximumFactory.get(); - averageFactory = DoubleAccumulator.DoubleAverageFactory.get(); - sumFactory = DoubleAccumulator.DoubleSumFactory.get(); - skewFactory = DoubleAccumulator.DoubleDataSkewFactory.get(); - } else { - for (MetricsAggregationParameter.AggregationMode aggregation : - requestedAggregations) { - switch (aggregation) { - case MIN: - minimumFactory = - DoubleAccumulator.DoubleMinimumFactory.get(); - break; - case MAX: - maximumFactory = - DoubleAccumulator.DoubleMaximumFactory.get(); - break; - case AVG: - averageFactory = - DoubleAccumulator.DoubleAverageFactory.get(); - break; - case SUM: - sumFactory = DoubleAccumulator.DoubleSumFactory.get(); - break; - case SKEW: - skewFactory = DoubleAccumulator.DoubleDataSkewFactory.get(); - break; - default: - log.warn( - "Unsupported aggregation specified: {}", - aggregation); - } - } - } - MetricAccumulatorFactory metricAccumulatorFactory = - new MetricAccumulatorFactory( - minimumFactory, - maximumFactory, - averageFactory, - sumFactory, - skewFactory); - - return getAggregatedMetricValues( - stores, requestedMetrics, metricAccumulatorFactory); + return AggregatedMetricsStoreHelper.getAggregatedMetricValues( + stores, requestedMetrics, requestedAggregations); } catch (Exception e) { log.warn("Could not retrieve metrics.", e); throw new CompletionException( @@ -181,155 +127,4 @@ protected CompletableFuture handleRequest( }, executor); } - - /** - * Returns a JSON string containing a list of all available metrics in the given stores. - * Effectively this method maps the union of all key-sets to JSON. - * - * @param stores metrics - * @return JSON string containing a list of all available metrics - */ - private static Collection getAvailableMetrics( - Collection stores) { - Set uniqueMetrics = CollectionUtil.newHashSetWithExpectedSize(32); - for (MetricStore.ComponentMetricStore store : stores) { - uniqueMetrics.addAll(store.metrics.keySet()); - } - return uniqueMetrics; - } - - /** - * Extracts and aggregates all requested metrics from the given metric stores, and maps the - * result to a JSON string. - * - * @param stores available metrics - * @param requestedMetrics ids of requested metrics - * @param requestedAggregationsFactories requested aggregations - * @return JSON string containing the requested metrics - */ - private AggregatedMetricsResponseBody getAggregatedMetricValues( - Collection stores, - List requestedMetrics, - MetricAccumulatorFactory requestedAggregationsFactories) { - - Collection aggregatedMetrics = new ArrayList<>(requestedMetrics.size()); - for (String requestedMetric : requestedMetrics) { - final Collection values = new ArrayList<>(stores.size()); - try { - for (MetricStore.ComponentMetricStore store : stores) { - String stringValue = store.metrics.get(requestedMetric); - if (stringValue != null) { - values.add(Double.valueOf(stringValue)); - } - } - } catch (NumberFormatException nfe) { - log.warn( - "The metric {} is not numeric and can't be aggregated.", - requestedMetric, - nfe); - // metric is not numeric so we can't perform aggregations => ignore it - continue; - } - if (!values.isEmpty()) { - - Iterator valuesIterator = values.iterator(); - MetricAccumulator acc = - requestedAggregationsFactories.get(requestedMetric, valuesIterator.next()); - valuesIterator.forEachRemaining(acc::add); - - aggregatedMetrics.add(acc.get()); - } else { - return new AggregatedMetricsResponseBody(Collections.emptyList()); - } - } - return new AggregatedMetricsResponseBody(aggregatedMetrics); - } - - private static class MetricAccumulatorFactory { - - @Nullable private final DoubleAccumulator.DoubleMinimumFactory minimumFactory; - - @Nullable private final DoubleAccumulator.DoubleMaximumFactory maximumFactory; - - @Nullable private final DoubleAccumulator.DoubleAverageFactory averageFactory; - - @Nullable private final DoubleAccumulator.DoubleSumFactory sumFactory; - @Nullable private final DoubleAccumulator.DoubleDataSkewFactory dataSkewFactory; - - private MetricAccumulatorFactory( - @Nullable DoubleAccumulator.DoubleMinimumFactory minimumFactory, - @Nullable DoubleAccumulator.DoubleMaximumFactory maximumFactory, - @Nullable DoubleAccumulator.DoubleAverageFactory averageFactory, - @Nullable DoubleAccumulator.DoubleSumFactory sumFactory, - @Nullable DoubleAccumulator.DoubleDataSkewFactory dataSkewFactory) { - this.minimumFactory = minimumFactory; - this.maximumFactory = maximumFactory; - this.averageFactory = averageFactory; - this.sumFactory = sumFactory; - this.dataSkewFactory = dataSkewFactory; - } - - MetricAccumulator get(String metricName, double init) { - return new MetricAccumulator( - metricName, - minimumFactory == null ? null : minimumFactory.get(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)); - } - } - - private static class MetricAccumulator { - private final String metricName; - - @Nullable private final DoubleAccumulator min; - @Nullable private final DoubleAccumulator max; - @Nullable private final DoubleAccumulator avg; - @Nullable private final DoubleAccumulator sum; - @Nullable private final DoubleAccumulator skew; - - private MetricAccumulator( - String metricName, - @Nullable DoubleAccumulator min, - @Nullable DoubleAccumulator max, - @Nullable DoubleAccumulator avg, - @Nullable DoubleAccumulator sum, - @Nullable DoubleAccumulator.DoubleDataSkew skew) { - this.metricName = Preconditions.checkNotNull(metricName); - this.min = min; - this.max = max; - this.avg = avg; - this.sum = sum; - this.skew = skew; - } - - void add(double value) { - if (min != null) { - min.add(value); - } - if (max != null) { - max.add(value); - } - if (avg != null) { - avg.add(value); - } - if (sum != null) { - sum.add(value); - } - if (skew != null) { - skew.add(value); - } - } - - AggregatedMetric get() { - return new AggregatedMetric( - metricName, - min == null ? null : min.getValue(), - max == null ? null : max.getValue(), - avg == null ? null : avg.getValue(), - sum == null ? null : sum.getValue(), - skew == null ? null : skew.getValue()); - } - } } diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/metrics/AggregatedMetricsStoreHelper.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/metrics/AggregatedMetricsStoreHelper.java new file mode 100644 index 0000000000000..993ca2fabfe7e --- /dev/null +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/metrics/AggregatedMetricsStoreHelper.java @@ -0,0 +1,240 @@ +/* + * 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.flink.runtime.rest.handler.job.metrics; + +import org.apache.flink.runtime.rest.handler.legacy.metrics.MetricStore; +import org.apache.flink.runtime.rest.messages.job.metrics.AggregatedMetric; +import org.apache.flink.runtime.rest.messages.job.metrics.AggregatedMetricsResponseBody; +import org.apache.flink.runtime.rest.messages.job.metrics.MetricsAggregationParameter; +import org.apache.flink.util.CollectionUtil; +import org.apache.flink.util.Preconditions; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import javax.annotation.Nullable; + +import java.util.ArrayList; +import java.util.Collection; +import java.util.Collections; +import java.util.Iterator; +import java.util.List; +import java.util.Set; + +/** Helper for aggregating metric values from metric stores. */ +final class AggregatedMetricsStoreHelper { + + private static final Logger LOG = LoggerFactory.getLogger(AggregatedMetricsStoreHelper.class); + + private AggregatedMetricsStoreHelper() {} + + /** + * Returns a JSON string containing a list of all available metrics in the given stores. + * Effectively this method maps the union of all key-sets to JSON. + * + * @param stores metrics + * @return JSON string containing a list of all available metrics + */ + static Collection getAvailableMetrics( + Collection stores) { + Set uniqueMetrics = CollectionUtil.newHashSetWithExpectedSize(32); + for (MetricStore.ComponentMetricStore store : stores) { + uniqueMetrics.addAll(store.metrics.keySet()); + } + return uniqueMetrics; + } + + /** + * Extracts and aggregates all requested metrics from the given metric stores, and maps the + * result to a JSON string. + * + * @param stores available metrics + * @param requestedMetrics ids of requested metrics + * @param requestedAggregations requested aggregation modes + * @return JSON string containing the requested metrics + */ + static AggregatedMetricsResponseBody getAggregatedMetricValues( + Collection stores, + List requestedMetrics, + List requestedAggregations) { + final MetricAccumulatorFactory requestedAggregationsFactories = + createMetricAccumulatorFactory(requestedAggregations); + + Collection aggregatedMetrics = new ArrayList<>(requestedMetrics.size()); + for (String requestedMetric : requestedMetrics) { + final Collection values = new ArrayList<>(stores.size()); + try { + for (MetricStore.ComponentMetricStore store : stores) { + String stringValue = store.metrics.get(requestedMetric); + if (stringValue != null) { + values.add(Double.valueOf(stringValue)); + } + } + } catch (NumberFormatException nfe) { + LOG.warn( + "The metric {} is not numeric and can't be aggregated.", + requestedMetric, + nfe); + // metric is not numeric so we can't perform aggregations => ignore it + continue; + } + if (!values.isEmpty()) { + + Iterator valuesIterator = values.iterator(); + MetricAccumulator acc = + requestedAggregationsFactories.get(requestedMetric, valuesIterator.next()); + valuesIterator.forEachRemaining(acc::add); + + aggregatedMetrics.add(acc.get()); + } else { + return new AggregatedMetricsResponseBody(Collections.emptyList()); + } + } + return new AggregatedMetricsResponseBody(aggregatedMetrics); + } + + private static MetricAccumulatorFactory createMetricAccumulatorFactory( + List requestedAggregations) { + DoubleAccumulator.DoubleMinimumFactory minimumFactory = null; + DoubleAccumulator.DoubleMaximumFactory maximumFactory = null; + DoubleAccumulator.DoubleAverageFactory averageFactory = null; + DoubleAccumulator.DoubleSumFactory sumFactory = null; + DoubleAccumulator.DoubleDataSkewFactory skewFactory = null; + // by default we return all aggregations + if (requestedAggregations.isEmpty()) { + minimumFactory = DoubleAccumulator.DoubleMinimumFactory.get(); + maximumFactory = DoubleAccumulator.DoubleMaximumFactory.get(); + averageFactory = DoubleAccumulator.DoubleAverageFactory.get(); + sumFactory = DoubleAccumulator.DoubleSumFactory.get(); + skewFactory = DoubleAccumulator.DoubleDataSkewFactory.get(); + } else { + for (MetricsAggregationParameter.AggregationMode aggregation : requestedAggregations) { + switch (aggregation) { + case MIN: + minimumFactory = DoubleAccumulator.DoubleMinimumFactory.get(); + break; + case MAX: + maximumFactory = DoubleAccumulator.DoubleMaximumFactory.get(); + break; + case AVG: + averageFactory = DoubleAccumulator.DoubleAverageFactory.get(); + break; + case SUM: + sumFactory = DoubleAccumulator.DoubleSumFactory.get(); + break; + case SKEW: + skewFactory = DoubleAccumulator.DoubleDataSkewFactory.get(); + break; + default: + LOG.warn("Unsupported aggregation specified: {}", aggregation); + } + } + } + return new MetricAccumulatorFactory( + minimumFactory, maximumFactory, averageFactory, sumFactory, skewFactory); + } + + private static class MetricAccumulatorFactory { + + @Nullable private final DoubleAccumulator.DoubleMinimumFactory minimumFactory; + + @Nullable private final DoubleAccumulator.DoubleMaximumFactory maximumFactory; + + @Nullable private final DoubleAccumulator.DoubleAverageFactory averageFactory; + + @Nullable private final DoubleAccumulator.DoubleSumFactory sumFactory; + @Nullable private final DoubleAccumulator.DoubleDataSkewFactory dataSkewFactory; + + private MetricAccumulatorFactory( + @Nullable DoubleAccumulator.DoubleMinimumFactory minimumFactory, + @Nullable DoubleAccumulator.DoubleMaximumFactory maximumFactory, + @Nullable DoubleAccumulator.DoubleAverageFactory averageFactory, + @Nullable DoubleAccumulator.DoubleSumFactory sumFactory, + @Nullable DoubleAccumulator.DoubleDataSkewFactory dataSkewFactory) { + this.minimumFactory = minimumFactory; + this.maximumFactory = maximumFactory; + this.averageFactory = averageFactory; + this.sumFactory = sumFactory; + this.dataSkewFactory = dataSkewFactory; + } + + MetricAccumulator get(String metricName, double init) { + return new MetricAccumulator( + metricName, + minimumFactory == null ? null : minimumFactory.get(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)); + } + } + + private static class MetricAccumulator { + private final String metricName; + + @Nullable private final DoubleAccumulator min; + @Nullable private final DoubleAccumulator max; + @Nullable private final DoubleAccumulator avg; + @Nullable private final DoubleAccumulator sum; + @Nullable private final DoubleAccumulator skew; + + private MetricAccumulator( + String metricName, + @Nullable DoubleAccumulator min, + @Nullable DoubleAccumulator max, + @Nullable DoubleAccumulator avg, + @Nullable DoubleAccumulator sum, + @Nullable DoubleAccumulator.DoubleDataSkew skew) { + this.metricName = Preconditions.checkNotNull(metricName); + this.min = min; + this.max = max; + this.avg = avg; + this.sum = sum; + this.skew = skew; + } + + void add(double value) { + if (min != null) { + min.add(value); + } + if (max != null) { + max.add(value); + } + if (avg != null) { + avg.add(value); + } + if (sum != null) { + sum.add(value); + } + if (skew != null) { + skew.add(value); + } + } + + AggregatedMetric get() { + return new AggregatedMetric( + metricName, + min == null ? null : min.getValue(), + max == null ? null : max.getValue(), + avg == null ? null : avg.getValue(), + sum == null ? null : sum.getValue(), + skew == null ? null : skew.getValue()); + } + } +} diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/metrics/AggregatingSubtasksMetricsBatchHandler.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/metrics/AggregatingSubtasksMetricsBatchHandler.java new file mode 100644 index 0000000000000..f574ab1f2940b --- /dev/null +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/metrics/AggregatingSubtasksMetricsBatchHandler.java @@ -0,0 +1,253 @@ +/* + * 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.flink.runtime.rest.handler.job.metrics; + +import org.apache.flink.api.common.JobID; +import org.apache.flink.runtime.jobgraph.JobVertexID; +import org.apache.flink.runtime.rest.handler.AbstractRestHandler; +import org.apache.flink.runtime.rest.handler.HandlerRequest; +import org.apache.flink.runtime.rest.handler.RestHandlerException; +import org.apache.flink.runtime.rest.handler.legacy.metrics.MetricFetcher; +import org.apache.flink.runtime.rest.handler.legacy.metrics.MetricStore; +import org.apache.flink.runtime.rest.messages.EmptyRequestBody; +import org.apache.flink.runtime.rest.messages.JobIDPathParameter; +import org.apache.flink.runtime.rest.messages.RuntimeMessageHeaders; +import org.apache.flink.runtime.rest.messages.job.metrics.AggregatedMetric; +import org.apache.flink.runtime.rest.messages.job.metrics.AggregatedMetricsResponseBody; +import org.apache.flink.runtime.rest.messages.job.metrics.AggregatedSubtaskMetricsBatchParameters; +import org.apache.flink.runtime.rest.messages.job.metrics.AggregatedSubtaskMetricsBatchResponseBody; +import org.apache.flink.runtime.rest.messages.job.metrics.AggregatedSubtaskMetricsNamesHeaders; +import org.apache.flink.runtime.rest.messages.job.metrics.AggregatedSubtaskMetricsNamesParameters; +import org.apache.flink.runtime.rest.messages.job.metrics.AggregatedSubtaskMetricsValuesHeaders; +import org.apache.flink.runtime.rest.messages.job.metrics.AggregatedSubtaskMetricsValuesParameters; +import org.apache.flink.runtime.rest.messages.job.metrics.JobVerticesFilterQueryParameter; +import org.apache.flink.runtime.rest.messages.job.metrics.MetricsAggregationParameter; +import org.apache.flink.runtime.rest.messages.job.metrics.MetricsFilterParameter; +import org.apache.flink.runtime.rest.messages.job.metrics.MetricsRegexFilterParameter; +import org.apache.flink.runtime.webmonitor.RestfulGateway; +import org.apache.flink.runtime.webmonitor.retriever.GatewayRetriever; +import org.apache.flink.util.Preconditions; + +import org.apache.flink.shaded.netty4.io.netty.handler.codec.http.HttpResponseStatus; + +import javax.annotation.Nonnull; + +import java.time.Duration; +import java.util.ArrayList; +import java.util.Collection; +import java.util.Collections; +import java.util.List; +import java.util.Map; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CompletionException; +import java.util.concurrent.Executor; +import java.util.regex.Pattern; +import java.util.regex.PatternSyntaxException; +import java.util.stream.Collectors; + +/** Batch handlers for aggregated subtask metrics. */ +public class AggregatingSubtasksMetricsBatchHandler { + + private AggregatingSubtasksMetricsBatchHandler() {} + + /** Handler for batch aggregated subtask metric name discovery. */ + public static class NamesHandler + extends AbstractBatchSubtasksMetricsHandler { + + public NamesHandler( + GatewayRetriever leaderRetriever, + Duration timeout, + Map responseHeaders, + Executor executor, + MetricFetcher fetcher) { + super( + leaderRetriever, + timeout, + responseHeaders, + AggregatedSubtaskMetricsNamesHeaders.getInstance(), + executor, + fetcher); + } + + @Override + AggregatedSubtaskMetricsBatchResponseBody handleBatchRequest( + MetricStore store, JobID jobId, HandlerRequest request) + throws RestHandlerException { + final List patterns = + compilePatterns(request.getQueryParameter(MetricsRegexFilterParameter.class)); + final List vertexIds = getVertexIds(request); + final Collection + metricsByVertex = new ArrayList<>(vertexIds.size()); + + for (JobVertexID vertexId : vertexIds) { + final Collection metrics = + AggregatedMetricsStoreHelper.getAvailableMetrics( + getSubtaskMetricStores(store, jobId, vertexId)) + .stream() + .filter(metric -> matchesAny(metric, patterns)) + .sorted() + .map(AggregatedMetric::new) + .collect(Collectors.toList()); + metricsByVertex.add( + new AggregatedSubtaskMetricsBatchResponseBody.VertexAggregatedMetrics( + vertexId, metrics)); + } + + return new AggregatedSubtaskMetricsBatchResponseBody(metricsByVertex); + } + + private static List compilePatterns(Collection regex) + throws RestHandlerException { + try { + return regex.stream().map(Pattern::compile).collect(Collectors.toList()); + } catch (PatternSyntaxException e) { + throw new RestHandlerException( + "Invalid metric name regex.", HttpResponseStatus.BAD_REQUEST, e); + } + } + + private static boolean matchesAny(String metric, List patterns) { + return patterns.isEmpty() + || patterns.stream().anyMatch(pattern -> pattern.matcher(metric).matches()); + } + } + + /** Handler for batch aggregated subtask metric values. */ + public static class ValuesHandler + extends AbstractBatchSubtasksMetricsHandler { + + public ValuesHandler( + GatewayRetriever leaderRetriever, + Duration timeout, + Map responseHeaders, + Executor executor, + MetricFetcher fetcher) { + super( + leaderRetriever, + timeout, + responseHeaders, + AggregatedSubtaskMetricsValuesHeaders.getInstance(), + executor, + fetcher); + } + + @Override + AggregatedSubtaskMetricsBatchResponseBody handleBatchRequest( + MetricStore store, JobID jobId, HandlerRequest request) + throws RestHandlerException { + final List aggregations = + request.getQueryParameter(MetricsAggregationParameter.class); + final List requestedMetrics = + request.getQueryParameter(MetricsFilterParameter.class); + final List vertexIds = getVertexIds(request); + final Collection + metricsByVertex = new ArrayList<>(vertexIds.size()); + + for (JobVertexID vertexId : vertexIds) { + final AggregatedMetricsResponseBody aggregatedMetrics = + AggregatedMetricsStoreHelper.getAggregatedMetricValues( + getSubtaskMetricStores(store, jobId, vertexId), + requestedMetrics, + aggregations); + metricsByVertex.add( + new AggregatedSubtaskMetricsBatchResponseBody.VertexAggregatedMetrics( + vertexId, aggregatedMetrics.getMetrics())); + } + + return new AggregatedSubtaskMetricsBatchResponseBody(metricsByVertex); + } + } + + private abstract static class AbstractBatchSubtasksMetricsHandler< + P extends AggregatedSubtaskMetricsBatchParameters> + extends AbstractRestHandler< + RestfulGateway, + EmptyRequestBody, + AggregatedSubtaskMetricsBatchResponseBody, + P> { + + private final Executor executor; + private final MetricFetcher fetcher; + + private AbstractBatchSubtasksMetricsHandler( + GatewayRetriever leaderRetriever, + Duration timeout, + Map responseHeaders, + RuntimeMessageHeaders< + EmptyRequestBody, AggregatedSubtaskMetricsBatchResponseBody, P> + messageHeaders, + Executor executor, + MetricFetcher fetcher) { + super(leaderRetriever, timeout, responseHeaders, messageHeaders); + this.executor = Preconditions.checkNotNull(executor); + this.fetcher = Preconditions.checkNotNull(fetcher); + } + + @Override + protected CompletableFuture handleRequest( + @Nonnull HandlerRequest request, @Nonnull RestfulGateway gateway) + throws RestHandlerException { + return CompletableFuture.supplyAsync( + () -> { + try { + fetcher.update(); + return handleBatchRequest( + fetcher.getMetricStore(), + request.getPathParameter(JobIDPathParameter.class), + request); + } catch (RestHandlerException e) { + throw new CompletionException(e); + } catch (Exception e) { + throw new CompletionException( + new RestHandlerException( + "Could not retrieve metrics.", + HttpResponseStatus.INTERNAL_SERVER_ERROR, + e)); + } + }, + executor); + } + + abstract AggregatedSubtaskMetricsBatchResponseBody handleBatchRequest( + MetricStore store, JobID jobId, HandlerRequest request) + throws RestHandlerException; + + final List getVertexIds(HandlerRequest request) + throws RestHandlerException { + final List vertexIds = + request.getQueryParameter(JobVerticesFilterQueryParameter.class); + if (vertexIds.isEmpty()) { + throw new RestHandlerException( + "At least one job vertex must be specified.", + HttpResponseStatus.BAD_REQUEST); + } + return vertexIds; + } + } + + private static Collection getSubtaskMetricStores( + MetricStore store, JobID jobId, JobVertexID vertexId) { + MetricStore.TaskMetricStore taskMetricStore = + store.getTaskMetricStore(jobId.toString(), vertexId.toString()); + if (taskMetricStore == null) { + return Collections.emptyList(); + } + return taskMetricStore.getAllSubtaskMetricStores().values(); + } +} diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/job/metrics/AggregatedSubtaskMetricsBatchParameters.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/job/metrics/AggregatedSubtaskMetricsBatchParameters.java new file mode 100644 index 0000000000000..0f7cc33cf0cb3 --- /dev/null +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/job/metrics/AggregatedSubtaskMetricsBatchParameters.java @@ -0,0 +1,37 @@ +/* + * 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.flink.runtime.rest.messages.job.metrics; + +import org.apache.flink.runtime.rest.messages.JobIDPathParameter; +import org.apache.flink.runtime.rest.messages.MessageParameters; +import org.apache.flink.runtime.rest.messages.MessagePathParameter; + +import java.util.Collection; +import java.util.Collections; + +/** Base parameters for batch aggregated subtask metrics. */ +public abstract class AggregatedSubtaskMetricsBatchParameters extends MessageParameters { + + private final JobIDPathParameter jobId = new JobIDPathParameter(); + + @Override + public Collection> getPathParameters() { + return Collections.singleton(jobId); + } +} diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/job/metrics/AggregatedSubtaskMetricsBatchResponseBody.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/job/metrics/AggregatedSubtaskMetricsBatchResponseBody.java new file mode 100644 index 0000000000000..5f2410a16538c --- /dev/null +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/job/metrics/AggregatedSubtaskMetricsBatchResponseBody.java @@ -0,0 +1,175 @@ +/* + * 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.flink.runtime.rest.messages.job.metrics; + +import org.apache.flink.runtime.jobgraph.JobVertexID; +import org.apache.flink.runtime.rest.messages.ResponseBody; +import org.apache.flink.runtime.rest.messages.json.JobVertexIDDeserializer; +import org.apache.flink.runtime.rest.messages.json.JobVertexIDSerializer; +import org.apache.flink.util.Preconditions; + +import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonCreator; +import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonIgnore; +import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonProperty; +import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.core.JsonGenerator; +import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.core.JsonParser; +import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.core.type.TypeReference; +import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.DeserializationContext; +import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.SerializerProvider; +import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.annotation.JsonDeserialize; +import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.annotation.JsonSerialize; +import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.deser.std.StdDeserializer; +import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ser.std.StdSerializer; + +import java.io.IOException; +import java.util.ArrayList; +import java.util.Collection; +import java.util.List; +import java.util.Objects; + +/** Response body for batch aggregated subtask metrics grouped by job vertex. */ +@JsonSerialize(using = AggregatedSubtaskMetricsBatchResponseBody.Serializer.class) +@JsonDeserialize(using = AggregatedSubtaskMetricsBatchResponseBody.Deserializer.class) +public class AggregatedSubtaskMetricsBatchResponseBody implements ResponseBody { + + private final Collection metricsByVertex; + + public AggregatedSubtaskMetricsBatchResponseBody( + Collection metricsByVertex) { + this.metricsByVertex = + new ArrayList<>( + Preconditions.checkNotNull( + metricsByVertex, "metricsByVertex must not be null")); + } + + @JsonIgnore + public Collection getMetricsByVertex() { + return metricsByVertex; + } + + @Override + public boolean equals(Object o) { + if (this == o) { + return true; + } + if (o == null || getClass() != o.getClass()) { + return false; + } + AggregatedSubtaskMetricsBatchResponseBody that = + (AggregatedSubtaskMetricsBatchResponseBody) o; + return Objects.equals(metricsByVertex, that.metricsByVertex); + } + + @Override + public int hashCode() { + return Objects.hash(metricsByVertex); + } + + /** JSON serializer for {@link AggregatedSubtaskMetricsBatchResponseBody}. */ + public static class Serializer + extends StdSerializer { + + private static final long serialVersionUID = 1L; + + protected Serializer() { + super(AggregatedSubtaskMetricsBatchResponseBody.class); + } + + @Override + public void serialize( + AggregatedSubtaskMetricsBatchResponseBody response, + JsonGenerator jsonGenerator, + SerializerProvider serializerProvider) + throws IOException { + jsonGenerator.writeObject(response.getMetricsByVertex()); + } + } + + /** JSON deserializer for {@link AggregatedSubtaskMetricsBatchResponseBody}. */ + public static class Deserializer + extends StdDeserializer { + + private static final long serialVersionUID = 1L; + + protected Deserializer() { + super(AggregatedSubtaskMetricsBatchResponseBody.class); + } + + @Override + public AggregatedSubtaskMetricsBatchResponseBody deserialize( + JsonParser jsonParser, DeserializationContext deserializationContext) + throws IOException { + return new AggregatedSubtaskMetricsBatchResponseBody( + jsonParser.readValueAs(new TypeReference>() {})); + } + } + + /** Aggregated metrics for one job vertex. */ + public static class VertexAggregatedMetrics { + + private static final String FIELD_NAME_VERTEX_ID = "vertexId"; + private static final String FIELD_NAME_METRICS = "metrics"; + + @JsonProperty(value = FIELD_NAME_VERTEX_ID, required = true) + @JsonSerialize(using = JobVertexIDSerializer.class) + @JsonDeserialize(using = JobVertexIDDeserializer.class) + private final JobVertexID vertexId; + + @JsonProperty(value = FIELD_NAME_METRICS, required = true) + private final Collection metrics; + + @JsonCreator + public VertexAggregatedMetrics( + @JsonProperty(value = FIELD_NAME_VERTEX_ID, required = true) JobVertexID vertexId, + @JsonProperty(value = FIELD_NAME_METRICS, required = true) + Collection metrics) { + this.vertexId = Preconditions.checkNotNull(vertexId, "vertexId must not be null"); + this.metrics = + new ArrayList<>( + Preconditions.checkNotNull(metrics, "metrics must not be null")); + } + + @JsonIgnore + public JobVertexID getVertexId() { + return vertexId; + } + + @JsonIgnore + public Collection getMetrics() { + return metrics; + } + + @Override + public boolean equals(Object o) { + if (this == o) { + return true; + } + if (o == null || getClass() != o.getClass()) { + return false; + } + VertexAggregatedMetrics that = (VertexAggregatedMetrics) o; + return Objects.equals(vertexId, that.vertexId) && Objects.equals(metrics, that.metrics); + } + + @Override + public int hashCode() { + return Objects.hash(vertexId, metrics); + } + } +} diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/job/metrics/AggregatedSubtaskMetricsNamesHeaders.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/job/metrics/AggregatedSubtaskMetricsNamesHeaders.java new file mode 100644 index 0000000000000..ee61b878c9e77 --- /dev/null +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/job/metrics/AggregatedSubtaskMetricsNamesHeaders.java @@ -0,0 +1,78 @@ +/* + * 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.flink.runtime.rest.messages.job.metrics; + +import org.apache.flink.runtime.rest.HttpMethodWrapper; +import org.apache.flink.runtime.rest.messages.EmptyRequestBody; +import org.apache.flink.runtime.rest.messages.JobIDPathParameter; +import org.apache.flink.runtime.rest.messages.RuntimeMessageHeaders; + +import org.apache.flink.shaded.netty4.io.netty.handler.codec.http.HttpResponseStatus; + +/** Headers for batch aggregated subtask metric name discovery. */ +public class AggregatedSubtaskMetricsNamesHeaders + implements RuntimeMessageHeaders< + EmptyRequestBody, + AggregatedSubtaskMetricsBatchResponseBody, + AggregatedSubtaskMetricsNamesParameters> { + + private static final AggregatedSubtaskMetricsNamesHeaders INSTANCE = + new AggregatedSubtaskMetricsNamesHeaders(); + + private AggregatedSubtaskMetricsNamesHeaders() {} + + @Override + public Class getRequestClass() { + return EmptyRequestBody.class; + } + + @Override + public Class getResponseClass() { + return AggregatedSubtaskMetricsBatchResponseBody.class; + } + + @Override + public HttpResponseStatus getResponseStatusCode() { + return HttpResponseStatus.OK; + } + + @Override + public AggregatedSubtaskMetricsNamesParameters getUnresolvedMessageParameters() { + return new AggregatedSubtaskMetricsNamesParameters(); + } + + @Override + public HttpMethodWrapper getHttpMethod() { + return HttpMethodWrapper.GET; + } + + @Override + public String getTargetRestEndpointURL() { + return "/jobs/:" + JobIDPathParameter.KEY + "/vertices/subtasks/metrics/names"; + } + + public static AggregatedSubtaskMetricsNamesHeaders getInstance() { + return INSTANCE; + } + + @Override + public String getDescription() { + return "Provides batch access to aggregated subtask metric names."; + } +} diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/job/metrics/AggregatedSubtaskMetricsNamesParameters.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/job/metrics/AggregatedSubtaskMetricsNamesParameters.java new file mode 100644 index 0000000000000..d970a6e62edc9 --- /dev/null +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/job/metrics/AggregatedSubtaskMetricsNamesParameters.java @@ -0,0 +1,38 @@ +/* + * 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.flink.runtime.rest.messages.job.metrics; + +import org.apache.flink.runtime.rest.messages.MessageQueryParameter; + +import java.util.Arrays; +import java.util.Collection; +import java.util.Collections; + +/** Parameters for batch aggregated subtask metric name discovery. */ +public class AggregatedSubtaskMetricsNamesParameters + extends AggregatedSubtaskMetricsBatchParameters { + + private final JobVerticesFilterQueryParameter vertices = new JobVerticesFilterQueryParameter(); + private final MetricsRegexFilterParameter regex = new MetricsRegexFilterParameter(); + + @Override + public Collection> getQueryParameters() { + return Collections.unmodifiableCollection(Arrays.asList(vertices, regex)); + } +} diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/job/metrics/AggregatedSubtaskMetricsValuesHeaders.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/job/metrics/AggregatedSubtaskMetricsValuesHeaders.java new file mode 100644 index 0000000000000..1453a1162d7c0 --- /dev/null +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/job/metrics/AggregatedSubtaskMetricsValuesHeaders.java @@ -0,0 +1,78 @@ +/* + * 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.flink.runtime.rest.messages.job.metrics; + +import org.apache.flink.runtime.rest.HttpMethodWrapper; +import org.apache.flink.runtime.rest.messages.EmptyRequestBody; +import org.apache.flink.runtime.rest.messages.JobIDPathParameter; +import org.apache.flink.runtime.rest.messages.RuntimeMessageHeaders; + +import org.apache.flink.shaded.netty4.io.netty.handler.codec.http.HttpResponseStatus; + +/** Headers for batch aggregated subtask metric values. */ +public class AggregatedSubtaskMetricsValuesHeaders + implements RuntimeMessageHeaders< + EmptyRequestBody, + AggregatedSubtaskMetricsBatchResponseBody, + AggregatedSubtaskMetricsValuesParameters> { + + private static final AggregatedSubtaskMetricsValuesHeaders INSTANCE = + new AggregatedSubtaskMetricsValuesHeaders(); + + private AggregatedSubtaskMetricsValuesHeaders() {} + + @Override + public Class getRequestClass() { + return EmptyRequestBody.class; + } + + @Override + public Class getResponseClass() { + return AggregatedSubtaskMetricsBatchResponseBody.class; + } + + @Override + public HttpResponseStatus getResponseStatusCode() { + return HttpResponseStatus.OK; + } + + @Override + public AggregatedSubtaskMetricsValuesParameters getUnresolvedMessageParameters() { + return new AggregatedSubtaskMetricsValuesParameters(); + } + + @Override + public HttpMethodWrapper getHttpMethod() { + return HttpMethodWrapper.GET; + } + + @Override + public String getTargetRestEndpointURL() { + return "/jobs/:" + JobIDPathParameter.KEY + "/vertices/subtasks/metrics/values"; + } + + public static AggregatedSubtaskMetricsValuesHeaders getInstance() { + return INSTANCE; + } + + @Override + public String getDescription() { + return "Provides batch access to aggregated subtask metric values."; + } +} diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/job/metrics/AggregatedSubtaskMetricsValuesParameters.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/job/metrics/AggregatedSubtaskMetricsValuesParameters.java new file mode 100644 index 0000000000000..dfcfedbfd3619 --- /dev/null +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/job/metrics/AggregatedSubtaskMetricsValuesParameters.java @@ -0,0 +1,39 @@ +/* + * 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.flink.runtime.rest.messages.job.metrics; + +import org.apache.flink.runtime.rest.messages.MessageQueryParameter; + +import java.util.Arrays; +import java.util.Collection; +import java.util.Collections; + +/** Parameters for batch aggregated subtask metric values. */ +public class AggregatedSubtaskMetricsValuesParameters + extends AggregatedSubtaskMetricsBatchParameters { + + private final JobVerticesFilterQueryParameter vertices = new JobVerticesFilterQueryParameter(); + private final MetricsFilterParameter metrics = new MetricsFilterParameter(); + private final MetricsAggregationParameter aggs = new MetricsAggregationParameter(); + + @Override + public Collection> getQueryParameters() { + return Collections.unmodifiableCollection(Arrays.asList(vertices, metrics, aggs)); + } +} diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/job/metrics/JobVerticesFilterQueryParameter.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/job/metrics/JobVerticesFilterQueryParameter.java new file mode 100644 index 0000000000000..1547a2de56629 --- /dev/null +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/job/metrics/JobVerticesFilterQueryParameter.java @@ -0,0 +1,50 @@ +/* + * 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.flink.runtime.rest.messages.job.metrics; + +import org.apache.flink.runtime.jobgraph.JobVertexID; +import org.apache.flink.runtime.rest.messages.ConversionException; +import org.apache.flink.runtime.rest.messages.MessageQueryParameter; + +/** {@link MessageQueryParameter} for selecting job vertices when aggregating subtask metrics. */ +public class JobVerticesFilterQueryParameter extends MessageQueryParameter { + + public JobVerticesFilterQueryParameter() { + super("vertices", MessageParameterRequisiteness.MANDATORY); + } + + @Override + public JobVertexID convertStringToValue(String value) throws ConversionException { + try { + return JobVertexID.fromHexString(value); + } catch (IllegalArgumentException iae) { + throw new ConversionException("Not a valid job vertex ID: " + value, iae); + } + } + + @Override + public String convertValueToString(JobVertexID value) { + return value.toString(); + } + + @Override + public String getDescription() { + return "Comma-separated list of 32-character hexadecimal strings to select specific job vertices."; + } +} diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/job/metrics/MetricsRegexFilterParameter.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/job/metrics/MetricsRegexFilterParameter.java new file mode 100644 index 0000000000000..5215e2168c458 --- /dev/null +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/job/metrics/MetricsRegexFilterParameter.java @@ -0,0 +1,44 @@ +/* + * 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.flink.runtime.rest.messages.job.metrics; + +import org.apache.flink.runtime.rest.messages.MessageQueryParameter; + +/** {@link MessageQueryParameter} for filtering metric names by full-match regular expressions. */ +public class MetricsRegexFilterParameter extends MessageQueryParameter { + + public MetricsRegexFilterParameter() { + super("regex", MessageParameterRequisiteness.OPTIONAL); + } + + @Override + public String convertStringToValue(String value) { + return value; + } + + @Override + public String convertValueToString(String value) { + return value; + } + + @Override + public String getDescription() { + return "Comma-separated list of Java regular expressions to filter metric names by full match."; + } +} diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/webmonitor/WebMonitorEndpoint.java b/flink-runtime/src/main/java/org/apache/flink/runtime/webmonitor/WebMonitorEndpoint.java index 16cf06d4dd966..de64c2f931833 100644 --- a/flink-runtime/src/main/java/org/apache/flink/runtime/webmonitor/WebMonitorEndpoint.java +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/webmonitor/WebMonitorEndpoint.java @@ -88,6 +88,7 @@ import org.apache.flink.runtime.rest.handler.job.checkpoints.TaskCheckpointStatisticDetailsHandler; import org.apache.flink.runtime.rest.handler.job.coordination.ClientCoordinationHandler; import org.apache.flink.runtime.rest.handler.job.metrics.AggregatingJobsMetricsHandler; +import org.apache.flink.runtime.rest.handler.job.metrics.AggregatingSubtasksMetricsBatchHandler; import org.apache.flink.runtime.rest.handler.job.metrics.AggregatingSubtasksMetricsHandler; import org.apache.flink.runtime.rest.handler.job.metrics.AggregatingTaskManagersMetricsHandler; import org.apache.flink.runtime.rest.handler.job.metrics.JobManagerMetricsHandler; @@ -598,6 +599,16 @@ protected List> initiali new AggregatingSubtasksMetricsHandler( leaderRetriever, timeout, responseHeaders, executor, metricFetcher); + final AggregatingSubtasksMetricsBatchHandler.NamesHandler + aggregatingSubtasksMetricsNamesHandler = + new AggregatingSubtasksMetricsBatchHandler.NamesHandler( + leaderRetriever, timeout, responseHeaders, executor, metricFetcher); + + final AggregatingSubtasksMetricsBatchHandler.ValuesHandler + aggregatingSubtasksMetricsValuesHandler = + new AggregatingSubtasksMetricsBatchHandler.ValuesHandler( + leaderRetriever, timeout, responseHeaders, executor, metricFetcher); + final JobVertexTaskManagersHandler jobVertexTaskManagersHandler = new JobVertexTaskManagersHandler( leaderRetriever, @@ -892,6 +903,14 @@ protected List> initiali Tuple2.of( aggregatingSubtasksMetricsHandler.getMessageHeaders(), aggregatingSubtasksMetricsHandler)); + handlers.add( + Tuple2.of( + aggregatingSubtasksMetricsNamesHandler.getMessageHeaders(), + aggregatingSubtasksMetricsNamesHandler)); + handlers.add( + Tuple2.of( + aggregatingSubtasksMetricsValuesHandler.getMessageHeaders(), + aggregatingSubtasksMetricsValuesHandler)); handlers.add( Tuple2.of( jobExecutionResultHandler.getMessageHeaders(), jobExecutionResultHandler)); diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/job/metrics/AggregatingSubtasksMetricsBatchHandlerTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/job/metrics/AggregatingSubtasksMetricsBatchHandlerTest.java new file mode 100644 index 0000000000000..029c32a587c35 --- /dev/null +++ b/flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/job/metrics/AggregatingSubtasksMetricsBatchHandlerTest.java @@ -0,0 +1,288 @@ +/* + * 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.flink.runtime.rest.handler.job.metrics; + +import org.apache.flink.api.common.JobID; +import org.apache.flink.configuration.MetricOptions; +import org.apache.flink.runtime.dispatcher.DispatcherGateway; +import org.apache.flink.runtime.jobgraph.JobVertexID; +import org.apache.flink.runtime.metrics.dump.MetricDump; +import org.apache.flink.runtime.metrics.dump.QueryScopeInfo; +import org.apache.flink.runtime.rest.handler.HandlerRequest; +import org.apache.flink.runtime.rest.handler.HandlerRequestException; +import org.apache.flink.runtime.rest.handler.RestHandlerException; +import org.apache.flink.runtime.rest.handler.legacy.metrics.MetricFetcher; +import org.apache.flink.runtime.rest.handler.legacy.metrics.MetricFetcherImpl; +import org.apache.flink.runtime.rest.messages.EmptyRequestBody; +import org.apache.flink.runtime.rest.messages.JobIDPathParameter; +import org.apache.flink.runtime.rest.messages.job.metrics.AggregatedMetric; +import org.apache.flink.runtime.rest.messages.job.metrics.AggregatedSubtaskMetricsBatchResponseBody; +import org.apache.flink.runtime.rest.messages.job.metrics.AggregatedSubtaskMetricsNamesHeaders; +import org.apache.flink.runtime.rest.messages.job.metrics.AggregatedSubtaskMetricsValuesHeaders; +import org.apache.flink.runtime.webmonitor.TestingDispatcherGateway; +import org.apache.flink.runtime.webmonitor.retriever.GatewayRetriever; +import org.apache.flink.testutils.TestingUtils; +import org.apache.flink.util.concurrent.Executors; + +import org.apache.flink.shaded.netty4.io.netty.handler.codec.http.HttpResponseStatus; + +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +import java.time.Duration; +import java.util.Collection; +import java.util.Collections; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ExecutionException; +import java.util.stream.Collectors; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; +import static org.assertj.core.api.AssertionsForClassTypes.catchThrowableOfType; +import static org.assertj.core.api.AssertionsForClassTypes.within; + +/** Tests for {@link AggregatingSubtasksMetricsBatchHandler}. */ +class AggregatingSubtasksMetricsBatchHandlerTest { + + private static final JobID JOB_ID = JobID.generate(); + private static final JobVertexID VERTEX_ID_1 = new JobVertexID(); + private static final JobVertexID VERTEX_ID_2 = new JobVertexID(); + private static final Duration TIMEOUT = Duration.ofMillis(50); + private static final Map TEST_HEADERS = Collections.emptyMap(); + + private static final DispatcherGateway MOCK_DISPATCHER_GATEWAY = new TestingDispatcherGateway(); + + private static final GatewayRetriever LEADER_RETRIEVER = + new GatewayRetriever() { + @Override + public CompletableFuture getFuture() { + return CompletableFuture.completedFuture(MOCK_DISPATCHER_GATEWAY); + } + }; + + private MetricFetcher fetcher; + private Map pathParameters; + private Map> queryParameters; + + @BeforeEach + void setUp() { + fetcher = + new MetricFetcherImpl<>( + () -> null, + rpcServiceAddress -> null, + Executors.directExecutor(), + TestingUtils.TIMEOUT, + MetricOptions.METRIC_FETCHER_UPDATE_INTERVAL.defaultValue().toMillis()); + + fetcher.getMetricStore().add(counter(VERTEX_ID_1, 0, "busyTimeMsPerSecond", 1)); + fetcher.getMetricStore().add(counter(VERTEX_ID_1, 1, "busyTimeMsPerSecond", 3)); + fetcher.getMetricStore().add(counter(VERTEX_ID_1, 0, "numRecordsInPerSecond", 5)); + fetcher.getMetricStore().add(counter(VERTEX_ID_2, 0, "busyTimeMsPerSecond", 2)); + fetcher.getMetricStore().add(counter(VERTEX_ID_2, 1, "numRecordsInPerSecond", 6)); + + pathParameters = new HashMap<>(); + pathParameters.put(JobIDPathParameter.KEY, JOB_ID.toString()); + queryParameters = new HashMap<>(); + } + + @Test + void testListMetricNamesForMultipleVertices() throws Exception { + final AggregatingSubtasksMetricsBatchHandler.NamesHandler handler = + new AggregatingSubtasksMetricsBatchHandler.NamesHandler( + LEADER_RETRIEVER, + TIMEOUT, + TEST_HEADERS, + Executors.directExecutor(), + fetcher); + + queryParameters.put("vertices", Collections.singletonList(VERTEX_ID_1 + "," + VERTEX_ID_2)); + queryParameters.put("regex", Collections.singletonList(".*busyTime.*")); + + final HandlerRequest request = + HandlerRequest.resolveParametersAndCreate( + EmptyRequestBody.getInstance(), + AggregatedSubtaskMetricsNamesHeaders.getInstance() + .getUnresolvedMessageParameters(), + pathParameters, + queryParameters, + Collections.emptyList()); + + final AggregatedSubtaskMetricsBatchResponseBody response = + handler.handleRequest(request, MOCK_DISPATCHER_GATEWAY).get(); + + assertThat(getMetricIds(response, VERTEX_ID_1)).containsExactly("abc.busyTimeMsPerSecond"); + assertThat(getMetricIds(response, VERTEX_ID_2)).containsExactly("abc.busyTimeMsPerSecond"); + } + + @Test + void testInvalidMetricNameRegexFailsWithBadRequest() throws Exception { + final AggregatingSubtasksMetricsBatchHandler.NamesHandler handler = + new AggregatingSubtasksMetricsBatchHandler.NamesHandler( + LEADER_RETRIEVER, + TIMEOUT, + TEST_HEADERS, + Executors.directExecutor(), + fetcher); + + queryParameters.put("vertices", Collections.singletonList(VERTEX_ID_1.toString())); + queryParameters.put("regex", Collections.singletonList("[")); + + final HandlerRequest request = + HandlerRequest.resolveParametersAndCreate( + EmptyRequestBody.getInstance(), + AggregatedSubtaskMetricsNamesHeaders.getInstance() + .getUnresolvedMessageParameters(), + pathParameters, + queryParameters, + Collections.emptyList()); + + final ExecutionException exception = + catchThrowableOfType( + () -> handler.handleRequest(request, MOCK_DISPATCHER_GATEWAY).get(), + ExecutionException.class); + + assertBadRequest(exception); + } + + @Test + void testMissingVerticesFailsWithBadRequest() throws Exception { + final AggregatingSubtasksMetricsBatchHandler.NamesHandler handler = + new AggregatingSubtasksMetricsBatchHandler.NamesHandler( + LEADER_RETRIEVER, + TIMEOUT, + TEST_HEADERS, + Executors.directExecutor(), + fetcher); + + final HandlerRequest request = + HandlerRequest.resolveParametersAndCreate( + EmptyRequestBody.getInstance(), + AggregatedSubtaskMetricsNamesHeaders.getInstance() + .getUnresolvedMessageParameters(), + pathParameters, + queryParameters, + Collections.emptyList()); + + final ExecutionException exception = + catchThrowableOfType( + () -> handler.handleRequest(request, MOCK_DISPATCHER_GATEWAY).get(), + ExecutionException.class); + + assertBadRequest(exception); + } + + @Test + void testAggregateMetricValuesForMultipleVertices() throws Exception { + final AggregatingSubtasksMetricsBatchHandler.ValuesHandler handler = + new AggregatingSubtasksMetricsBatchHandler.ValuesHandler( + LEADER_RETRIEVER, + TIMEOUT, + TEST_HEADERS, + Executors.directExecutor(), + fetcher); + + queryParameters.put("vertices", Collections.singletonList(VERTEX_ID_1 + "," + VERTEX_ID_2)); + queryParameters.put("get", Collections.singletonList("abc.busyTimeMsPerSecond")); + queryParameters.put("agg", Collections.singletonList("min,max,avg")); + + final HandlerRequest request = + HandlerRequest.resolveParametersAndCreate( + EmptyRequestBody.getInstance(), + AggregatedSubtaskMetricsValuesHeaders.getInstance() + .getUnresolvedMessageParameters(), + pathParameters, + queryParameters, + Collections.emptyList()); + + final AggregatedSubtaskMetricsBatchResponseBody response = + handler.handleRequest(request, MOCK_DISPATCHER_GATEWAY).get(); + + final AggregatedMetric vertex1Metric = getOnlyMetric(response, VERTEX_ID_1); + assertThat(vertex1Metric.getId()).isEqualTo("abc.busyTimeMsPerSecond"); + assertThat(vertex1Metric.getMin()).isCloseTo(1.0, within(0.1)); + assertThat(vertex1Metric.getMax()).isCloseTo(3.0, within(0.1)); + assertThat(vertex1Metric.getAvg()).isCloseTo(2.0, within(0.1)); + assertThat(vertex1Metric.getSum()).isNull(); + + final AggregatedMetric vertex2Metric = getOnlyMetric(response, VERTEX_ID_2); + assertThat(vertex2Metric.getId()).isEqualTo("abc.busyTimeMsPerSecond"); + assertThat(vertex2Metric.getMin()).isCloseTo(2.0, within(0.1)); + assertThat(vertex2Metric.getMax()).isCloseTo(2.0, within(0.1)); + assertThat(vertex2Metric.getAvg()).isCloseTo(2.0, within(0.1)); + assertThat(vertex2Metric.getSum()).isNull(); + } + + @Test + void testInvalidAggregationFailsDuringRequestParameterResolution() { + queryParameters.put("vertices", Collections.singletonList(VERTEX_ID_1.toString())); + queryParameters.put("get", Collections.singletonList("abc.busyTimeMsPerSecond")); + queryParameters.put("agg", Collections.singletonList("median")); + + assertThatThrownBy( + () -> + HandlerRequest.resolveParametersAndCreate( + EmptyRequestBody.getInstance(), + AggregatedSubtaskMetricsValuesHeaders.getInstance() + .getUnresolvedMessageParameters(), + pathParameters, + queryParameters, + Collections.emptyList())) + .isInstanceOf(HandlerRequestException.class); + } + + private static MetricDump.CounterDump counter( + JobVertexID vertexId, int subtaskIndex, String name, long value) { + return new MetricDump.CounterDump( + new QueryScopeInfo.TaskQueryScopeInfo( + JOB_ID.toString(), vertexId.toString(), subtaskIndex, 0, "abc"), + name, + value); + } + + private static List getMetricIds( + AggregatedSubtaskMetricsBatchResponseBody response, JobVertexID vertexId) { + return getMetrics(response, vertexId).stream() + .map(AggregatedMetric::getId) + .collect(Collectors.toList()); + } + + private static AggregatedMetric getOnlyMetric( + AggregatedSubtaskMetricsBatchResponseBody response, JobVertexID vertexId) { + return getMetrics(response, vertexId).iterator().next(); + } + + private static Collection getMetrics( + AggregatedSubtaskMetricsBatchResponseBody response, JobVertexID vertexId) { + return response.getMetricsByVertex().stream() + .filter(vertexMetrics -> vertexMetrics.getVertexId().equals(vertexId)) + .findFirst() + .orElseThrow(AssertionError::new) + .getMetrics(); + } + + private static void assertBadRequest(ExecutionException exception) { + assertThat(exception).isNotNull(); + assertThat(exception.getCause()).isInstanceOf(RestHandlerException.class); + assertThat(((RestHandlerException) exception.getCause()).getHttpResponseStatus()) + .isEqualTo(HttpResponseStatus.BAD_REQUEST); + } +} diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/rest/messages/job/metrics/AggregatedSubtaskMetricsBatchHeadersTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/rest/messages/job/metrics/AggregatedSubtaskMetricsBatchHeadersTest.java new file mode 100644 index 0000000000000..a5503ef81abb9 --- /dev/null +++ b/flink-runtime/src/test/java/org/apache/flink/runtime/rest/messages/job/metrics/AggregatedSubtaskMetricsBatchHeadersTest.java @@ -0,0 +1,58 @@ +/* + * 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.flink.runtime.rest.messages.job.metrics; + +import org.apache.flink.runtime.rest.HttpMethodWrapper; +import org.apache.flink.runtime.rest.messages.EmptyRequestBody; +import org.apache.flink.runtime.rest.messages.JobIDPathParameter; + +import org.junit.jupiter.api.Test; + +import static org.assertj.core.api.Assertions.assertThat; + +/** Tests for batch aggregated subtask metrics headers. */ +class AggregatedSubtaskMetricsBatchHeadersTest { + + @Test + void testMetricNamesHeaders() { + final AggregatedSubtaskMetricsNamesHeaders headers = + AggregatedSubtaskMetricsNamesHeaders.getInstance(); + + assertThat(headers.getHttpMethod()).isEqualTo(HttpMethodWrapper.GET); + assertThat(headers.getTargetRestEndpointURL()) + .isEqualTo("/jobs/:" + JobIDPathParameter.KEY + "/vertices/subtasks/metrics/names"); + assertThat(headers.getRequestClass()).isEqualTo(EmptyRequestBody.class); + assertThat(headers.getResponseClass()) + .isEqualTo(AggregatedSubtaskMetricsBatchResponseBody.class); + } + + @Test + void testMetricValuesHeaders() { + final AggregatedSubtaskMetricsValuesHeaders headers = + AggregatedSubtaskMetricsValuesHeaders.getInstance(); + + assertThat(headers.getHttpMethod()).isEqualTo(HttpMethodWrapper.GET); + assertThat(headers.getTargetRestEndpointURL()) + .isEqualTo( + "/jobs/:" + JobIDPathParameter.KEY + "/vertices/subtasks/metrics/values"); + assertThat(headers.getRequestClass()).isEqualTo(EmptyRequestBody.class); + assertThat(headers.getResponseClass()) + .isEqualTo(AggregatedSubtaskMetricsBatchResponseBody.class); + } +} diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/rest/messages/job/metrics/AggregatedSubtaskMetricsBatchParametersTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/rest/messages/job/metrics/AggregatedSubtaskMetricsBatchParametersTest.java new file mode 100644 index 0000000000000..5f52a866c8537 --- /dev/null +++ b/flink-runtime/src/test/java/org/apache/flink/runtime/rest/messages/job/metrics/AggregatedSubtaskMetricsBatchParametersTest.java @@ -0,0 +1,98 @@ +/* + * 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.flink.runtime.rest.messages.job.metrics; + +import org.apache.flink.runtime.jobgraph.JobVertexID; +import org.apache.flink.runtime.rest.handler.HandlerRequest; +import org.apache.flink.runtime.rest.messages.EmptyRequestBody; +import org.apache.flink.runtime.rest.messages.JobIDPathParameter; + +import org.junit.jupiter.api.Test; + +import java.util.Collections; +import java.util.HashMap; +import java.util.List; +import java.util.Map; + +import static org.assertj.core.api.Assertions.assertThat; + +/** Tests for batch aggregated subtask metrics query parameters. */ +class AggregatedSubtaskMetricsBatchParametersTest { + + @Test + void testMetricNamesQueryParameters() throws Exception { + final JobVertexID vertexId1 = new JobVertexID(); + final JobVertexID vertexId2 = new JobVertexID(); + final HandlerRequest request = + HandlerRequest.resolveParametersAndCreate( + EmptyRequestBody.getInstance(), + AggregatedSubtaskMetricsNamesHeaders.getInstance() + .getUnresolvedMessageParameters(), + pathParameters(), + queryParameters( + "vertices", vertexId1 + "," + vertexId2, "regex", ".*busyTime.*"), + Collections.emptyList()); + + assertThat(request.getQueryParameter(JobVerticesFilterQueryParameter.class)) + .containsExactly(vertexId1, vertexId2); + assertThat(request.getQueryParameter(MetricsRegexFilterParameter.class)) + .containsExactly(".*busyTime.*"); + } + + @Test + void testMetricValuesQueryParameters() throws Exception { + final JobVertexID vertexId1 = new JobVertexID(); + final JobVertexID vertexId2 = new JobVertexID(); + final HandlerRequest request = + HandlerRequest.resolveParametersAndCreate( + EmptyRequestBody.getInstance(), + AggregatedSubtaskMetricsValuesHeaders.getInstance() + .getUnresolvedMessageParameters(), + pathParameters(), + queryParameters( + "vertices", + vertexId1 + "," + vertexId2, + "get", + "abc.busyTimeMsPerSecond,abc.numRecordsInPerSecond", + "agg", + "min,max"), + Collections.emptyList()); + + assertThat(request.getQueryParameter(JobVerticesFilterQueryParameter.class)) + .containsExactly(vertexId1, vertexId2); + assertThat(request.getQueryParameter(MetricsFilterParameter.class)) + .containsExactly("abc.busyTimeMsPerSecond", "abc.numRecordsInPerSecond"); + assertThat(request.getQueryParameter(MetricsAggregationParameter.class)) + .containsExactly( + MetricsAggregationParameter.AggregationMode.MIN, + MetricsAggregationParameter.AggregationMode.MAX); + } + + private static Map pathParameters() { + return Collections.singletonMap(JobIDPathParameter.KEY, "00000000000000000000000000000000"); + } + + private static Map> queryParameters(String... keyValues) { + final Map> queryParameters = new HashMap<>(); + for (int index = 0; index < keyValues.length; index += 2) { + queryParameters.put(keyValues[index], Collections.singletonList(keyValues[index + 1])); + } + return queryParameters; + } +} diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/rest/messages/job/metrics/AggregatedSubtaskMetricsBatchResponseBodyTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/rest/messages/job/metrics/AggregatedSubtaskMetricsBatchResponseBodyTest.java new file mode 100644 index 0000000000000..49b94722c58e5 --- /dev/null +++ b/flink-runtime/src/test/java/org/apache/flink/runtime/rest/messages/job/metrics/AggregatedSubtaskMetricsBatchResponseBodyTest.java @@ -0,0 +1,94 @@ +/* + * 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.flink.runtime.rest.messages.job.metrics; + +import org.apache.flink.runtime.jobgraph.JobVertexID; +import org.apache.flink.runtime.rest.messages.RestResponseMarshallingTestBase; +import org.apache.flink.runtime.rest.util.RestMapperUtils; +import org.apache.flink.testutils.junit.extensions.parameterized.NoOpTestExtension; + +import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.JsonNode; +import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ObjectMapper; + +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; + +import java.util.Collections; + +import static org.assertj.core.api.Assertions.assertThat; + +/** Tests for {@link AggregatedSubtaskMetricsBatchResponseBody}. */ +@ExtendWith(NoOpTestExtension.class) +class AggregatedSubtaskMetricsBatchResponseBodyTest + extends RestResponseMarshallingTestBase { + + private static final JobVertexID VERTEX_ID = new JobVertexID(); + private static final String METRIC_ID = "abc.busyTimeMsPerSecond"; + + @Override + protected Class getTestResponseClass() { + return AggregatedSubtaskMetricsBatchResponseBody.class; + } + + @Override + protected AggregatedSubtaskMetricsBatchResponseBody getTestResponseInstance() { + return createResponse(); + } + + @Override + protected void assertOriginalEqualsToUnmarshalled( + AggregatedSubtaskMetricsBatchResponseBody expected, + AggregatedSubtaskMetricsBatchResponseBody actual) { + assertThat(actual.getMetricsByVertex()).hasSize(1); + + final AggregatedSubtaskMetricsBatchResponseBody.VertexAggregatedMetrics vertexMetrics = + actual.getMetricsByVertex().iterator().next(); + assertThat(vertexMetrics.getVertexId()).isEqualTo(VERTEX_ID); + assertThat(vertexMetrics.getMetrics()).hasSize(1); + + final AggregatedMetric metric = vertexMetrics.getMetrics().iterator().next(); + assertThat(metric.getId()).isEqualTo(METRIC_ID); + assertThat(metric.getMin()).isEqualTo(1.0); + assertThat(metric.getMax()).isEqualTo(3.0); + assertThat(metric.getAvg()).isEqualTo(2.0); + assertThat(metric.getSum()).isEqualTo(4.0); + } + + @Test + void testSerializesAsTopLevelArray() throws Exception { + final ObjectMapper objectMapper = RestMapperUtils.getStrictObjectMapper(); + final JsonNode rootNode = + objectMapper.readTree(objectMapper.writeValueAsString(createResponse())); + + assertThat(rootNode.isArray()).isTrue(); + assertThat(rootNode).hasSize(1); + assertThat(rootNode.get(0).has("vertexId")).isTrue(); + assertThat(rootNode.get(0).has("metrics")).isTrue(); + } + + private static AggregatedSubtaskMetricsBatchResponseBody createResponse() { + return new AggregatedSubtaskMetricsBatchResponseBody( + Collections.singletonList( + new AggregatedSubtaskMetricsBatchResponseBody.VertexAggregatedMetrics( + VERTEX_ID, + Collections.singletonList( + new AggregatedMetric( + METRIC_ID, 1.0, 3.0, 2.0, 4.0, null))))); + } +}