fix: primary cache utils mechanism

csviri · csviri · commit 56cda35ac9ce · 2025-05-21T16:54:02.000+02:00
Signed-off-by: Attila Mészáros &lt;a_meszaros@apple.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 RESOURCE_CACHE_POLL_TIMEOUT = 10000;
+  public static final int DEFAULT_SLEEP_FOR_CACHE_POLL_MILLIS = 30;
 
   private PrimaryUpdateAndCacheUtils() {}
 
@@ -90,8 +94,8 @@ 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}.
    *
    * @param resourceToUpdate original resource to update
    * @param context of reconciliation
@@ -106,7 +110,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,
+        RESOURCE_CACHE_POLL_TIMEOUT,
+        DEFAULT_SLEEP_FOR_CACHE_POLL_MILLIS);
   }
 
   /**
@@ -133,7 +143,9 @@ public static <P extends HasMetadata> P updateAndCacheResource(
       Context<P> context,
       UnaryOperator<P> modificationFunction,
       UnaryOperator<P> updateMethod,
-      int maxRetry) {
+      int maxRetry,
+      long cachePollTimeoutMillis,
+      long pollDelayMillis) {
 
     if (log.isDebugEnabled()) {
       log.debug("Conflict retrying update for: {}", ResourceID.fromResource(resourceToUpdate));
@@ -180,14 +192,35 @@ 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, pollDelayMillis);
+      }
+    }
+  }
+
+  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();
+      while (startTime.plus(timeoutMillis, ChronoUnit.MILLIS).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`	`}`