Informer pools

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,16 @@ 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();
+      informerPool.setConfigurationService(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

@@ -76,6 +76,7 @@ default boolean followControllerNamespaceChanges() {
 
   <P extends HasMetadata> PrimaryToSecondaryMapper<P> getPrimaryToSecondaryMapper();
 
+  // todo deprecate
   Optional<GroupVersionKind> getGroupVersionKind();
 
   default String name() {
@@ -167,7 +168,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

@@ -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");
@@ -249,8 +249,6 @@ public <R> EventSource<R, P> dynamicallyRegisterEventSource(EventSource<R, P> ev
         registerEventSource(eventSource);
       }
     }
-    // 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();
     return eventSource;
   }

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;
@@ -54,20 +52,18 @@ public class InformerEventSource<R extends HasMetadata, P extends HasMetadata>
   private final PrimaryToSecondaryIndex<R> primaryToSecondaryIndex;
   private final PrimaryToSecondaryMapper<P> primaryToSecondaryMapper;
 
+  /**
+   * @deprecated use {@link InformerEventSource(InformerEventSourceConfiguration)}
+   */
+  // todo migrate sample, separate PR?
+  @Deprecated(forRemoval = true)
   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);
+  public 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()) {