feat: Allow configuration of informer pagination through list limit (#1976)

Author: Bijnagte

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

@@ -157,7 +157,8 @@ protected <P extends HasMetadata> ControllerConfiguration<P> configFor(Reconcile
             Constants.NO_VALUE_SET),
         null,
         Utils.instantiate(annotation.itemStore(), ItemStore.class, context), dependentFieldManager,
-        this);
+        this,
+        null);
 
     ResourceEventFilter<P> answer = deprecatedEventFilter(annotation);
     config.setEventFilter(answer != null ? answer : ResourceEventFilters.passthrough());

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

@@ -40,6 +40,7 @@ public class ControllerConfigurationOverrider<R extends HasMetadata> {
   private ItemStore<R> itemStore;
   private String name;
   private String fieldManager;
+  private Long informerListLimit;
 
   private ControllerConfigurationOverrider(ControllerConfiguration<R> original) {
     this.finalizer = original.getFinalizerName();
@@ -56,6 +57,7 @@ private ControllerConfigurationOverrider(ControllerConfiguration<R> original) {
     this.rateLimiter = original.getRateLimiter();
     this.name = original.getName();
     this.fieldManager = original.fieldManager();
+    this.informerListLimit = original.getInformerListLimit().orElse(null);
   }
 
   public ControllerConfigurationOverrider<R> withFinalizer(String finalizer) {
@@ -176,6 +178,20 @@ public ControllerConfigurationOverrider<R> withFieldManager(
     return this;
   }
 
+
+  /**
+   * Sets a max page size limit when starting the informer. This will result in pagination while
+   * populating the cache. This means that longer lists will take multiple requests to fetch. See
+   * {@link io.fabric8.kubernetes.client.dsl.Informable#withLimit(Long)} for more details.
+   *
+   * @param informerListLimit null (the default) results in no pagination
+   */
+  public ControllerConfigurationOverrider<R> withInformerListLimit(
+      Long informerListLimit) {
+    this.informerListLimit = informerListLimit;
+    return this;
+  }
+
   public ControllerConfigurationOverrider<R> replacingNamedDependentResourceConfig(String name,
       Object dependentResourceConfig) {
 
@@ -199,7 +215,7 @@ public ControllerConfiguration<R> build() {
         reconciliationMaxInterval, onAddFilter, onUpdateFilter, genericFilter,
         original.getDependentResources(),
         namespaces, finalizer, labelSelector, configurations, itemStore, fieldManager,
-        original.getConfigurationService());
+        original.getConfigurationService(), informerListLimit);
     overridden.setEventFilter(customResourcePredicate);
     return overridden;
   }

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

@@ -21,11 +21,12 @@ public class DefaultResourceConfiguration<R extends HasMetadata>
   private final String labelSelector;
   private final Set<String> namespaces;
   private final ItemStore<R> itemStore;
+  private final Long informerListLimit;
 
   protected DefaultResourceConfiguration(Class<R> resourceClass,
       Set<String> namespaces, String labelSelector, OnAddFilter<? super R> onAddFilter,
       OnUpdateFilter<? super R> onUpdateFilter, GenericFilter<? super R> genericFilter,
-      ItemStore<R> itemStore) {
+      ItemStore<R> itemStore, Long informerListLimit) {
     this.resourceClass = resourceClass;
     this.resourceTypeName = ReconcilerUtils.getResourceTypeName(resourceClass);
     this.onAddFilter = onAddFilter;
@@ -35,6 +36,7 @@ protected DefaultResourceConfiguration(Class<R> resourceClass,
     this.namespaces = ResourceConfiguration.ensureValidNamespaces(namespaces);
     this.labelSelector = ResourceConfiguration.ensureValidLabelSelector(labelSelector);
     this.itemStore = itemStore;
+    this.informerListLimit = informerListLimit;
   }
 
   @Override
@@ -76,4 +78,9 @@ public Optional<GenericFilter<? super R>> genericFilter() {
   public Optional<ItemStore<R>> getItemStore() {
     return Optional.ofNullable(itemStore);
   }
+
+  @Override
+  public Optional<Long> getInformerListLimit() {
+    return Optional.ofNullable(informerListLimit);
+  }
 }

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

@@ -46,7 +46,8 @@ public ResolvedControllerConfiguration(Class<P> resourceClass, ControllerConfigu
         other.getDependentResources(), other.getNamespaces(),
         other.getFinalizerName(), other.getLabelSelector(), Collections.emptyMap(),
         other.getItemStore().orElse(null), other.fieldManager(),
-        other.getConfigurationService());
+        other.getConfigurationService(),
+        other.getInformerListLimit().orElse(null));
   }
 
   public static Duration getMaxReconciliationInterval(long interval, TimeUnit timeUnit) {
@@ -75,11 +76,11 @@ public ResolvedControllerConfiguration(Class<P> resourceClass, String name,
       Set<String> namespaces, String finalizer, String labelSelector,
       Map<DependentResourceSpec, Object> configurations, ItemStore<P> itemStore,
       String fieldManager,
-      ConfigurationService configurationService) {
+      ConfigurationService configurationService, Long informerListLimit) {
     this(resourceClass, name, generationAware, associatedReconcilerClassName, retry, rateLimiter,
         maxReconciliationInterval, onAddFilter, onUpdateFilter, genericFilter,
         namespaces, finalizer, labelSelector, configurations, itemStore, fieldManager,
-        configurationService);
+        configurationService, informerListLimit);
     setDependentResources(dependentResources);
   }
 
@@ -91,9 +92,9 @@ protected ResolvedControllerConfiguration(Class<P> resourceClass, String name,
       Set<String> namespaces, String finalizer, String labelSelector,
       Map<DependentResourceSpec, Object> configurations, ItemStore<P> itemStore,
       String fieldManager,
-      ConfigurationService configurationService) {
+      ConfigurationService configurationService, Long informerListLimit) {
     super(resourceClass, namespaces, labelSelector, onAddFilter, onUpdateFilter, genericFilter,
-        itemStore);
+        itemStore, informerListLimit);
     this.configurationService = configurationService;
     this.name = ControllerConfiguration.ensureValidName(name, associatedReconcilerClassName);
     this.generationAware = generationAware;
@@ -112,7 +113,7 @@ protected ResolvedControllerConfiguration(Class<P> resourceClass, String name,
       Class<? extends Reconciler> reconcilerClas, ConfigurationService configurationService) {
     this(resourceClass, name, false, getAssociatedReconcilerClassName(reconcilerClas), null, null,
         null, null, null, null, null,
-        null, null, null, null, null, configurationService);
+        null, null, null, null, null, configurationService, null);
   }
 
   @Override

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

