refactor: manage state associated with a resource in a single spot

Fixes #1314

Author: metacosm

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

@@ -1,11 +1,7 @@
 package io.javaoperatorsdk.operator.processing.event;
 
 import java.time.Duration;
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.Map;
 import java.util.Optional;
-import java.util.Set;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.ScheduledThreadPoolExecutor;
 
@@ -22,6 +18,7 @@
 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.rate.RateLimiter.RateLimitState;
 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;
@@ -37,17 +34,17 @@ class EventProcessor<R extends HasMetadata> implements EventHandler, LifecycleAw
   private static final long MINIMAL_RATE_LIMIT_RESCHEDULE_DURATION = 50;
 
   private volatile boolean running;
-  private final Set<ResourceID> underProcessing = new HashSet<>();
   private final ReconciliationDispatcher<R> reconciliationDispatcher;
   private final Retry retry;
-  private final Map<ResourceID, RetryExecution> retryState = new HashMap<>();
   private final ExecutorService executor;
   private final String controllerName;
   private final Metrics metrics;
   private final Cache<R> cache;
   private final EventSourceManager<R> eventSourceManager;
   private final EventMarker eventMarker = new EventMarker();
-  private final RateLimiter rateLimiter;
+  private final RateLimiter<? extends RateLimitState> rateLimiter;
+
+  private final ResourceStateManager resourceStateManager = new ResourceStateManager();
 
   EventProcessor(EventSourceManager<R> eventSourceManager) {
     this(
@@ -61,6 +58,7 @@ class EventProcessor<R extends HasMetadata> implements EventHandler, LifecycleAw
         eventSourceManager);
   }
 
+  @SuppressWarnings("rawtypes")
   EventProcessor(
       ReconciliationDispatcher<R> reconciliationDispatcher,
       EventSourceManager<R> eventSourceManager,
@@ -79,6 +77,7 @@ class EventProcessor<R extends HasMetadata> implements EventHandler, LifecycleAw
         eventSourceManager);
   }
 
