[WIP] Informer pools

bootstrapper-maven-plugin/pom.xml

@@ -22,7 +22,7 @@
   <parent>
     <groupId>io.javaoperatorsdk</groupId>
     <artifactId>java-operator-sdk</artifactId>
-    <version>5.4.1-SNAPSHOT</version>
+    <version>999-SNAPSHOT</version>
   </parent>
 
   <artifactId>bootstrapper</artifactId>

caffeine-bounded-cache-support/pom.xml

@@ -21,7 +21,7 @@
   <parent>
     <groupId>io.javaoperatorsdk</groupId>
     <artifactId>java-operator-sdk</artifactId>
-    <version>5.4.1-SNAPSHOT</version>
+    <version>999-SNAPSHOT</version>
   </parent>
 
   <artifactId>caffeine-bounded-cache-support</artifactId>

micrometer-support/pom.xml

@@ -21,7 +21,7 @@
   <parent>
     <groupId>io.javaoperatorsdk</groupId>
     <artifactId>java-operator-sdk</artifactId>
-    <version>5.4.1-SNAPSHOT</version>
+    <version>999-SNAPSHOT</version>
   </parent>
 
   <artifactId>micrometer-support</artifactId>

migration/pom.xml

@@ -21,7 +21,7 @@
   <parent>
     <groupId>io.javaoperatorsdk</groupId>
     <artifactId>java-operator-sdk</artifactId>
-    <version>5.4.1-SNAPSHOT</version>
+    <version>999-SNAPSHOT</version>
   </parent>
 
   <artifactId>migration</artifactId>

operator-framework-bom/pom.xml

@@ -21,7 +21,7 @@
 
   <groupId>io.javaoperatorsdk</groupId>
   <artifactId>operator-framework-bom</artifactId>
-  <version>5.4.1-SNAPSHOT</version>
+  <version>999-SNAPSHOT</version>
   <packaging>pom</packaging>
   <name>Operator SDK - Bill of Materials</name>
   <description>Java SDK for implementing Kubernetes operators</description>

operator-framework-core/pom.xml

@@ -21,7 +21,7 @@
   <parent>
     <groupId>io.javaoperatorsdk</groupId>
     <artifactId>java-operator-sdk</artifactId>
-    <version>5.4.1-SNAPSHOT</version>
+    <version>999-SNAPSHOT</version>
     <relativePath>../pom.xml</relativePath>
   </parent>
 

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/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
@@ -97,11 +100,14 @@ private void initSources() {
         configuration.getInformerConfig().getEffectiveNamespaces(controllerConfiguration);
     if (InformerConfiguration.allNamespacesWatched(targetNamespaces)) {
       var source = createEventSourceForNamespace(WATCH_ALL_NAMESPACES);
+      source.start();
       log.debug("Registered {} -> {} for any namespace", this, source);
     } else {
       targetNamespaces.forEach(
           ns -> {
+            // todo should we start parallel
             final var source = createEventSourceForNamespace(ns);
+            source.start();
             log.debug("Registered {} -> {} for namespace: {}", this, source, ns);
           });
     }
@@ -111,6 +117,8 @@ 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);
+    // todo remove event handler
+    // remove from pool
     sourcesToRemove.forEach(k -> sources.remove(k).stop());
 
     var newNamespaces =
@@ -133,56 +141,27 @@ public void changeNamespaces(Set<String> namespaces) {
             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 =
+        new InformerClassifier<>(
+            configuration.getInformerConfig().getLabelSelector(),
+            configuration.getInformerConfig().getShardSelector(),
+            namespaceIdentifier,
+            configuration.getResourceClass(),
+            configuration.getInformerConfig().getFieldSelector(),
+            configuration.getInformerConfig().getInformerListLimit(),
+            configuration.getInformerConfig().getItemStore());
     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;
   }
 

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

@@ -30,7 +30,6 @@
 
 import io.fabric8.kubernetes.api.model.GenericKubernetesResource;
 import io.fabric8.kubernetes.api.model.HasMetadata;
-import io.fabric8.kubernetes.client.informers.ExceptionHandler;
 import io.fabric8.kubernetes.client.informers.ResourceEventHandler;
 import io.fabric8.kubernetes.client.informers.SharedIndexInformer;
 import io.fabric8.kubernetes.client.informers.cache.Cache;
@@ -55,86 +54,56 @@ class InformerWrapper<T extends HasMetadata>
 
   public InformerWrapper(
       SharedIndexInformer<T> informer,
-      ConfigurationService configurationService,
-      String namespaceIdentifier) {
+      String namespaceIdentifier,
+      ConfigurationService configurationService) {
     this.informer = informer;
     this.namespaceIdentifier = namespaceIdentifier;
     this.cache = (Cache<T>) informer.getStore();
     this.configurationService = configurationService;
   }
 
+  // todo sync running multiple informers
   @Override