@@ -145,4 +145,12 @@ default Set<String> getEffectiveNamespaces(ConfigurationService configurationSer
   default Optional<ItemStore<R>> getItemStore() {
     return Optional.empty();
   }
+
+  /**
+   * The maximum amount of items to return for a single list call when starting an informer. If this
+   * is a not null it will result in paginating for the initial load of the informer cache.
+   */
+  default Optional<Long> getInformerListLimit() {
+    return Optional.empty();
+  }
 }

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

@@ -40,9 +40,9 @@ protected DefaultInformerConfiguration(String labelSelector,
         OnUpdateFilter<? super R> onUpdateFilter,
         OnDeleteFilter<? super R> onDeleteFilter,
         GenericFilter<? super R> genericFilter,
-        ItemStore<R> itemStore) {
+        ItemStore<R> itemStore, Long informerListLimit) {
       super(resourceClass, namespaces, labelSelector, onAddFilter, onUpdateFilter, genericFilter,
-          itemStore);
+          itemStore, informerListLimit);
       this.followControllerNamespaceChanges = followControllerNamespaceChanges;
 
       this.primaryToSecondaryMapper = primaryToSecondaryMapper;
@@ -119,6 +119,7 @@ class InformerConfigurationBuilder<R extends HasMetadata> {
     private GenericFilter<? super R> genericFilter;
     private boolean inheritControllerNamespacesOnChange = false;
     private ItemStore<R> itemStore;
+    private Long informerListLimit;
 
     private InformerConfigurationBuilder(Class<R> resourceClass) {
       this.resourceClass = resourceClass;
@@ -226,12 +227,24 @@ public InformerConfigurationBuilder<R> withItemStore(ItemStore<R> itemStore) {
       return this;
     }
 
+    /**
+     * Sets a max page size limit when starting the informer. This will result in pagination while
+     * populating the cache. This means that longer lists will take multiple requests to fetch. See
+     * {@link io.fabric8.kubernetes.client.dsl.Informable#withLimit(Long)} for more details.
+     *
+     * @param informerListLimit null (the default) results in no pagination
+     */
+    public InformerConfigurationBuilder<R> withInformerListLimit(Long informerListLimit) {
+      this.informerListLimit = informerListLimit;
+      return this;
+    }
+
     public InformerConfiguration<R> build() {
       return new DefaultInformerConfiguration<>(labelSelector, resourceClass,
           primaryToSecondaryMapper,
           secondaryToPrimaryMapper,
           namespaces, inheritControllerNamespacesOnChange, onAddFilter, onUpdateFilter,
-          onDeleteFilter, genericFilter, itemStore);
+          onDeleteFilter, genericFilter, itemStore, informerListLimit);
     }
   }
 

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

