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/11/13 03:44:48 UTC
(flink-kubernetes-operator) branch main updated: [FLINK-33527][autoscaler] Clear all physical states after autoscaler is disabled
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 bf5bb94b [FLINK-33527][autoscaler] Clear all physical states after autoscaler is disabled
bf5bb94b is described below
commit bf5bb94bc87295e5ecb8ace8f579a22dca440e2d
Author: Rui Fan <19...@gmail.com>
AuthorDate: Mon Nov 13 11:44:43 2023 +0800
[FLINK-33527][autoscaler] Clear all physical states after autoscaler is disabled
---
.../apache/flink/autoscaler/JobAutoScalerImpl.java | 45 ++++++++++++--
.../autoscaler/metrics/ScalingHistoryUtils.java | 49 ++++++++-------
.../flink/autoscaler/JobAutoScalerImplTest.java | 72 ++++++++++++++++++++++
3 files changed, 139 insertions(+), 27 deletions(-)
diff --git a/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/JobAutoScalerImpl.java b/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/JobAutoScalerImpl.java
index 69c646f5..663a8669 100644
--- a/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/JobAutoScalerImpl.java
+++ b/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/JobAutoScalerImpl.java
@@ -29,10 +29,12 @@ import org.apache.flink.autoscaler.realizer.ScalingRealizer;
import org.apache.flink.autoscaler.state.AutoScalerStateStore;
import org.apache.flink.configuration.PipelineOptions;
import org.apache.flink.runtime.jobgraph.JobVertexID;
+import org.apache.flink.util.Preconditions;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import java.time.Clock;
import java.util.HashMap;
import java.util.Map;
import java.util.Optional;
@@ -41,6 +43,7 @@ import java.util.concurrent.ConcurrentHashMap;
import static org.apache.flink.autoscaler.config.AutoScalerOptions.AUTOSCALER_ENABLED;
import static org.apache.flink.autoscaler.metrics.AutoscalerFlinkMetrics.initRecommendedParallelism;
import static org.apache.flink.autoscaler.metrics.AutoscalerFlinkMetrics.resetRecommendedParallelism;
+import static org.apache.flink.autoscaler.metrics.ScalingHistoryUtils.trimScalingHistory;
/** The default implementation of {@link JobAutoScaler}. */
public class JobAutoScalerImpl<KEY, Context extends JobAutoScalerContext<KEY>>
@@ -57,6 +60,8 @@ public class JobAutoScalerImpl<KEY, Context extends JobAutoScalerContext<KEY>>
private final ScalingRealizer<KEY, Context> scalingRealizer;
private final AutoScalerStateStore<KEY, Context> stateStore;
+ private Clock clock = Clock.systemDefaultZone();
+
@VisibleForTesting
final Map<KEY, Map<JobVertexID, Map<ScalingMetric, EvaluatedScalingMetric>>>
lastEvaluatedMetrics = new ConcurrentHashMap<>();
@@ -86,7 +91,7 @@ public class JobAutoScalerImpl<KEY, Context extends JobAutoScalerContext<KEY>>
try {
if (!ctx.getConfiguration().getBoolean(AUTOSCALER_ENABLED)) {
LOG.debug("Autoscaler is disabled");
- clearParallelismOverrides(ctx);
+ clearStatesAfterAutoscalerDisabled(ctx);
return;
}
@@ -116,10 +121,37 @@ public class JobAutoScalerImpl<KEY, Context extends JobAutoScalerContext<KEY>>
stateStore.removeInfoFromCache(jobKey);
}
- private void clearParallelismOverrides(Context ctx) throws Exception {
- var parallelismOverrides = stateStore.getParallelismOverrides(ctx);
- if (parallelismOverrides.isPresent()) {
+ private void clearStatesAfterAutoscalerDisabled(Context ctx) throws Exception {
+ var needFlush = false;
+ var parallelismOverridesOpt = stateStore.getParallelismOverrides(ctx);
+ if (parallelismOverridesOpt.isPresent()) {
+ needFlush = true;
stateStore.removeParallelismOverrides(ctx);
+ }
+
+ var collectedMetricsOpt = stateStore.getCollectedMetrics(ctx);
+ if (collectedMetricsOpt.isPresent()) {
+ needFlush = true;
+ stateStore.removeCollectedMetrics(ctx);
+ }
+
+ var scalingHistoryOpt = stateStore.getScalingHistory(ctx);
+ if (scalingHistoryOpt.isPresent()) {
+ var scalingHistory = scalingHistoryOpt.get();
+ var trimmedScalingHistory =
+ trimScalingHistory(clock.instant(), ctx.getConfiguration(), scalingHistory);
+ if (trimmedScalingHistory.isEmpty()) {
+ // All scaling histories are trimmed.
+ needFlush = true;
+ stateStore.removeScalingHistory(ctx);
+ } else if (!scalingHistory.equals(trimmedScalingHistory)) {
+ // Some scaling histories are trimmed.
+ needFlush = true;
+ stateStore.storeScalingHistory(ctx, trimmedScalingHistory);
+ }
+ }
+
+ if (needFlush) {
stateStore.flush(ctx);
}
}
@@ -211,4 +243,9 @@ public class JobAutoScalerImpl<KEY, Context extends JobAutoScalerContext<KEY>>
ctx.getJobKey(),
id -> new AutoscalerFlinkMetrics(ctx.getMetricGroup().addGroup("AutoScaler")));
}
+
+ @VisibleForTesting
+ void setClock(Clock clock) {
+ this.clock = Preconditions.checkNotNull(clock);
+ }
}
diff --git a/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/metrics/ScalingHistoryUtils.java b/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/metrics/ScalingHistoryUtils.java
index 1f786ee9..91aedff4 100644
--- a/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/metrics/ScalingHistoryUtils.java
+++ b/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/metrics/ScalingHistoryUtils.java
@@ -21,6 +21,7 @@ import org.apache.flink.autoscaler.JobAutoScalerContext;
import org.apache.flink.autoscaler.ScalingSummary;
import org.apache.flink.autoscaler.config.AutoScalerOptions;
import org.apache.flink.autoscaler.state.AutoScalerStateStore;
+import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.jobgraph.JobVertexID;
import javax.annotation.Nonnull;
@@ -87,29 +88,31 @@ public class ScalingHistoryUtils {
var conf = context.getConfiguration();
return autoScalerStateStore
.getScalingHistory(context)
- .map(
- scalingHistory -> {
- Instant expectedStartTime =
- now.minus(
- conf.get(AutoScalerOptions.VERTEX_SCALING_HISTORY_AGE));
- var entryIt = scalingHistory.entrySet().iterator();
- while (entryIt.hasNext()) {
- var entry = entryIt.next();
- // Limit how long past scaling decisions are remembered
- entry.setValue(
- new TreeMap<>(entry.getValue().tailMap(expectedStartTime)));
- var vertexHistory = entry.getValue();
- while (vertexHistory.size()
- > conf.get(
- AutoScalerOptions.VERTEX_SCALING_HISTORY_COUNT)) {
- vertexHistory.remove(vertexHistory.firstKey());
- }
- if (vertexHistory.isEmpty()) {
- entryIt.remove();
- }
- }
- return scalingHistory;
- })
+ .map(scalingHistory -> trimScalingHistory(now, conf, scalingHistory))
.orElse(new HashMap<>());
}
+
+ public static Map<JobVertexID, SortedMap<Instant, ScalingSummary>> trimScalingHistory(
+ Instant now,
+ Configuration conf,
+ Map<JobVertexID, SortedMap<Instant, ScalingSummary>> scalingHistory) {
+ Instant expectedStartTime =
+ now.minus(conf.get(AutoScalerOptions.VERTEX_SCALING_HISTORY_AGE));
+ var result = new TreeMap<>(scalingHistory);
+ var entryIt = result.entrySet().iterator();
+ while (entryIt.hasNext()) {
+ var entry = entryIt.next();
+ // Limit how long past scaling decisions are remembered
+ entry.setValue(new TreeMap<>(entry.getValue().tailMap(expectedStartTime)));
+ var vertexHistory = entry.getValue();
+ while (vertexHistory.size()
+ > conf.get(AutoScalerOptions.VERTEX_SCALING_HISTORY_COUNT)) {
+ vertexHistory.remove(vertexHistory.firstKey());
+ }
+ if (vertexHistory.isEmpty()) {
+ entryIt.remove();
+ }
+ }
+ return result;
+ }
}
diff --git a/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/JobAutoScalerImplTest.java b/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/JobAutoScalerImplTest.java
index 2d2ed88d..fd61ccdb 100644
--- a/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/JobAutoScalerImplTest.java
+++ b/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/JobAutoScalerImplTest.java
@@ -22,6 +22,7 @@ import org.apache.flink.autoscaler.config.AutoScalerOptions;
import org.apache.flink.autoscaler.event.TestingEventCollector;
import org.apache.flink.autoscaler.exceptions.NotReadyException;
import org.apache.flink.autoscaler.metrics.AutoscalerFlinkMetrics;
+import org.apache.flink.autoscaler.metrics.CollectedMetrics;
import org.apache.flink.autoscaler.metrics.FlinkMetric;
import org.apache.flink.autoscaler.metrics.ScalingMetric;
import org.apache.flink.autoscaler.realizer.TestingScalingRealizer;
@@ -42,17 +43,23 @@ import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
+import java.time.Clock;
+import java.time.Duration;
import java.time.Instant;
+import java.time.ZoneId;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
import java.util.Optional;
import java.util.Set;
+import java.util.SortedMap;
+import java.util.TreeMap;
import static org.apache.flink.autoscaler.TestingAutoscalerUtils.createDefaultJobAutoScalerContext;
import static org.apache.flink.autoscaler.config.AutoScalerOptions.AUTOSCALER_ENABLED;
import static org.apache.flink.autoscaler.config.AutoScalerOptions.SCALING_ENABLED;
+import static org.apache.flink.autoscaler.config.AutoScalerOptions.VERTEX_SCALING_HISTORY_AGE;
import static org.assertj.core.api.Assertions.assertThat;
import static org.junit.jupiter.api.Assertions.assertEquals;
@@ -284,6 +291,71 @@ public class JobAutoScalerImplTest {
assertParallelismOverrides(Map.of(v1.toString(), "1", v2.toString(), "4"));
}
+ @Test
+ void testAutoscalerDisabled() throws Exception {
+ context.getConfiguration().setBoolean(AUTOSCALER_ENABLED, false);
+ context.getConfiguration().set(VERTEX_SCALING_HISTORY_AGE, Duration.ofMillis(200));
+
+ var scalingHistory = new TreeMap<Instant, ScalingSummary>();
+ scalingHistory.put(Instant.ofEpochMilli(100), new ScalingSummary());
+ scalingHistory.put(Instant.ofEpochMilli(200), new ScalingSummary());
+
+ // Test all scaling aren't expired
+ getInstantScalingSummaryTreeMap(
+ scalingHistory, Clock.fixed(Instant.ofEpochMilli(250), ZoneId.systemDefault()), 2);
+
+ // Test one scaling aren't expired
+ getInstantScalingSummaryTreeMap(
+ scalingHistory, Clock.fixed(Instant.ofEpochMilli(350), ZoneId.systemDefault()), 1);
+
+ // Test all scaling are expired
+ getInstantScalingSummaryTreeMap(
+ scalingHistory, Clock.fixed(Instant.ofEpochMilli(450), ZoneId.systemDefault()), 0);
+ }
+
+ private void getInstantScalingSummaryTreeMap(
+ SortedMap<Instant, ScalingSummary> scalingHistory,
+ Clock clock,
+ int expectedScalingHistorySize)
+ throws Exception {
+ stateStore = new TestingAutoscalerStateStore<>();
+ var autoscaler =
+ new JobAutoScalerImpl<>(
+ null, null, null, eventCollector, scalingRealizer, stateStore);
+
+ enrichStateStore(scalingHistory);
+ stateStore.flush(context);
+ assertThat(stateStore.getFlushCount()).isEqualTo(1);
+
+ autoscaler.setClock(clock);
+ autoscaler.scale(context);
+
+ assertThat(stateStore.getParallelismOverrides(context)).isEmpty();
+ assertThat(stateStore.getCollectedMetrics(context)).isEmpty();
+
+ if (expectedScalingHistorySize > 0) {
+ Optional<Map<JobVertexID, SortedMap<Instant, ScalingSummary>>> scalingHistoryOpt =
+ stateStore.getScalingHistory(context);
+ assertThat(scalingHistoryOpt).isPresent();
+ assertThat(scalingHistoryOpt.get().values())
+ .allMatch(aa -> aa.size() == expectedScalingHistorySize);
+ } else {
+ assertThat(stateStore.getScalingHistory(context)).isEmpty();
+ }
+ assertThat(stateStore.getFlushCount()).isEqualTo(2);
+ }
+
+ private void enrichStateStore(SortedMap<Instant, ScalingSummary> scalingHistory) {
+ var v1 = new JobVertexID();
+ var v2 = new JobVertexID();
+ stateStore.storeParallelismOverrides(
+ context, Map.of(v1.toString(), "1", v2.toString(), "2"));
+
+ var metricHistory = new TreeMap<Instant, CollectedMetrics>();
+ stateStore.storeCollectedMetrics(context, metricHistory);
+ stateStore.storeScalingHistory(context, Map.of(v1, scalingHistory, v2, scalingHistory));
+ }
+
private void assertParallelismOverrides(Map<String, String> expectedOverrides) {
TestingScalingRealizer.Event<JobID, JobAutoScalerContext<JobID>> scalingEvent;
scalingEvent = scalingRealizer.events.poll();