feat: Dependent Resources for External Resources

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

@@ -5,6 +5,7 @@
 
 import io.fabric8.kubernetes.api.model.HasMetadata;
 import io.javaoperatorsdk.operator.api.reconciler.Context;
+import io.javaoperatorsdk.operator.processing.event.ResourceID;
 
 public abstract class AbstractDependentResource<R, P extends HasMetadata>
     implements DependentResource<R, P> {
@@ -13,21 +14,25 @@ public abstract class AbstractDependentResource<R, P extends HasMetadata>
   private final boolean creatable = this instanceof Creator;
   private final boolean updatable = this instanceof Updater;
   private final boolean deletable = this instanceof Deleter;
+  private final boolean filteringEventSource;
+  private final boolean cachingEventSource;
   protected Creator<R, P> creator;
   protected Updater<R, P> updater;
   protected Deleter<P> deleter;
 
   @SuppressWarnings("unchecked")
   public AbstractDependentResource() {
-    init(Creator.NOOP, Updater.NOOP, Deleter.NOOP);
-  }
-
-  @SuppressWarnings({"unchecked"})
-  protected void init(Creator<R, P> defaultCreator, Updater<R, P> defaultUpdater,
-      Deleter<P> defaultDeleter) {
-    creator = creatable ? (Creator<R, P>) this : defaultCreator;
-    updater = updatable ? (Updater<R, P>) this : defaultUpdater;
-    deleter = deletable ? (Deleter<P>) this : defaultDeleter;
+    if (this instanceof EventSourceProvider) {
+      final var eventSource = ((EventSourceProvider<P>) this).getEventSource();
+      filteringEventSource = eventSource instanceof RecentOperationEventFilter;
+      cachingEventSource = eventSource instanceof RecentOperationCacheFiller;
+    } else {
+      filteringEventSource = false;
+      cachingEventSource = false;
+    }
+    creator = creatable ? (Creator<R, P>) this : null;
+    updater = updatable ? (Updater<R, P>) this : null;
+    deleter = deletable ? (Deleter<P>) this : null;
   }
 
   @Override
@@ -40,7 +45,7 @@ public void reconcile(P primary, Context context) {
         if (creatable) {
           var desired = desired(primary, context);
           log.debug("Creating dependent {} for primary {}", desired, primary);
-          creator.create(desired, primary, context);
+          handleCreate(desired, primary, context);
         }
       } else {
         final var actual = maybeActual.get();
@@ -49,7 +54,7 @@ public void reconcile(P primary, Context context) {
           if (!match.matched()) {
             final var desired = match.computedDesired().orElse(desired(primary, context));
             log.debug("Updating dependent {} for primary {}", desired, primary);
-            updater.update(actual, desired, primary, context);
+            handleUpdate(actual, desired, primary, context);
           }
         } else {
           log.debug("Update skipped for dependent {} as it matched the existing one", actual);
@@ -62,8 +67,71 @@ public void reconcile(P primary, Context context) {
     }
   }
 
+  protected void handleCreate(R desired, P primary, Context context) {
+    ResourceID resourceID = ResourceID.fromResource(primary);
+    R created = null;
+    try {
+      prepareEventFiltering(desired, resourceID);
+      created = creator.create(desired, primary, context);
+      cacheAfterCreate(resourceID, created);
+    } catch (RuntimeException e) {
+      cleanupAfterEventFiltering(desired, resourceID, created);
+      throw e;
+    }
+  }
+
+  private void cleanupAfterEventFiltering(R desired, ResourceID resourceID, R created) {
+    if (filteringEventSource) {
+      eventSourceAsRecentOperationEventFilter()
+          .cleanupOnCreateOrUpdateEventFiltering(resourceID, created);
+    }
+  }
+
+  private void cacheAfterCreate(ResourceID resourceID, R created) {
+    if (cachingEventSource) {
+      eventSourceAsRecentOperationCacheFiller().handleRecentResourceCreate(resourceID, created);
+    }
+  }
+
+  private void cacheAfterUpdate(R actual, ResourceID resourceID, R updated) {
+    if (cachingEventSource) {
+      eventSourceAsRecentOperationCacheFiller().handleRecentResourceUpdate(resourceID, updated,
+          actual);
+    }
+  }
+
+  private void prepareEventFiltering(R desired, ResourceID resourceID) {
+    if (filteringEventSource) {
+      eventSourceAsRecentOperationEventFilter().prepareForCreateOrUpdateEventFiltering(resourceID,
+          desired);
+    }
+  }
+
+  protected void handleUpdate(R actual, R desired, P primary, Context context) {
+    ResourceID resourceID = ResourceID.fromResource(primary);
+    R updated = null;
+    try {
+      prepareEventFiltering(desired, resourceID);
+      updated = updater.update(actual, desired, primary, context);
+      cacheAfterUpdate(actual, resourceID, updated);
+    } catch (RuntimeException e) {
+      cleanupAfterEventFiltering(desired, resourceID, updated);
+      throw e;
+    }
+  }
+
+  @SuppressWarnings("unchecked")
+  private RecentOperationEventFilter<R> eventSourceAsRecentOperationEventFilter() {
+    return (RecentOperationEventFilter<R>) ((EventSourceProvider<P>) this).getEventSource();
+  }
+
+  @SuppressWarnings("unchecked")
+  private RecentOperationCacheFiller<R> eventSourceAsRecentOperationCacheFiller() {
+    return (RecentOperationCacheFiller<R>) ((EventSourceProvider<P>) this).getEventSource();
+  }
+
   @Override
-  public void delete(P primary, Context context) {
+  public void cleanup(P primary, Context context) {
     if (isDeletable(primary, context)) {
       deleter.delete(primary, context);
     }

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

@@ -3,10 +3,7 @@
 import io.fabric8.kubernetes.api.model.HasMetadata;
 import io.javaoperatorsdk.operator.api.reconciler.Context;
 
-@SuppressWarnings("rawtypes")
+@FunctionalInterface
 public interface Creator<R, P extends HasMetadata> {
-  Creator NOOP = (desired, primary, context) -> {
-  };
-
-  void create(R desired, P primary, Context context);
+  R create(R desired, P primary, Context context);
 }

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

@@ -3,10 +3,7 @@
 import io.fabric8.kubernetes.api.model.HasMetadata;
 import io.javaoperatorsdk.operator.api.reconciler.Context;
 
-@SuppressWarnings("rawtypes")
+@FunctionalInterface
 public interface Deleter<P extends HasMetadata> {
-  Deleter NOOP = (primary, context) -> {
-  };
-
   void delete(P primary, Context context);
 }

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

@@ -8,7 +8,7 @@
 public interface DependentResource<R, P extends HasMetadata> {
   void reconcile(P primary, Context context);
 
-  default void delete(P primary, Context context) {}
+  default void cleanup(P primary, Context context) {}
 
   Optional<R> getResource(P primaryResource);
 }

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

@@ -0,0 +1,19 @@
+package io.javaoperatorsdk.operator.api.reconciler.dependent;
+
+import io.fabric8.kubernetes.api.model.HasMetadata;
+import io.javaoperatorsdk.operator.api.reconciler.Context;
+
+public class DesiredEqualsMatcher<R, P extends HasMetadata> implements Matcher<R, P> {
+
+  private final AbstractDependentResource<R, P> abstractDependentResource;
+
+  public DesiredEqualsMatcher(AbstractDependentResource<R, P> abstractDependentResource) {
+    this.abstractDependentResource = abstractDependentResource;
+  }
+
+  @Override
+  public Result<R> match(R actualResource, P primary, Context context) {
+    var desired = abstractDependentResource.desired(primary, context);
+    return Result.computed(actualResource.equals(desired), desired);
+  }
+}

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

@@ -5,6 +5,11 @@
 import io.javaoperatorsdk.operator.processing.event.source.EventSource;
 
 public interface EventSourceProvider<P extends HasMetadata> {
+  /**
+   * @param context - event source context where the event source is initialized
+   * @return the initiated event source.
+   */
+  EventSource initEventSource(EventSourceContext<P> context);
 
-  EventSource eventSource(EventSourceContext<P> context);
+  EventSource getEventSource();
 }

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

@@ -0,0 +1,10 @@
+package io.javaoperatorsdk.operator.api.reconciler.dependent;
+
+import io.javaoperatorsdk.operator.processing.event.ResourceID;
+
+public interface RecentOperationCacheFiller<R> {
+
+  void handleRecentResourceCreate(ResourceID resourceID, R resource);
+
+  void handleRecentResourceUpdate(ResourceID resourceID, R resource, R previousResourceVersion);
+}

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

@@ -0,0 +1,11 @@
+package io.javaoperatorsdk.operator.api.reconciler.dependent;
+
+import io.javaoperatorsdk.operator.processing.event.ResourceID;
+
+public interface RecentOperationEventFilter<R> extends RecentOperationCacheFiller<R> {
+
+  void prepareForCreateOrUpdateEventFiltering(ResourceID resourceID, R resource);
+
+  void cleanupOnCreateOrUpdateEventFiltering(ResourceID resourceID, R resource);
+
+}

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

@@ -4,19 +4,8 @@
 import io.javaoperatorsdk.operator.api.reconciler.Context;
 import io.javaoperatorsdk.operator.api.reconciler.dependent.Matcher.Result;
 
-@SuppressWarnings("rawtypes")
 public interface Updater<R, P extends HasMetadata> {
-  Updater NOOP = new Updater() {
-    @Override
-    public void update(Object actual, Object desired, HasMetadata primary, Context context) {}
-
-    @Override
-    public Result match(Object actualResource, HasMetadata primary, Context context) {
-      return Result.nonComputed(true);
-    }
-  };
-
-  void update(R actual, R desired, P primary, Context context);
+  R update(R actual, R desired, P primary, Context context);
 
   Result<R> match(R actualResource, P primary, Context context);
 }

operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/DependentResourceManager.java

@@ -54,7 +54,7 @@ public List<EventSource> prepareEventSources(EventSourceContext<P> context) {
                   final var dependentResource = createAndConfigureFrom(drc, context.getClient());
                   if (dependentResource instanceof EventSourceProvider) {
                     EventSourceProvider provider = (EventSourceProvider) dependentResource;
-                    sources.add(provider.eventSource(context));
+                    sources.add(provider.initEventSource(context));
                   }
                   return dependentResource;
                 })
@@ -72,7 +72,7 @@ public UpdateControl<P> reconcile(P resource, Context context) {
   @Override
   public DeleteControl cleanup(P resource, Context context) {
     initContextIfNeeded(resource, context);
-    dependents.forEach(dependent -> dependent.delete(resource, context));
+    dependents.forEach(dependent -> dependent.cleanup(resource, context));
     return Reconciler.super.cleanup(resource, context);
   }
 

operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/external/AbstractCachingDependentResource.java

@@ -0,0 +1,34 @@
+package io.javaoperatorsdk.operator.processing.dependent.external;
+
+import java.util.Optional;
+
+import io.fabric8.kubernetes.api.model.HasMetadata;
+import io.javaoperatorsdk.operator.api.config.Utils;
+import io.javaoperatorsdk.operator.api.reconciler.dependent.AbstractDependentResource;
+import io.javaoperatorsdk.operator.api.reconciler.dependent.EventSourceProvider;
+import io.javaoperatorsdk.operator.processing.event.ExternalResourceCachingEventSource;
+import io.javaoperatorsdk.operator.processing.event.source.EventSource;
+
+public abstract class AbstractCachingDependentResource<R, P extends HasMetadata>
+    extends AbstractDependentResource<R, P> implements EventSourceProvider<P> {
+
+  protected ExternalResourceCachingEventSource<R, P> eventSource;
+
+  public Optional<R> fetchResource(P primaryResource) {
+    return eventSource.getAssociated(primaryResource);
+  }
+
+  @Override
+  public EventSource getEventSource() {
+    return eventSource;
+  }
+
+  protected Class<R> resourceType() {
+    return (Class<R>) Utils.getFirstTypeArgumentFromExtendedClass(getClass());
+  }
+
+  @Override
+  public Optional<R> getResource(P primaryResource) {
+    return eventSource.getAssociated(primaryResource);
+  }
+}

operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/external/AbstractSimpleDependentResource.java

@@ -0,0 +1,73 @@
+package io.javaoperatorsdk.operator.processing.dependent.external;
+
+import java.util.Optional;
+
+import io.fabric8.kubernetes.api.model.HasMetadata;
+import io.javaoperatorsdk.operator.api.reconciler.Context;
+import io.javaoperatorsdk.operator.api.reconciler.dependent.AbstractDependentResource;
+import io.javaoperatorsdk.operator.api.reconciler.dependent.DesiredEqualsMatcher;
+import io.javaoperatorsdk.operator.api.reconciler.dependent.Matcher;
+import io.javaoperatorsdk.operator.processing.event.ResourceID;
+import io.javaoperatorsdk.operator.processing.event.source.ConcurrentHashMapCache;
+import io.javaoperatorsdk.operator.processing.event.source.UpdatableCache;
+
+/** A base class for external dependent resources that don't have an event source. */
+public abstract class AbstractSimpleDependentResource<R, P extends HasMetadata>
+    extends AbstractDependentResource<R, P> {
+
+  // cache serves only to keep the resource readable again until next reconciliation when the
+  // new resource is read again.
+  protected final UpdatableCache<R> cache;
+  protected Matcher<R, P> matcher;
+
+  public AbstractSimpleDependentResource() {
+    this(new ConcurrentHashMapCache<>());
+  }
+
+  public AbstractSimpleDependentResource(UpdatableCache<R> cache) {
+    this.cache = cache;
+    initMatcher();
+  }
+
+  @Override
+  public Optional<R> getResource(HasMetadata primaryResource) {
+    return cache.get(ResourceID.fromResource(primaryResource));
+  }
+
+  /** Actually read the resource from the target API */
+  public abstract Optional<R> fetchResource(HasMetadata primaryResource);
+
+  @Override
+  public void reconcile(P primary, Context context) {
+    var resourceId = ResourceID.fromResource(primary);
+    Optional<R> resource = fetchResource(primary);
+    resource.ifPresentOrElse(r -> cache.put(resourceId, r), () -> cache.remove(resourceId));
+    super.reconcile(primary, context);
+  }
+
+  public void cleanup(P primary, Context context) {
+    super.cleanup(primary, context);
+    cache.remove(ResourceID.fromResource(primary));
+  }
+
+  @Override
+  protected void handleCreate(R desired, P primary, Context context) {
+    var res = this.creator.create(desired, primary, context);
+    cache.put(ResourceID.fromResource(primary), res);
+  }
+
+  @Override
+  protected void handleUpdate(R actual, R desired, P primary, Context context) {
+    var res = updater.update(actual, desired, primary, context);
+    cache.put(ResourceID.fromResource(primary), res);
+  }
+
+  public Matcher.Result<R> match(R actualResource, P primary, Context context) {
+    return matcher.match(actualResource, primary, context);
+  }
+
+  protected void initMatcher() {
+    matcher = new DesiredEqualsMatcher<>(this);
+  }
+
+}