[WIP] Informer pools

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

@@ -24,6 +24,7 @@
 import io.fabric8.kubernetes.client.KubernetesClient;
 import io.javaoperatorsdk.operator.ReconcilerUtilsInternal;
 import io.javaoperatorsdk.operator.api.reconciler.Reconciler;
+import io.javaoperatorsdk.operator.processing.event.source.informer.pool.InformerPool;
 
 /**
  * An abstract implementation of {@link ConfigurationService} meant to ease custom implementations
@@ -35,6 +36,7 @@ public class AbstractConfigurationService implements ConfigurationService {
   private KubernetesClient client;
   private Cloner cloner;
   private ExecutorServiceManager executorServiceManager;
+  private InformerPool informerPool;
 
   protected AbstractConfigurationService(Version version) {
     this(version, null);
@@ -190,4 +192,15 @@ public ExecutorServiceManager getExecutorServiceManager() {
     }
     return executorServiceManager;
   }
+
+  @Override
+  public synchronized InformerPool informerPool() {
+    // cached so that all controllers backed by this ConfigurationService share the same pool and
+    // can therefore share the underlying informers; synchronized so concurrent first-access from
+    // multiple controllers cannot create (and share out) more than one pool instance
+    if (informerPool == null) {
+      informerPool = ConfigurationService.super.informerPool();
+    }
+    return informerPool;
+  }
 }

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

@@ -43,6 +43,8 @@
 import io.javaoperatorsdk.operator.processing.dependent.kubernetes.KubernetesDependentResourceConfig;
 import io.javaoperatorsdk.operator.processing.dependent.workflow.ManagedWorkflowFactory;
 import io.javaoperatorsdk.operator.processing.event.source.controller.ControllerEventSource;
+import io.javaoperatorsdk.operator.processing.event.source.informer.pool.DefaultInformerPool;
+import io.javaoperatorsdk.operator.processing.event.source.informer.pool.InformerPool;
 
 /** An interface from which to retrieve configuration information. */
 public interface ConfigurationService {
@@ -476,4 +478,9 @@ default boolean useSSAToPatchPrimaryResource() {
   default boolean cloneSecondaryResourcesWhenGettingFromCache() {
     return false;
   }
+
+  default InformerPool informerPool() {
+    // todo this is probably wrong here
+    return new DefaultInformerPool(this);
+  }
 }

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

@@ -29,6 +29,7 @@
 import io.javaoperatorsdk.operator.Operator;
 import io.javaoperatorsdk.operator.api.monitoring.Metrics;
 import io.javaoperatorsdk.operator.api.reconciler.dependent.DependentResourceFactory;
+import io.javaoperatorsdk.operator.processing.event.source.informer.pool.InformerPool;
 
 @SuppressWarnings({"unused", "UnusedReturnValue"})
 public class ConfigurationServiceOverrider {
@@ -53,6 +54,7 @@ public class ConfigurationServiceOverrider {
   private Set<Class<? extends HasMetadata>> defaultNonSSAResource;
   private Boolean useSSAToPatchPrimaryResource;
   private Boolean cloneSecondaryResourcesWhenGettingFromCache;
+  private InformerPool informerPool;
 
   @SuppressWarnings("rawtypes")
   private DependentResourceFactory dependentResourceFactory;
@@ -176,6 +178,15 @@ public ConfigurationServiceOverrider withCloneSecondaryResourcesWhenGettingFromC
     return this;
   }
 
+  /**
+   * Overrides the {@link InformerPool} strategy used to create/share the informers backing the
+   * event sources. When not set, the default (informer-sharing) pool is used.
+   */
+  public ConfigurationServiceOverrider withInformerPool(InformerPool informerPool) {
+    this.informerPool = informerPool;
+    return this;
+  }
+
   public ConfigurationService build() {
     return new BaseConfigurationService(original.getVersion(), cloner, client) {
       @Override
@@ -309,6 +320,15 @@ public boolean cloneSecondaryResourcesWhenGettingFromCache() {
             cloneSecondaryResourcesWhenGettingFromCache,
             ConfigurationService::cloneSecondaryResourcesWhenGettingFromCache);
       }
+
+      @Override
+      public InformerPool informerPool() {
+        if (informerPool == null) {
+          return super.informerPool();
+        }
+        informerPool.setConfigurationService(this);
+        return informerPool;
+      }
     };
   }
 }

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

@@ -17,6 +17,7 @@
 
 import java.util.Arrays;
 import java.util.List;
+import java.util.Objects;
 
 public class FieldSelector {
   private final List<Field> fields;
@@ -38,4 +39,16 @@ public Field(String path, String value) {
       this(path, value, false);
     }
   }
+
+  @Override
+  public boolean equals(Object o) {
+    if (o == null || getClass() != o.getClass()) return false;
+    FieldSelector that = (FieldSelector) o;
+    return Objects.equals(fields, that.fields);
+  }
+
+  @Override
+  public int hashCode() {
+    return Objects.hashCode(fields);
+  }
 }

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

