You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by gy...@apache.org on 2023/02/28 11:08:28 UTC
[flink-kubernetes-operator] branch main updated: [FLINK-30920] [AutoScaler] adds exclude vertex ids for autoscaler to ignore for evaluation and scaling
This is an automated email from the ASF dual-hosted git repository.
gyfora pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/flink-kubernetes-operator.git
The following commit(s) were added to refs/heads/main by this push:
new f7e353c2 [FLINK-30920] [AutoScaler] adds exclude vertex ids for autoscaler to ignore for evaluation and scaling
f7e353c2 is described below
commit f7e353c26254c3ccc4a8c2a291da7697f1fbec5d
Author: gauravmiglanid11 <85...@users.noreply.github.com>
AuthorDate: Tue Feb 28 16:38:22 2023 +0530
[FLINK-30920] [AutoScaler] adds exclude vertex ids for autoscaler to ignore for evaluation and scaling
---
.../generated/auto_scaler_configuration.html | 6 ++++
.../operator/autoscaler/ScalingExecutor.java | 32 +++++++++++++++-------
.../autoscaler/config/AutoScalerOptions.java | 9 ++++++
.../operator/autoscaler/ScalingExecutorTest.java | 28 +++++++++++++++++++
4 files changed, 65 insertions(+), 10 deletions(-)
diff --git a/docs/layouts/shortcodes/generated/auto_scaler_configuration.html b/docs/layouts/shortcodes/generated/auto_scaler_configuration.html
index ca468948..85d30e78 100644
--- a/docs/layouts/shortcodes/generated/auto_scaler_configuration.html
+++ b/docs/layouts/shortcodes/generated/auto_scaler_configuration.html
@@ -104,6 +104,12 @@
<td>Double</td>
<td>Target vertex utilization boundary. Scaling won't be performed if utilization is within (target - boundary, target + boundary)</td>
</tr>
+ <tr>
+ <td><h5>kubernetes.operator.job.autoscaler.vertex.exclude.ids</h5></td>
+ <td style="word-wrap: break-word;"></td>
+ <td>List<String></td>
+ <td>A (semicolon-separated) list of vertex ids in hexstring for which to disable scaling. Caution: For non-sink vertices this will still scale their downstream operators until https://issues.apache.org/jira/browse/FLINK-31215 is implemented.</td>
+ </tr>
<tr>
<td><h5>kubernetes.operator.job.autoscaler.vertex.max-parallelism</h5></td>
<td style="word-wrap: break-word;">2147483647</td>
diff --git a/flink-kubernetes-operator-autoscaler/src/main/java/org/apache/flink/kubernetes/operator/autoscaler/ScalingExecutor.java b/flink-kubernetes-operator-autoscaler/src/main/java/org/apache/flink/kubernetes/operator/autoscaler/ScalingExecutor.java
index a1f5b554..d0df57b7 100644
--- a/flink-kubernetes-operator-autoscaler/src/main/java/org/apache/flink/kubernetes/operator/autoscaler/ScalingExecutor.java
+++ b/flink-kubernetes-operator-autoscaler/src/main/java/org/apache/flink/kubernetes/operator/autoscaler/ScalingExecutor.java
@@ -23,6 +23,7 @@ import org.apache.flink.configuration.ConfigOption;
import org.apache.flink.configuration.ConfigOptions;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.kubernetes.operator.api.AbstractFlinkResource;
+import org.apache.flink.kubernetes.operator.autoscaler.config.AutoScalerOptions;
import org.apache.flink.kubernetes.operator.autoscaler.metrics.EvaluatedScalingMetric;
import org.apache.flink.kubernetes.operator.autoscaler.metrics.ScalingMetric;
import org.apache.flink.kubernetes.operator.utils.EventRecorder;
@@ -222,19 +223,30 @@ public class ScalingExecutor {
Map<JobVertexID, SortedMap<Instant, ScalingSummary>> scalingHistory) {
var out = new HashMap<JobVertexID, ScalingSummary>();
+ var excludeVertexIdList = conf.get(AutoScalerOptions.VERTEX_EXCLUDE_IDS);
evaluatedMetrics.forEach(
(v, metrics) -> {
- var currentParallelism =
- (int) metrics.get(ScalingMetric.PARALLELISM).getCurrent();
- var newParallelism =
- jobVertexScaler.computeScaleTargetParallelism(
- resource,
- conf,
+ if (excludeVertexIdList.contains(v.toHexString())) {
+ LOG.info(
+ "Vertex {} is part of `vertex.exclude.ids` config, Ignoring it for scaling",
+ v);
+ } else {
+ var currentParallelism =
+ (int) metrics.get(ScalingMetric.PARALLELISM).getCurrent();
+ var newParallelism =
+ jobVertexScaler.computeScaleTargetParallelism(
+ resource,
+ conf,
+ v,
+ metrics,
+ scalingHistory.getOrDefault(
+ v, Collections.emptySortedMap()));
+ if (currentParallelism != newParallelism) {
+ out.put(
v,
- metrics,
- scalingHistory.getOrDefault(v, Collections.emptySortedMap()));
- if (currentParallelism != newParallelism) {
- out.put(v, new ScalingSummary(currentParallelism, newParallelism, metrics));
+ new ScalingSummary(
+ currentParallelism, newParallelism, metrics));
+ }
}
});
diff --git a/flink-kubernetes-operator-autoscaler/src/main/java/org/apache/flink/kubernetes/operator/autoscaler/config/AutoScalerOptions.java b/flink-kubernetes-operator-autoscaler/src/main/java/org/apache/flink/kubernetes/operator/autoscaler/config/AutoScalerOptions.java
index c457b462..0d5e5435 100644
--- a/flink-kubernetes-operator-autoscaler/src/main/java/org/apache/flink/kubernetes/operator/autoscaler/config/AutoScalerOptions.java
+++ b/flink-kubernetes-operator-autoscaler/src/main/java/org/apache/flink/kubernetes/operator/autoscaler/config/AutoScalerOptions.java
@@ -22,6 +22,7 @@ import org.apache.flink.configuration.ConfigOptions;
import org.apache.flink.kubernetes.operator.autoscaler.metrics.MetricAggregator;
import java.time.Duration;
+import java.util.List;
import static org.apache.flink.kubernetes.operator.config.KubernetesOperatorConfigOptions.operatorConfig;
@@ -153,4 +154,12 @@ public class AutoScalerOptions {
.defaultValue(MetricAggregator.MAX)
.withDescription(
"Metric aggregator to use for busyTime metrics. This affects how true processing/output rate will be computed. Using max allows us to handle jobs with data skew more robustly, while avg may provide better stability when we know that the load distribution is even.");
+
+ public static final ConfigOption<List<String>> VERTEX_EXCLUDE_IDS =
+ autoScalerConfig("vertex.exclude.ids")
+ .stringType()
+ .asList()
+ .defaultValues(new String[0])
+ .withDescription(
+ "A (semicolon-separated) list of vertex ids in hexstring for which to disable scaling. Caution: For non-sink vertices this will still scale their downstream operators until https://issues.apache.org/jira/browse/FLINK-31215 is implemented.");
}
diff --git a/flink-kubernetes-operator-autoscaler/src/test/java/org/apache/flink/kubernetes/operator/autoscaler/ScalingExecutorTest.java b/flink-kubernetes-operator-autoscaler/src/test/java/org/apache/flink/kubernetes/operator/autoscaler/ScalingExecutorTest.java
index 447c262d..0bc1ecb7 100644
--- a/flink-kubernetes-operator-autoscaler/src/test/java/org/apache/flink/kubernetes/operator/autoscaler/ScalingExecutorTest.java
+++ b/flink-kubernetes-operator-autoscaler/src/test/java/org/apache/flink/kubernetes/operator/autoscaler/ScalingExecutorTest.java
@@ -41,6 +41,7 @@ import java.time.Duration;
import java.time.Instant;
import java.time.ZoneId;
import java.util.HashMap;
+import java.util.List;
import java.util.Map;
import static org.apache.flink.kubernetes.operator.autoscaler.ScalingExecutor.PARALLELISM_OVERRIDES;
@@ -183,6 +184,33 @@ public class ScalingExecutorTest {
assertFalse(ScalingExecutor.allVerticesWithinUtilizationTarget(evaluated, scalingSummary));
}
+ @Test
+ public void testVertexesExclusionForScaling() {
+ var sourceHexString = "0bfd135746ac8efb3cce668b12e16d3a";
+ var source = JobVertexID.fromHexString(sourceHexString);
+ var filterOperatorHexString = "869fb403873411306404e9f2e4438c0e";
+ var filterOperator = JobVertexID.fromHexString(filterOperatorHexString);
+ var sinkHexString = "a6b7102b8d3e3a9564998c1ffeb5e2b7";
+ var sink = JobVertexID.fromHexString(sinkHexString);
+
+ var scalingInfo = new AutoScalerInfo(new HashMap<>());
+ conf.set(AutoScalerOptions.TARGET_UTILIZATION, .8);
+ var metrics =
+ Map.of(
+ source,
+ evaluated(10, 80, 100),
+ filterOperator,
+ evaluated(10, 60, 100),
+ sink,
+ evaluated(10, 80, 100));
+ // filter operator should not scale
+ conf.set(AutoScalerOptions.VERTEX_EXCLUDE_IDS, List.of(filterOperatorHexString));
+ assertFalse(scalingDecisionExecutor.scaleResource(flinkDep, scalingInfo, conf, metrics));
+ // filter operator should scale
+ conf.set(AutoScalerOptions.VERTEX_EXCLUDE_IDS, List.of());
+ assertTrue(scalingDecisionExecutor.scaleResource(flinkDep, scalingInfo, conf, metrics));
+ }
+
@ParameterizedTest
@ValueSource(booleans = {true, false})
public void testScalingEvents(boolean scalingEnabled) {