rate limiter

Signed-off-by: Attila Mészáros <a_meszaros@apple.com>

Author: csviri

operator-framework/src/main/java/io/javaoperatorsdk/operator/config/loader/ConfigLoader.java

@@ -30,6 +30,7 @@
 import io.javaoperatorsdk.operator.config.loader.provider.AgregatePrirityListConfigProvider;
 import io.javaoperatorsdk.operator.config.loader.provider.EnvVarConfigProvider;
 import io.javaoperatorsdk.operator.config.loader.provider.SystemPropertyConfigProvider;
+import io.javaoperatorsdk.operator.processing.event.rate.LinearRateLimiter;
 import io.javaoperatorsdk.operator.processing.retry.GenericRetry;
 
 public class ConfigLoader {
@@ -121,6 +122,12 @@ public static ConfigLoader getDefault() {
   static final String RETRY_INTERVAL_MULTIPLIER_SUFFIX = "retry.interval-multiplier";
   static final String RETRY_MAX_INTERVAL_SUFFIX = "retry.max-interval";
 
+  // ---------------------------------------------------------------------------
+  // Controller-level rate-limiter property suffixes
+  // ---------------------------------------------------------------------------
+  static final String RATE_LIMITER_REFRESH_PERIOD_SUFFIX = "rate-limiter.refresh-period";
+  static final String RATE_LIMITER_LIMIT_FOR_PERIOD_SUFFIX = "rate-limiter.limit-for-period";
+
   // ---------------------------------------------------------------------------
   // Controller-level (ControllerConfigurationOverrider) bindings
   // The key used at runtime is built as:
@@ -147,7 +154,11 @@ public static ConfigLoader getDefault() {
               Boolean.class,
               ControllerConfigurationOverrider::withTriggerReconcilerOnAllEvents),
           new ConfigBinding<>(
-              "informer-list-limit",
+              "informer.label-selector",
+              String.class,
+              ControllerConfigurationOverrider::withLabelSelector),
+          new ConfigBinding<>(
+              "informer.list-limit",
               Long.class,
               ControllerConfigurationOverrider::withInformerListLimit));
 
@@ -210,6 +221,11 @@ Consumer<ControllerConfigurationOverrider<R>> applyControllerConfigs(String cont
     if (retryStep != null) {
       consumer = consumer == null ? retryStep : consumer.andThen(retryStep);
     }
+    Consumer<ControllerConfigurationOverrider<R>> rateLimiterStep =
+        buildRateLimiterConsumer(prefix);
+    if (rateLimiterStep != null) {
+      consumer = consumer.andThen(rateLimiterStep);
+    }
     return consumer;
   }
 
@@ -246,6 +262,32 @@ private <R extends HasMetadata> Consumer<ControllerConfigurationOverrider<R>> bu
     };
   }
 
+  /**
+   * Returns a {@link Consumer} that builds a {@link LinearRateLimiter} only if {@code
+   * rate-limiter.limit-for-period} is present and positive (a non-positive value would deactivate
+   * the limiter and is therefore treated as absent). {@code rate-limiter.refresh-period} is applied
+   * when also present; otherwise the default refresh period is used. Returns {@code null} when no
+   * effective rate-limiter configuration is found.
+   */
+  private <R extends HasMetadata>
+      Consumer<ControllerConfigurationOverrider<R>> buildRateLimiterConsumer(String prefix) {
+    Optional<Duration> refreshPeriod =
+        configProvider.getValue(prefix + RATE_LIMITER_REFRESH_PERIOD_SUFFIX, Duration.class);
+    Optional<Integer> limitForPeriod =
+        configProvider.getValue(prefix + RATE_LIMITER_LIMIT_FOR_PERIOD_SUFFIX, Integer.class);
+
+    if (limitForPeriod.isEmpty() || limitForPeriod.get() <= 0) {
+      return null;
+    }
+
+    return overrider -> {
+      var rateLimiter =
+          new LinearRateLimiter(
+              refreshPeriod.orElse(LinearRateLimiter.DEFAULT_REFRESH_PERIOD), limitForPeriod.get());
+      overrider.withRateLimiter(rateLimiter);
+    };
+  }
+
   /**
    * If leader election is explicitly disabled via {@code leader-election.enabled=false}, returns
    * {@code null}. Otherwise, if at least one leader-election property is present (with {@code

operator-framework/src/test/java/io/javaoperatorsdk/operator/config/loader/ConfigLoaderTest.java

@@ -201,7 +201,10 @@ public <T> Optional<T> getValue(String key, Class<T> type) {
             "josdk.controller.ctrl.max-reconciliation-interval",
             "josdk.controller.ctrl.field-manager",
             "josdk.controller.ctrl.trigger-reconciler-on-all-events",
-            "josdk.controller.ctrl.informer-list-limit");
+            "josdk.controller.ctrl.informer.label-selector",
+            "josdk.controller.ctrl.informer.list-limit",
+            "josdk.controller.ctrl.rate-limiter.refresh-period",
+            "josdk.controller.ctrl.rate-limiter.limit-for-period");
   }
 
   @Test
@@ -214,6 +217,26 @@ void controllerKeyPrefixIsJosdkControllerDot() {
     assertThat(ConfigLoader.DEFAULT_CONTROLLER_KEY_PREFIX).isEqualTo("josdk.controller.");
   }
 
+  // -- rate limiter -----------------------------------------------------------
+
+  @Test
+  void rateLimiterQueriesExpectedKeys() {
+    var queriedKeys = new ArrayList<String>();
+    ConfigProvider recordingProvider =
+        new ConfigProvider() {
+          @Override
+          public <T> Optional<T> getValue(String key, Class<T> type) {
+            queriedKeys.add(key);
+            return Optional.empty();
+          }
+        };
+    new ConfigLoader(recordingProvider).applyControllerConfigs("ctrl");
+    assertThat(queriedKeys)
+        .contains(
+            "josdk.controller.ctrl.rate-limiter.refresh-period",
+            "josdk.controller.ctrl.rate-limiter.limit-for-period");
+  }
+
   // -- binding coverage -------------------------------------------------------
 
   /**