This is an automated email from the ASF dual-hosted git repository.
fanrui 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 249a6945 [FLINK-35105][autoscaler] Support setting default Autoscaler
options at autoscaler standalone level
249a6945 is described below
commit 249a6945aaa847a506c42111d7332ad072578095
Author: Rui Fan <[email protected]>
AuthorDate: Mon Apr 15 15:27:54 2024 +0800
[FLINK-35105][autoscaler] Support setting default Autoscaler options at
autoscaler standalone level
---
docs/content.zh/docs/custom-resource/autoscaler.md | 7 ++++
docs/content/docs/custom-resource/autoscaler.md | 7 ++++
.../autoscaler/standalone/JobListFetcher.java | 9 ++++-
.../standalone/StandaloneAutoscalerExecutor.java | 5 ++-
.../flinkcluster/FlinkClusterJobListFetcher.java | 16 +++++---
.../StandaloneAutoscalerExecutorTest.java | 8 ++--
.../FlinkClusterJobListFetcherTest.java | 46 ++++++++++++++++++----
7 files changed, 78 insertions(+), 20 deletions(-)
diff --git a/docs/content.zh/docs/custom-resource/autoscaler.md
b/docs/content.zh/docs/custom-resource/autoscaler.md
index 08b33f29..4359c97f 100644
--- a/docs/content.zh/docs/custom-resource/autoscaler.md
+++ b/docs/content.zh/docs/custom-resource/autoscaler.md
@@ -275,6 +275,13 @@
org.apache.flink.autoscaler.standalone.StandaloneAutoscalerEntrypoint \
Updating the `autoscaler.standalone.fetcher.flink-cluster.host` and
`autoscaler.standalone.fetcher.flink-cluster.port`
based on your flink cluster. In general, the host and port are the same as
Flink WebUI.
+All autoscaler related options can be set at autoscaler standalone level, and
the configuration at job-level can
+override the default value provided in the autoscaler standalone, such as:
+
+- job.autoscaler.enabled
+- job.autoscaler.metrics.window
+- etc
+
### Using the JDBC Autoscaler State Store & Event Handler
A driver dependency is required to connect to a specified database. Here are
drivers currently supported,
diff --git a/docs/content/docs/custom-resource/autoscaler.md
b/docs/content/docs/custom-resource/autoscaler.md
index 08b33f29..df7207b3 100644
--- a/docs/content/docs/custom-resource/autoscaler.md
+++ b/docs/content/docs/custom-resource/autoscaler.md
@@ -275,6 +275,13 @@
org.apache.flink.autoscaler.standalone.StandaloneAutoscalerEntrypoint \
Updating the `autoscaler.standalone.fetcher.flink-cluster.host` and
`autoscaler.standalone.fetcher.flink-cluster.port`
based on your flink cluster. In general, the host and port are the same as
Flink WebUI.
+All autoscaler related options can be set at autoscaler standalone level, and
the configuration at job-level can
+override the default value provided in the autoscaler standalone, such as:
+
+- job.autoscaler.enabled
+- job.autoscaler.metrics.window
+- etc
+
### Using the JDBC Autoscaler State Store & Event Handler
A driver dependency is required to connect to a specified database. Here are
drivers currently supported,
diff --git
a/flink-autoscaler-standalone/src/main/java/org/apache/flink/autoscaler/standalone/JobListFetcher.java
b/flink-autoscaler-standalone/src/main/java/org/apache/flink/autoscaler/standalone/JobListFetcher.java
index 9e0ff4fe..2b772e97 100644
---
a/flink-autoscaler-standalone/src/main/java/org/apache/flink/autoscaler/standalone/JobListFetcher.java
+++
b/flink-autoscaler-standalone/src/main/java/org/apache/flink/autoscaler/standalone/JobListFetcher.java
@@ -19,6 +19,7 @@ package org.apache.flink.autoscaler.standalone;
import org.apache.flink.annotation.Experimental;
import org.apache.flink.autoscaler.JobAutoScalerContext;
+import org.apache.flink.configuration.Configuration;
import java.util.Collection;
@@ -26,5 +27,11 @@ import java.util.Collection;
@Experimental
public interface JobListFetcher<KEY, Context extends
JobAutoScalerContext<KEY>> {
- Collection<Context> fetch() throws Exception;
+ /**
+ * Fetch the job context.
+ *
+ * @param baseConf The basic configuration for standalone autoscaler. The
basic configuration
+ * can be overridden by the configuration at job-level.
+ */
+ Collection<Context> fetch(Configuration baseConf) throws Exception;
}
diff --git
a/flink-autoscaler-standalone/src/main/java/org/apache/flink/autoscaler/standalone/StandaloneAutoscalerExecutor.java
b/flink-autoscaler-standalone/src/main/java/org/apache/flink/autoscaler/standalone/StandaloneAutoscalerExecutor.java
index fa057e3e..741be348 100644
---
a/flink-autoscaler-standalone/src/main/java/org/apache/flink/autoscaler/standalone/StandaloneAutoscalerExecutor.java
+++
b/flink-autoscaler-standalone/src/main/java/org/apache/flink/autoscaler/standalone/StandaloneAutoscalerExecutor.java
@@ -22,6 +22,7 @@ import org.apache.flink.autoscaler.JobAutoScaler;
import org.apache.flink.autoscaler.JobAutoScalerContext;
import org.apache.flink.autoscaler.event.AutoScalerEventHandler;
import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.UnmodifiableConfiguration;
import org.apache.flink.util.concurrent.ExecutorThreadFactory;
import
org.apache.flink.shaded.guava31.com.google.common.util.concurrent.ThreadFactoryBuilder;
@@ -60,6 +61,7 @@ public class StandaloneAutoscalerExecutor<KEY, Context
extends JobAutoScalerCont
private final JobAutoScaler<KEY, Context> autoScaler;
private final ScheduledExecutorService scheduledExecutorService;
private final ExecutorService scalingThreadPool;
+ private final UnmodifiableConfiguration baseConf;
/**
* Maintain a set of job keys that during scaling, it should be updated at
{@link
@@ -88,6 +90,7 @@ public class StandaloneAutoscalerExecutor<KEY, Context
extends JobAutoScalerCont
Executors.newFixedThreadPool(
parallelism, new
ExecutorThreadFactory("autoscaler-standalone-scaling"));
this.scalingJobKeys = new HashSet<>();
+ this.baseConf = new UnmodifiableConfiguration(conf);
}
public void start() {
@@ -107,7 +110,7 @@ public class StandaloneAutoscalerExecutor<KEY, Context
extends JobAutoScalerCont
LOG.info("Standalone autoscaler starts scaling.");
Collection<Context> jobList;
try {
- jobList = jobListFetcher.fetch();
+ jobList = jobListFetcher.fetch(baseConf);
} catch (Throwable e) {
LOG.error("Error while fetch job list.", e);
return;
diff --git
a/flink-autoscaler-standalone/src/main/java/org/apache/flink/autoscaler/standalone/flinkcluster/FlinkClusterJobListFetcher.java
b/flink-autoscaler-standalone/src/main/java/org/apache/flink/autoscaler/standalone/flinkcluster/FlinkClusterJobListFetcher.java
index ba12cded..87e948e5 100644
---
a/flink-autoscaler-standalone/src/main/java/org/apache/flink/autoscaler/standalone/flinkcluster/FlinkClusterJobListFetcher.java
+++
b/flink-autoscaler-standalone/src/main/java/org/apache/flink/autoscaler/standalone/flinkcluster/FlinkClusterJobListFetcher.java
@@ -54,7 +54,7 @@ public class FlinkClusterJobListFetcher
}
@Override
- public Collection<JobAutoScalerContext<JobID>> fetch() throws Exception {
+ public Collection<JobAutoScalerContext<JobID>> fetch(Configuration
baseConf) throws Exception {
try (var restClusterClient = restClientGetter.apply(new
Configuration())) {
return restClusterClient
.sendRequest(
@@ -67,7 +67,8 @@ public class FlinkClusterJobListFetcher
.map(
jobStatusMessage -> {
try {
- return
generateJobContext(restClusterClient, jobStatusMessage);
+ return generateJobContext(
+ baseConf, restClusterClient,
jobStatusMessage);
} catch (Throwable e) {
throw new RuntimeException(
"generateJobContext throw
exception", e);
@@ -78,10 +79,12 @@ public class FlinkClusterJobListFetcher
}
private JobAutoScalerContext<JobID> generateJobContext(
- RestClusterClient<String> restClusterClient, JobStatusMessage
jobStatusMessage)
+ Configuration baseConf,
+ RestClusterClient<String> restClusterClient,
+ JobStatusMessage jobStatusMessage)
throws Exception {
var jobId = jobStatusMessage.getJobId();
- var conf = getConfiguration(restClusterClient, jobId);
+ var conf = getConfiguration(baseConf, restClusterClient, jobId);
return new JobAutoScalerContext<>(
jobId,
@@ -92,7 +95,8 @@ public class FlinkClusterJobListFetcher
() -> restClientGetter.apply(conf));
}
- private Configuration getConfiguration(RestClusterClient<String>
restClusterClient, JobID jobId)
+ private Configuration getConfiguration(
+ Configuration baseConf, RestClusterClient<String>
restClusterClient, JobID jobId)
throws Exception {
var jobParameters = new JobMessageParameters();
jobParameters.jobPathParameter.resolve(jobId);
@@ -105,7 +109,7 @@ public class FlinkClusterJobListFetcher
EmptyRequestBody.getInstance())
.get(restClientTimeout.toSeconds(), TimeUnit.SECONDS);
- var conf = new Configuration();
+ var conf = new Configuration(baseConf);
configurationInfo.forEach(entry -> conf.setString(entry.getKey(),
entry.getValue()));
return conf;
}
diff --git
a/flink-autoscaler-standalone/src/test/java/org/apache/flink/autoscaler/standalone/StandaloneAutoscalerExecutorTest.java
b/flink-autoscaler-standalone/src/test/java/org/apache/flink/autoscaler/standalone/StandaloneAutoscalerExecutorTest.java
index ca905071..74154a72 100644
---
a/flink-autoscaler-standalone/src/test/java/org/apache/flink/autoscaler/standalone/StandaloneAutoscalerExecutorTest.java
+++
b/flink-autoscaler-standalone/src/test/java/org/apache/flink/autoscaler/standalone/StandaloneAutoscalerExecutorTest.java
@@ -84,7 +84,7 @@ class StandaloneAutoscalerExecutorTest {
try (var autoscalerExecutor =
new StandaloneAutoscalerExecutor<>(
- conf, () -> jobList, eventCollector, jobAutoScaler) {
+ conf, baseConf -> jobList, eventCollector,
jobAutoScaler) {
@Override
protected void
scalingSingleJob(JobAutoScalerContext<JobID> jobContext) {
super.scalingSingleJob(jobContext);
@@ -112,7 +112,7 @@ class StandaloneAutoscalerExecutorTest {
try (var autoscalerExecutor =
new StandaloneAutoscalerExecutor<>(
new Configuration(),
- () -> {
+ baseConf -> {
throw new RuntimeException("Excepted exception.");
},
eventCollector,
@@ -149,7 +149,7 @@ class StandaloneAutoscalerExecutorTest {
try (var autoscalerExecutor =
new StandaloneAutoscalerExecutor<>(
conf,
- () -> jobList,
+ baseConf -> jobList,
new TestingEventCollector<>(),
new JobAutoScaler<>() {
@Override
@@ -197,7 +197,7 @@ class StandaloneAutoscalerExecutorTest {
try (var autoscalerExecutor =
new StandaloneAutoscalerExecutor<>(
conf,
- jobContextWithIndex::keySet,
+ baseConf -> jobContextWithIndex.keySet(),
new TestingEventCollector<>(),
new JobAutoScaler<>() {
@Override
diff --git
a/flink-autoscaler-standalone/src/test/java/org/apache/flink/autoscaler/standalone/flinkcluster/FlinkClusterJobListFetcherTest.java
b/flink-autoscaler-standalone/src/test/java/org/apache/flink/autoscaler/standalone/flinkcluster/FlinkClusterJobListFetcherTest.java
index 9192e521..2a549795 100644
---
a/flink-autoscaler-standalone/src/test/java/org/apache/flink/autoscaler/standalone/flinkcluster/FlinkClusterJobListFetcherTest.java
+++
b/flink-autoscaler-standalone/src/test/java/org/apache/flink/autoscaler/standalone/flinkcluster/FlinkClusterJobListFetcherTest.java
@@ -84,17 +84,42 @@ class FlinkClusterJobListFetcherTest {
closeCounter),
Duration.ofSeconds(10));
+ // Test for empty base conf
+ assertFetcherResult(
+ new Configuration(), jobs, configurations, closeCounter,
jobListFetcher);
+
+ // Add the new option key, and old option key with different value.
+ var baseConf = new Configuration();
+ baseConf.setString("option_key4", "option_value4");
+ baseConf.setString("option_key3", "option_value5");
+
+ closeCounter.set(0);
+ // Test for mixed base conf
+ assertFetcherResult(
+ new Configuration(), jobs, configurations, closeCounter,
jobListFetcher);
+ }
+
+ private void assertFetcherResult(
+ Configuration baseConf,
+ Map<JobID, JobStatus> jobs,
+ Map<JobID, Configuration> configurations,
+ AtomicLong closeCounter,
+ FlinkClusterJobListFetcher jobListFetcher)
+ throws Exception {
// Fetch multiple times and check whether the results are as expected
each time
for (int i = 1; i <= 3; i++) {
- var fetchedJobList = jobListFetcher.fetch();
+ var fetchedJobList = jobListFetcher.fetch(baseConf);
// Check whether rest client is closed.
assertThat(closeCounter).hasValue(i);
assertThat(fetchedJobList).hasSize(2);
for (var jobContext : fetchedJobList) {
- var expectedJobState = jobs.get(jobContext.getJobID());
- Configuration expectedConf =
configurations.get(jobContext.getJobID());
-
assertThat(jobContext.getJobStatus()).isEqualTo(expectedJobState);
+ var expectedJobStatus = jobs.get(jobContext.getJobID());
+
+ var expectedConf = new Configuration(baseConf);
+ expectedConf.addAll(configurations.get(jobContext.getJobID()));
+
assertThat(jobContext.getJobStatus()).isNotNull().isEqualTo(expectedJobStatus);
+
assertThat(jobContext.getConfiguration()).isNotNull().isEqualTo(expectedConf);
}
}
@@ -116,7 +141,9 @@ class FlinkClusterJobListFetcherTest {
Either.Left(Map.of()),
closeCounter),
Duration.ofSeconds(10));
-
assertThatThrownBy(jobListFetcher::fetch).getCause().isEqualTo(expectedException);
+ assertThatThrownBy(() -> jobListFetcher.fetch(new Configuration()))
+ .getCause()
+ .isEqualTo(expectedException);
assertThat(closeCounter).hasValue(1);
}
@@ -137,7 +164,9 @@ class FlinkClusterJobListFetcherTest {
closeCounter),
Duration.ofSeconds(10));
-
assertThatThrownBy(jobListFetcher::fetch).getRootCause().isEqualTo(expectedException);
+ assertThatThrownBy(() -> jobListFetcher.fetch(new Configuration()))
+ .getRootCause()
+ .isEqualTo(expectedException);
assertThat(closeCounter).hasValue(1);
}
@@ -154,7 +183,8 @@ class FlinkClusterJobListFetcherTest {
Duration.ofSeconds(2));
assertThat(closeFuture).isNotDone();
-
assertThatThrownBy(jobListFetcher::fetch).isInstanceOf(TimeoutException.class);
+ assertThatThrownBy(() -> jobListFetcher.fetch(new Configuration()))
+ .isInstanceOf(TimeoutException.class);
assertThat(closeFuture).isDone();
}
@@ -173,7 +203,7 @@ class FlinkClusterJobListFetcherTest {
Duration.ofSeconds(2));
assertThat(closeFuture).isNotDone();
- assertThatThrownBy(jobListFetcher::fetch)
+ assertThatThrownBy(() -> jobListFetcher.fetch(new Configuration()))
.getRootCause()
.isInstanceOf(TimeoutException.class);
assertThat(closeFuture).isDone();
