fix: primary cache utils mechanism (#2814)

csviri · metacosm · web-flow · commit f51c65b7b8ac · 2025-05-23T12:40:43.000+02:00
Reading from API server was not correct, this works in all cases only if the informer cache has the up to date resources.
If we don't have the up to date resource in the cache, and don't do the update based on that, we cannot say for sure if we can remove the resource for the next event or not from overlay cache.

Signed-off-by: Attila Mészáros &lt;a_meszaros@apple.com&gt;
Signed-off-by: Chris Laprun &lt;claprun@redhat.com&gt;
Co-authored-by: Chris Laprun &lt;claprun@redhat.com&gt;
diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/reconciler/PrimaryUpdateAndCacheUtils.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/reconciler/PrimaryUpdateAndCacheUtils.java
@@ -1,5 +1,7 @@
 package io.javaoperatorsdk.operator.api.reconciler;
 
+import java.time.LocalTime;
+import java.time.temporal.ChronoUnit;
 import java.util.function.UnaryOperator;
 
 import org.slf4j.Logger;
@@ -25,6 +27,8 @@
 public class PrimaryUpdateAndCacheUtils {
 
   public static final int DEFAULT_MAX_RETRY = 10;
+  public static final int DEFAULT_RESOURCE_CACHE_TIMEOUT_MILLIS = 10000;
+  public static final int DEFAULT_RESOURCE_CACHE_POLL_PERIOD_MILLIS = 50;
 
   private PrimaryUpdateAndCacheUtils() {}
 
@@ -90,8 +94,10 @@ public static <P extends HasMetadata> P ssaPatchStatusAndCacheResource(
   }
 
   /**
-   * Same as {@link #updateAndCacheResource(HasMetadata, Context, UnaryOperator, UnaryOperator,
-   * int)} using the default maximum retry number as defined by {@link #DEFAULT_MAX_RETRY}.
+   * Same as {@link #updateAndCacheResource(HasMetadata, Context, UnaryOperator, UnaryOperator, int,
+   * long,long)} using the default maximum retry number as defined by {@link #DEFAULT_MAX_RETRY} and
+   * default cache maximum polling time and period as defined, respectively by {@link
+   * #DEFAULT_RESOURCE_CACHE_TIMEOUT_MILLIS} and {@link #DEFAULT_RESOURCE_CACHE_POLL_PERIOD_MILLIS}.
    *
    * @param resourceToUpdate original resource to update
    * @param context of reconciliation
@@ -106,7 +112,13 @@ public static <P extends HasMetadata> P updateAndCacheResource(
       UnaryOperator<P> modificationFunction,
       UnaryOperator<P> updateMethod) {
     return updateAndCacheResource(
-        resourceToUpdate, context, modificationFunction, updateMethod, DEFAULT_MAX_RETRY);
+        resourceToUpdate,
+        context,
+        modificationFunction,
+        updateMethod,
+        DEFAULT_MAX_RETRY,
+        DEFAULT_RESOURCE_CACHE_TIMEOUT_MILLIS,
+        DEFAULT_RESOURCE_CACHE_POLL_PERIOD_MILLIS);
   }
 
   /**
@@ -124,16 +136,20 @@ public static <P extends HasMetadata> P updateAndCacheResource(
    * @param modificationFunction modifications to make on primary
    * @param updateMethod the update method implementation
    * @param maxRetry maximum number of retries before giving up
+   * @param cachePollTimeoutMillis maximum amount of milliseconds to wait for the updated resource
+   *     to appear in cache
+   * @param cachePollPeriodMillis cache polling period, in milliseconds
    * @param <P> primary type
    * @return the updated resource
    */
-  @SuppressWarnings("unchecked")
   public static <P extends HasMetadata> P updateAndCacheResource(
       P resourceToUpdate,
       Context<P> context,
       UnaryOperator<P> modificationFunction,
       UnaryOperator<P> updateMethod,
-      int maxRetry) {
+      int maxRetry,
+      long cachePollTimeoutMillis,
+      long cachePollPeriodMillis) {
 
     if (log.isDebugEnabled()) {
       log.debug("Conflict retrying update for: {}", ResourceID.fromResource(resourceToUpdate));
@@ -180,14 +196,37 @@ public static <P extends HasMetadata> P updateAndCacheResource(
             resourceToUpdate.getMetadata().getNamespace(),
             e.getCode());
         resourceToUpdate =
-            (P)
-                context
-                    .getClient()
-                    .resources(resourceToUpdate.getClass())
-                    .inNamespace(resourceToUpdate.getMetadata().getNamespace())
-                    .withName(resourceToUpdate.getMetadata().getName())
-                    .get();
+            pollLocalCache(
+                context, resourceToUpdate, cachePollTimeoutMillis, cachePollPeriodMillis);
+      }
+    }
+  }
+
+  private static <P extends HasMetadata> P pollLocalCache(
+      Context<P> context, P staleResource, long timeoutMillis, long pollDelayMillis) {
+    try {
+      var resourceId = ResourceID.fromResource(staleResource);
+      var startTime = LocalTime.now();
+      final var timeoutTime = startTime.plus(timeoutMillis, ChronoUnit.MILLIS);
+      while (timeoutTime.isAfter(LocalTime.now())) {
+        log.debug("Polling cache for resource: {}", resourceId);
+        var cachedResource = context.getPrimaryCache().get(resourceId).orElseThrow();
+        if (!cachedResource
+            .getMetadata()
+            .getResourceVersion()
+            .equals(staleResource.getMetadata().getResourceVersion())) {
+          return context
+              .getControllerConfiguration()
+              .getConfigurationService()
+              .getResourceCloner()
+              .clone(cachedResource);
+        }
+        Thread.sleep(pollDelayMillis);
       }
+      throw new OperatorException("Timeout of resource polling from cache for resource");
+    } catch (InterruptedException e) {
+      Thread.currentThread().interrupt();
+      throw new OperatorException(e);
     }
   }
 }
diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/informer/TemporaryResourceCache.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/informer/TemporaryResourceCache.java
@@ -126,9 +126,7 @@ public synchronized void putResource(T newResource, String previousResourceVersi
       knownResourceVersions.add(newResource.getMetadata().getResourceVersion());
     }
     var resourceId = ResourceID.fromResource(newResource);
-    var cachedResource =
-        getResourceFromCache(resourceId)
-            .orElse(managedInformerEventSource.get(resourceId).orElse(null));
+    var cachedResource = managedInformerEventSource.get(resourceId).orElse(null);
 
     boolean moveAhead = false;
     if (previousResourceVersion == null && cachedResource == null) {
diff --git a/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/api/reconciler/PrimaryUpdateAndCacheUtilsTest.java b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/api/reconciler/PrimaryUpdateAndCacheUtilsTest.java
@@ -1,16 +1,22 @@
 package io.javaoperatorsdk.operator.api.reconciler;
 
+import java.util.Optional;
 import java.util.function.UnaryOperator;
 
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 
+import io.fabric8.kubernetes.api.model.HasMetadata;
 import io.fabric8.kubernetes.client.KubernetesClient;
 import io.fabric8.kubernetes.client.KubernetesClientException;
 import io.fabric8.kubernetes.client.dsl.MixedOperation;
 import io.fabric8.kubernetes.client.dsl.Resource;
+import io.fabric8.kubernetes.client.utils.KubernetesSerialization;
 import io.javaoperatorsdk.operator.OperatorException;
 import io.javaoperatorsdk.operator.TestUtils;
+import io.javaoperatorsdk.operator.api.config.Cloner;
+import io.javaoperatorsdk.operator.api.config.ConfigurationService;
+import io.javaoperatorsdk.operator.api.config.ControllerConfiguration;
 import io.javaoperatorsdk.operator.processing.event.EventSourceRetriever;
 import io.javaoperatorsdk.operator.processing.event.source.controller.ControllerEventSource;
 import io.javaoperatorsdk.operator.sample.simple.TestCustomResource;
@@ -29,6 +35,7 @@ class PrimaryUpdateAndCacheUtilsTest {
   Context<TestCustomResource> context = mock(Context.class);
   KubernetesClient client = mock(KubernetesClient.class);
   Resource resource = mock(Resource.class);
+  IndexedResourceCache<TestCustomResource> primaryCache = mock(IndexedResourceCache.class);
 
   @BeforeEach
   void setup() {
@@ -41,6 +48,20 @@ void setup() {
     when(mixedOp.inNamespace(any())).thenReturn(mixedOp);
     when(mixedOp.withName(any())).thenReturn(resource);
     when(resource.get()).thenReturn(TestUtils.testCustomResource1());
+    when(context.getPrimaryCache()).thenReturn(primaryCache);
+
+    var controllerConfiguration = mock(ControllerConfiguration.class);
+    when(context.getControllerConfiguration()).thenReturn(controllerConfiguration);
+    var configService = mock(ConfigurationService.class);
+    when(controllerConfiguration.getConfigurationService()).thenReturn(configService);
+    when(configService.getResourceCloner())
+        .thenReturn(
+            new Cloner() {
+              @Override
+              public <R extends HasMetadata> R clone(R object) {
+                return new KubernetesSerialization().clone(object);
+              }
+            });
   }
 
   @Test
@@ -76,6 +97,10 @@ void retriesConflicts() {
     when(updateOperation.apply(any()))
         .thenThrow(new KubernetesClientException("", 409, null))
         .thenReturn(TestUtils.testCustomResource1());
+    var freshResource = TestUtils.testCustomResource1();
+
+    freshResource.getMetadata().setResourceVersion("2");
+    when(primaryCache.get(any())).thenReturn(Optional.of(freshResource));
 
     var updated =
         PrimaryUpdateAndCacheUtils.updateAndCacheResource(
@@ -89,15 +114,21 @@ void retriesConflicts() {
             updateOperation);
 
     assertThat(updated).isNotNull();
-    verify(resource, times(1)).get();
+    verify(primaryCache, times(1)).get(any());
   }
 
   @Test
   void throwsIfRetryExhausted() {
     var updateOperation = mock(UnaryOperator.class);
 
     when(updateOperation.apply(any())).thenThrow(new KubernetesClientException("", 409, null));
+    var stubbing = when(primaryCache.get(any()));
 
+    for (int i = 0; i < DEFAULT_MAX_RETRY; i++) {
+      var resource = TestUtils.testCustomResource1();
+      resource.getMetadata().setResourceVersion("" + i);
+      stubbing = stubbing.thenReturn(Optional.of(resource));
+    }
     assertThrows(
         OperatorException.class,
         () ->
@@ -106,6 +137,28 @@ void throwsIfRetryExhausted() {
                 context,
                 UnaryOperator.identity(),
                 updateOperation));
-    verify(resource, times(DEFAULT_MAX_RETRY)).get();
+    verify(primaryCache, times(DEFAULT_MAX_RETRY)).get(any());
+  }
+
+  @Test
+  void cachePollTimeouts() {
+    var updateOperation = mock(UnaryOperator.class);
+
+    when(updateOperation.apply(any())).thenThrow(new KubernetesClientException("", 409, null));
+    when(primaryCache.get(any())).thenReturn(Optional.of(TestUtils.testCustomResource1()));
+
+    var ex =
+        assertThrows(
+            OperatorException.class,
+            () ->
+                PrimaryUpdateAndCacheUtils.updateAndCacheResource(
+                    TestUtils.testCustomResource1(),
+                    context,
+                    UnaryOperator.identity(),
+                    updateOperation,
+                    2,
+                    50L,
+                    10L));
+    assertThat(ex.getMessage()).contains("Timeout");
   }
 }
diff --git a/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/statuscache/StatusPatchCacheCustomResource.java b/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/statuscache/StatusPatchCacheCustomResource.java
@@ -9,6 +9,5 @@
 @Group("sample.javaoperatorsdk")
 @Version("v1")
 @ShortNames("spwl")
-public class StatusPatchCacheWithLockCustomResource
-    extends CustomResource<StatusPatchCacheWithLockSpec, StatusPatchCacheWithLockStatus>
-    implements Namespaced {}
+public class StatusPatchCacheCustomResource
+    extends CustomResource<StatusPatchCacheSpec, StatusPatchCacheStatus> implements Namespaced {}
diff --git a/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/statuscache/StatusPatchCacheIT.java b/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/statuscache/StatusPatchCacheIT.java
@@ -11,19 +11,19 @@
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.awaitility.Awaitility.await;
 
-public class StatusPatchCacheWithLockIT {
+public class StatusPatchCacheIT {
 
   public static final String TEST_1 = "test1";
 
   @RegisterExtension
   LocallyRunOperatorExtension extension =
       LocallyRunOperatorExtension.builder()
-          .withReconciler(StatusPatchCacheWithLockReconciler.class)
+          .withReconciler(StatusPatchCacheReconciler.class)
           .build();
 
   @Test
   void testStatusAlwaysUpToDate() {
-    var reconciler = extension.getReconcilerOfType(StatusPatchCacheWithLockReconciler.class);
+    var reconciler = extension.getReconcilerOfType(StatusPatchCacheReconciler.class);
 
     extension.create(testResource());
 
@@ -39,10 +39,10 @@ void testStatusAlwaysUpToDate() {
             });
   }
 
-  StatusPatchCacheWithLockCustomResource testResource() {
-    var res = new StatusPatchCacheWithLockCustomResource();
+  StatusPatchCacheCustomResource testResource() {
+    var res = new StatusPatchCacheCustomResource();
     res.setMetadata(new ObjectMetaBuilder().withName(TEST_1).build());
-    res.setSpec(new StatusPatchCacheWithLockSpec());
+    res.setSpec(new StatusPatchCacheSpec());
     return res;
   }
 }
diff --git a/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/statuscache/StatusPatchCacheReconciler.java b/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/statuscache/StatusPatchCacheReconciler.java
@@ -12,16 +12,14 @@
 import io.javaoperatorsdk.operator.processing.event.source.EventSource;
 
 @ControllerConfiguration
-public class StatusPatchCacheWithLockReconciler
-    implements Reconciler<StatusPatchCacheWithLockCustomResource> {
+public class StatusPatchCacheReconciler implements Reconciler<StatusPatchCacheCustomResource> {
 
   public volatile int latestValue = 0;
   public volatile boolean errorPresent = false;
 
   @Override
-  public UpdateControl<StatusPatchCacheWithLockCustomResource> reconcile(
-      StatusPatchCacheWithLockCustomResource resource,
-      Context<StatusPatchCacheWithLockCustomResource> context) {
+  public UpdateControl<StatusPatchCacheCustomResource> reconcile(
+      StatusPatchCacheCustomResource resource, Context<StatusPatchCacheCustomResource> context) {
 
     if (resource.getStatus() != null && resource.getStatus().getValue() != latestValue) {
       errorPresent = true;
@@ -50,22 +48,20 @@ public UpdateControl<StatusPatchCacheWithLockCustomResource> reconcile(
   }
 
   @Override
-  public List<EventSource<?, StatusPatchCacheWithLockCustomResource>> prepareEventSources(
-      EventSourceContext<StatusPatchCacheWithLockCustomResource> context) {
+  public List<EventSource<?, StatusPatchCacheCustomResource>> prepareEventSources(
+      EventSourceContext<StatusPatchCacheCustomResource> context) {
     // periodic event triggering for testing purposes
     return List.of(new PeriodicTriggerEventSource<>(context.getPrimaryCache()));
   }
 
-  private StatusPatchCacheWithLockCustomResource createFreshCopy(
-      StatusPatchCacheWithLockCustomResource resource) {
-    var res = new StatusPatchCacheWithLockCustomResource();
+  private StatusPatchCacheCustomResource createFreshCopy(StatusPatchCacheCustomResource resource) {
+    var res = new StatusPatchCacheCustomResource();
     res.setMetadata(
         new ObjectMetaBuilder()
             .withName(resource.getMetadata().getName())
             .withNamespace(resource.getMetadata().getNamespace())
             .build());
-    res.setStatus(new StatusPatchCacheWithLockStatus());
-
+    res.setStatus(new StatusPatchCacheStatus());
     return res;
   }
 }
diff --git a/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/statuscache/StatusPatchCacheSpec.java b/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/statuscache/StatusPatchCacheSpec.java
@@ -1,6 +1,6 @@
 package io.javaoperatorsdk.operator.baseapi.statuscache;
 
-public class StatusPatchCacheWithLockSpec {
+public class StatusPatchCacheSpec {
 
   private int counter = 0;
 
diff --git a/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/statuscache/StatusPatchCacheStatus.java b/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/statuscache/StatusPatchCacheStatus.java
@@ -1,14 +1,14 @@
 package io.javaoperatorsdk.operator.baseapi.statuscache;
 
-public class StatusPatchCacheWithLockStatus {
+public class StatusPatchCacheStatus {
 
   private Integer value = 0;
 
   public Integer getValue() {
     return value;
   }
 
-  public StatusPatchCacheWithLockStatus setValue(Integer value) {
+  public StatusPatchCacheStatus setValue(Integer value) {
     this.value = value;
     return this;
   }

Original file line number	Diff line number	Diff line change
`@@ -1,14 +1,14 @@`
`1`	`1`	`package io.javaoperatorsdk.operator.baseapi.statuscache;`
`2`	`2`
`3`		`-public class StatusPatchCacheWithLockStatus {`
	`3`	`+public class StatusPatchCacheStatus {`
`4`	`4`
`5`	`5`	`private Integer value = 0;`
`6`	`6`
`7`	`7`	`public Integer getValue() {`
`8`	`8`	`return value;`
`9`	`9`	`}`
`10`	`10`
`11`		`- public StatusPatchCacheWithLockStatus setValue(Integer value) {`
	`11`	`+ public StatusPatchCacheStatus setValue(Integer value) {`
`12`	`12`	`this.value = value;`
`13`	`13`	`return this;`
`14`	`14`	`}`