@@ -128,7 +128,8 @@ private InformerWrapper<T> createEventSourceForNamespace(String namespace) {
   private InformerWrapper<T> createEventSource(
       FilterWatchListDeletable<T, KubernetesResourceList<T>, Resource<T>> filteredBySelectorClient,
       ResourceEventHandler<T> eventHandler, String namespaceIdentifier) {
-    var informer = filteredBySelectorClient.runnableInformer(0);
+    var informer = configuration.getInformerListLimit().map(filteredBySelectorClient::withLimit)
+        .orElse(filteredBySelectorClient).runnableInformer(0);
     configuration.getItemStore().ifPresent(informer::itemStore);
     var source = new InformerWrapper<>(informer, configurationService, namespaceIdentifier);
     source.addEventHandler(eventHandler);

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

@@ -12,13 +12,7 @@
 import io.fabric8.kubernetes.api.model.authorization.v1.SubjectRulesReviewStatus;
 import io.fabric8.kubernetes.client.KubernetesClient;
 import io.fabric8.kubernetes.client.V1ApiextensionAPIGroupDSL;
-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.MixedOperation;
-import io.fabric8.kubernetes.client.dsl.NamespaceableResource;
-import io.fabric8.kubernetes.client.dsl.NonNamespaceOperation;
-import io.fabric8.kubernetes.client.dsl.Resource;
+import io.fabric8.kubernetes.client.dsl.*;
 import io.fabric8.kubernetes.client.extended.leaderelection.LeaderElectorBuilder;
 import io.fabric8.kubernetes.client.informers.SharedIndexInformer;
 import io.fabric8.kubernetes.client.informers.cache.Indexer;
@@ -76,8 +70,15 @@ public static <T extends HasMetadata> KubernetesClient client(Class<T> clazz,
     }
     doAnswer(invocation -> null).when(informer).stop();
     Indexer mockIndexer = mock(Indexer.class);
+
     when(informer.getIndexer()).thenReturn(mockIndexer);
+
     when(filterable.runnableInformer(anyLong())).thenReturn(informer);
+
+    Informable<T> informable = mock(Informable.class);
+    when(filterable.withLimit(anyLong())).thenReturn(informable);
+    when(informable.runnableInformer(anyLong())).thenReturn(informer);
+
     when(client.resources(clazz)).thenReturn(resources);
     when(client.leaderElector())
         .thenReturn(new LeaderElectorBuilder(client, Executors.newSingleThreadExecutor()));

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

@@ -145,7 +145,7 @@ public ControllerConfig(String finalizer, boolean generationAware,
           null,
           null,
           null,
-          null, null, null, finalizer, null, null, null, new BaseConfigurationService());
+          null, null, null, finalizer, null, null, null, new BaseConfigurationService(), null);
       setEventFilter(eventFilter);
     }
   }

operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/source/controller/ControllerResourceEventSourceTest.java

@@ -198,7 +198,8 @@ public TestConfiguration(boolean generationAware, OnAddFilter<TestCustomResource
           null,
           null,
           FINALIZER,
-          null, null, null, new BaseConfigurationService());
+          null, null, null, new BaseConfigurationService(),
+          null);
     }
   }
 }