feat: optional event source from dependent resources

Author: csviri

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

@@ -2,9 +2,12 @@
 
 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 +42,18 @@ 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) {
+
+    Map<String, EventSource> eventSourceMap = new HashMap<>(dependentResources.length);
+    for (DependentResource dependentResource : dependentResources) {
+      Optional<ResourceEventSource> es = dependentResource.provideEventSource(context);
+      es.ifPresent(e -> eventSourceMap.put(generateNameFor(e), e));
+    }
+    return eventSourceMap;
+  }
+
   /**
    * 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/dependent/DependentResource.java

@@ -1,7 +1,11 @@
 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.api.reconciler.EventSourceContext;
+import io.javaoperatorsdk.operator.processing.event.source.ResourceEventSource;
 
 /**
  * An interface to implement and provide dependent resource support.
@@ -27,6 +31,11 @@ public interface DependentResource<R, P extends HasMetadata> {
    */
   Class<R> resourceType();
 
+  default Optional<ResourceEventSource<R, P>> provideEventSource(
+      EventSourceContext<P> eventSourceContext) {
+    return Optional.empty();
+  }
+
   /**
    * Computes a default name for the specified DependentResource class
    *

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

@@ -39,6 +39,7 @@
 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;
 
@@ -209,13 +210,19 @@ 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);
+          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().provideEventSource(context);
+            eventSource.ifPresent(es -> {
+              eventSourceManager.registerEventSource(drEntry.getKey(), es);
+            });
+          }
         });
-
     // add manually defined event sources
     if (reconciler instanceof EventSourceInitializer) {
       final var provider = (EventSourceInitializer<P>) this.reconciler;

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

@@ -1,12 +1,12 @@
 package io.javaoperatorsdk.operator.processing.dependent;
 
+import java.util.Optional;
+
 import io.fabric8.kubernetes.api.model.HasMetadata;
 import io.javaoperatorsdk.operator.api.reconciler.EventSourceContext;
 import io.javaoperatorsdk.operator.api.reconciler.Ignore;
-import io.javaoperatorsdk.operator.api.reconciler.dependent.EventSourceProvider;
 import io.javaoperatorsdk.operator.api.reconciler.dependent.RecentOperationCacheFiller;
 import io.javaoperatorsdk.operator.processing.event.ResourceID;
-import io.javaoperatorsdk.operator.processing.event.source.EventSource;
 import io.javaoperatorsdk.operator.processing.event.source.ResourceEventSource;
 import io.javaoperatorsdk.operator.processing.event.source.filter.GenericFilter;
 import io.javaoperatorsdk.operator.processing.event.source.filter.OnAddFilter;
@@ -15,8 +15,7 @@
 
 @Ignore
 public abstract class AbstractEventSourceHolderDependentResource<R, P extends HasMetadata, T extends ResourceEventSource<R, P>>
-    extends AbstractDependentResource<R, P>
-    implements EventSourceProvider<P> {
+    extends AbstractDependentResource<R, P> {
 
   private T eventSource;
   private boolean isCacheFillerEventSource;
@@ -25,26 +24,42 @@ public abstract class AbstractEventSourceHolderDependentResource<R, P extends Ha
   protected OnDeleteFilter<R> onDeleteFilter;
   protected GenericFilter<R> genericFilter;
 
+  /**
+   * Even a dependent resource holds an event source, it might not provide it. For example if event
+   * sources are shared between multiple dependent resources. Typically
+   * {@link io.javaoperatorsdk.operator.processing.dependent.kubernetes.KubernetesDependentResource}
+   * needs to be aware of event source even if the actual state of the resource is not read directly
+   * from the event source.
+   */
+  private boolean provideEventSource = true;
+
 
-  public EventSource initEventSource(EventSourceContext<P> context) {
+  public Optional<ResourceEventSource<R, P>> provideEventSource(EventSourceContext<P> context) {
+    if (!provideEventSource) {
+      return Optional.empty();
+    }
     // some sub-classes (e.g. KubernetesDependentResource) can have their event source created
     // before this method is called in the managed case, so only create the event source if it
     // hasn't already been set.
     // The filters are applied automatically only if event source is created automatically. So if an
     // event source
     // is shared between dependent resources this does not override the existing filters.
     if (eventSource == null) {
-      eventSource = createEventSource(context);
+      setEventSource(createEventSource(context));
       applyFilters();
     }
+    return Optional.of(eventSource);
+  }
 
-    isCacheFillerEventSource = eventSource instanceof RecentOperationCacheFiller;
-    return eventSource;
+  /** To make this backwards compatible even for respect of overriding */
+  public T initEventSource(EventSourceContext<P> context) {
+    return (T) provideEventSource(context).orElseThrow();
   }
 
   protected abstract T createEventSource(EventSourceContext<P> context);
 
   protected void setEventSource(T eventSource) {
+    isCacheFillerEventSource = eventSource instanceof RecentOperationCacheFiller;
     this.eventSource = eventSource;
   }
 
@@ -55,8 +70,8 @@ protected void applyFilters() {
     this.eventSource.setGenericFilter(genericFilter);
   }
 
-  protected T eventSource() {
-    return eventSource;
+  public Optional<ResourceEventSource<R, P>> eventSource() {
+    return Optional.ofNullable(eventSource);
   }
 
   protected void onCreated(ResourceID primaryResourceId, R created) {
@@ -87,4 +102,8 @@ public void setOnUpdateFilter(OnUpdateFilter<R> onUpdateFilter) {
   public void setOnDeleteFilter(OnDeleteFilter<R> onDeleteFilter) {
     this.onDeleteFilter = onDeleteFilter;
   }
+
+  public void setProvideEventSource(boolean provideEventSource) {
+    this.provideEventSource = provideEventSource;
+  }
 }

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

@@ -179,7 +179,7 @@ protected InformerEventSource<R, P> createEventSource(EventSourceContext<P> cont
           "Using default configuration for {} KubernetesDependentResource, call configureWith to provide configuration",
           resourceType().getSimpleName());
     }
-    return eventSource();
+    return (InformerEventSource<R, P>) eventSource().orElseThrow();
   }
 
   private boolean useDefaultAnnotationsToIdentifyPrimary() {
@@ -227,11 +227,13 @@ protected R desired(P primary, Context<P> context) {
   }
 
   private void prepareEventFiltering(R desired, ResourceID resourceID) {
-    eventSource().prepareForCreateOrUpdateEventFiltering(resourceID, desired);
+    ((InformerEventSource<R, P>) eventSource().orElseThrow())
+        .prepareForCreateOrUpdateEventFiltering(resourceID, desired);
   }
 
   private void cleanupAfterEventFiltering(ResourceID resourceID) {
-    eventSource().cleanupOnCreateOrUpdateEventFiltering(resourceID);
+    ((InformerEventSource<R, P>) eventSource().orElseThrow())
+        .cleanupOnCreateOrUpdateEventFiltering(resourceID);
   }
 
 }