refactor: simplify handling of reused event sources (#1518)

Co-authored-by: Chris Laprun <metacosm@gmail.com>

Author: csviri

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

@@ -219,13 +219,15 @@ public List<DependentResourceSpec> getDependentResources() {
           throw new IllegalArgumentException(
               "A DependentResource named '" + name + "' already exists: " + spec);
         }
+        var eventSourceName = dependent.useEventSourceWithName();
+        eventSourceName = Constants.NO_VALUE_SET.equals(eventSourceName) ? null : eventSourceName;
         final var context = Utils.contextFor(this, dependentType, null);
         spec = new DependentResourceSpec(dependentType, config, name,
             Set.of(dependent.dependsOn()),
             Utils.instantiate(dependent.readyPostcondition(), Condition.class, context),
             Utils.instantiate(dependent.reconcilePrecondition(), Condition.class, context),
             Utils.instantiate(dependent.deletePostcondition(), Condition.class, context),
-            dependent.provideEventSource());
+                eventSourceName);
         specsMap.put(name, spec);
       }
 
@@ -256,7 +258,6 @@ private Object createKubernetesResourceConfig(Class<? extends DependentResource>
     OnDeleteFilter<? extends HasMetadata> onDeleteFilter = null;
     GenericFilter<? extends HasMetadata> genericFilter = null;
     ResourceDiscriminator<?, ? extends HasMetadata> resourceDiscriminator = null;
-    String eventSourceNameToUse = null;
     if (kubeDependent != null) {
       if (!Arrays.equals(KubernetesDependent.DEFAULT_NAMESPACES,
           kubeDependent.namespaces())) {
@@ -279,14 +280,12 @@ private Object createKubernetesResourceConfig(Class<? extends DependentResource>
       resourceDiscriminator =
               Utils.instantiate(kubeDependent.resourceDiscriminator(),
                       ResourceDiscriminator.class, context);
-      eventSourceNameToUse = Constants.NO_VALUE_SET.equals(kubeDependent.eventSourceToUse()) ? null
-              : kubeDependent.eventSourceToUse();
     }
 
     config =
         new KubernetesDependentResourceConfig(namespaces, labelSelector, configuredNS,
             resourceDiscriminator, onAddFilter,
-            onUpdateFilter, onDeleteFilter, genericFilter, eventSourceNameToUse);
+            onUpdateFilter, onDeleteFilter, genericFilter);
 
     return config;
   }

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

@@ -174,7 +174,7 @@ private void replaceConfig(String name, Object newConfig, DependentResourceSpec<
     namedDependentResourceSpecs.put(name,
         new DependentResourceSpec<>(current.getDependentResourceClass(), newConfig, name,
             current.getDependsOn(), current.getReadyCondition(), current.getReconcileCondition(),
-            current.getDeletePostCondition(), current.provideEventSource()));
+            current.getDeletePostCondition(), current.getUseEventSourceWithName().orElse(null)));
   }
 
   @SuppressWarnings("unchecked")
