wip

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

Author: csviri

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

@@ -120,9 +120,11 @@ public void changeNamespaces(Set<String> namespaces) {
     // remove from pool
     sourcesToRemove.forEach(
         k -> {
-          informerPool.releaseInformer(
-              configuration.getInformerConfig().getName(), getClassifier(k));
+          var optionalInformer =
+              informerPool.releaseInformer(
+                  configuration.getInformerConfig().getName(), getClassifier(k));
           sources.remove(k);
+          optionalInformer.ifPresent(i -> i.removeEventHandler(eventHandler));
         });
 
     var newNamespaces =

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

@@ -16,14 +16,20 @@
 package io.javaoperatorsdk.operator.processing.event.source.informer.pool;
 
 import java.util.Map;
+import java.util.Optional;
 import java.util.concurrent.ConcurrentHashMap;
 
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
 import io.fabric8.kubernetes.api.model.HasMetadata;
 import io.fabric8.kubernetes.client.informers.SharedIndexInformer;
 
 @SuppressWarnings({"unchecked", "rawtypes"})
 public class AlwaysCreateInformerPool extends AbstractInformerPool {
 
+  private static final Logger log = LoggerFactory.getLogger(AlwaysCreateInformerPool.class);
+
   private Map<ClassifierWithName, SharedIndexInformer> informers = new ConcurrentHashMap();
 
   @Override
@@ -38,9 +44,15 @@ public <R extends HasMetadata> SharedIndexInformer<R> getInformer(
   // todo controller name should be added here? for better naming => referencing and have guaranteed
   // uniqueness
   @Override
-  public <R extends HasMetadata> void releaseInformer(
+  public <R extends HasMetadata> Optional<SharedIndexInformer<R>> releaseInformer(
       String name, InformerClassifier<R> classifier) {
-    informers.get(new ClassifierWithName(name, classifier)).stop();
+    var informer = informers.get(new ClassifierWithName(name, classifier));
+    if (informer != null) {
+      informer.stop();
+    } else {
+      log.warn("Informer was not found for classifier: {}", classifier);
+    }
+    return Optional.ofNullable(informer);
   }
 
   public record ClassifierWithName<R extends HasMetadata>(

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

@@ -17,6 +17,7 @@
 
 import java.util.HashMap;
 import java.util.Map;
+import java.util.Optional;
 import java.util.concurrent.atomic.AtomicInteger;
 
 import org.slf4j.Logger;
@@ -41,6 +42,7 @@ public DefaultInformerPool(KubernetesClient client, ConfigurationService service
   }
 
   @SuppressWarnings("unchecked")
+  @Override
   public <R extends HasMetadata> SharedIndexInformer<R> getInformer(
       String name, InformerClassifier<R> classifier) {
     SharedIndexInformer<R> informer;
@@ -51,29 +53,58 @@ public <R extends HasMetadata> SharedIndexInformer<R> getInformer(
         informers.put(classifier, informer);
         counters.put(classifier, new AtomicInteger(1));
       } else {
+        informers.keySet().stream()
+            .filter(existing -> existing.differsOnlyByInformerListLimit(classifier))
+            .findFirst()
+            .ifPresent(
+                existing ->
+                    log.warn(
+                        "Reusing informer for classifier {} that differs only by informerListLimit"
+                            + " (existing: {}, requested: {}). The existing informerListLimit is"
+                            + " kept.",
+                        classifier,
+                        existing.informerListLimit(),
+                        classifier.informerListLimit()));
         counters.get(classifier).incrementAndGet();
       }
     }
     start(informer, classifier);
     return informer;
   }
 
-  @SuppressWarnings("rawtypes")
-  public synchronized <R extends HasMetadata> void releaseInformer(
+  @SuppressWarnings("unchecked")
+  public synchronized <R extends HasMetadata> Optional<SharedIndexInformer<R>> releaseInformer(
       String name, InformerClassifier<R> classifier) {
-    SharedIndexInformer informer = null;
+    SharedIndexInformer<R> informer = null;
     synchronized (this) {
       var counter = counters.get(classifier);
+      informer = (SharedIndexInformer<R>) informers.get(classifier);
       if (counter != null && counter.decrementAndGet() == 0) {
-        informer = informers.get(classifier);
         counters.remove(classifier);
         informers.remove(classifier);
+        // todo check if we can remove event handled if informer stopped
+        informer.stop();
       } else {
         log.warn("No informer found in the pool.");
       }
     }
-    if (informer != null) {
-      informer.stop();
-    }
+    return Optional.ofNullable(informer);
+  }
+
+  /** Total number of distinct informers currently held in the pool. */
+  public synchronized int size() {
+    return informers.size();
+  }
+
+  /**
+   * Number of distinct informers currently held in the pool for the given resource type. When
+   * multiple controllers share a single informer for a resource, this returns {@code 1} for that
+   * resource type regardless of how many controllers use it.
+   */
+  public synchronized long numberOfInformersForResource(
+      Class<? extends HasMetadata> resourceClass) {
+    return informers.keySet().stream()
+        .filter(classifier -> resourceClass.equals(classifier.resourceClass()))
+        .count();
   }
 }

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

@@ -15,12 +15,13 @@
  */
 package io.javaoperatorsdk.operator.processing.event.source.informer.pool;
 
+import java.util.Objects;
+
 import io.fabric8.kubernetes.api.model.HasMetadata;
 import io.fabric8.kubernetes.client.informers.cache.ItemStore;
 import io.javaoperatorsdk.operator.api.config.informer.FieldSelector;
 import io.javaoperatorsdk.operator.processing.GroupVersionKind;
 
-// todo add this to docs
 /**
  * Indexers are not part of classifier since those can be added dynamically to an informer, also can
  * live side by side from different controllers, just users have to avoid naming collision.
@@ -33,4 +34,42 @@ public record InformerClassifier<R extends HasMetadata>(
     GroupVersionKind groupVersionKind,
     FieldSelector fieldSelector,
     Long informerListLimit,
-    ItemStore<R> itemStore) {}
+    ItemStore<R> itemStore) {
+
+  @Override
+  public boolean equals(Object o) {
+    if (this == o) {
+      return true;
+    }
+    if (!(o instanceof InformerClassifier<?> that)) {
+      return false;
+    }
+    return Objects.equals(labelSelector, that.labelSelector)
+        && Objects.equals(shardSelector, that.shardSelector)
+        && Objects.equals(namespaceIdentifier, that.namespaceIdentifier)
+        && Objects.equals(resourceClass, that.resourceClass)
+        && Objects.equals(groupVersionKind, that.groupVersionKind)
+        && Objects.equals(fieldSelector, that.fieldSelector)
+        && Objects.equals(itemStore, that.itemStore);
+  }
+
+  @Override
+  public int hashCode() {
+    return Objects.hash(
+        labelSelector,
+        shardSelector,
+        namespaceIdentifier,
+        resourceClass,
+        groupVersionKind,
+        fieldSelector,
+        itemStore);
+  }
+
+  /**
+   * Checks whether this classifier and the other are equal in every attribute except for the {@link
+   * #informerListLimit()}, which differs between them.
+   */
+  public boolean differsOnlyByInformerListLimit(InformerClassifier<?> other) {
+    return equals(other) && !Objects.equals(informerListLimit, other.informerListLimit);
+  }
+}

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

@@ -15,6 +15,8 @@
  */
 package io.javaoperatorsdk.operator.processing.event.source.informer.pool;
 
+import java.util.Optional;
+
 import io.fabric8.kubernetes.api.model.HasMetadata;
 import io.fabric8.kubernetes.client.KubernetesClient;
 import io.fabric8.kubernetes.client.informers.SharedIndexInformer;
@@ -29,7 +31,8 @@ public interface InformerPool {
   <R extends HasMetadata> SharedIndexInformer<R> getInformer(
       String name, InformerClassifier<R> classifier);
 
-  <R extends HasMetadata> void releaseInformer(String name, InformerClassifier<R> classifier);
+  <R extends HasMetadata> Optional<SharedIndexInformer<R>> releaseInformer(
+      String name, InformerClassifier<R> classifier);
 
   void setClient(KubernetesClient client);
 

operator-framework-core/src/test/java/io/javaoperatorsdk/operator/MockKubernetesClient.java

@@ -31,7 +31,6 @@
 import io.fabric8.kubernetes.client.dsl.AnyNamespaceOperation;
 import io.fabric8.kubernetes.client.dsl.ApiextensionsAPIGroupDSL;
 import io.fabric8.kubernetes.client.dsl.FilterWatchListDeletable;
-import io.fabric8.kubernetes.client.dsl.Informable;
 import io.fabric8.kubernetes.client.dsl.MixedOperation;
 import io.fabric8.kubernetes.client.dsl.NamespaceableResource;
 import io.fabric8.kubernetes.client.dsl.NonNamespaceOperation;
@@ -112,9 +111,9 @@ public static <T extends HasMetadata> KubernetesClient client(
 
     when(filterable.runnableInformer(anyLong())).thenReturn(informer);
 
-    Informable<T> informable = mock(Informable.class);
-    when(filterable.withLimit(anyLong())).thenReturn(informable);
-    when(informable.runnableInformer(anyLong())).thenReturn(informer);
+    // The informer pool casts the result of withLimit() back to FilterWatchListDeletable, so it has
+    // to return the filterable mock (which is one) rather than a plain Informable mock.
+    when(filterable.withLimit(anyLong())).thenReturn(filterable);
 
     when(client.resources(clazz)).thenReturn(resources);
     when(client.leaderElector())

operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/ControllerTest.java

@@ -62,6 +62,9 @@ class ControllerTest {
   @Test
   void crdShouldNotBeCheckedForNativeResources() {
     final var client = MockKubernetesClient.client(Secret.class);
+    final var configurationService =
+        ConfigurationService.newOverriddenConfigurationService(
+            this.configurationService, o -> o.withKubernetesClient(client));
     final var configuration =
         MockControllerConfiguration.forResource(Secret.class, configurationService);
     final var controller = new Controller<Secret>(reconciler, configuration, client);
@@ -75,7 +78,8 @@ void notifiesMetricsWhenEventProcessorStarts() {
     final var metrics = mock(Metrics.class);
     final var configurationService =
         ConfigurationService.newOverriddenConfigurationService(
-            new BaseConfigurationService(), o -> o.withMetrics(metrics));
+            new BaseConfigurationService(),
+            o -> o.withMetrics(metrics).withKubernetesClient(client));
     final var configuration =
         MockControllerConfiguration.forResource(Secret.class, configurationService);
     final var controller = new Controller<Secret>(reconciler, configuration, client);
@@ -95,7 +99,8 @@ void doesNotNotifyMetricsWhenEventProcessorNotStarted() {
     final var metrics = mock(Metrics.class);
     final var configurationService =
         ConfigurationService.newOverriddenConfigurationService(
-            new BaseConfigurationService(), o -> o.withMetrics(metrics));
+            new BaseConfigurationService(),
+            o -> o.withMetrics(metrics).withKubernetesClient(client));
     final var configuration =
         MockControllerConfiguration.forResource(Secret.class, configurationService);
     final var controller = new Controller<Secret>(reconciler, configuration, client);
@@ -110,7 +115,8 @@ void crdShouldNotBeCheckedForCustomResourcesIfDisabled() {
     final var client = MockKubernetesClient.client(TestCustomResource.class);
     ConfigurationService configurationService =
         ConfigurationService.newOverriddenConfigurationService(
-            new BaseConfigurationService(), o -> o.checkingCRDAndValidateLocalModel(false));
+            new BaseConfigurationService(),
+            o -> o.checkingCRDAndValidateLocalModel(false).withKubernetesClient(client));
 
     final var configuration =
         MockControllerConfiguration.forResource(TestCustomResource.class, configurationService);

operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/EventSourceManagerTest.java

@@ -24,6 +24,7 @@
 import io.javaoperatorsdk.operator.MockKubernetesClient;
 import io.javaoperatorsdk.operator.OperatorException;
 import io.javaoperatorsdk.operator.api.config.BaseConfigurationService;
+import io.javaoperatorsdk.operator.api.config.ConfigurationService;
 import io.javaoperatorsdk.operator.api.config.MockControllerConfiguration;
 import io.javaoperatorsdk.operator.api.config.informer.InformerEventSourceConfiguration;
 import io.javaoperatorsdk.operator.api.reconciler.Reconciler;
@@ -202,12 +203,14 @@ void changesNamespacesOnControllerAndInformerEventSources() {
 
   private EventSourceManager initManager() {
     final var configuration = MockControllerConfiguration.forResource(ConfigMap.class);
-    final var configService = new BaseConfigurationService();
+    final var mockClient = MockKubernetesClient.client(ConfigMap.class);
+    final var configService =
+        ConfigurationService.newOverriddenConfigurationService(
+            new BaseConfigurationService(),
+            overrider -> overrider.withKubernetesClient(mockClient));
     when(configuration.getConfigurationService()).thenReturn(configService);
 
-    final Controller controller =
-        new Controller(
-            mock(Reconciler.class), configuration, MockKubernetesClient.client(ConfigMap.class));
+    final Controller controller = new Controller(mock(Reconciler.class), configuration, mockClient);
     return new EventSourceManager(controller);
   }
 }