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/09/12 14:03:13 UTC
[flink-kubernetes-operator] branch main updated: [FLINK-33067] Expose rate limiter config and enable by default
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 4d95de3f [FLINK-33067] Expose rate limiter config and enable by default
4d95de3f is described below
commit 4d95de3fe29de7d8465d9240d04f878358635e3d
Author: Gyula Fora <g_...@apple.com>
AuthorDate: Fri Sep 8 11:43:11 2023 +0200
[FLINK-33067] Expose rate limiter config and enable by default
---
.../kubernetes_operator_config_configuration.html | 28 ++++++++--
.../shortcodes/generated/system_section.html | 28 ++++++++--
.../flink/kubernetes/operator/FlinkOperator.java | 1 +
.../config/FlinkOperatorConfiguration.java | 46 ++++++++++++----
.../config/KubernetesOperatorConfigOptions.java | 32 ++++++++---
.../kubernetes/operator/FlinkOperatorTest.java | 6 +++
...st.java => FlinkOperatorConfigurationTest.java} | 63 +++++++++++++---------
7 files changed, 152 insertions(+), 52 deletions(-)
diff --git a/docs/layouts/shortcodes/generated/kubernetes_operator_config_configuration.html b/docs/layouts/shortcodes/generated/kubernetes_operator_config_configuration.html
index aab4573e..893d5eab 100644
--- a/docs/layouts/shortcodes/generated/kubernetes_operator_config_configuration.html
+++ b/docs/layouts/shortcodes/generated/kubernetes_operator_config_configuration.html
@@ -278,6 +278,18 @@
<td>Boolean</td>
<td>Configure the array merge behaviour during pod merging. Arrays can be either merged by position or name matching.</td>
</tr>
+ <tr>
+ <td><h5>kubernetes.operator.rate-limiter.limit</h5></td>
+ <td style="word-wrap: break-word;">5</td>
+ <td>Integer</td>
+ <td>Max number of reconcile loops triggered within the rate limiter refresh period for each resource. Setting the limit <= 0 disables the limiter.</td>
+ </tr>
+ <tr>
+ <td><h5>kubernetes.operator.rate-limiter.refresh-period</h5></td>
+ <td style="word-wrap: break-word;">15 s</td>
+ <td>Duration</td>
+ <td>Operator rate limiter refresh period for each resource.</td>
+ </tr>
<tr>
<td><h5>kubernetes.operator.reconcile.interval</h5></td>
<td style="word-wrap: break-word;">1 min</td>
@@ -306,19 +318,25 @@
<td><h5>kubernetes.operator.retry.initial.interval</h5></td>
<td style="word-wrap: break-word;">5 s</td>
<td>Duration</td>
- <td>Initial interval of automatic reconcile retries on recoverable errors.</td>
+ <td>Initial interval of retries on unhandled controller errors.</td>
</tr>
<tr>
<td><h5>kubernetes.operator.retry.interval.multiplier</h5></td>
- <td style="word-wrap: break-word;">2.0</td>
+ <td style="word-wrap: break-word;">1.5</td>
<td>Double</td>
- <td>Interval multiplier of automatic reconcile retries on recoverable errors.</td>
+ <td>Interval multiplier of retries on unhandled controller errors.</td>
</tr>
<tr>
<td><h5>kubernetes.operator.retry.max.attempts</h5></td>
- <td style="word-wrap: break-word;">10</td>
+ <td style="word-wrap: break-word;">15</td>
<td>Integer</td>
- <td>Max attempts of automatic reconcile retries on recoverable errors.</td>
+ <td>Max attempts of retries on unhandled controller errors.</td>
+ </tr>
+ <tr>
+ <td><h5>kubernetes.operator.retry.max.interval</h5></td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <td>Duration</td>
+ <td>Max interval of retries on unhandled controller errors.</td>
</tr>
<tr>
<td><h5>kubernetes.operator.savepoint.cleanup.enabled</h5></td>
diff --git a/docs/layouts/shortcodes/generated/system_section.html b/docs/layouts/shortcodes/generated/system_section.html
index aa317587..9a0350e6 100644
--- a/docs/layouts/shortcodes/generated/system_section.html
+++ b/docs/layouts/shortcodes/generated/system_section.html
@@ -80,6 +80,18 @@
<td>Duration</td>
<td>Leader election retry period.</td>
</tr>
+ <tr>
+ <td><h5>kubernetes.operator.rate-limiter.limit</h5></td>
+ <td style="word-wrap: break-word;">5</td>
+ <td>Integer</td>
+ <td>Max number of reconcile loops triggered within the rate limiter refresh period for each resource. Setting the limit <= 0 disables the limiter.</td>
+ </tr>
+ <tr>
+ <td><h5>kubernetes.operator.rate-limiter.refresh-period</h5></td>
+ <td style="word-wrap: break-word;">15 s</td>
+ <td>Duration</td>
+ <td>Operator rate limiter refresh period for each resource.</td>
+ </tr>
<tr>
<td><h5>kubernetes.operator.reconcile.interval</h5></td>
<td style="word-wrap: break-word;">1 min</td>
@@ -102,19 +114,25 @@
<td><h5>kubernetes.operator.retry.initial.interval</h5></td>
<td style="word-wrap: break-word;">5 s</td>
<td>Duration</td>
- <td>Initial interval of automatic reconcile retries on recoverable errors.</td>
+ <td>Initial interval of retries on unhandled controller errors.</td>
</tr>
<tr>
<td><h5>kubernetes.operator.retry.interval.multiplier</h5></td>
- <td style="word-wrap: break-word;">2.0</td>
+ <td style="word-wrap: break-word;">1.5</td>
<td>Double</td>
- <td>Interval multiplier of automatic reconcile retries on recoverable errors.</td>
+ <td>Interval multiplier of retries on unhandled controller errors.</td>
</tr>
<tr>
<td><h5>kubernetes.operator.retry.max.attempts</h5></td>
- <td style="word-wrap: break-word;">10</td>
+ <td style="word-wrap: break-word;">15</td>
<td>Integer</td>
- <td>Max attempts of automatic reconcile retries on recoverable errors.</td>
+ <td>Max attempts of retries on unhandled controller errors.</td>
+ </tr>
+ <tr>
+ <td><h5>kubernetes.operator.retry.max.interval</h5></td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <td>Duration</td>
+ <td>Max interval of retries on unhandled controller errors.</td>
</tr>
<tr>
<td><h5>kubernetes.operator.user.artifacts.base.dir</h5></td>
diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/FlinkOperator.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/FlinkOperator.java
index 8c42035d..f6333153 100644
--- a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/FlinkOperator.java
+++ b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/FlinkOperator.java
@@ -215,6 +215,7 @@ public class FlinkOperator {
overrider.settingNamespaces(operatorConf.getWatchedNamespaces());
overrider.withRetry(operatorConf.getRetryConfiguration());
+ overrider.withRateLimiter(operatorConf.getRateLimiter());
var labelSelector = operatorConf.getLabelSelector();
LOG.info(
diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/config/FlinkOperatorConfiguration.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/config/FlinkOperatorConfiguration.java
index 3cad8669..9fcdb6a1 100644
--- a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/config/FlinkOperatorConfiguration.java
+++ b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/config/FlinkOperatorConfiguration.java
@@ -25,6 +25,8 @@ import org.apache.flink.kubernetes.operator.utils.EnvUtils;
import io.fabric8.kubernetes.api.model.DeletionPropagation;
import io.javaoperatorsdk.operator.api.config.LeaderElectionConfiguration;
+import io.javaoperatorsdk.operator.processing.event.rate.LinearRateLimiter;
+import io.javaoperatorsdk.operator.processing.event.rate.RateLimiter;
import io.javaoperatorsdk.operator.processing.retry.GenericRetry;
import lombok.Value;
import org.apache.commons.lang3.StringUtils;
@@ -63,6 +65,7 @@ public class FlinkOperatorConfiguration {
Integer savepointHistoryCountThreshold;
Duration savepointHistoryAgeThreshold;
GenericRetry retryConfiguration;
+ RateLimiter<?> rateLimiter;
boolean exceptionStackTraceEnabled;
int exceptionStackTraceLengthThreshold;
int exceptionFieldLengthThreshold;
@@ -174,6 +177,7 @@ public class FlinkOperatorConfiguration {
KubernetesOperatorMetricOptions.OPERATOR_METRICS_HISTOGRAM_SAMPLE_SIZE);
GenericRetry retryConfiguration = getRetryConfig(operatorConfig);
+ RateLimiter rateLimiter = getRateLimiter(operatorConfig);
String labelSelector =
operatorConfig.getString(KubernetesOperatorConfigOptions.OPERATOR_LABEL_SELECTOR);
@@ -200,6 +204,7 @@ public class FlinkOperatorConfiguration {
savepointHistoryCountThreshold,
savepointHistoryAgeThreshold,
retryConfiguration,
+ rateLimiter,
exceptionStackTraceEnabled,
exceptionStackTraceLengthThreshold,
exceptionFieldLengthThreshold,
@@ -211,17 +216,36 @@ public class FlinkOperatorConfiguration {
}
private static GenericRetry getRetryConfig(Configuration conf) {
- return new GenericRetry()
- .setMaxAttempts(
- conf.getInteger(
- KubernetesOperatorConfigOptions.OPERATOR_RETRY_MAX_ATTEMPTS))
- .setInitialInterval(
- conf.get(KubernetesOperatorConfigOptions.OPERATOR_RETRY_INITIAL_INTERVAL)
- .toMillis())
- .setIntervalMultiplier(
- conf.getDouble(
- KubernetesOperatorConfigOptions
- .OPERATOR_RETRY_INTERVAL_MULTIPLIER));
+ var genericRetry =
+ new GenericRetry()
+ .setMaxAttempts(
+ conf.getInteger(
+ KubernetesOperatorConfigOptions
+ .OPERATOR_RETRY_MAX_ATTEMPTS))
+ .setInitialInterval(
+ conf.get(
+ KubernetesOperatorConfigOptions
+ .OPERATOR_RETRY_INITIAL_INTERVAL)
+ .toMillis())
+ .setIntervalMultiplier(
+ conf.getDouble(
+ KubernetesOperatorConfigOptions
+ .OPERATOR_RETRY_INTERVAL_MULTIPLIER));
+
+ if (conf.contains(KubernetesOperatorConfigOptions.OPERATOR_RETRY_MAX_INTERVAL)) {
+ genericRetry.setMaxInterval(
+ conf.get(KubernetesOperatorConfigOptions.OPERATOR_RETRY_MAX_INTERVAL)
+ .toMillis());
+ } else {
+ genericRetry.withoutMaxInterval();
+ }
+ return genericRetry;
+ }
+
+ private static RateLimiter<?> getRateLimiter(Configuration conf) {
+ return new LinearRateLimiter(
+ conf.get(KubernetesOperatorConfigOptions.OPERATOR_RATE_LIMITER_PERIOD),
+ conf.get(KubernetesOperatorConfigOptions.OPERATOR_RATE_LIMITER_LIMIT));
}
private static LeaderElectionConfiguration getLeaderElectionConfig(Configuration conf) {
diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/config/KubernetesOperatorConfigOptions.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/config/KubernetesOperatorConfigOptions.java
index 06ab1270..629df78b 100644
--- a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/config/KubernetesOperatorConfigOptions.java
+++ b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/config/KubernetesOperatorConfigOptions.java
@@ -362,24 +362,44 @@ public class KubernetesOperatorConfigOptions {
operatorConfig("retry.initial.interval")
.durationType()
.defaultValue(Duration.ofSeconds(5))
- .withDescription(
- "Initial interval of automatic reconcile retries on recoverable errors.");
+ .withDescription("Initial interval of retries on unhandled controller errors.");
+
+ @Documentation.Section(SECTION_SYSTEM)
+ public static final ConfigOption<Duration> OPERATOR_RETRY_MAX_INTERVAL =
+ operatorConfig("retry.max.interval")
+ .durationType()
+ .noDefaultValue()
+ .withDescription("Max interval of retries on unhandled controller errors.");
@Documentation.Section(SECTION_SYSTEM)
public static final ConfigOption<Double> OPERATOR_RETRY_INTERVAL_MULTIPLIER =
operatorConfig("retry.interval.multiplier")
.doubleType()
- .defaultValue(2.0)
+ .defaultValue(1.5)
.withDescription(
- "Interval multiplier of automatic reconcile retries on recoverable errors.");
+ "Interval multiplier of retries on unhandled controller errors.");
@Documentation.Section(SECTION_SYSTEM)
public static final ConfigOption<Integer> OPERATOR_RETRY_MAX_ATTEMPTS =
operatorConfig("retry.max.attempts")
.intType()
- .defaultValue(10)
+ .defaultValue(15)
+ .withDescription("Max attempts of retries on unhandled controller errors.");
+
+ @Documentation.Section(SECTION_SYSTEM)
+ public static final ConfigOption<Duration> OPERATOR_RATE_LIMITER_PERIOD =
+ operatorConfig("rate-limiter.refresh-period")
+ .durationType()
+ .defaultValue(Duration.ofSeconds(15))
+ .withDescription("Operator rate limiter refresh period for each resource.");
+
+ @Documentation.Section(SECTION_SYSTEM)
+ public static final ConfigOption<Integer> OPERATOR_RATE_LIMITER_LIMIT =
+ operatorConfig("rate-limiter.limit")
+ .intType()
+ .defaultValue(5)
.withDescription(
- "Max attempts of automatic reconcile retries on recoverable errors.");
+ "Max number of reconcile loops triggered within the rate limiter refresh period for each resource. Setting the limit <= 0 disables the limiter.");
@Documentation.Section(SECTION_DYNAMIC)
public static final ConfigOption<Boolean> OPERATOR_JOB_UPGRADE_LAST_STATE_FALLBACK_ENABLED =
diff --git a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/FlinkOperatorTest.java b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/FlinkOperatorTest.java
index edbb5366..b60551a9 100644
--- a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/FlinkOperatorTest.java
+++ b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/FlinkOperatorTest.java
@@ -24,6 +24,7 @@ import org.apache.flink.kubernetes.operator.config.KubernetesOperatorConfigOptio
import io.fabric8.kubernetes.client.Config;
import io.javaoperatorsdk.operator.RegisteredController;
import io.javaoperatorsdk.operator.api.config.ControllerConfiguration;
+import io.javaoperatorsdk.operator.processing.event.rate.LinearRateLimiter;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.Test;
@@ -85,6 +86,11 @@ public class FlinkOperatorTest {
labelSelectors.forEach(selector -> Assertions.assertEquals(testSelector, selector));
Assertions.assertFalse(configService.stopOnInformerErrorDuringStartup());
+ testOperator.registeredControllers.stream()
+ .map(RegisteredController::getConfiguration)
+ .map(ControllerConfiguration::getRateLimiter)
+ .forEach(rl -> Assertions.assertTrue(((LinearRateLimiter) rl).isActivated()));
+
var leaderElectionConfiguration = configService.getLeaderElectionConfiguration().get();
Assertions.assertEquals(testLeaseName, leaderElectionConfiguration.getLeaseName());
diff --git a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/config/FlinkOperatorRetryConfigurationTest.java b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/config/FlinkOperatorConfigurationTest.java
similarity index 54%
rename from flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/config/FlinkOperatorRetryConfigurationTest.java
rename to flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/config/FlinkOperatorConfigurationTest.java
index ba503cba..240a54f5 100644
--- a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/config/FlinkOperatorRetryConfigurationTest.java
+++ b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/config/FlinkOperatorConfigurationTest.java
@@ -19,36 +19,33 @@ package org.apache.flink.kubernetes.operator.config;
import org.apache.flink.configuration.Configuration;
+import io.javaoperatorsdk.operator.processing.event.rate.LinearRateLimiter;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
import java.util.Map;
/** Operator retry config tests. */
-public class FlinkOperatorRetryConfigurationTest {
+public class FlinkOperatorConfigurationTest {
@Test
public void testRetryConfiguration() {
-
// default values
FlinkConfigManager configManager = new FlinkConfigManager(new Configuration());
+ var retryConf = configManager.getOperatorConfiguration().getRetryConfiguration();
+
Assertions.assertEquals(
KubernetesOperatorConfigOptions.OPERATOR_RETRY_INITIAL_INTERVAL
.defaultValue()
.toMillis(),
- configManager
- .getOperatorConfiguration()
- .getRetryConfiguration()
- .getInitialInterval());
+ retryConf.getInitialInterval());
Assertions.assertEquals(
KubernetesOperatorConfigOptions.OPERATOR_RETRY_INTERVAL_MULTIPLIER.defaultValue(),
- configManager
- .getOperatorConfiguration()
- .getRetryConfiguration()
- .getIntervalMultiplier());
+ retryConf.getIntervalMultiplier());
Assertions.assertEquals(
KubernetesOperatorConfigOptions.OPERATOR_RETRY_MAX_ATTEMPTS.defaultValue(),
- configManager.getOperatorConfiguration().getRetryConfiguration().getMaxAttempts());
+ retryConf.getMaxAttempts());
+ Assertions.assertEquals(-1, retryConf.getMaxInterval());
// overrides
var overrides =
@@ -61,22 +58,38 @@ public class FlinkOperatorRetryConfigurationTest {
.key(),
"2.0",
KubernetesOperatorConfigOptions.OPERATOR_RETRY_MAX_ATTEMPTS.key(),
- "3"));
+ "3",
+ KubernetesOperatorConfigOptions.OPERATOR_RETRY_MAX_INTERVAL.key(),
+ "4"));
configManager.updateDefaultConfig(overrides);
+ retryConf = configManager.getOperatorConfiguration().getRetryConfiguration();
+ Assertions.assertEquals(1000L, retryConf.getInitialInterval());
+ Assertions.assertEquals(2.0, retryConf.getIntervalMultiplier());
+ Assertions.assertEquals(3, retryConf.getMaxAttempts());
+ Assertions.assertEquals(4, retryConf.getMaxInterval());
+ }
+
+ @Test
+ public void testRateLimiter() {
+ // default values
+ FlinkConfigManager configManager = new FlinkConfigManager(new Configuration());
+ var rateLimiter =
+ (LinearRateLimiter) configManager.getOperatorConfiguration().getRateLimiter();
+
+ Assertions.assertTrue(rateLimiter.isActivated());
Assertions.assertEquals(
- 1000L,
- configManager
- .getOperatorConfiguration()
- .getRetryConfiguration()
- .getInitialInterval());
- Assertions.assertEquals(
- 2.0,
- configManager
- .getOperatorConfiguration()
- .getRetryConfiguration()
- .getIntervalMultiplier());
+ KubernetesOperatorConfigOptions.OPERATOR_RATE_LIMITER_PERIOD.defaultValue(),
+ rateLimiter.getRefreshPeriod());
Assertions.assertEquals(
- 3,
- configManager.getOperatorConfiguration().getRetryConfiguration().getMaxAttempts());
+ KubernetesOperatorConfigOptions.OPERATOR_RATE_LIMITER_LIMIT.defaultValue(),
+ rateLimiter.getLimitForPeriod());
+ configManager.updateDefaultConfig(
+ Configuration.fromMap(
+ Map.of(
+ KubernetesOperatorConfigOptions.OPERATOR_RATE_LIMITER_LIMIT.key(),
+ "0")));
+
+ rateLimiter = (LinearRateLimiter) configManager.getOperatorConfiguration().getRateLimiter();
+ Assertions.assertFalse(rateLimiter.isActivated());
}
}