feat: primary resource caching for followup reconciliation(s)

operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/reconciler/PrimaryUpdateAndCacheUtils.java

@@ -0,0 +1,232 @@
+package io.javaoperatorsdk.operator.api.reconciler;
+
+import java.util.function.BiFunction;
+import java.util.function.UnaryOperator;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import io.fabric8.kubernetes.api.model.HasMetadata;
+import io.fabric8.kubernetes.client.KubernetesClient;
+import io.fabric8.kubernetes.client.dsl.base.PatchContext;
+import io.fabric8.kubernetes.client.dsl.base.PatchType;
+import io.javaoperatorsdk.operator.api.reconciler.support.PrimaryResourceCache;
+import io.javaoperatorsdk.operator.processing.event.ResourceID;
+
+public class PrimaryUpdateAndCacheUtils {
+
+  private PrimaryUpdateAndCacheUtils() {}
+
+  private static final Logger log = LoggerFactory.getLogger(PrimaryUpdateAndCacheUtils.class);
+
+  /**
+   * Makes sure that the up-to-date primary resource will be present during the next reconciliation.
+   * Using update (PUT) method.
+   *
+   * @param primary resource
+   * @param context of reconciliation
+   * @return updated resource
+   * @param <P> primary resource type
+   */
+  public static <P extends HasMetadata> P updateAndCacheStatusWith(P primary, Context<P> context) {
+    return patchAndCacheStatusWith(primary, context, (p, c) -> c.resource(primary).updateStatus());
+  }
+
+  /**
+   * Makes sure that the up-to-date primary resource will be present during the next reconciliation.
+   * Using JSON Merge patch.
+   *
+   * @param primary resource
+   * @param context of reconciliation
+   * @return updated resource
+   * @param <P> primary resource type
+   */
+  public static <P extends HasMetadata> P patchAndCacheStatusWith(P primary, Context<P> context) {
+    return patchAndCacheStatusWith(primary, context, (p, c) -> c.resource(primary).patchStatus());
+  }
+
+  /**
+   * Makes sure that the up-to-date primary resource will be present during the next reconciliation.
+   * Using JSON Patch.
+   *
+   * @param primary resource
+   * @param context of reconciliation
+   * @return updated resource
+   * @param <P> primary resource type
+   */
+  public static <P extends HasMetadata> P editAndCacheStatusWith(
+      P primary, Context<P> context, UnaryOperator<P> operation) {
+    return patchAndCacheStatusWith(
+        primary, context, (p, c) -> c.resource(primary).editStatus(operation));
+  }
+
+  /**
+   * Makes sure that the up-to-date primary resource will be present during the next reconciliation.
+   *
+   * @param primary resource
+   * @param context of reconciliation
+   * @param patch free implementation of cache - make sure you use optimistic locking during the
+   *     update
+   * @return the updated resource.
+   * @param <P> primary resource type
+   */
+  public static <P extends HasMetadata> P patchAndCacheStatusWith(
+      P primary, Context<P> context, BiFunction<P, KubernetesClient, P> patch) {
+    checkResourceVersionPresent(primary);
+    var updatedResource = patch.apply(primary, context.getClient());
+    context
+        .eventSourceRetriever()
+        .getControllerEventSource()
+        .handleRecentResourceUpdate(ResourceID.fromResource(primary), updatedResource, primary);
+    return updatedResource;
+  }
+
+  /**
+   * Makes sure that the up-to-date primary resource will be present during the next reconciliation.
+   * Using Server Side Apply.
+   *
+   * @param primary resource
+   * @param freshResourceWithStatus - fresh resource with target state
+   * @param context of reconciliation
+   * @return the updated resource.
+   * @param <P> primary resource type
+   */
+  public static <P extends HasMetadata> P ssaPatchAndCacheStatusWith(
+      P primary, P freshResourceWithStatus, Context<P> context) {
+    checkResourceVersionPresent(freshResourceWithStatus);
+    var res =
+        context
+            .getClient()
+            .resource(freshResourceWithStatus)
+            .subresource("status")
+            .patch(
+                new PatchContext.Builder()
+                    .withForce(true)
+                    .withFieldManager(context.getControllerConfiguration().fieldManager())
+                    .withPatchType(PatchType.SERVER_SIDE_APPLY)
+                    .build());
+
+    context
+        .eventSourceRetriever()
+        .getControllerEventSource()
+        .handleRecentResourceUpdate(ResourceID.fromResource(primary), res, primary);
+    return res;
+  }
+
+  /**
+   * Patches the resource and adds it to the {@link PrimaryResourceCache} provided. Optimistic
+   * locking is not required.
+   *
+   * @param primary resource
+   * @param freshResourceWithStatus - fresh resource with target state
+   * @param context of reconciliation
+   * @param cache - resource cache managed by user
+   * @return the updated resource.
+   * @param <P> primary resource type
+   */
+  public static <P extends HasMetadata> P ssaPatchAndCacheStatus(
+      P primary, P freshResourceWithStatus, Context<P> context, PrimaryResourceCache<P> cache) {
+    logWarnIfResourceVersionPresent(freshResourceWithStatus);
+    return patchAndCacheStatus(
+        primary,
+        context.getClient(),
+        cache,
+        (P p, KubernetesClient c) ->
+            c.resource(freshResourceWithStatus)
+                .subresource("status")
+                .patch(
+                    new PatchContext.Builder()
+                        .withForce(true)
+                        .withFieldManager(context.getControllerConfiguration().fieldManager())
+                        .withPatchType(PatchType.SERVER_SIDE_APPLY)
+                        .build()));
+  }
+
+  /**
+   * Patches the resource with JSON Patch and adds it to the {@link PrimaryResourceCache} provided.
+   * Optimistic locking is not required.
+   *
+   * @param primary resource*
+   * @param context of reconciliation
+   * @param cache - resource cache managed by user
+   * @return the updated resource.
+   * @param <P> primary resource type
+   */
+  public static <P extends HasMetadata> P editAndCacheStatus(
+      P primary, Context<P> context, PrimaryResourceCache<P> cache, UnaryOperator<P> operation) {
+    logWarnIfResourceVersionPresent(primary);
+    return patchAndCacheStatus(
+        primary,
+        context.getClient(),
+        cache,
+        (P p, KubernetesClient c) -> c.resource(primary).editStatus(operation));
+  }
+
+  /**
+   * Patches the resource with JSON Merge patch and adds it to the {@link PrimaryResourceCache}
+   * provided. Optimistic locking is not required.
+   *
+   * @param primary resource*
+   * @param context of reconciliation
+   * @param cache - resource cache managed by user
+   * @return the updated resource.
+   * @param <P> primary resource type
+   */
+  public static <P extends HasMetadata> P patchAndCacheStatus(
+      P primary, Context<P> context, PrimaryResourceCache<P> cache) {
+    logWarnIfResourceVersionPresent(primary);
+    return patchAndCacheStatus(
+        primary,
+        context.getClient(),
+        cache,
+        (P p, KubernetesClient c) -> c.resource(primary).patchStatus());
+  }
+
+  /**
+   * Updates the resource and adds it to the {@link PrimaryResourceCache} provided. Optimistic
+   * locking is not required.
+   *
+   * @param primary resource*
+   * @param context of reconciliation
+   * @param cache - resource cache managed by user
+   * @return the updated resource.
+   * @param <P> primary resource type
+   */
+  public static <P extends HasMetadata> P updateAndCacheStatus(
+      P primary, Context<P> context, PrimaryResourceCache<P> cache) {
+    logWarnIfResourceVersionPresent(primary);
+    return patchAndCacheStatus(
+        primary,
+        context.getClient(),
+        cache,
+        (P p, KubernetesClient c) -> c.resource(primary).updateStatus());
+  }
+
+  public static <P extends HasMetadata> P patchAndCacheStatus(
+      P primary,
+      KubernetesClient client,
+      PrimaryResourceCache<P> cache,
+      BiFunction<P, KubernetesClient, P> patch) {
+    var updatedResource = patch.apply(primary, client);
+    cache.cacheResource(primary, updatedResource);
+    return updatedResource;
+  }
+
+  private static <P extends HasMetadata> void checkResourceVersionPresent(P primary) {
+    if (primary.getMetadata().getResourceVersion() == null) {
+      throw new IllegalStateException(
+          "Primary resource version is null, it is expected to set resource version for updates for caching. Name: %s namespace: %s"
+              .formatted(primary.getMetadata().getName(), primary.getMetadata().getNamespace()));
+    }
+  }
+
+  private static <P extends HasMetadata> void logWarnIfResourceVersionPresent(P primary) {
+    if (primary.getMetadata().getResourceVersion() != null) {
+      log.warn(
+          "Primary resource version is NOT null, for caching with optimistic locking use"
+              + " alternative methods. Name: {} namespace: {}",
+          primary.getMetadata().getName(),
+          primary.getMetadata().getNamespace());
+    }
+  }
+}

operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/reconciler/support/PrimaryResourceCache.java

@@ -0,0 +1,65 @@
+package io.javaoperatorsdk.operator.api.reconciler.support;
+
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.function.BiPredicate;
+
+import io.fabric8.kubernetes.api.model.HasMetadata;
+import io.javaoperatorsdk.operator.processing.event.ResourceID;
+
+public class PrimaryResourceCache<P extends HasMetadata> {
+
+  private final BiPredicate<Pair<P>, P> evictionPredicate;
+  private final ConcurrentHashMap<ResourceID, Pair<P>> cache = new ConcurrentHashMap<>();
+
+  public PrimaryResourceCache(BiPredicate<Pair<P>, P> evictionPredicate) {
+    this.evictionPredicate = evictionPredicate;
+  }
+
+  public PrimaryResourceCache() {
+    this(new ResourceVersionParsingEvictionPredicate<>());
+  }
+
+  public void cacheResource(P afterUpdate) {
+    var resourceId = ResourceID.fromResource(afterUpdate);
+    cache.put(resourceId, new Pair<>(null, afterUpdate));
+  }
+
+  public void cacheResource(P beforeUpdate, P afterUpdate) {
+    var resourceId = ResourceID.fromResource(beforeUpdate);
+    cache.put(resourceId, new Pair<>(beforeUpdate, afterUpdate));
+  }
+
+  public P getFreshResource(P newVersion) {
+    var resourceId = ResourceID.fromResource(newVersion);
+    var pair = cache.get(resourceId);
+    if (pair == null) {
+      return newVersion;
+    }
+    if (!newVersion.getMetadata().getUid().equals(pair.afterUpdate().getMetadata().getUid())) {
+      cache.remove(resourceId);
+      return newVersion;
+    }
+    if (evictionPredicate.test(pair, newVersion)) {
+      cache.remove(resourceId);
+      return newVersion;
+    } else {
+      return pair.afterUpdate();
+    }
+  }
+
+  public void cleanup(P resource) {
+    cache.remove(ResourceID.fromResource(resource));
+  }
+
+  public record Pair<T extends HasMetadata>(T beforeUpdate, T afterUpdate) {}
+
+  /** This works in general, but it does not strictly follow the contract with k8s API */
+  public static class ResourceVersionParsingEvictionPredicate<T extends HasMetadata>
+      implements BiPredicate<Pair<T>, T> {
+    @Override
+    public boolean test(Pair<T> updatePair, T newVersion) {
+      return Long.parseLong(updatePair.afterUpdate().getMetadata().getResourceVersion())
+          <= Long.parseLong(newVersion.getMetadata().getResourceVersion());
+    }
+  }
+}

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

@@ -208,6 +208,7 @@ public Stream<? extends EventSource<?, P>> getEventSourcesStream() {
     return eventSources.flatMappedSources();
   }
 
+  @Override
   public ControllerEventSource<P> getControllerEventSource() {
     return eventSources.controllerEventSource();
   }

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

@@ -6,6 +6,7 @@
 import io.fabric8.kubernetes.api.model.HasMetadata;
 import io.javaoperatorsdk.operator.api.reconciler.EventSourceContext;
 import io.javaoperatorsdk.operator.processing.event.source.EventSource;
+import io.javaoperatorsdk.operator.processing.event.source.controller.ControllerEventSource;
 
 public interface EventSourceRetriever<P extends HasMetadata> {
 
@@ -17,6 +18,8 @@ default <R> EventSource<R, P> getEventSourceFor(Class<R> dependentType) {
 
   <R> List<EventSource<R, P>> getEventSourcesFor(Class<R> dependentType);
 
+  ControllerEventSource<P> getControllerEventSource();
+
   /**
    * Registers (and starts) the specified {@link EventSource} dynamically during the reconciliation.
    * If an EventSource is already registered with the specified name, the registration will be