feat: optional eventsource on dependent resources (#1479)

Author: csviri

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

@@ -224,7 +224,8 @@ public List<DependentResourceSpec> getDependentResources() {
             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));
+            Utils.instantiate(dependent.deletePostcondition(), Condition.class, context),
+            dependent.provideEventSource());
         specsMap.put(name, spec);
       }
 
@@ -255,13 +256,13 @@ 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())) {
         namespaces = Set.of(kubeDependent.namespaces());
         configuredNS = true;
       }
-
       final var fromAnnotation = kubeDependent.labelSelector();
       labelSelector = Constants.NO_VALUE_SET.equals(fromAnnotation) ? null : fromAnnotation;
 
@@ -278,12 +279,14 @@ 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);
+            onUpdateFilter, onDeleteFilter, genericFilter, eventSourceNameToUse);
 
     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.getDeletePostCondition(), current.provideEventSource()));
   }
 
   @SuppressWarnings("unchecked")
@@ -220,7 +220,7 @@ public ControllerConfiguration<R> build() {
       KubernetesDependentResourceConfig c) {
     return new DependentResourceSpec(spec.getDependentResourceClass(),
         c.setNamespaces(namespaces), name, spec.getDependsOn(), spec.getReadyCondition(),
-        spec.getReconcileCondition(), spec.getDeletePostCondition());
+        spec.getReconcileCondition(), spec.getDeletePostCondition(), spec.provideEventSource());
   }
 
   public static <R extends HasMetadata> ControllerConfigurationOverrider<R> override(

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

@@ -23,16 +23,20 @@ public class DependentResourceSpec<T extends DependentResource<?, ?>, C> {
 
   private final Condition<?, ?> deletePostCondition;
 
+  private final boolean provideEventSource;
+
   public DependentResourceSpec(Class<T> dependentResourceClass, C dependentResourceConfig,
       String name, Set<String> dependsOn, Condition<?, ?> readyCondition,
-      Condition<?, ?> reconcileCondition, Condition<?, ?> deletePostCondition) {
+      Condition<?, ?> reconcileCondition, Condition<?, ?> deletePostCondition,
+      boolean provideEventSource) {
     this.dependentResourceClass = dependentResourceClass;
     this.dependentResourceConfig = dependentResourceConfig;
     this.name = name;
     this.dependsOn = dependsOn;
     this.readyCondition = readyCondition;
     this.reconcileCondition = reconcileCondition;
     this.deletePostCondition = deletePostCondition;
+    this.provideEventSource = provideEventSource;
   }
 
   public Class<T> getDependentResourceClass() {
@@ -89,4 +93,8 @@ public Condition getReconcileCondition() {
   public Condition getDeletePostCondition() {
     return deletePostCondition;
   }
+
+  public boolean provideEventSource() {
+    return provideEventSource;
+  }
 }

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

@@ -1,10 +1,14 @@
 package io.javaoperatorsdk.operator.api.reconciler;
 
+import java.util.Collections;
 import java.util.HashMap;
 import java.util.Map;
+import java.util.Optional;
 
 import io.fabric8.kubernetes.api.model.HasMetadata;
+import io.javaoperatorsdk.operator.api.reconciler.dependent.DependentResource;
 import io.javaoperatorsdk.operator.processing.event.source.EventSource;
+import io.javaoperatorsdk.operator.processing.event.source.ResourceEventSource;
 
 /**
  * An interface that a {@link Reconciler} can implement to have the SDK register the provided
@@ -39,6 +43,22 @@ static Map<String, EventSource> nameEventSources(EventSource... eventSources) {
     return eventSourceMap;
   }
 
+  @SuppressWarnings("unchecked,rawtypes")
+  static <K extends HasMetadata> Map<String, EventSource> nameEventSourcesFromDependentResource(
+      EventSourceContext<K> context, DependentResource... dependentResources) {
+
+    if (dependentResources != null) {
+      Map<String, EventSource> eventSourceMap = new HashMap<>(dependentResources.length);
+      for (DependentResource dependentResource : dependentResources) {
+        Optional<ResourceEventSource> es = dependentResource.eventSource(context);
+        es.ifPresent(e -> eventSourceMap.put(generateNameFor(e), e));
+      }
+      return eventSourceMap;
+    } else {
+      return Collections.emptyMap();
+    }
+  }
+
   /**
    * This is for the use case when the event sources are not access explicitly by name in the
    * reconciler.

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

@@ -7,4 +7,5 @@
 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/dependent/Dependent.java

@@ -57,4 +57,13 @@
    *         one can be
    */
   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.
+   *
+   * @return if the event source (if any) provided by the dependent resource should be used or not.
+   */
+  boolean provideEventSource() default true;
 }

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

@@ -4,6 +4,8 @@
 
 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.processing.event.source.ResourceEventSource;
 
 /**
  * An interface to implement and provide dependent resource support.
@@ -29,6 +31,31 @@ public interface DependentResource<R, P extends HasMetadata> {
    */
   Class<R> resourceType();
 
+  /**
+   * Dependent resources are designed to by default provide event sources. There are cases where it
+   * might not:
+   * <ul>
+   * <li>If an event source is shared between multiple dependent resources. In this case only one or
+   * none of the dependent resources sharing the event source should provide one.</li>
+   * <li>Some special implementation of an event source. That just execute some action might not
+   * provide one.</li>
+   * </ul>
+   *
+   * @param eventSourceContext context of event source initialization
+   * @return an optional event source
+   */
+  default Optional<ResourceEventSource<R, P>> eventSource(
+      EventSourceContext<P> eventSourceContext) {
+    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,10 @@
+package io.javaoperatorsdk.operator.api.reconciler.dependent;
+
+import io.fabric8.kubernetes.api.model.HasMetadata;
+import io.javaoperatorsdk.operator.processing.event.EventSourceRetriever;
+
+public interface EventSourceAware<P extends HasMetadata> {
+
+  void selectEventSources(EventSourceRetriever<P> eventSourceRetriever);
+
+}

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

@@ -4,6 +4,11 @@
 import io.javaoperatorsdk.operator.api.reconciler.EventSourceContext;
 import io.javaoperatorsdk.operator.processing.event.source.EventSource;
 
+/**
+ * @deprecated now event source related methods are directly on {@link DependentResource}
+ * @param <P> primary resource
+ */
+@Deprecated(forRemoval = true)
 public interface EventSourceProvider<P extends HasMetadata> {
   /**
    * @param context - event source context where the event source is initialized

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

@@ -32,13 +32,15 @@
 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.EventSourceProvider;
 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;
 import io.javaoperatorsdk.operator.processing.event.EventProcessor;
 import io.javaoperatorsdk.operator.processing.event.EventSourceManager;
 import io.javaoperatorsdk.operator.processing.event.ResourceID;
+import io.javaoperatorsdk.operator.processing.event.source.ResourceEventSource;
 
 import static io.javaoperatorsdk.operator.api.reconciler.Constants.WATCH_CURRENT_NAMESPACE;
 
@@ -207,21 +209,30 @@ private void initContextIfNeeded(P resource, Context<P> context) {
   }
 
   public void initAndRegisterEventSources(EventSourceContext<P> context) {
-    managedWorkflow
-        .getDependentResourcesByName().entrySet().stream()
-        .filter(drEntry -> drEntry.getValue() instanceof EventSourceProvider)
-        .forEach(drEntry -> {
-          final var provider = (EventSourceProvider) drEntry.getValue();
-          final var source = provider.initEventSource(context);
-          eventSourceManager.registerEventSource(drEntry.getKey(), source);
-        });
-
-    // add manually defined event sources
     if (reconciler instanceof EventSourceInitializer) {
       final var provider = (EventSourceInitializer<P>) this.reconciler;
       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));
   }
 
   @Override

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

@@ -9,11 +9,13 @@
 
 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>
@@ -27,8 +29,9 @@ 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;
 
-  private final List<ResourceDiscriminator<R, P>> resourceDiscriminator = new ArrayList<>(1);
+  protected List<ResourceDiscriminator<R, P>> resourceDiscriminator = new ArrayList<>(1);
 
   @SuppressWarnings("unchecked")
   public AbstractDependentResource() {
@@ -38,6 +41,23 @@ 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) {
@@ -172,7 +192,7 @@ protected R handleCreate(R desired, P primary, Context<P> context) {
   protected abstract void onCreated(ResourceID primaryResourceId, R created);
 
   /**
-   * Allows sub-classes to perform additional processing on the updated resource if needed.
+   * Allows subclasses to perform additional processing on the updated resource if needed.
    *
    * @param primaryResourceId the {@link ResourceID} of the primary resource associated with the
    *        newly updated resource
@@ -219,4 +239,7 @@ protected int lastKnownBulkSize() {
     return resourceDiscriminator.size();
   }
 
+  protected boolean getReturnEventSource() {
+    return returnEventSource;
+  }
 }