Merge branch 'rate-limiting' of github.com:java-operator-sdk/java-operator-sdk into rate-limiting

Author: csviri

docs/documentation/features.md

@@ -326,28 +326,37 @@ intersections:
 
 ## Rate Limiting
 
-It is possible to rate limit reconciliation for a resource. Thus rate limiting is per resource, 
-and it takes precedence over retry and re-schedule configurations. So for example event a retry is scheduled in
-1 seconds but this does not meet the rate limit, the next reconciliation will be postponed according rate limiting rules;
-however never cancelled, just executed as early as possible according rate limit configuration.
-
-Rate limiting is by default turned off, since correct configuration depends on the reconciler implementation, and 
-how long an execution takes. 
-(The parallelism of reconciliation itself can be limited  [`ConfigurationService`](https://github.com/java-operator-sdk/java-operator-sdk/blob/ce4d996ee073ebef5715737995fc3d33f4751275/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/ConfigurationService.java#L120-L120)
-by setting appropriate ExecutorService.)
-
-A default implementation of rate limiter is provided, see: [`PeriodRateLimiter`](https://github.com/java-operator-sdk/java-operator-sdk/blob/ce4d996ee073ebef5715737995fc3d33f4751275/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/rate/PeriodRateLimiter.java#L14-L14). 
-Users can override it with a custom implementation of 
+It is possible to rate limit reconciliation on a per-resource basis. The rate limit also takes
+precedence over retry/re-schedule configurations: for example, even if a retry was scheduled for
+the next second but this request would make the resource go over its rate limit, the next
+reconciliation will be postponed according to the rate limiting rules. Note that the
+reconciliation is never cancelled, it will just be executed as early as possible based on rate
+limitations.
+
+Rate limiting is by default turned **off**, since correct configuration depends on the reconciler
+implementation, in particular, on how long a typical reconciliation takes.
+(The parallelism of reconciliation itself can be
+limited  [`ConfigurationService`](https://github.com/java-operator-sdk/java-operator-sdk/blob/ce4d996ee073ebef5715737995fc3d33f4751275/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/ConfigurationService.java#L120-L120)
+by configuring the `ExecutorService` appropriately.)
+
+A default rate limiter implementation is provided, see:
+[`PeriodRateLimiter`](https://github.com/java-operator-sdk/java-operator-sdk/blob/ce4d996ee073ebef5715737995fc3d33f4751275/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/rate/PeriodRateLimiter.java#L14-L14)
+.
+Users can override it by implementing their own
 [`RateLimiter`](https://github.com/java-operator-sdk/java-operator-sdk/blob/ce4d996ee073ebef5715737995fc3d33f4751275/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/rate/RateLimiter.java)
-interface.
+.
 
-To configure the default rate limiter use `@ControllerConfiguration` annotation. The following configuration limits
-the reconciliation to 2 in 3 seconds: 
+To configure the default rate limiter use `@ControllerConfiguration` annotation. The following
+configuration limits
+each resource to reconcile at most twice within a 3 second interval:
 
-`@ControllerConfiguration(rateLimit = @RateLimit(limitForPeriod = 2,refreshPeriod = 3,refreshPeriodTimeUnit = TimeUnit.SECONDS))`.
+`@ControllerConfiguration(rateLimit = @RateLimit(limitForPeriod = 2,refreshPeriod = 3,refreshPeriodTimeUnit = TimeUnit.SECONDS))`
+.
 
-That means if the reconciler executed twice in one second, it will wait at least additional two seconds before it is
-reconciled again.
+Thus, if a given resource was reconciled twice in one second, no further reconciliation for this
+resource will happen before two seconds have elapsed. Note that, since rate is limited on a
+per-resource basis, other resources can still be reconciled at the same time, as long, of course,
+that they stay within their own rate limits.
 
 
 ## Handling Related Events with Event Sources

operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/ControllerConfiguration.java

@@ -17,6 +17,8 @@
 
 public interface ControllerConfiguration<R extends HasMetadata> extends ResourceConfiguration<R> {
 
+  RateLimiter DEFAULT_RATE_LIMITER = new PeriodRateLimiter();
+
   default String getName() {
     return ReconcilerUtils.getDefaultReconcilerName(getAssociatedReconcilerClassName());
   }
@@ -46,7 +48,7 @@ default RetryConfiguration getRetryConfiguration() {
   }
 
   default RateLimiter getRateLimiter() {
-    return new PeriodRateLimiter();
+    return DEFAULT_RATE_LIMITER;
   }
 
   /**

operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/timer/TimerEventSource.java

@@ -24,6 +24,9 @@ public class TimerEventSource<R extends HasMetadata>
   private final AtomicBoolean running = new AtomicBoolean();
   private final Map<ResourceID, EventProducerTimeTask> onceTasks = new ConcurrentHashMap<>();
 
+  public void scheduleOnce(R resource, long delay) {
+    scheduleOnce(ResourceID.fromResource(resource), delay);
+  }
 
   public void scheduleOnce(ResourceID resourceID, long delay) {
     if (!running.get()) {

operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/EventProcessorTest.java

@@ -32,8 +32,19 @@
 import static org.awaitility.Awaitility.await;
 import static org.mockito.ArgumentMatchers.eq;
 import static org.mockito.ArgumentMatchers.isNull;
-import static org.mockito.Mockito.*;
-
+import static org.mockito.Mockito.after;
+import static org.mockito.Mockito.any;
+import static org.mockito.Mockito.anyLong;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.timeout;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+@SuppressWarnings({"rawtypes", "unchecked"})
 class EventProcessorTest {
 
   private static final Logger log = LoggerFactory.getLogger(EventProcessorTest.class);
@@ -42,16 +53,16 @@ class EventProcessorTest {
   public static final int SEPARATE_EXECUTION_TIMEOUT = 450;
   public static final String TEST_NAMESPACE = "default-event-handler-test";
 
-  private ReconciliationDispatcher reconciliationDispatcherMock =
+  private final ReconciliationDispatcher reconciliationDispatcherMock =
       mock(ReconciliationDispatcher.class);
-  private EventSourceManager eventSourceManagerMock = mock(EventSourceManager.class);
-  private TimerEventSource retryTimerEventSourceMock = mock(TimerEventSource.class);
-  private ControllerResourceEventSource controllerResourceEventSourceMock =
+  private final EventSourceManager eventSourceManagerMock = mock(EventSourceManager.class);
+  private final TimerEventSource retryTimerEventSourceMock = mock(TimerEventSource.class);
+  private final ControllerResourceEventSource controllerResourceEventSourceMock =
       mock(ControllerResourceEventSource.class);
-  private Metrics metricsMock = mock(Metrics.class);
+  private final Metrics metricsMock = mock(Metrics.class);
   private EventProcessor eventProcessor;
   private EventProcessor eventProcessorWithRetry;
-  private RateLimiter rateLimiterMock = mock(RateLimiter.class);
+  private final RateLimiter rateLimiterMock = mock(RateLimiter.class);
 
   @BeforeEach
   void setup() {
@@ -89,7 +100,7 @@ void skipProcessingIfLatestCustomResourceNotInCache() {
   }
 
   @Test
-  void ifExecutionInProgressWaitsUntilItsFinished() throws InterruptedException {
+  void ifExecutionInProgressWaitsUntilItsFinished() {
     ResourceID resourceUid = eventAlreadyUnderProcessing();
 
     eventProcessor.handleEvent(nonCREvent(resourceUid));
@@ -204,7 +215,7 @@ void scheduleTimedEventIfInstructedByPostExecutionControl() {
     eventProcessor.handleEvent(prepareCREvent());
 
     verify(retryTimerEventSourceMock, timeout(SEPARATE_EXECUTION_TIMEOUT).times(1))
-        .scheduleOnce(any(), eq(testDelay));
+        .scheduleOnce((ResourceID) any(), eq(testDelay));
   }
 
   @Test
@@ -220,7 +231,7 @@ void reScheduleOnlyIfNotExecutedEventsReceivedMeanwhile() throws InterruptedExce
 
     verify(retryTimerEventSourceMock,
         after((long) (FAKE_CONTROLLER_EXECUTION_DURATION * 1.5)).times(0))
-        .scheduleOnce(any(), eq(testDelay));
+            .scheduleOnce((ResourceID) any(), eq(testDelay));
   }
 
   @Test
@@ -335,7 +346,7 @@ void newResourceAfterMissedDeleteEvent() {
   }
 
   @Test
-  void rateLimitsReconciliationSubmission() throws InterruptedException {
+  void rateLimitsReconciliationSubmission() {
     // the refresh period value does not matter here
     var refreshPeriod = Duration.ofMillis(100);
     var event = prepareCREvent();
@@ -347,10 +358,10 @@ void rateLimitsReconciliationSubmission() throws InterruptedException {
     eventProcessor.handleEvent(event);
     verify(reconciliationDispatcherMock, after(FAKE_CONTROLLER_EXECUTION_DURATION).times(1))
         .handleExecution(any());
-    verify(retryTimerEventSourceMock, times(0)).scheduleOnce(any(), anyLong());
+    verify(retryTimerEventSourceMock, times(0)).scheduleOnce((ResourceID) any(), anyLong());
 
     eventProcessor.handleEvent(event);
-    verify(retryTimerEventSourceMock, times(1)).scheduleOnce(any(), anyLong());
+    verify(retryTimerEventSourceMock, times(1)).scheduleOnce((ResourceID) any(), anyLong());
   }
 
   private ResourceID eventAlreadyUnderProcessing() {