@@ -220,7 +220,8 @@ public ControllerConfiguration<R> build() {
       KubernetesDependentResourceConfig c) {
     return new DependentResourceSpec(spec.getDependentResourceClass(),
         c.setNamespaces(namespaces), name, spec.getDependsOn(), spec.getReadyCondition(),
-        spec.getReconcileCondition(), spec.getDeletePostCondition(), spec.provideEventSource());
+        spec.getReconcileCondition(), spec.getDeletePostCondition(),
+        (String) spec.getUseEventSourceWithName().orElse(null));
   }
 
   public static <R extends HasMetadata> ControllerConfigurationOverrider<R> override(

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

@@ -23,20 +23,20 @@ public class DependentResourceSpec<T extends DependentResource<?, ?>, C> {
 
   private final Condition<?, ?> deletePostCondition;
 
-  private final boolean provideEventSource;
+  private final String useEventSourceWithName;
 
   public DependentResourceSpec(Class<T> dependentResourceClass, C dependentResourceConfig,
       String name, Set<String> dependsOn, Condition<?, ?> readyCondition,
       Condition<?, ?> reconcileCondition, Condition<?, ?> deletePostCondition,
-      boolean provideEventSource) {
+      String useEventSourceWithName) {
     this.dependentResourceClass = dependentResourceClass;
     this.dependentResourceConfig = dependentResourceConfig;
     this.name = name;
     this.dependsOn = dependsOn;
     this.readyCondition = readyCondition;
     this.reconcileCondition = reconcileCondition;
     this.deletePostCondition = deletePostCondition;
-    this.provideEventSource = provideEventSource;
+    this.useEventSourceWithName = useEventSourceWithName;
   }
 
   public Class<T> getDependentResourceClass() {
@@ -94,7 +94,7 @@ public Condition getDeletePostCondition() {
     return deletePostCondition;
   }
 
-  public boolean provideEventSource() {
-    return provideEventSource;
+  public Optional<String> getUseEventSourceWithName() {
+    return Optional.ofNullable(useEventSourceWithName);
   }
 }

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

@@ -1,5 +1,6 @@
 package io.javaoperatorsdk.operator.api.reconciler.dependent;
 
+import io.javaoperatorsdk.operator.api.reconciler.Constants;
 import io.javaoperatorsdk.operator.processing.dependent.workflow.Condition;
 
 import static io.javaoperatorsdk.operator.api.reconciler.Constants.NO_VALUE_SET;
@@ -59,11 +60,11 @@
   String[] dependsOn() default {};
 
   /**
-   * Setting this to false means that the event source provided by the dependent resource won't be
-   * used. This is helpful if more dependent resources created for the same type, and want to share
-   * a common event source. In that case an event source needs to be explicitly registered.
+   * Setting here a name of the event source means that dependent resource will use an event source
+   * registered with that name. So won't create one. This is helpful if more dependent resources
+   * created for the same type, and want to share a common event source.
    *
-   * @return if the event source (if any) provided by the dependent resource should be used or not.
+   * @return event source name (if any) provided by the dependent resource should be used.
    */
-  boolean provideEventSource() default true;
+  String useEventSourceWithName() default NO_VALUE_SET;
 }

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

@@ -49,13 +49,6 @@ default Optional<ResourceEventSource<R, P>> eventSource(
     return Optional.empty();
   }
 
-  /**
-   * Calling this method, instructs the implementation to not provide an event source, even if it
-   * normally does.
-   */
-  void doNotProvideEventSource();
-
-
   default Optional<R> getSecondaryResource(P primary, Context<P> context) {
     return Optional.empty();
   }

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

@@ -0,0 +1,16 @@
+package io.javaoperatorsdk.operator.api.reconciler.dependent;
+
+import io.javaoperatorsdk.operator.OperatorException;
+
+public class EventSourceNotFoundException extends OperatorException {
+
+  private String eventSourceName;
+
+  public EventSourceNotFoundException(String eventSourceName) {
+    this.eventSourceName = eventSourceName;
+  }
+
+  public String getEventSourceName() {
+    return eventSourceName;
+  }
+}

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

@@ -0,0 +1,17 @@
+package io.javaoperatorsdk.operator.api.reconciler.dependent;
+
+import io.fabric8.kubernetes.api.model.HasMetadata;
+import io.javaoperatorsdk.operator.processing.event.EventSourceRetriever;
+
+public interface EventSourceReferencer<P extends HasMetadata> {
+
+  default void useEventSourceWithName(String name) {}
+
+  /**
+   * Throws {@link EventSourceNotFoundException} an exception if the target event source to use is
+   * not found.
+   */
+  void resolveEventSource(EventSourceRetriever<P> eventSourceRetriever)
+      throws EventSourceNotFoundException;
+
+}

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

@@ -1,5 +1,8 @@
 package io.javaoperatorsdk.operator.processing;
 
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
 import java.util.Map;
 import java.util.Optional;
 import java.util.Set;
@@ -32,8 +35,9 @@
 import io.javaoperatorsdk.operator.api.reconciler.Ignore;
 import io.javaoperatorsdk.operator.api.reconciler.Reconciler;
 import io.javaoperatorsdk.operator.api.reconciler.UpdateControl;
-import io.javaoperatorsdk.operator.api.reconciler.dependent.EventSourceAware;
+import io.javaoperatorsdk.operator.api.reconciler.dependent.EventSourceNotFoundException;
 import io.javaoperatorsdk.operator.api.reconciler.dependent.EventSourceProvider;
+import io.javaoperatorsdk.operator.api.reconciler.dependent.EventSourceReferencer;
 import io.javaoperatorsdk.operator.api.reconciler.dependent.managed.DefaultManagedDependentResourceContext;
 import io.javaoperatorsdk.operator.processing.dependent.workflow.ManagedWorkflow;
 import io.javaoperatorsdk.operator.processing.dependent.workflow.WorkflowCleanupResult;
@@ -214,25 +218,40 @@ public void initAndRegisterEventSources(EventSourceContext<P> context) {
       final var ownSources = provider.prepareEventSources(context);
       ownSources.forEach(eventSourceManager::registerEventSource);
     }
-    managedWorkflow
-        .getDependentResourcesByName().entrySet().stream()
-        .forEach(drEntry -> {
-          if (drEntry.getValue() instanceof EventSourceProvider) {
-            final var provider = (EventSourceProvider) drEntry.getValue();
-            final var source = provider.initEventSource(context);
-            eventSourceManager.registerEventSource(drEntry.getKey(), source);
-          } else {
-            Optional<ResourceEventSource> eventSource =
-                drEntry.getValue().eventSource(context);
-            eventSource.ifPresent(es -> {
-              eventSourceManager.registerEventSource(drEntry.getKey(), es);
-            });
-          }
-        });
-    managedWorkflow.getDependentResourcesByName().entrySet().stream().map(Map.Entry::getValue)
-        .filter(EventSourceAware.class::isInstance)
-        .forEach(dr -> ((EventSourceAware) dr)
-            .selectEventSources(eventSourceManager));
+
+    // register created event sources
+    final var dependentResourcesByName = managedWorkflow.getDependentResourcesByName();
+    final var size = dependentResourcesByName.size();
+    if (size > 0) {
+      dependentResourcesByName.forEach((key, value) -> {
+        if (value instanceof EventSourceProvider) {
+          final var provider = (EventSourceProvider) value;
+          final var source = provider.initEventSource(context);
+          eventSourceManager.registerEventSource(key, source);
+        } else {
+          Optional<ResourceEventSource> eventSource = value.eventSource(context);
+          eventSource.ifPresent(es -> eventSourceManager.registerEventSource(key, es));
+        }
+      });
+
+      // resolve event sources referenced by name for dependents that reuse an existing event source
+      final Map<String, List<EventSourceReferencer>> unresolvable = new HashMap<>(size);
+      dependentResourcesByName.values().stream()
+          .filter(EventSourceReferencer.class::isInstance)
+          .map(EventSourceReferencer.class::cast)
+          .forEach(dr -> {
+            try {
+              ((EventSourceReferencer<P>) dr)
+                  .resolveEventSource(eventSourceManager);
+            } catch (EventSourceNotFoundException e) {
+              unresolvable.computeIfAbsent(e.getEventSourceName(), s -> new ArrayList<>()).add(dr);
+            }
+          });
+      if (!unresolvable.isEmpty()) {
+        throw new IllegalStateException(
+            "Couldn't resolve referenced EventSources: " + unresolvable);
+      }
+    }
   }
 
   @Override

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

@@ -9,13 +9,11 @@
 
 import io.fabric8.kubernetes.api.model.HasMetadata;
 import io.javaoperatorsdk.operator.api.reconciler.Context;
-import io.javaoperatorsdk.operator.api.reconciler.EventSourceContext;
 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;
-import io.javaoperatorsdk.operator.processing.event.source.ResourceEventSource;
 
 @Ignore
 public abstract class AbstractDependentResource<R, P extends HasMetadata>
@@ -29,7 +27,6 @@ public abstract class AbstractDependentResource<R, P extends HasMetadata>
   protected Creator<R, P> creator;
   protected Updater<R, P> updater;
   protected BulkDependentResource<R, P> bulkDependentResource;
-  private boolean returnEventSource = true;
 
   protected List<ResourceDiscriminator<R, P>> resourceDiscriminator = new ArrayList<>(1);
 
@@ -41,23 +38,6 @@ public AbstractDependentResource() {
     bulkDependentResource = bulk ? (BulkDependentResource<R, P>) this : null;
   }
 
-  @Override
-  public void doNotProvideEventSource() {
-    this.returnEventSource = false;
-  }
-
-  @Override
-  public Optional<ResourceEventSource<R, P>> eventSource(EventSourceContext<P> eventSourceContext) {
-    if (!returnEventSource) {
-      return Optional.empty();
-    } else {
-      return Optional.of(provideEventSource(eventSourceContext));
-    }
-  }
-
-  protected abstract ResourceEventSource<R, P> provideEventSource(
-      EventSourceContext<P> eventSourceContext);
-
   @Override
   public ReconcileResult<R> reconcile(P primary, Context<P> context) {
     if (bulk) {
@@ -239,7 +219,4 @@ protected int lastKnownBulkSize() {
     return resourceDiscriminator.size();
   }
 
-  protected boolean getReturnEventSource() {
-    return returnEventSource;
-  }
 }