fix: create state only on resource event (operator-framework#2899)

* fix: create state only on resource event

In general we cleanup caches on delete event, so although in practice this probably not causing
issues in theory it can happen that we receive events where there is no related custom resource.
In that case we would create state, what can lead to a memory leak.

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

Author: csviri

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

@@ -102,8 +102,16 @@ public synchronized void handleEvent(Event event) {
     try {
       log.debug("Received event: {}", event);
 
+      final var optionalState = resourceStateManager.getOrCreateOnResourceEvent(event);
+      if (optionalState.isEmpty()) {
+        log.debug(
+            "Skipping event, since no state present and it is not a resource event. Resource ID:"
+                + " {}",
+            event.getRelatedCustomResourceID());
+        return;
+      }
+      var state = optionalState.orElseThrow();
       final var resourceID = event.getRelatedCustomResourceID();
-      final var state = resourceStateManager.getOrCreate(event.getRelatedCustomResourceID());
       MDCUtils.addResourceIDInfo(resourceID);
       metrics.receivedEvent(event, metricsMetadata);
       handleEventMarking(event, state);

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

@@ -2,15 +2,33 @@
 
 import java.util.List;
 import java.util.Map;
+import java.util.Optional;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.stream.Collectors;
 
+import io.javaoperatorsdk.operator.processing.event.source.controller.ResourceEvent;
+
 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 Optional<ResourceState> getOrCreateOnResourceEvent(Event event) {
+    var resourceId = event.getRelatedCustomResourceID();
+    var state = states.get(event.getRelatedCustomResourceID());
+    if (state != null) {
+      return Optional.of(state);
+    }
+    if (event instanceof ResourceEvent) {
+      state = new ResourceState(resourceId);
+      states.put(resourceId, state);
+      return Optional.of(state);
+    } else {
+      return Optional.empty();
+    }
+  }
+
   public ResourceState getOrCreate(ResourceID resourceID) {
     return states.computeIfAbsent(resourceID, ResourceState::new);
   }

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

@@ -276,6 +276,30 @@ void cancelScheduleOnceEventsOnSuccessfulExecution() {
     verify(retryTimerEventSourceMock, times(1)).cancelOnceSchedule(eq(crID));
   }
 
+  @Test
+  void skipsGenericEventIfNoResourceEventReceivedBefore() {
+    var crID = new ResourceID("test-cr", TEST_NAMESPACE);
+    eventProcessor =
+        spy(
+            new EventProcessor(
+                controllerConfiguration(null, LinearRateLimiter.deactivatedRateLimiter()),
+                reconciliationDispatcherMock,
+                eventSourceManagerMock,
+                metricsMock));
+
+    verify(reconciliationDispatcherMock, timeout(100).times(0)).handleExecution(any());
+
+    eventProcessor.start();
+    eventProcessor.handleEvent(new Event(crID));
+
+    await()
+        .pollDelay(Duration.ofMillis(100))
+        .untilAsserted(
+            () -> {
+              verify(reconciliationDispatcherMock, never()).handleExecution(any());
+            });
+  }
+
   @Test
   void startProcessedMarkedEventReceivedBefore() {
     var crID = new ResourceID("test-cr", TEST_NAMESPACE);
@@ -287,7 +311,7 @@ void startProcessedMarkedEventReceivedBefore() {
                 eventSourceManagerMock,
                 metricsMock));
     when(controllerEventSourceMock.get(eq(crID))).thenReturn(Optional.of(testCustomResource()));
-    eventProcessor.handleEvent(new Event(crID));
+    eventProcessor.handleEvent(new ResourceEvent(ResourceAction.ADDED, crID, testCustomResource()));
 
     verify(reconciliationDispatcherMock, timeout(100).times(0)).handleExecution(any());
 

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

@@ -4,6 +4,10 @@
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 
+import io.javaoperatorsdk.operator.TestUtils;
+import io.javaoperatorsdk.operator.processing.event.source.controller.ResourceAction;
+import io.javaoperatorsdk.operator.processing.event.source.controller.ResourceEvent;
+
 import static org.assertj.core.api.Assertions.assertThat;
 
 class ResourceStateManagerTest {
@@ -87,4 +91,26 @@ public void listsResourceIDSWithEventsPresent() {
     assertThat(res).hasSize(1);
     assertThat(res.get(0).getId()).isEqualTo(sampleResourceID2);
   }
+
+  @Test
+  void createStateOnlyOnResourceEvent() {
+    var state = manager.getOrCreateOnResourceEvent(new Event(new ResourceID("newEvent")));
+
+    assertThat(state).isEmpty();
+
+    state =
+        manager.getOrCreateOnResourceEvent(
+            new ResourceEvent(
+                ResourceAction.ADDED, new ResourceID("newEvent"), TestUtils.testCustomResource()));
+
+    assertThat(state).isNotNull();
+  }
+
+  @Test
+  void createsOnlyResourceEventReturnsPreviouslyCreatedState() {
+    manager.getOrCreate(new ResourceID("newEvent"));
+
+    var res = manager.getOrCreateOnResourceEvent(new Event(new ResourceID("newEvent")));
+    assertThat(res).isNotNull();
+  }
 }