Skip to content

[FLINK-39617] [runtime] Add batch aggregated subtask metrics endpoints #29012

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
Jackeyzhe wants to merge 1 commit into
base: master
Choose a base branch
from
Open

[FLINK-39617] [runtime] Add batch aggregated subtask metrics endpoints #29012

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
213 changes: 4 additions & 209 deletions .../org/apache/flink/runtime/rest/handler/job/metrics/AbstractAggregatingMetricsHandler.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -113,64 +107,16 @@ protected CompletableFuture<AggregatedMetricsResponseBody> handleRequest(
getStores(store, request);

if (requestedMetrics.isEmpty()) {
Collection<String> list = getAvailableMetrics(stores);
Collection<String> 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(
Expand All @@ -181,155 +127,4 @@ protected CompletableFuture<AggregatedMetricsResponseBody> 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<String> getAvailableMetrics(
Collection<? extends MetricStore.ComponentMetricStore> stores) {
Set<String> 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<? extends MetricStore.ComponentMetricStore> stores,
List<String> requestedMetrics,
MetricAccumulatorFactory requestedAggregationsFactories) {

Collection<AggregatedMetric> aggregatedMetrics = new ArrayList<>(requestedMetrics.size());
for (String requestedMetric : requestedMetrics) {
final Collection<Double> 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<Double> 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());
}
}
}
Loading