feat: decouple event source from cache + list discriminator (#1378)

Author: csviri

operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/AnnotationControllerConfiguration.java

@@ -15,9 +15,8 @@
 import io.javaoperatorsdk.operator.OperatorException;
 import io.javaoperatorsdk.operator.ReconcilerUtils;
 import io.javaoperatorsdk.operator.api.config.dependent.DependentResourceSpec;
-import io.javaoperatorsdk.operator.api.reconciler.Constants;
+import io.javaoperatorsdk.operator.api.reconciler.*;
 import io.javaoperatorsdk.operator.api.reconciler.ControllerConfiguration;
-import io.javaoperatorsdk.operator.api.reconciler.Reconciler;
 import io.javaoperatorsdk.operator.api.reconciler.dependent.Dependent;
 import io.javaoperatorsdk.operator.api.reconciler.dependent.DependentResource;
 import io.javaoperatorsdk.operator.processing.dependent.kubernetes.KubernetesDependent;
@@ -48,9 +47,8 @@ public AnnotationControllerConfiguration(Reconciler<P> reconciler) {
     this.reconciler = reconciler;
     this.annotation = reconciler.getClass().getAnnotation(ControllerConfiguration.class);
     if (annotation == null) {
-      throw new OperatorException(
-          "Missing mandatory @" + ControllerConfiguration.class.getSimpleName() +
-              " annotation for reconciler:  " + reconciler);
+      throw new OperatorException("Missing mandatory @" + CONTROLLER_CONFIG_ANNOTATION +
+          " annotation for reconciler:  " + reconciler);
     }
   }
 
@@ -256,6 +254,7 @@ private Object createKubernetesResourceConfig(Class<? extends DependentResource>
     OnUpdateFilter<? extends HasMetadata> onUpdateFilter = null;
     OnDeleteFilter<? extends HasMetadata> onDeleteFilter = null;
     GenericFilter<? extends HasMetadata> genericFilter = null;
+    ResourceDiscriminator<?, ? extends HasMetadata> resourceDiscriminator = null;
     if (kubeDependent != null) {
       if (!Arrays.equals(KubernetesDependent.DEFAULT_NAMESPACES,
           kubeDependent.namespaces())) {
@@ -266,7 +265,6 @@ private Object createKubernetesResourceConfig(Class<? extends DependentResource>
       final var fromAnnotation = kubeDependent.labelSelector();
       labelSelector = Constants.NO_VALUE_SET.equals(fromAnnotation) ? null : fromAnnotation;
 
-
       final var context =
           Utils.contextFor(this, dependentType, null);
       onAddFilter = Utils.instantiate(kubeDependent.onAddFilter(), OnAddFilter.class, context);
@@ -276,10 +274,15 @@ private Object createKubernetesResourceConfig(Class<? extends DependentResource>
           Utils.instantiate(kubeDependent.onDeleteFilter(), OnDeleteFilter.class, context);
       genericFilter =
           Utils.instantiate(kubeDependent.genericFilter(), GenericFilter.class, context);
+
+      resourceDiscriminator =
+              Utils.instantiate(kubeDependent.resourceDiscriminator(),
+                      ResourceDiscriminator.class, context);
     }
 
     config =
-        new KubernetesDependentResourceConfig(namespaces, labelSelector, configuredNS, onAddFilter,
+        new KubernetesDependentResourceConfig(namespaces, labelSelector, configuredNS,
+            resourceDiscriminator, onAddFilter,
             onUpdateFilter, onDeleteFilter, genericFilter);
 
     return config;

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

@@ -6,20 +6,27 @@
 import io.fabric8.kubernetes.api.model.HasMetadata;
 import io.javaoperatorsdk.operator.api.config.ControllerConfiguration;
 import io.javaoperatorsdk.operator.api.reconciler.dependent.managed.ManagedDependentResourceContext;
+import io.javaoperatorsdk.operator.processing.event.EventSourceRetriever;
 
 public interface Context<P extends HasMetadata> {
 
   Optional<RetryInfo> getRetryInfo();
 
-  default <T> Optional<T> getSecondaryResource(Class<T> expectedType) {
-    return getSecondaryResource(expectedType, null);
+  default <R> Optional<R> getSecondaryResource(Class<R> expectedType) {
+    return getSecondaryResource(expectedType, (String) null);
   }
 
-  <T> Set<T> getSecondaryResources(Class<T> expectedType);
+  <R> Set<R> getSecondaryResources(Class<R> expectedType);
 
-  <T> Optional<T> getSecondaryResource(Class<T> expectedType, String eventSourceName);
+  @Deprecated(forRemoval = true)
+  <R> Optional<R> getSecondaryResource(Class<R> expectedType, String eventSourceName);
+
+  <R> Optional<R> getSecondaryResource(Class<R> expectedType,
+      ResourceDiscriminator<R, P> discriminator);
 
   ControllerConfiguration<P> getControllerConfiguration();
 
   ManagedDependentResourceContext managedDependentResourceContext();
+
+  EventSourceRetriever<P> eventSourceRetriever();
 }

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

@@ -9,6 +9,7 @@
 import io.javaoperatorsdk.operator.api.reconciler.dependent.managed.DefaultManagedDependentResourceContext;
 import io.javaoperatorsdk.operator.api.reconciler.dependent.managed.ManagedDependentResourceContext;
 import io.javaoperatorsdk.operator.processing.Controller;
+import io.javaoperatorsdk.operator.processing.event.EventSourceRetriever;
 
 public class DefaultContext<P extends HasMetadata> implements Context<P> {
 
@@ -47,6 +48,12 @@ public <T> Optional<T> getSecondaryResource(Class<T> expectedType, String eventS
         .getSecondaryResource(primaryResource);
   }
 
+  @Override
+  public <R> Optional<R> getSecondaryResource(Class<R> expectedType,
+      ResourceDiscriminator<R, P> discriminator) {
+    return discriminator.distinguish(expectedType, primaryResource, this);
+  }
+
   @Override
   public ControllerConfiguration<P> getControllerConfiguration() {
     return controllerConfiguration;
@@ -57,6 +64,11 @@ public ManagedDependentResourceContext managedDependentResourceContext() {
     return defaultManagedDependentResourceContext;
   }
 
+  @Override
+  public EventSourceRetriever<P> eventSourceRetriever() {
+    return controller.getEventSourceManager();
+  }
+
   public DefaultContext<P> setRetryInfo(RetryInfo retryInfo) {
     this.retryInfo = retryInfo;
     return this;

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

@@ -0,0 +1,11 @@
+package io.javaoperatorsdk.operator.api.reconciler;
+
+import java.util.Optional;
+
+import io.fabric8.kubernetes.api.model.HasMetadata;
+
+public interface ResourceDiscriminator<R, P extends HasMetadata> {
+
+  Optional<R> distinguish(Class<R> resource, P primary, Context<P> context);
+
+}

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

@@ -0,0 +1,25 @@
+package io.javaoperatorsdk.operator.api.reconciler;
+
+import java.util.Optional;
+import java.util.function.Function;
+
+import io.fabric8.kubernetes.api.model.HasMetadata;
+import io.javaoperatorsdk.operator.processing.event.ResourceID;
+
+public class ResourceIDMatcherDiscriminator<R extends HasMetadata, P extends HasMetadata>
+    implements ResourceDiscriminator<R, P> {
+
+  private final Function<P, ResourceID> mapper;
+
+  public ResourceIDMatcherDiscriminator(Function<P, ResourceID> mapper) {
+    this.mapper = mapper;
+  }
+
+  @Override
+  public Optional<R> distinguish(Class<R> resource, P primary, Context<P> context) {
+    var resourceID = mapper.apply(primary);
+    return context.getSecondaryResources(resource).stream()
+        .filter(resourceID::isSameResource)
+        .findFirst();
+  }
+}

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

@@ -1,16 +1,17 @@
 package io.javaoperatorsdk.operator.api.reconciler.dependent;
 
+import java.util.Optional;
+
 import io.fabric8.kubernetes.api.model.HasMetadata;
 import io.javaoperatorsdk.operator.api.reconciler.Context;
-import io.javaoperatorsdk.operator.processing.ResourceOwner;
 
 /**
  * An interface to implement and provide dependent resource support.
  *
  * @param <R> the dependent resource type
  * @param <P> the associated primary resource type
  */
-public interface DependentResource<R, P extends HasMetadata> extends ResourceOwner<R, P> {
+public interface DependentResource<R, P extends HasMetadata> {
 
   /**
    * Reconciles the dependent resource given the desired primary state
@@ -21,6 +22,17 @@ public interface DependentResource<R, P extends HasMetadata> extends ResourceOwn
    */
   ReconcileResult<R> reconcile(P primary, Context<P> context);
 
+  /**
+   * Retrieves the resource type associated with this DependentResource
+   *
+   * @return the resource type associated with this DependentResource
+   */
+  Class<R> resourceType();
+
+  default Optional<R> getSecondaryResource(P primary, Context<P> context) {
+    return Optional.empty();
+  }
+
   /**
    * Computes a default name for the specified DependentResource class
    *

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

@@ -1,11 +1,14 @@
 package io.javaoperatorsdk.operator.processing.dependent;
 
+import java.util.Optional;
+
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import io.fabric8.kubernetes.api.model.HasMetadata;
 import io.javaoperatorsdk.operator.api.reconciler.Context;
 import io.javaoperatorsdk.operator.api.reconciler.Ignore;
+import io.javaoperatorsdk.operator.api.reconciler.ResourceDiscriminator;
 import io.javaoperatorsdk.operator.api.reconciler.dependent.DependentResource;
 import io.javaoperatorsdk.operator.api.reconciler.dependent.ReconcileResult;
 import io.javaoperatorsdk.operator.processing.event.ResourceID;
@@ -21,6 +24,8 @@ public abstract class AbstractDependentResource<R, P extends HasMetadata>
   protected Creator<R, P> creator;
   protected Updater<R, P> updater;
 
+  private ResourceDiscriminator<R, P> resourceDiscriminator;
+
   @SuppressWarnings("unchecked")
   public AbstractDependentResource() {
     creator = creatable ? (Creator<R, P>) this : null;
@@ -29,7 +34,7 @@ public AbstractDependentResource() {
 
   @Override
   public ReconcileResult<R> reconcile(P primary, Context<P> context) {
-    var maybeActual = getSecondaryResource(primary);
+    Optional<R> maybeActual = getSecondaryResource(primary, context);
     if (creatable || updatable) {
       if (maybeActual.isEmpty()) {
         if (creatable) {
@@ -62,6 +67,11 @@ public ReconcileResult<R> reconcile(P primary, Context<P> context) {
     return ReconcileResult.noOperation(maybeActual.orElse(null));
   }
 
+  public Optional<R> getSecondaryResource(P primary, Context<P> context) {
+    return resourceDiscriminator == null ? context.getSecondaryResource(resourceType())
+        : resourceDiscriminator.distinguish(resourceType(), primary, context);
+  }
+
   private void throwIfNull(R desired, P primary, String descriptor) {
     if (desired == null) {
       throw new DependentResourceException(
@@ -118,4 +128,13 @@ protected R desired(P primary, Context<P> context) {
     throw new IllegalStateException(
         "desired method must be implemented if this DependentResource can be created and/or updated");
   }
+
+  public void setResourceDiscriminator(
+      ResourceDiscriminator<R, P> resourceDiscriminator) {
+    this.resourceDiscriminator = resourceDiscriminator;
+  }
+
+  public ResourceDiscriminator<R, P> getResourceDiscriminator() {
+    return resourceDiscriminator;
+  }
 }

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

@@ -19,12 +19,16 @@ public abstract class AbstractEventSourceHolderDependentResource<R, P extends Ha
     implements EventSourceProvider<P> {
 
   private T eventSource;
+  private final Class<R> resourceType;
   private boolean isCacheFillerEventSource;
   protected OnAddFilter<R> onAddFilter;
   protected OnUpdateFilter<R> onUpdateFilter;
   protected OnDeleteFilter<R> onDeleteFilter;
   protected GenericFilter<R> genericFilter;
 
+  protected AbstractEventSourceHolderDependentResource(Class<R> resourceType) {
+    this.resourceType = resourceType;
+  }
 
   public EventSource initEventSource(EventSourceContext<P> context) {
     // some sub-classes (e.g. KubernetesDependentResource) can have their event source created
@@ -42,6 +46,11 @@ public EventSource initEventSource(EventSourceContext<P> context) {
     return eventSource;
   }
 
+  @Override
+  public Class<R> resourceType() {
+    return resourceType;
+  }
+
   protected abstract T createEventSource(EventSourceContext<P> context);
 
   protected void setEventSource(T eventSource) {

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

@@ -2,11 +2,15 @@
 
 import io.fabric8.kubernetes.api.model.HasMetadata;
 import io.javaoperatorsdk.operator.api.reconciler.Ignore;
+import io.javaoperatorsdk.operator.processing.dependent.AbstractEventSourceHolderDependentResource;
 import io.javaoperatorsdk.operator.processing.event.source.CacheKeyMapper;
+import io.javaoperatorsdk.operator.processing.event.source.ExternalResourceCachingEventSource;
 
 @Ignore
 public abstract class AbstractPollingDependentResource<R, P extends HasMetadata>
-    extends AbstractCachingDependentResource<R, P> implements CacheKeyMapper<R> {
+    extends
+    AbstractEventSourceHolderDependentResource<R, P, ExternalResourceCachingEventSource<R, P>>
+    implements CacheKeyMapper<R> {
 
   public static final int DEFAULT_POLLING_PERIOD = 5000;
   private long pollingPeriod;

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

@@ -33,7 +33,7 @@ public AbstractSimpleDependentResource(UpdatableCache<R> cache) {
   }
 
   @Override
-  public Optional<R> getSecondaryResource(HasMetadata primaryResource) {
+  public Optional<R> getSecondaryResource(P primaryResource, Context<P> context) {
     return cache.get(ResourceID.fromResource(primaryResource));
   }
 

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

@@ -6,6 +6,8 @@
 import java.lang.annotation.Target;
 
 import io.javaoperatorsdk.operator.api.reconciler.Constants;
+import io.javaoperatorsdk.operator.api.reconciler.ResourceDiscriminator;
+import io.javaoperatorsdk.operator.processing.event.source.filter.*;
 import io.javaoperatorsdk.operator.processing.event.source.filter.GenericFilter;
 import io.javaoperatorsdk.operator.processing.event.source.filter.OnAddFilter;
 import io.javaoperatorsdk.operator.processing.event.source.filter.OnDeleteFilter;
@@ -68,4 +70,6 @@
    *         itself if no value is set
    */
   Class<? extends GenericFilter> genericFilter() default GenericFilter.class;
+
+  Class<? extends ResourceDiscriminator> resourceDiscriminator() default ResourceDiscriminator.class;
 }