controller integration

Author: csviri

operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/EventProcessor.java

@@ -1,5 +1,6 @@
 package io.javaoperatorsdk.operator.processing.event;
 
+import java.time.Duration;
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.Map;
@@ -20,6 +21,7 @@
 import io.javaoperatorsdk.operator.api.reconciler.RetryInfo;
 import io.javaoperatorsdk.operator.processing.LifecycleAware;
 import io.javaoperatorsdk.operator.processing.MDCUtils;
+import io.javaoperatorsdk.operator.processing.event.rate.RateLimiter;
 import io.javaoperatorsdk.operator.processing.event.source.Cache;
 import io.javaoperatorsdk.operator.processing.event.source.controller.ResourceAction;
 import io.javaoperatorsdk.operator.processing.event.source.controller.ResourceEvent;
@@ -44,6 +46,7 @@ class EventProcessor<R extends HasMetadata> implements EventHandler, LifecycleAw
   private final Cache<R> cache;
   private final EventSourceManager<R> eventSourceManager;
   private final EventMarker eventMarker = new EventMarker();
+  private final RateLimiter rateLimiter;
 
   EventProcessor(EventSourceManager<R> eventSourceManager) {
     this(
@@ -92,6 +95,8 @@ private EventProcessor(
     this.cache = cache;
     this.metrics = metrics != null ? metrics : Metrics.NOOP;
     this.eventSourceManager = eventSourceManager;
+    // todo configure
+    this.rateLimiter = new RateLimiter(Duration.ofSeconds(1), 5);
   }
 
   @Override
@@ -128,6 +133,11 @@ private void submitReconciliationExecution(ResourceID resourceID) {
       Optional<R> latest = cache.get(resourceID);
       latest.ifPresent(MDCUtils::addResourceInfo);
       if (!controllerUnderExecution && latest.isPresent()) {
+        var rateLimiterPermission = rateLimiter.acquirePermission(resourceID);
+        if (rateLimiterPermission.isPresent()) {
+          handleRateLimitedSubmission(resourceID, rateLimiterPermission.get());
+          return;
+        }
         setUnderExecutionProcessing(resourceID);
         final var retryInfo = retryInfo(resourceID);
         ExecutionScope<R> executionScope = new ExecutionScope<>(latest.get(), retryInfo);
@@ -193,6 +203,13 @@ private boolean isResourceMarkedForDeletion(ResourceEvent resourceEvent) {
     return resourceEvent.getResource().map(HasMetadata::isMarkedForDeletion).orElse(false);
   }
 
+  private void handleRateLimitedSubmission(ResourceID resourceID, Duration minimalDuration) {
+    var minimalDurationMillis = minimalDuration.toMillis();
+    log.debug("Rate limited resource: {}, rescheduled in {} millis", resourceID,
+        minimalDurationMillis);
+    retryEventSource().scheduleOnce(resourceID, minimalDurationMillis);
+  }
+
   private RetryInfo retryInfo(ResourceID resourceID) {
     return retryState.get(resourceID);
   }
@@ -251,11 +268,10 @@ private void reScheduleExecutionIfInstructed(
     postExecutionControl
         .getReScheduleDelay()
         .ifPresent(delay -> {
-          if (log.isDebugEnabled()) {
-            log.debug("ReScheduling event for resource: {} with delay: {}",
-                ResourceID.fromResource(customResource), delay);
-          }
-          retryEventSource().scheduleOnce(customResource, delay);
+          var resourceID = ResourceID.fromResource(customResource);
+          log.debug("ReScheduling event for resource: {} with delay: {}",
+              resourceID, delay);
+          retryEventSource().scheduleOnce(resourceID, delay);
         });
   }
 
@@ -289,7 +305,7 @@ private void handleRetryOnException(
               delay,
               resourceID);
           metrics.failedReconciliation(resourceID, exception);
-          retryEventSource().scheduleOnce(executionScope.getResource(), delay);
+          retryEventSource().scheduleOnce(resourceID, delay);
         },
         () -> log.error("Exhausted retries for {}", executionScope));
   }

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

@@ -25,16 +25,16 @@ public class TimerEventSource<R extends HasMetadata>
   private final Map<ResourceID, EventProducerTimeTask> onceTasks = new ConcurrentHashMap<>();
 
 
-  public void scheduleOnce(R resource, long delay) {
+  public void scheduleOnce(ResourceID resourceID, long delay) {
     if (!running.get()) {
       throw new IllegalStateException("The TimerEventSource is not running");
     }
-    ResourceID resourceUid = ResourceID.fromResource(resource);
-    if (onceTasks.containsKey(resourceUid)) {
-      cancelOnceSchedule(resourceUid);
+
+    if (onceTasks.containsKey(resourceID)) {
+      cancelOnceSchedule(resourceID);
     }
-    EventProducerTimeTask task = new EventProducerTimeTask(resourceUid);
-    onceTasks.put(resourceUid, task);
+    EventProducerTimeTask task = new EventProducerTimeTask(resourceID);
+    onceTasks.put(resourceID, task);
     timer.schedule(task, delay);
   }
 

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

@@ -105,7 +105,8 @@ void schedulesAnEventRetryOnException() {
     eventProcessorWithRetry.eventProcessingFinished(executionScope, postExecutionControl);
 
     verify(retryTimerEventSourceMock, times(1))
-        .scheduleOnce(eq(customResource), eq(RetryConfiguration.DEFAULT_INITIAL_INTERVAL));
+        .scheduleOnce(eq(ResourceID.fromResource(customResource)),
+            eq(RetryConfiguration.DEFAULT_INITIAL_INTERVAL));
   }
 
   @Test
@@ -136,7 +137,8 @@ void executesTheControllerInstantlyAfterErrorIfNewEventsReceived() {
     List<ExecutionScope> allValues = executionScopeArgumentCaptor.getAllValues();
     assertThat(allValues).hasSize(2);
     verify(retryTimerEventSourceMock, never())
-        .scheduleOnce(eq(customResource), eq(RetryConfiguration.DEFAULT_INITIAL_INTERVAL));
+        .scheduleOnce(eq(ResourceID.fromResource(customResource)),
+            eq(RetryConfiguration.DEFAULT_INITIAL_INTERVAL));
   }
 
   @Test

operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/source/timer/TimerEventSourceTest.java

@@ -37,30 +37,30 @@ public void setup() {
 
   @Test
   public void schedulesOnce() {
-    TestCustomResource customResource = TestUtils.testCustomResource();
+    var resourceID = ResourceID.fromResource(TestUtils.testCustomResource());
 
-    source.scheduleOnce(customResource, PERIOD);
+    source.scheduleOnce(resourceID, PERIOD);
 
     untilAsserted(() -> assertThat(eventHandler.events).hasSize(1));
     untilAsserted(PERIOD * 2, 0, () -> assertThat(eventHandler.events).hasSize(1));
   }
 
   @Test
   public void canCancelOnce() {
-    TestCustomResource customResource = TestUtils.testCustomResource();
+    var resourceID = ResourceID.fromResource(TestUtils.testCustomResource());
 
-    source.scheduleOnce(customResource, PERIOD);
-    source.cancelOnceSchedule(ResourceID.fromResource(customResource));
+    source.scheduleOnce(resourceID, PERIOD);
+    source.cancelOnceSchedule(resourceID);
 
     untilAsserted(() -> assertThat(eventHandler.events).isEmpty());
   }
 
   @Test
   public void canRescheduleOnceEvent() {
-    TestCustomResource customResource = TestUtils.testCustomResource();
+    var resourceID = ResourceID.fromResource(TestUtils.testCustomResource());
 
-    source.scheduleOnce(customResource, PERIOD);
-    source.scheduleOnce(customResource, 2 * PERIOD);
+    source.scheduleOnce(resourceID, PERIOD);
+    source.scheduleOnce(resourceID, 2 * PERIOD);
 
     untilAsserted(PERIOD * 2, PERIOD, () -> assertThat(eventHandler.events).hasSize(1));
   }
@@ -69,24 +69,24 @@ public void canRescheduleOnceEvent() {
   public void deRegistersOnceEventSources() {
     TestCustomResource customResource = TestUtils.testCustomResource();
 
-    source.scheduleOnce(customResource, PERIOD);
+    source.scheduleOnce(ResourceID.fromResource(customResource), PERIOD);
     source.onResourceDeleted(customResource);
 
     untilAsserted(() -> assertThat(eventHandler.events).isEmpty());
   }
 
   @Test
   public void eventNotRegisteredIfStopped() throws IOException {
-    TestCustomResource customResource = TestUtils.testCustomResource();
+    var resourceID = ResourceID.fromResource(TestUtils.testCustomResource());
 
     source.stop();
     assertThatExceptionOfType(IllegalStateException.class).isThrownBy(
-        () -> source.scheduleOnce(customResource, PERIOD));
+        () -> source.scheduleOnce(resourceID, PERIOD));
   }
 
   @Test
   public void eventNotFiredIfStopped() throws IOException {
-    source.scheduleOnce(TestUtils.testCustomResource(), PERIOD);
+    source.scheduleOnce(ResourceID.fromResource(TestUtils.testCustomResource()), PERIOD);
     source.stop();
 
     untilAsserted(() -> assertThat(eventHandler.events).isEmpty());