@@ -30,6 +30,7 @@
 import io.javaoperatorsdk.operator.api.config.ControllerConfiguration;
 import io.javaoperatorsdk.operator.api.config.Utils;
 import io.javaoperatorsdk.operator.api.reconciler.Constants;
+import io.javaoperatorsdk.operator.processing.GroupVersionKind;
 import io.javaoperatorsdk.operator.processing.event.source.cache.BoundedItemStore;
 import io.javaoperatorsdk.operator.processing.event.source.filter.GenericFilter;
 import io.javaoperatorsdk.operator.processing.event.source.filter.OnAddFilter;
@@ -42,6 +43,7 @@
 public class InformerConfiguration<R extends HasMetadata> {
   private final Builder builder = new Builder();
   private final Class<R> resourceClass;
+  private final GroupVersionKind resourceGroupVersionKind;
   private final String resourceTypeName;
   private String name;
   private Set<String> namespaces;
@@ -59,6 +61,7 @@ public class InformerConfiguration<R extends HasMetadata> {
 
   protected InformerConfiguration(
       Class<R> resourceClass,
+      GroupVersionKind resourceGroupVersionKind,
       String name,
       Set<String> namespaces,
       boolean followControllerNamespaceChanges,
@@ -74,7 +77,7 @@ protected InformerConfiguration(
       Boolean comparableResourceVersions,
       // TODO for removal in major release
       Duration ghostResourceCacheCheckInterval) {
-    this(resourceClass);
+    this(resourceClass, resourceGroupVersionKind);
     this.name = name;
     this.namespaces = namespaces;
     this.followControllerNamespaceChanges = followControllerNamespaceChanges;
@@ -90,8 +93,9 @@ protected InformerConfiguration(
     this.comparableResourceVersions = comparableResourceVersions;
   }
 
-  private InformerConfiguration(Class<R> resourceClass) {
+  private InformerConfiguration(Class<R> resourceClass, GroupVersionKind resourceGroupVersionKind) {
     this.resourceClass = resourceClass;
+    this.resourceGroupVersionKind = resourceGroupVersionKind;
     this.resourceTypeName =
         resourceClass.isAssignableFrom(GenericKubernetesResource.class)
             // in general this is irrelevant now for secondary resources it is used just by
@@ -101,17 +105,24 @@ private InformerConfiguration(Class<R> resourceClass) {
             : ReconcilerUtilsInternal.getResourceTypeName(resourceClass);
   }
 
+  @SuppressWarnings({"rawtypes", "unchecked"})
+  public static <R extends HasMetadata> InformerConfiguration<R>.Builder builder(
+      Class<R> resourceClass, GroupVersionKind groupVersionKind) {
+    return new InformerConfiguration(resourceClass, groupVersionKind).builder;
+  }
+
   @SuppressWarnings({"rawtypes", "unchecked"})
   public static <R extends HasMetadata> InformerConfiguration<R>.Builder builder(
       Class<R> resourceClass) {
-    return new InformerConfiguration(resourceClass).builder;
+    return new InformerConfiguration(resourceClass, null).builder;
   }
 
   @SuppressWarnings({"rawtypes", "unchecked"})
   public static <R extends HasMetadata> InformerConfiguration<R>.Builder builder(
       InformerConfiguration<R> original) {
     return new InformerConfiguration(
             original.resourceClass,
+            original.resourceGroupVersionKind,
             original.name,
             original.namespaces,
             original.followControllerNamespaceChanges,
@@ -305,6 +316,10 @@ public Long getInformerListLimit() {
     return informerListLimit;
   }
 
+  public GroupVersionKind getResourceGroupVersionKind() {
+    return resourceGroupVersionKind;
+  }
+
   public FieldSelector getFieldSelector() {
     return fieldSelector;
   }

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

@@ -167,7 +167,7 @@ private Builder(
       this.resourceClass = resourceClass;
       this.groupVersionKind = groupVersionKind;
       this.primaryResourceClass = primaryResourceClass;
-      this.config = InformerConfiguration.builder(resourceClass);
+      this.config = InformerConfiguration.builder(resourceClass, groupVersionKind);
     }
 
     public Builder<R> withName(String name) {

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

@@ -136,4 +136,8 @@ public int hashCode() {
   public String toString() {
     return toGVKString();
   }
+
+  public String getApiVersion() {
+    return apiVersion;
+  }
 }

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

@@ -127,7 +127,7 @@ private void logEventSourceEvent(EventSource eventSource, String event) {
   private <R> Void startEventSource(EventSource<R, P> eventSource) {
     try {
       logEventSourceEvent(eventSource, "Starting");
-      eventSource.start();
+      eventSource.start(false);
       logEventSourceEvent(eventSource, "Started");
     } catch (MissingCRDException e) {
       throw e; // leave untouched
@@ -148,7 +148,7 @@ private <R> Void stopEventSource(EventSource<R, P> eventSource) {
     return null;
   }
 
-  @SuppressWarnings("rawtypes")
+  @SuppressWarnings({"rawtypes", "unchecked"})
   public final synchronized <R> void registerEventSource(EventSource<R, P> eventSource)
       throws OperatorException {
     Objects.requireNonNull(eventSource, "EventSource must not be null");
@@ -251,7 +251,7 @@ public <R> EventSource<R, P> dynamicallyRegisterEventSource(EventSource<R, P> ev
     }
     // The start itself is blocking thus blocking only the threads which are attempt to start the
     // actual event source. Think of this as a form of lock striping.
-    eventSource.start();
+    eventSource.start(true);
     return eventSource;
   }
 

operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/EventSource.java

@@ -19,9 +19,9 @@
 import java.util.Set;
 
 import io.fabric8.kubernetes.api.model.HasMetadata;
+import io.javaoperatorsdk.operator.OperatorException;
 import io.javaoperatorsdk.operator.health.EventSourceHealthIndicator;
 import io.javaoperatorsdk.operator.health.Status;
-import io.javaoperatorsdk.operator.processing.LifecycleAware;
 import io.javaoperatorsdk.operator.processing.event.EventHandler;
 import io.javaoperatorsdk.operator.processing.event.source.filter.GenericFilter;
 import io.javaoperatorsdk.operator.processing.event.source.filter.OnAddFilter;
@@ -37,8 +37,7 @@
  * @param <P> the primary resource type which reconciler needs to be triggered when events occur on
  *     resources of type R
  */
-public interface EventSource<R, P extends HasMetadata>
-    extends LifecycleAware, EventSourceHealthIndicator {
+public interface EventSource<R, P extends HasMetadata> extends EventSourceHealthIndicator {
 
   static String generateName(EventSource<?, ?> eventSource) {
     return eventSource.getClass().getName() + "@" + Integer.toHexString(eventSource.hashCode());
@@ -119,4 +118,22 @@ default Optional<R> getSecondaryResource(P primary) {
   default Status getStatus() {
     return Status.UNKNOWN;
   }
+
+  /**
+   * Start the event source. Normally, should synchronously populate caches, before the method
+   * returns.
+   *
+   * @since 5.6.0
+   * @param dynamicRegistration true if event source registered dynamically, false otherwise
+   */
+  default void start(boolean dynamicRegistration) {
+    start();
+  }
+
+  /** Only for backwards compatible purposes. Use {@link #start(boolean)}. */
+  @Deprecated(forRemoval = true)
+  void start() throws OperatorException;
+
+  /** Strop the event source */
+  void stop() throws OperatorException;
 }

operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/controller/ControllerEventSource.java

@@ -48,7 +48,7 @@ public class ControllerEventSource<T extends HasMetadata>
 
   @SuppressWarnings({"unchecked", "rawtypes"})
   public ControllerEventSource(Controller<T> controller) {
-    super(NAME, controller.getCRClient(), controller.getConfiguration());
+    super(NAME, controller.getConfiguration());
     this.controller = controller;
 
     final var config = controller.getConfiguration();

operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/informer/InformerEventSource.java

@@ -24,8 +24,6 @@
 import org.slf4j.LoggerFactory;
 
 import io.fabric8.kubernetes.api.model.HasMetadata;
-import io.fabric8.kubernetes.client.KubernetesClient;
-import io.fabric8.kubernetes.client.dsl.MixedOperation;
 import io.fabric8.kubernetes.client.informers.ResourceEventHandler;
 import io.javaoperatorsdk.operator.api.config.informer.InformerEventSourceConfiguration;
 import io.javaoperatorsdk.operator.api.reconciler.EventSourceContext;
@@ -56,18 +54,12 @@ public class InformerEventSource<R extends HasMetadata, P extends HasMetadata>
 
   public InformerEventSource(
       InformerEventSourceConfiguration<R> configuration, EventSourceContext<P> context) {
-    this(configuration, configuration.getKubernetesClient().orElse(context.getClient()));
+    this(configuration);
   }
 
   @SuppressWarnings({"unchecked", "rawtypes"})
-  InformerEventSource(InformerEventSourceConfiguration<R> configuration, KubernetesClient client) {
-    super(
-        configuration.name(),
-        configuration
-            .getGroupVersionKind()
-            .map(gvk -> client.genericKubernetesResources(gvk.apiVersion(), gvk.getKind()))
-            .orElseGet(() -> (MixedOperation) client.resources(configuration.getResourceClass())),
-        configuration);
+  InformerEventSource(InformerEventSourceConfiguration<R> configuration) {
+    super(configuration.name(), configuration);
     // If there is a primary to secondary mapper there is no need for primary to secondary index.
     primaryToSecondaryMapper = configuration.getPrimaryToSecondaryMapper();
     if (useSecondaryToPrimaryIndex()) {
@@ -175,11 +167,11 @@ protected void handleEvent(
   }
 
   @Override
-  public synchronized void start() {
+  public synchronized void start(boolean dynamicRegistration) {
     if (isRunning()) {
       return;
     }
-    super.start();
+    super.start(dynamicRegistration);
     // this makes sure that on first reconciliation all resources are
     // present on the index
     manager().list().forEach(r -> primaryToSecondaryIndex.onAddOrUpdate(r, null));