+  @SuppressWarnings({"rawtypes", "unchecked"})
   private EventProcessor(
       Cache<R> cache,
       ExecutorService executor,
@@ -137,7 +136,13 @@ private void submitReconciliationExecution(ResourceID resourceID) {
       Optional<R> latest = cache.get(resourceID);
       latest.ifPresent(MDCUtils::addResourceInfo);
       if (!controllerUnderExecution && latest.isPresent()) {
-        var rateLimiterPermission = rateLimiter.acquirePermission(resourceID);
+        final var resourceState = resourceStateManager.getOrCreate(resourceID);
+        var rateLimit = resourceState.getRateLimit();
+        if (rateLimit == null) {
+          rateLimit = rateLimiter.initState();
+          resourceState.setRateLimit(rateLimit);
+        }
+        var rateLimiterPermission = rateLimiter.isLimited(rateLimit);
         if (rateLimiterPermission.isPresent()) {
           handleRateLimitedSubmission(resourceID, rateLimiterPermission.get());
           return;
@@ -216,7 +221,7 @@ private void handleRateLimitedSubmission(ResourceID resourceID, Duration minimal
   }
 
   private RetryInfo retryInfo(ResourceID resourceID) {
-    return retryState.get(resourceID);
+    return resourceStateManager.getOrCreate(resourceID).getRetry();
   }
 
   synchronized void eventProcessingFinished(
@@ -319,37 +324,38 @@ private void cleanupOnSuccessfulExecution(ExecutionScope<R> executionScope) {
     log.debug(
         "Cleanup for successful execution for resource: {}", getName(executionScope.getResource()));
     if (isRetryConfigured()) {
-      retryState.remove(executionScope.getResourceID());
+      resourceStateManager.getOrCreate(executionScope.getResourceID()).setRetry(null);
     }
     retryEventSource().cancelOnceSchedule(executionScope.getResourceID());
   }
 
   private RetryExecution getOrInitRetryExecution(ExecutionScope<R> executionScope) {
-    RetryExecution retryExecution = retryState.get(executionScope.getResourceID());
+    final var state = resourceStateManager.getOrCreate(executionScope.getResourceID());
+    RetryExecution retryExecution = state.getRetry();
     if (retryExecution == null) {
       retryExecution = retry.initExecution();
-      retryState.put(executionScope.getResourceID(), retryExecution);
+      state.setRetry(retryExecution);
     }
     return retryExecution;
   }
 
   private void cleanupForDeletedEvent(ResourceID resourceID) {
     log.debug("Cleaning up for delete event for: {}", resourceID);
     eventMarker.cleanup(resourceID);
-    rateLimiter.clear(resourceID);
+    resourceStateManager.remove(resourceID);
     metrics.cleanupDoneFor(resourceID);
   }
 
   private boolean isControllerUnderExecution(ResourceID resourceID) {
-    return underProcessing.contains(resourceID);
+    return resourceStateManager.getOrCreate(resourceID).isUnderProcessing();
   }
 
   private void setUnderExecutionProcessing(ResourceID resourceID) {
-    underProcessing.add(resourceID);
+    resourceStateManager.getOrCreate(resourceID).setUnderProcessing(true);
   }
 
   private void unsetUnderExecution(ResourceID resourceID) {
-    underProcessing.remove(resourceID);
+    resourceStateManager.getOrCreate(resourceID).setUnderProcessing(false);
   }
 
   private boolean isRetryConfigured() {
@@ -405,6 +411,6 @@ public String toString() {
   }
 
   public synchronized boolean isUnderProcessing(ResourceID resourceID) {
-    return underProcessing.contains(resourceID);
+    return isControllerUnderExecution(resourceID);
   }
 }

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

@@ -0,0 +1,46 @@
+package io.javaoperatorsdk.operator.processing.event;
+
+import io.javaoperatorsdk.operator.processing.event.EventMarker.EventingState;
+import io.javaoperatorsdk.operator.processing.event.rate.RateLimiter.RateLimitState;
+import io.javaoperatorsdk.operator.processing.retry.RetryExecution;
+
+class ResourceState {
+  private final ResourceID id;
+
+  private boolean underProcessing;
+  private RetryExecution retry;
+  private EventingState eventing;
+  private RateLimitState rateLimit;
+
+  public ResourceState(ResourceID id) {
+    this.id = id;
+  }
+
+  public ResourceID getId() {
+    return id;
+  }
+
+  public RateLimitState getRateLimit() {
+    return rateLimit;
+  }
+
+  public void setRateLimit(RateLimitState rateLimit) {
+    this.rateLimit = rateLimit;
+  }
+
+  public RetryExecution getRetry() {
+    return retry;
+  }
+
+  public void setRetry(RetryExecution retry) {
+    this.retry = retry;
+  }
+
+  public boolean isUnderProcessing() {
+    return underProcessing;
+  }
+
+  public void setUnderProcessing(boolean underProcessing) {
+    this.underProcessing = underProcessing;
+  }
+}

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

@@ -0,0 +1,24 @@
+package io.javaoperatorsdk.operator.processing.event;
+
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
+class ResourceStateManager {
+  // maybe we should have a way for users to specify a hint on the amount of CRs their reconciler
+  // will process to avoid under- or over-sizing the state maps and avoid too many resizing that
+  // take time and memory?
+  private final Map<ResourceID, ResourceState> states = new ConcurrentHashMap<>(100);
+
+
+  public ResourceState getOrCreate(ResourceID resourceID) {
+    return states.computeIfAbsent(resourceID, ResourceState::new);
+  }
+
+  public void remove(ResourceID resourceID) {
+    states.remove(resourceID);
+  }
+
+  public boolean contains(ResourceID resourceID) {
+    return states.containsKey(resourceID);
+  }
+}

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

@@ -2,16 +2,12 @@
 
 import java.time.Duration;
 import java.time.LocalDateTime;
-import java.util.HashMap;
-import java.util.Map;
 import java.util.Optional;
 
-import io.javaoperatorsdk.operator.processing.event.ResourceID;
-
 /**
  * A Simple rate limiter that limits the number of permission for a time interval.
  */
-public class PeriodRateLimiter implements RateLimiter {
+public class PeriodRateLimiter implements RateLimiter<RateState> {
 
   public static final int DEFAULT_REFRESH_PERIOD_SECONDS = 10;
   public static final int DEFAULT_LIMIT_FOR_PERIOD = 3;
@@ -24,8 +20,6 @@ public class PeriodRateLimiter implements RateLimiter {
   private Duration refreshPeriod;
   private int limitForPeriod;
 
-  private Map<ResourceID, RateState> limitData = new HashMap<>();
-
   public PeriodRateLimiter() {
     this(DEFAULT_REFRESH_PERIOD, DEFAULT_LIMIT_FOR_PERIOD);
   }
@@ -36,11 +30,12 @@ public PeriodRateLimiter(Duration refreshPeriod, int limitForPeriod) {
   }
 
   @Override
-  public Optional<Duration> acquirePermission(ResourceID resourceID) {
-    if (limitForPeriod <= 0) {
+  public Optional<Duration> isLimited(RateLimitState rateLimitState) {
+    if (limitForPeriod <= 0 || !(rateLimitState instanceof RateState)) {
       return Optional.empty();
     }
-    var actualState = limitData.computeIfAbsent(resourceID, r -> RateState.initialState());
+
+    var actualState = (RateState) rateLimitState;
     if (actualState.getCount() < limitForPeriod) {
       actualState.increaseCount();
       return Optional.empty();
@@ -55,7 +50,7 @@ public Optional<Duration> acquirePermission(ResourceID resourceID) {
   }
 
   @Override
-  public void clear(ResourceID resourceID) {
-    limitData.remove(resourceID);
+  public RateState initState() {
+    return RateState.initialState();
   }
 }

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

@@ -3,20 +3,17 @@
 import java.time.Duration;
 import java.util.Optional;
 
-import io.javaoperatorsdk.operator.processing.event.ResourceID;
+import io.javaoperatorsdk.operator.processing.event.rate.RateLimiter.RateLimitState;
 
-public interface RateLimiter {
+public interface RateLimiter<S extends RateLimitState> {
+  interface RateLimitState {
+  }
 
   /**
-   * @param resourceID id of the resource
    * @return empty if permission acquired or minimal duration until a permission could be acquired
    *         again
    */
-  Optional<Duration> acquirePermission(ResourceID resourceID);
-
-  /**
-   * Cleanup state. Called when resource is deleted.
-   */
-  void clear(ResourceID resourceID);
+  Optional<Duration> isLimited(RateLimitState rateLimitState);
 
+  S initState();
 }

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

@@ -2,7 +2,9 @@
 
 import java.time.LocalDateTime;
 
-class RateState {
+import io.javaoperatorsdk.operator.processing.event.rate.RateLimiter.RateLimitState;
+
+class RateState implements RateLimitState {
 
   private LocalDateTime lastRefreshTime;
   private int count;

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

@@ -19,6 +19,7 @@
 import io.javaoperatorsdk.operator.api.monitoring.Metrics;
 import io.javaoperatorsdk.operator.processing.event.rate.PeriodRateLimiter;
 import io.javaoperatorsdk.operator.processing.event.rate.RateLimiter;
+import io.javaoperatorsdk.operator.processing.event.rate.RateLimiter.RateLimitState;
 import io.javaoperatorsdk.operator.processing.event.source.controller.ControllerResourceEventSource;
 import io.javaoperatorsdk.operator.processing.event.source.controller.ResourceAction;
 import io.javaoperatorsdk.operator.processing.event.source.controller.ResourceEvent;
@@ -78,7 +79,7 @@ void setup() {
     eventProcessorWithRetry.start();
     when(eventProcessor.retryEventSource()).thenReturn(retryTimerEventSourceMock);
     when(eventProcessorWithRetry.retryEventSource()).thenReturn(retryTimerEventSourceMock);
-    when(rateLimiterMock.acquirePermission(any())).thenReturn(Optional.empty());
+    when(rateLimiterMock.isLimited(any())).thenReturn(Optional.empty());
   }
 
   @Test
@@ -351,7 +352,9 @@ void rateLimitsReconciliationSubmission() {
     var refreshPeriod = Duration.ofMillis(100);
     var event = prepareCREvent();
 
-    when(rateLimiterMock.acquirePermission(event.getRelatedCustomResourceID()))
+    final var rateLimit = new RateLimitState() {};
+    when(rateLimiterMock.initState()).thenReturn(rateLimit);
+    when(rateLimiterMock.isLimited(rateLimit))
         .thenReturn(Optional.empty())
         .thenReturn(Optional.of(refreshPeriod));