[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,14 @@ public ExecutorServiceManager getExecutorServiceManager() {
     }
     return executorServiceManager;
   }
+
+  @Override
+  public InformerPool informerPool() {
+    // cached so that all controllers backed by this ConfigurationService share the same pool and
+    // can therefore share the underlying informers
+    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,8 @@ default boolean useSSAToPatchPrimaryResource() {
   default boolean cloneSecondaryResourcesWhenGettingFromCache() {
     return false;
   }
+
+  default InformerPool informerPool() {
+    return new DefaultInformerPool(getKubernetesClient(), this);
+  }
 }

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/source/informer/InformerManager.java

@@ -27,7 +27,6 @@
 
 import io.fabric8.kubernetes.api.model.HasMetadata;
 import io.fabric8.kubernetes.api.model.KubernetesResourceList;
-import io.fabric8.kubernetes.client.dsl.FilterWatchListDeletable;
 import io.fabric8.kubernetes.client.dsl.MixedOperation;
 import io.fabric8.kubernetes.client.dsl.Resource;
 import io.fabric8.kubernetes.client.informers.ResourceEventHandler;
@@ -41,6 +40,8 @@
 import io.javaoperatorsdk.operator.processing.event.ResourceID;
 import io.javaoperatorsdk.operator.processing.event.source.Cache;
 import io.javaoperatorsdk.operator.processing.event.source.IndexerResourceCache;
+import io.javaoperatorsdk.operator.processing.event.source.informer.pool.InformerClassifier;
+import io.javaoperatorsdk.operator.processing.event.source.informer.pool.InformerPool;
 
 import static io.javaoperatorsdk.operator.api.reconciler.Constants.WATCH_ALL_NAMESPACES;
 
@@ -55,6 +56,7 @@ class InformerManager<R extends HasMetadata, C extends Informable<R>>
   private final ResourceEventHandler<R> eventHandler;
   private final Map<String, Function<R, List<String>>> indexers = new HashMap<>();
   private ControllerConfiguration<R> controllerConfiguration;
+  private InformerPool informerPool;
 
   InformerManager(
       MixedOperation<R, KubernetesResourceList<R>, Resource<R>> client,
@@ -67,6 +69,7 @@ class InformerManager<R extends HasMetadata, C extends Informable<R>>
 
   void setControllerConfiguration(ControllerConfiguration<R> controllerConfiguration) {
     this.controllerConfiguration = controllerConfiguration;
+    this.informerPool = controllerConfiguration.getConfigurationService().informerPool();
   }
 
   @Override
@@ -101,7 +104,9 @@ private void initSources() {
     } else {
       targetNamespaces.forEach(
           ns -> {
+            // todo should we start parallel
             final var source = createEventSourceForNamespace(ns);
+
             log.debug("Registered {} -> {} for namespace: {}", this, source, ns);
           });
     }
@@ -111,7 +116,16 @@ public void changeNamespaces(Set<String> namespaces) {
     var sourcesToRemove =
         sources.keySet().stream().filter(k -> !namespaces.contains(k)).collect(Collectors.toSet());
     log.debug("Stopped informer {} for namespaces: {}", this, sourcesToRemove);
-    sourcesToRemove.forEach(k -> sources.remove(k).stop());
+    // todo remove event handler
+    // remove from pool
+    sourcesToRemove.forEach(
+        k -> {
+          var optionalInformer =
+              informerPool.releaseInformer(
+                  configuration.getInformerConfig().getName(), getClassifier(k));
+          sources.remove(k);
+          optionalInformer.ifPresent(i -> i.removeEventHandler(eventHandler));
+        });
 
     var newNamespaces =
         namespaces.stream().filter(ns -> !sources.containsKey(ns)).collect(Collectors.toList());
@@ -126,66 +140,38 @@ public void changeNamespaces(Set<String> namespaces) {
             newNamespaces.stream(),
             ns -> {
               final var source = createEventSourceForNamespace(ns);
-              source.start();
               log.debug("Registered new {} -> {} for namespace: {}", this, source, ns);
               return null;
             },
             ns -> "InformerStarter-" + ns + "-" + configuration.getResourceClass().getSimpleName());
   }
 
-  private InformerWrapper<R> createEventSourceForNamespace(String namespace) {
+  private InformerWrapper<R> createEventSourceForNamespace(String namespaceIdentifier) {
     final InformerWrapper<R> source;
-    final var labelSelector = configuration.getInformerConfig().getLabelSelector();
-    final var shardSelector = configuration.getInformerConfig().getShardSelector();
-    if (namespace.equals(WATCH_ALL_NAMESPACES)) {
-      final var filteredBySelectorClient =
-          client.inAnyNamespace().withLabelSelector(labelSelector).withShardSelector(shardSelector);
-      source = createEventSource(filteredBySelectorClient, eventHandler, WATCH_ALL_NAMESPACES);
-    } else {
-      source =
-          createEventSource(
-              client
-                  .inNamespace(namespace)
-                  .withLabelSelector(labelSelector)
-                  .withShardSelector(shardSelector),
-              eventHandler,
-              namespace);
-    }
-    source.addIndexers(indexers);
-    return source;
-  }
-
-  private InformerWrapper<R> createEventSource(
-      FilterWatchListDeletable<R, KubernetesResourceList<R>, Resource<R>> filteredBySelectorClient,
-      ResourceEventHandler<R> eventHandler,
-      String namespaceIdentifier) {
-    final var informerConfig = configuration.getInformerConfig();
-
-    if (informerConfig.getFieldSelector() != null
-        && !informerConfig.getFieldSelector().getFields().isEmpty()) {
-      for (var f : informerConfig.getFieldSelector().getFields()) {
-        if (f.negated()) {
-          filteredBySelectorClient = filteredBySelectorClient.withoutField(f.path(), f.value());
-        } else {
-          filteredBySelectorClient = filteredBySelectorClient.withField(f.path(), f.value());
-        }
-      }
-    }
-
+    InformerClassifier<R> classifier = getClassifier(namespaceIdentifier);
     var informer =
-        Optional.ofNullable(informerConfig.getInformerListLimit())
-            .map(filteredBySelectorClient::withLimit)
-            .orElse(filteredBySelectorClient)
-            .runnableInformer(0);
-    Optional.ofNullable(informerConfig.getItemStore()).ifPresent(informer::itemStore);
-    var source =
+        informerPool.getInformer(configuration.getInformerConfig().getName(), classifier);
+    source =
         new InformerWrapper<>(
-            informer, controllerConfiguration.getConfigurationService(), namespaceIdentifier);
+            informer, namespaceIdentifier, controllerConfiguration.getConfigurationService());
     source.addEventHandler(eventHandler);
     sources.put(namespaceIdentifier, source);
+    source.addIndexers(indexers);
     return source;
   }
 
+  private InformerClassifier<R> getClassifier(String namespaceIdentifier) {
+    return new InformerClassifier<>(
+        configuration.getInformerConfig().getLabelSelector(),
+        configuration.getInformerConfig().getShardSelector(),
+        namespaceIdentifier,
+        configuration.getResourceClass(),
+        configuration.getInformerConfig().getResourceGroupVersionKind(),
+        configuration.getInformerConfig().getFieldSelector(),
+        configuration.getInformerConfig().getInformerListLimit(),
+        configuration.getInformerConfig().getItemStore());
+  }
+
   @Override
   public void stop() {
     sources.forEach(