-  public void start() throws OperatorException {
+  public void start() {
+    // change thread name for easier debugging
+    final var thread = Thread.currentThread();
+    final var name = thread.getName();
     try {
-
-      // register stopped handler if we have one defined
-      configurationService
-          .getInformerStoppedHandler()
-          .ifPresent(
-              ish -> {
-                final var stopped = informer.stopped();
-                if (stopped != null) {
-                  stopped.handle(
-                      (res, ex) -> {
-                        ish.onStop(informer, ex);
-                        return null;
-                      });
-                } else {
-                  final var apiTypeClass = informer.getApiTypeClass();
-                  final var fullResourceName = HasMetadata.getFullResourceName(apiTypeClass);
-                  final var version = HasMetadata.getVersion(apiTypeClass);
-                  throw new IllegalStateException(
-                      "Cannot retrieve 'stopped' callback to listen to informer stopping for"
-                          + " informer for "
-                          + fullResourceName
-                          + "/"
-                          + version);
-                }
-              });
-      if (!configurationService.stopOnInformerErrorDuringStartup()) {
-        informer.exceptionHandler((b, t) -> !ExceptionHandler.isDeserializationException(t));
-      }
-      // change thread name for easier debugging
-      final var thread = Thread.currentThread();
-      final var name = thread.getName();
-      try {
-        thread.setName(informerInfo() + " " + thread.getId());
-        final var resourceName = informer.getApiTypeClass().getSimpleName();
-        log.debug(
-            "Starting informer for namespace: {} resource: {}", namespaceIdentifier, resourceName);
-        var start = informer.start();
-        // note that in case we don't put here timeout and stopOnInformerErrorDuringStartup is
-        // false, and there is a rbac issue the get never returns; therefore operator never really
-        // starts
-        log.trace(
-            "Waiting informer to start namespace: {} resource: {}",
-            namespaceIdentifier,
-            resourceName);
-        start
-            .toCompletableFuture()
-            .get(configurationService.cacheSyncTimeout().toMillis(), TimeUnit.MILLISECONDS);
-        log.debug(
-            "Started informer for namespace: {} resource: {}", namespaceIdentifier, resourceName);
-      } catch (TimeoutException | ExecutionException e) {
-        if (configurationService.stopOnInformerErrorDuringStartup()) {
-          log.error("Informer startup error. Operator will be stopped. Informer: {}", informer, e);
-          throw new OperatorException(e);
-        } else {
-          log.warn("Informer startup error. Will periodically retry. Informer: {}", informer, e);
-        }
-      } catch (InterruptedException e) {
-        thread.interrupt();
-        throw new IllegalStateException(e);
-      } finally {
-        // restore original name
-        thread.setName(name);
+      thread.setName(informerInfo() + " " + thread.getId());
+      final var resourceName = informer.getApiTypeClass().getSimpleName();
+      log.debug(
+          "Starting informer for namespace: {} resource: {}", namespaceIdentifier, resourceName);
+      var start = informer.start();
+      // note that in case we don't put here timeout and stopOnInformerErrorDuringStartup is
+      // false, and there is a rbac issue the get never returns; therefore operator never really
+      // starts
+      log.trace(
+          "Waiting informer to start namespace: {} resource: {}",
+          namespaceIdentifier,
+          resourceName);
+      start
+          .toCompletableFuture()
+          .get(configurationService.cacheSyncTimeout().toMillis(), TimeUnit.MILLISECONDS);
+      log.debug(
+          "Started informer for namespace: {} resource: {}", namespaceIdentifier, resourceName);
+    } catch (TimeoutException | ExecutionException e) {
+      if (configurationService.stopOnInformerErrorDuringStartup()) {
+        log.error("Informer startup error. Operator will be stopped. Informer: {}", informer, e);
+        throw new OperatorException(e);
+      } else {
+        log.warn("Informer startup error. Will periodically retry. Informer: {}", informer, e);
       }
-
+    } catch (InterruptedException e) {
+      thread.interrupt();
+      throw new IllegalStateException(e);
     } catch (Exception e) {
       ReconcilerUtilsInternal.handleKubernetesClientException(
           e, HasMetadata.getFullResourceName(informer.getApiTypeClass()));
       throw new OperatorException(
           "Couldn't start informer for " + versionedFullResourceName() + " resources", e);
+    } finally {
+      // restore original name
+      thread.setName(name);
     }
   }
 
@@ -201,7 +170,7 @@ public String toString() {
   }
 
   private String informerInfo() {
-    return "InformerWrapper [" + versionedFullResourceName() + "]";
+    return "InformerWrapper [" + informer.getApiTypeClass().getSimpleName() + "]";
   }
 
   @Override

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

@@ -0,0 +1,18 @@
+/*
+ * Copyright Java Operator SDK Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *         http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.javaoperatorsdk.operator.processing.event.source.informer.pool;
+
+public class AbstractInformerPool {}