feat: informer pool to share informers across controllers and event sources

Introduce an informer pool so that InformerEventSources watching the same
resource type with an equivalent configuration share a single underlying
fabric8 SharedIndexInformer instead of creating one per event source. This
reduces memory usage and the number of watch connections opened against the API
server when many controllers watch the same (secondary) resource type.

- Add an InformerPool abstraction with two strategies:
  - DefaultInformerPool (default): reference-counted sharing of informers keyed
    by an InformerClassifier (API server URL, resource type / GVK, namespace,
    label/field/shard selectors, item store). The informer is created on first
    use and stopped only when the last user releases it. informerListLimit is
    intentionally not part of the identity; indexers are added independently.
  - AlwaysNewInformerPool: never shares (one informer per event source), to opt
    out of pooling and keep the previous behavior.
- Resolve the pool via ConfigurationService.informerPool() (an abstract method,
  cached as a per-ConfigurationService singleton in AbstractConfigurationService)
  and expose ConfigurationServiceOverrider.withInformerPool(...) to select the
  strategy.
- Route InformerManager / ManagedInformerEventSource through the pool. Dynamic
  event source registration reuses an already-running informer and relies on the
  informer replaying its cache to the newly added handler for initial state.
- Add group/version/kind and field-selector support to the classifier.
- Tests: unit tests for InformerClassifier equality and pool reference counting;
  integration tests for basic sharing, dynamic registration and de-registration,
  each run against both pool strategies.
- Add documentation for informer pooling.

The feature is marked @experimental: the runtime behavior is production-ready,
but the configuration API may still change in a non-backwards-compatible way.

Signed-off-by: Attila Mészáros <a_meszaros@apple.com>

Author: csviri

docs/content/en/docs/documentation/eventing.md

@@ -346,4 +346,57 @@ for [primary resources](https://github.com/operator-framework/java-operator-sdk/
 
 See
 also [CaffeineBoundedItemStores](https://github.com/operator-framework/java-operator-sdk/blob/main/caffeine-bounded-cache-support/src/main/java/io/javaoperatorsdk/operator/processing/event/source/cache/CaffeineBoundedItemStores.java)
-for more details.
+for more details.
+
+### Sharing Informers Between Controllers (Informer Pool)
+
+{{% alert title="Experimental" color="warning" %}}
+Informer pooling is marked `@Experimental`: the feature itself is production ready, but its
+configuration API may still change in a non-backwards-compatible way.
+{{% /alert %}}
+
+By default JOSDK maintains an *informer pool* so that informers are **shared** across controllers
+and event sources. When several `InformerEventSource`s (whether belonging to different controllers,
+or dynamically registered at runtime) watch the same resource type with an equivalent configuration,
+they are all backed by a single underlying `SharedIndexInformer` instead of one informer each. This
+reduces memory usage and the number of watch connections opened against the API server — which
+matters in operators where many controllers watch the same secondary resource type (for example
+`ConfigMap` or `Secret`).
+
+Two event sources share an informer when their effective informer configuration matches on all of:
+
+- the target cluster (API server URL, including the remote client used for
+  [multi-cluster](#informereventsource-multi-cluster-support) event sources),
+- the resource type (or the group/version/kind for generic resources),
+- the watched namespace,
+- the label, field and shard selectors,
+- the configured [item store](#bounded-caches-for-informers).
+
+The `informerListLimit` is intentionally *not* part of this identity: if two otherwise-equivalent
+event sources request a different list limit, the existing informer is reused (a warning is logged
+and the first-configured limit is kept). Indexers are also not part of the identity, since they can
+be added to a shared informer independently; just make sure indexer names don't collide.
+
+The pool is reference-counted: the shared informer is created on first use and only stopped once the
+last event source using it is de-registered (or its controller stops). Dynamically registering an
+event source for a resource that is already backed by a running informer reuses that informer, and
+the initial state already in its cache is replayed to the newly added handler.
+
+#### Selecting the pooling strategy
+
+The strategy is provided by the
+[`InformerPool`](https://github.com/operator-framework/java-operator-sdk/blob/main/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/informer/pool/InformerPool.java)
+configured on the `ConfigurationService`. Two implementations are available:
+
+- [`DefaultInformerPool`](https://github.com/operator-framework/java-operator-sdk/blob/main/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/informer/pool/DefaultInformerPool.java)
+  (the default): shares informers as described above.
+- [`AlwaysNewInformerPool`](https://github.com/operator-framework/java-operator-sdk/blob/main/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/informer/pool/AlwaysNewInformerPool.java):
+  never shares informers, creating a dedicated informer for every event source. Use this to opt out
+  of pooling and restore the pre-pooling behavior.
+
+You can override the strategy through the `ConfigurationService`:
+
+```java
+Operator operator = new Operator(overrider ->
+    overrider.withInformerPool(new AlwaysNewInformerPool()));
+```

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

@@ -24,6 +24,8 @@
 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.DefaultInformerPool;
+import io.javaoperatorsdk.operator.processing.event.source.informer.pool.InformerPool;
 
 /**
  * An abstract implementation of {@link ConfigurationService} meant to ease custom implementations
@@ -35,6 +37,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 +193,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 = new DefaultInformerPool(this);
+    }
+    return informerPool;
+  }
 }

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

@@ -43,6 +43,7 @@
 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.InformerPool;
 
 /** An interface from which to retrieve configuration information. */
 public interface ConfigurationService {
@@ -476,4 +477,21 @@ default boolean useSSAToPatchPrimaryResource() {
   default boolean cloneSecondaryResourcesWhenGettingFromCache() {
     return false;
   }
+
+  /**
+   * The {@link InformerPool} used to create and (when using the default, sharing pool) share the
+   * informers backing the event sources of all controllers managed by this {@code
+   * ConfigurationService}.
+   *
+   * <p><strong>Implementations must return the same instance on every call.</strong> The pool is
+   * effectively a per-{@code ConfigurationService} singleton: controllers share informers only if
+   * they resolve the same pool, and reference counting / informer shutdown are only correct if
+   * {@code getInformer} and {@code releaseInformer} operate on that same instance. This is
+   * intentionally not a {@code default} method, since a {@code default} could not cache the result
+   * and would hand out a fresh (unshared) pool on each call; {@link AbstractConfigurationService}
+   * provides a cached implementation backed by the default sharing pool.
+   *
+   * @return the informer pool for this configuration service
+   */
+  InformerPool informerPool();
 }

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;
 }