You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by mx...@apache.org on 2024/03/11 20:06:11 UTC
(flink-kubernetes-operator) branch main updated: [FLINK-34642] Add a test for zero lower utilization boundary (#796)
This is an automated email from the ASF dual-hosted git repository.
mxm 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 ff7c0df4 [FLINK-34642] Add a test for zero lower utilization boundary (#796)
ff7c0df4 is described below
commit ff7c0df4f1cec369680ecbffeb1743b30fbebb99
Author: Maximilian Michels <mx...@apache.org>
AuthorDate: Mon Mar 11 21:06:05 2024 +0100
[FLINK-34642] Add a test for zero lower utilization boundary (#796)
Users recently reported that scale down as "broken". Turned out that
`target.utilization` and `target.utilization.boundary` were set to the same
value. This prevents scale down entirely, which is expected and desired.
We should add a test to verify this behavior to avoid breaking it in the future.
---
.../generated/auto_scaler_configuration.html | 2 +-
.../apache/flink/autoscaler/ScalingExecutor.java | 9 +-
.../flink/autoscaler/config/AutoScalerOptions.java | 3 +-
.../flink/autoscaler/ScalingExecutorTest.java | 96 +++++++++++++++++-----
4 files changed, 81 insertions(+), 29 deletions(-)
diff --git a/docs/layouts/shortcodes/generated/auto_scaler_configuration.html b/docs/layouts/shortcodes/generated/auto_scaler_configuration.html
index 5dc38861..b3466b63 100644
--- a/docs/layouts/shortcodes/generated/auto_scaler_configuration.html
+++ b/docs/layouts/shortcodes/generated/auto_scaler_configuration.html
@@ -192,7 +192,7 @@
<td><h5>job.autoscaler.target.utilization.boundary</h5></td>
<td style="word-wrap: break-word;">0.3</td>
<td>Double</td>
- <td>Target vertex utilization boundary. Scaling won't be performed if the current processing rate is within [target_rate / (target_utilization - boundary), (target_rate / (target_utilization + boundary)]</td>
+ <td>Target vertex utilization boundary. Scaling won't be performed if the processing capacity is within [target_rate / (target_utilization - boundary), (target_rate / (target_utilization + boundary)]</td>
</tr>
<tr>
<td><h5>job.autoscaler.vertex.exclude.ids</h5></td>
diff --git a/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/ScalingExecutor.java b/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/ScalingExecutor.java
index 7acb482d..71efc160 100644
--- a/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/ScalingExecutor.java
+++ b/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/ScalingExecutor.java
@@ -174,15 +174,16 @@ public class ScalingExecutor<KEY, Context extends JobAutoScalerContext<KEY>> {
var vertex = entry.getKey();
var metrics = evaluatedMetrics.get(vertex);
- double processingRate = metrics.get(TRUE_PROCESSING_RATE).getAverage();
+ double trueProcessingRate = metrics.get(TRUE_PROCESSING_RATE).getAverage();
double scaleUpRateThreshold = metrics.get(SCALE_UP_RATE_THRESHOLD).getCurrent();
double scaleDownRateThreshold = metrics.get(SCALE_DOWN_RATE_THRESHOLD).getCurrent();
- if (processingRate < scaleUpRateThreshold || processingRate > scaleDownRateThreshold) {
+ if (trueProcessingRate < scaleUpRateThreshold
+ || trueProcessingRate > scaleDownRateThreshold) {
LOG.debug(
"Vertex {} processing rate {} is outside ({}, {})",
vertex,
- processingRate,
+ trueProcessingRate,
scaleUpRateThreshold,
scaleDownRateThreshold);
return false;
@@ -190,7 +191,7 @@ public class ScalingExecutor<KEY, Context extends JobAutoScalerContext<KEY>> {
LOG.debug(
"Vertex {} processing rate {} is within target ({}, {})",
vertex,
- processingRate,
+ trueProcessingRate,
scaleUpRateThreshold,
scaleDownRateThreshold);
}
diff --git a/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/config/AutoScalerOptions.java b/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/config/AutoScalerOptions.java
index 080472fd..aa095e1c 100644
--- a/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/config/AutoScalerOptions.java
+++ b/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/config/AutoScalerOptions.java
@@ -100,8 +100,7 @@ public class AutoScalerOptions {
.defaultValue(0.3)
.withFallbackKeys(oldOperatorConfigKey("target.utilization.boundary"))
.withDescription(
- "Target vertex utilization boundary. Scaling won't be performed if the current processing rate is within [target_rate / (target_utilization - boundary), (target_rate / (target_utilization + boundary)]");
-
+ "Target vertex utilization boundary. Scaling won't be performed if the processing capacity is within [target_rate / (target_utilization - boundary), (target_rate / (target_utilization + boundary)]");
public static final ConfigOption<Duration> SCALE_UP_GRACE_PERIOD =
autoScalerConfig("scale-up.grace-period")
.durationType()
diff --git a/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/ScalingExecutorTest.java b/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/ScalingExecutorTest.java
index 7c8ead4e..26a92f3e 100644
--- a/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/ScalingExecutorTest.java
+++ b/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/ScalingExecutorTest.java
@@ -26,6 +26,7 @@ import org.apache.flink.autoscaler.metrics.ScalingMetric;
import org.apache.flink.autoscaler.resources.ResourceCheck;
import org.apache.flink.autoscaler.state.AutoScalerStateStore;
import org.apache.flink.autoscaler.state.InMemoryAutoScalerStateStore;
+import org.apache.flink.autoscaler.topology.IOMetrics;
import org.apache.flink.autoscaler.topology.JobTopology;
import org.apache.flink.autoscaler.topology.VertexInfo;
import org.apache.flink.configuration.Configuration;
@@ -66,12 +67,14 @@ public class ScalingExecutorTest {
private JobAutoScalerContext<JobID> context;
private TestingEventCollector<JobID, JobAutoScalerContext<JobID>> eventCollector;
- private ScalingExecutor<JobID, JobAutoScalerContext<JobID>> scalingDecisionExecutor;
+ private ScalingExecutor<JobID, JobAutoScalerContext<JobID>> scalingExecutor;
private InMemoryAutoScalerStateStore<JobID, JobAutoScalerContext<JobID>> stateStore;
private Configuration conf;
+ private ScalingTracking scalingTracking = new ScalingTracking();
+
private static final Map<ScalingMetric, EvaluatedScalingMetric> dummyGlobalMetrics =
Map.of(
ScalingMetric.GC_PRESSURE, EvaluatedScalingMetric.of(Double.NaN),
@@ -83,7 +86,7 @@ public class ScalingExecutorTest {
context = createDefaultJobAutoScalerContext();
stateStore = new InMemoryAutoScalerStateStore<>();
- scalingDecisionExecutor = new ScalingExecutor<>(eventCollector, stateStore);
+ scalingExecutor = new ScalingExecutor<>(eventCollector, stateStore);
conf = context.getConfiguration();
conf.set(AutoScalerOptions.STABILIZATION_INTERVAL, Duration.ZERO);
conf.set(AutoScalerOptions.SCALING_ENABLED, true);
@@ -146,6 +149,54 @@ public class ScalingExecutorTest {
assertFalse(ScalingExecutor.allVerticesWithinUtilizationTarget(evaluated, scalingSummary));
}
+ @Test
+ public void testNoScaleDownOnZeroLowerUtilizationBoundary() throws Exception {
+ var conf = context.getConfiguration();
+ // Target utilization and boundary are identical
+ // which will set the scale down boundary to infinity
+ conf.set(AutoScalerOptions.TARGET_UTILIZATION, 0.6);
+ conf.set(AutoScalerOptions.TARGET_UTILIZATION_BOUNDARY, 0.6);
+
+ var vertex = new JobVertexID();
+ int parallelism = 100;
+ int expectedParallelism = 1;
+ int targetRate = 1000;
+ // Intentionally also set the true processing rate to infinity
+ // to test the boundaries of the scaling condition.
+ double trueProcessingRate = Double.POSITIVE_INFINITY;
+
+ var evaluated =
+ new EvaluatedMetrics(
+ Map.of(vertex, evaluated(parallelism, targetRate, trueProcessingRate)),
+ dummyGlobalMetrics);
+
+ // Verify precondition
+ var scalingSummary =
+ Map.of(
+ vertex,
+ new ScalingSummary(
+ parallelism,
+ expectedParallelism,
+ evaluated.getVertexMetrics().get(vertex)));
+ assertTrue(
+ ScalingExecutor.allVerticesWithinUtilizationTarget(
+ evaluated.getVertexMetrics(), scalingSummary));
+
+ // Execute the full scaling path
+ var now = Instant.now();
+ var jobTopology =
+ new JobTopology(
+ new VertexInfo(
+ vertex,
+ Map.of(),
+ parallelism,
+ Integer.MAX_VALUE,
+ new IOMetrics(10000, 10000, 100)));
+ assertFalse(
+ scalingExecutor.scaleResource(
+ context, evaluated, new HashMap<>(), scalingTracking, now, jobTopology));
+ }
+
@Test
public void testVertexesExclusionForScaling() throws Exception {
var sourceHexString = "0bfd135746ac8efb3cce668b12e16d3a";
@@ -177,7 +228,7 @@ public class ScalingExecutorTest {
conf.set(AutoScalerOptions.VERTEX_EXCLUDE_IDS, List.of(filterOperatorHexString));
var now = Instant.now();
assertFalse(
- scalingDecisionExecutor.scaleResource(
+ scalingExecutor.scaleResource(
context,
metrics,
new HashMap<>(),
@@ -187,7 +238,7 @@ public class ScalingExecutorTest {
// filter operator should scale
conf.set(AutoScalerOptions.VERTEX_EXCLUDE_IDS, List.of());
assertTrue(
- scalingDecisionExecutor.scaleResource(
+ scalingExecutor.scaleResource(
context,
metrics,
new HashMap<>(),
@@ -223,7 +274,7 @@ public class ScalingExecutorTest {
Map.of(source, evaluated(10, 110, 100), sink, evaluated(10, 110, 100)),
dummyGlobalMetrics);
assertFalse(
- scalingDecisionExecutor.scaleResource(
+ scalingExecutor.scaleResource(
context,
metrics,
new HashMap<>(),
@@ -238,7 +289,7 @@ public class ScalingExecutorTest {
.toString();
conf.set(AutoScalerOptions.EXCLUDED_PERIODS, List.of(excludedPeriod));
assertTrue(
- scalingDecisionExecutor.scaleResource(
+ scalingExecutor.scaleResource(
context,
metrics,
new HashMap<>(),
@@ -273,7 +324,7 @@ public class ScalingExecutorTest {
// Would normally scale without resource usage check
assertTrue(
- scalingDecisionExecutor.scaleResource(
+ scalingExecutor.scaleResource(
context,
metrics,
new HashMap<>(),
@@ -281,7 +332,7 @@ public class ScalingExecutorTest {
now,
jobTopology));
- scalingDecisionExecutor =
+ scalingExecutor =
new ScalingExecutor<>(
eventCollector,
stateStore,
@@ -298,7 +349,7 @@ public class ScalingExecutorTest {
// Scaling blocked due to unavailable resources
assertFalse(
- scalingDecisionExecutor.scaleResource(
+ scalingExecutor.scaleResource(
TestingAutoscalerUtils.createResourceAwareContext(),
metrics,
new HashMap<>(),
@@ -343,7 +394,7 @@ public class ScalingExecutorTest {
new VertexInfo(sink, Map.of(source, REBALANCE), 10, 1000, false, null));
assertTrue(
- scalingDecisionExecutor.scaleResource(
+ scalingExecutor.scaleResource(
context,
metrics,
new HashMap<>(),
@@ -406,7 +457,7 @@ public class ScalingExecutorTest {
Map.of(jobVertexID, evaluated(1, 110, 100)), dummyGlobalMetrics);
assertEquals(
scalingEnabled,
- scalingDecisionExecutor.scaleResource(
+ scalingExecutor.scaleResource(
context,
metrics,
new HashMap<>(),
@@ -415,7 +466,7 @@ public class ScalingExecutorTest {
jobTopology));
assertEquals(
scalingEnabled,
- scalingDecisionExecutor.scaleResource(
+ scalingExecutor.scaleResource(
context,
metrics,
new HashMap<>(),
@@ -454,7 +505,7 @@ public class ScalingExecutorTest {
Map.of(jobVertexID, evaluated(1, 110, 101)), dummyGlobalMetrics);
assertEquals(
scalingEnabled,
- scalingDecisionExecutor.scaleResource(
+ scalingExecutor.scaleResource(
context,
metrics,
new HashMap<>(),
@@ -489,7 +540,7 @@ public class ScalingExecutorTest {
// Baseline, no GC/Heap metrics
assertTrue(
- scalingDecisionExecutor.scaleResource(
+ scalingExecutor.scaleResource(
context,
metrics,
new HashMap<>(),
@@ -507,7 +558,7 @@ public class ScalingExecutorTest {
ScalingMetric.HEAP_MAX_USAGE_RATIO,
new EvaluatedScalingMetric(0.9, 0.79)));
assertTrue(
- scalingDecisionExecutor.scaleResource(
+ scalingExecutor.scaleResource(
context,
metrics,
new HashMap<>(),
@@ -527,7 +578,7 @@ public class ScalingExecutorTest {
ScalingMetric.HEAP_MAX_USAGE_RATIO,
new EvaluatedScalingMetric(0.9, 0.79)));
assertFalse(
- scalingDecisionExecutor.scaleResource(
+ scalingExecutor.scaleResource(
context,
metrics,
new HashMap<>(),
@@ -547,7 +598,7 @@ public class ScalingExecutorTest {
ScalingMetric.HEAP_MAX_USAGE_RATIO,
new EvaluatedScalingMetric(0.6, 0.81)));
assertFalse(
- scalingDecisionExecutor.scaleResource(
+ scalingExecutor.scaleResource(
context,
metrics,
new HashMap<>(),
@@ -601,7 +652,7 @@ public class ScalingExecutorTest {
dummyGlobalMetrics);
var now = Instant.now();
assertThat(
- scalingDecisionExecutor.scaleResource(
+ scalingExecutor.scaleResource(
context,
metrics,
new HashMap<>(),
@@ -625,14 +676,15 @@ public class ScalingExecutorTest {
}
private Map<ScalingMetric, EvaluatedScalingMetric> evaluated(
- int parallelism, double target, double procRate, double catchupRate) {
+ int parallelism, double target, double trueProcessingRate, double catchupRate) {
var metrics = new HashMap<ScalingMetric, EvaluatedScalingMetric>();
metrics.put(ScalingMetric.PARALLELISM, EvaluatedScalingMetric.of(parallelism));
metrics.put(ScalingMetric.MAX_PARALLELISM, EvaluatedScalingMetric.of(MAX_PARALLELISM));
metrics.put(ScalingMetric.TARGET_DATA_RATE, new EvaluatedScalingMetric(target, target));
metrics.put(ScalingMetric.CATCH_UP_DATA_RATE, EvaluatedScalingMetric.of(catchupRate));
metrics.put(
- ScalingMetric.TRUE_PROCESSING_RATE, new EvaluatedScalingMetric(procRate, procRate));
+ ScalingMetric.TRUE_PROCESSING_RATE,
+ new EvaluatedScalingMetric(trueProcessingRate, trueProcessingRate));
var restartTime = context.getConfiguration().get(AutoScalerOptions.RESTART_TIME);
ScalingMetricEvaluator.computeProcessingRateThresholds(
@@ -641,8 +693,8 @@ public class ScalingExecutorTest {
}
private Map<ScalingMetric, EvaluatedScalingMetric> evaluated(
- int parallelism, double target, double procRate) {
- return evaluated(parallelism, target, procRate, 0.);
+ int parallelism, double target, double trueProcessingRate) {
+ return evaluated(parallelism, target, trueProcessingRate, 0.);
}
protected static <KEY, Context extends JobAutoScalerContext<KEY>>