Add cache-level predicate support for informers

Co-authored-by: brendandburns <5751682+brendandburns@users.noreply.github.com>

Author: Copilot

e2e/src/test/java/io/kubernetes/client/e2e/informer/NamespaceInformerTest.java

@@ -19,10 +19,13 @@
 import io.kubernetes.client.informer.SharedInformerFactory;
 import io.kubernetes.client.informer.cache.Lister;
 import io.kubernetes.client.openapi.ApiClient;
+import io.kubernetes.client.openapi.apis.CoreV1Api;
 import io.kubernetes.client.openapi.models.V1Namespace;
 import io.kubernetes.client.openapi.models.V1NamespaceList;
+import io.kubernetes.client.openapi.models.V1ObjectMeta;
 import io.kubernetes.client.util.ClientBuilder;
 import io.kubernetes.client.util.generic.GenericKubernetesApi;
+import java.util.stream.Collectors;
 import org.junit.jupiter.api.Test;
 
 class NamespaceInformerTest {
@@ -57,4 +60,61 @@ void listWatchingNamespaces() throws Exception {
       informerFactory.stopAllRegisteredInformers(true);
     }
   }
+
+  @Test
+  void listWatchingNamespacesWithCachePredicate() throws Exception {
+    ApiClient client = ClientBuilder.defaultClient();
+    CoreV1Api coreV1Api = new CoreV1Api(client);
+    SharedInformerFactory informerFactory = new SharedInformerFactory(client);
+    String selectedNamespace = "e2e-cache-selected";
+    String ignoredNamespace = "e2e-cache-ignored";
+
+    coreV1Api
+        .createNamespace(
+            new V1Namespace()
+                .metadata(
+                    new V1ObjectMeta()
+                        .name(selectedNamespace)
+                        .labels(java.util.Map.of("cache-filter", "keep"))))
+        .execute();
+    coreV1Api
+        .createNamespace(
+            new V1Namespace()
+                .metadata(
+                    new V1ObjectMeta()
+                        .name(ignoredNamespace)
+                        .labels(java.util.Map.of("cache-filter", "drop"))))
+        .execute();
+
+    GenericKubernetesApi<V1Namespace, V1NamespaceList> api =
+        new GenericKubernetesApi<>(V1Namespace.class, V1NamespaceList.class, "", "v1", "namespaces", client);
+    SharedIndexInformer<V1Namespace> nsInformer =
+        informerFactory.sharedIndexInformerFor(
+            api,
+            V1Namespace.class,
+            0,
+            ns ->
+                ns.getMetadata() != null
+                    && ns.getMetadata().getLabels() != null
+                    && "keep".equals(ns.getMetadata().getLabels().get("cache-filter")));
+
+    try {
+      informerFactory.startAllRegisteredInformers();
+      await().untilAsserted(() -> assertThat(nsInformer.hasSynced()).isTrue());
+      await()
+          .untilAsserted(
+              () -> {
+                java.util.List<String> cachedNamespaceNames =
+                    nsInformer.getIndexer().list().stream()
+                        .map(ns -> ns.getMetadata().getName())
+                        .collect(Collectors.toList());
+                assertThat(cachedNamespaceNames).contains(selectedNamespace);
+                assertThat(cachedNamespaceNames).doesNotContain(ignoredNamespace);
+              });
+    } finally {
+      informerFactory.stopAllRegisteredInformers(true);
+      coreV1Api.deleteNamespace(selectedNamespace).execute();
+      coreV1Api.deleteNamespace(ignoredNamespace).execute();
+    }
+  }
 }

util/src/main/java/io/kubernetes/client/informer/SharedInformerFactory.java

@@ -37,6 +37,7 @@
 import java.util.concurrent.Executors;
 import java.util.concurrent.Future;
 import java.util.function.BiConsumer;
+import java.util.function.Predicate;
 
 /**
  * SharedInformerFactory class constructs and caches informers for api types.
@@ -200,6 +201,17 @@ SharedIndexInformer<ApiType> sharedIndexInformerFor(
           Class<ApiType> apiTypeClass,
           long resyncPeriodInMillis,
           BiConsumer<Class<ApiType>, Throwable> exceptionHandler) {
+    return sharedIndexInformerFor(
+        listerWatcher, apiTypeClass, resyncPeriodInMillis, exceptionHandler, obj -> true);
+  }
+
+  public synchronized <ApiType extends KubernetesObject, ApiListType extends KubernetesListObject>
+      SharedIndexInformer<ApiType> sharedIndexInformerFor(
+          ListerWatcher<ApiType, ApiListType> listerWatcher,
+          Class<ApiType> apiTypeClass,
+          long resyncPeriodInMillis,
+          BiConsumer<Class<ApiType>, Throwable> exceptionHandler,
+          Predicate<ApiType> cachePredicate) {
     Type apiType = TypeToken.get(apiTypeClass).getType();
 
     if(informers.containsKey(apiType) && reuseExistingCachedInformer) {
@@ -208,7 +220,11 @@ SharedIndexInformer<ApiType> sharedIndexInformerFor(
 
     SharedIndexInformer<ApiType> informer =
         new DefaultSharedIndexInformer<>(
-            apiTypeClass, listerWatcher, resyncPeriodInMillis, new Cache<>(), exceptionHandler);
+            apiTypeClass,
+            listerWatcher,
+            resyncPeriodInMillis,
+            new Cache<>(cachePredicate),
+            exceptionHandler);
 
     this.informers.putIfAbsent(apiType, informer);
     return informer;
@@ -233,6 +249,21 @@ SharedIndexInformer<ApiType> sharedIndexInformerFor(
         genericKubernetesApi, apiTypeClass, resyncPeriodInMillis, Namespaces.NAMESPACE_ALL);
   }
 
+  public synchronized <ApiType extends KubernetesObject, ApiListType extends KubernetesListObject>
+      SharedIndexInformer<ApiType> sharedIndexInformerFor(
+          GenericKubernetesApi<ApiType, ApiListType> genericKubernetesApi,
+          Class<ApiType> apiTypeClass,
+          long resyncPeriodInMillis,
+          Predicate<ApiType> cachePredicate) {
+    return sharedIndexInformerFor(
+        genericKubernetesApi,
+        apiTypeClass,
+        resyncPeriodInMillis,
+        Namespaces.NAMESPACE_ALL,
+        null,
+        cachePredicate);
+  }
+
   /**
    * Working the same as {@link SharedInformerFactory#sharedIndexInformerFor} above.
    *
@@ -253,7 +284,27 @@ SharedIndexInformer<ApiType> sharedIndexInformerFor(
           long resyncPeriodInMillis,
           String namespace) {
     return sharedIndexInformerFor(
-        genericKubernetesApi, apiTypeClass, resyncPeriodInMillis, namespace, null);
+        genericKubernetesApi,
+        apiTypeClass,
+        resyncPeriodInMillis,
+        namespace,
+        (BiConsumer<Class<ApiType>, Throwable>) null);
+  }
+
+  public synchronized <ApiType extends KubernetesObject, ApiListType extends KubernetesListObject>
+      SharedIndexInformer<ApiType> sharedIndexInformerFor(
+          GenericKubernetesApi<ApiType, ApiListType> genericKubernetesApi,
+          Class<ApiType> apiTypeClass,
+          long resyncPeriodInMillis,
+          String namespace,
+          Predicate<ApiType> cachePredicate) {
+    return sharedIndexInformerFor(
+        genericKubernetesApi,
+        apiTypeClass,
+        resyncPeriodInMillis,
+        namespace,
+        null,
+        cachePredicate);
   }
 
   /**
@@ -277,10 +328,27 @@ SharedIndexInformer<ApiType> sharedIndexInformerFor(
           long resyncPeriodInMillis,
           String namespace,
           BiConsumer<Class<ApiType>, Throwable> exceptionHandler) {
+    return sharedIndexInformerFor(
+        genericKubernetesApi,
+        apiTypeClass,
+        resyncPeriodInMillis,
+        namespace,
+        exceptionHandler,
+        obj -> true);
+  }
+
+  public synchronized <ApiType extends KubernetesObject, ApiListType extends KubernetesListObject>
+      SharedIndexInformer<ApiType> sharedIndexInformerFor(
+          GenericKubernetesApi<ApiType, ApiListType> genericKubernetesApi,
+          Class<ApiType> apiTypeClass,
+          long resyncPeriodInMillis,
+          String namespace,
+          BiConsumer<Class<ApiType>, Throwable> exceptionHandler,
+          Predicate<ApiType> cachePredicate) {
     ListerWatcher<ApiType, ApiListType> listerWatcher =
         listerWatcherFor(genericKubernetesApi, namespace);
     return sharedIndexInformerFor(
-        listerWatcher, apiTypeClass, resyncPeriodInMillis, exceptionHandler);
+        listerWatcher, apiTypeClass, resyncPeriodInMillis, exceptionHandler, cachePredicate);
   }
 
   private <ApiType extends KubernetesObject, ApiListType extends KubernetesListObject>

util/src/main/java/io/kubernetes/client/informer/cache/Cache.java

@@ -21,6 +21,7 @@
 import java.util.Map;
 import java.util.Set;
 import java.util.function.Function;
+import java.util.function.Predicate;
 import org.apache.commons.collections4.CollectionUtils;
 import org.apache.commons.collections4.MapUtils;
 
@@ -43,11 +44,22 @@ public class Cache<ApiType extends KubernetesObject> implements Indexer<ApiType>
   /** indices stores objects' keys by their indices */
   private Map<String, Map<String, Set<String>>> indices = new HashMap<>();
 
+  private Predicate<ApiType> storePredicate;
+
   public Cache() {
     this(
         Caches.NAMESPACE_INDEX,
         Caches::metaNamespaceIndexFunc,
-        Caches::deletionHandlingMetaNamespaceKeyFunc);
+        Caches::deletionHandlingMetaNamespaceKeyFunc,
+        obj -> true);
+  }
+
+  public Cache(Predicate<ApiType> storePredicate) {
+    this(
+        Caches.NAMESPACE_INDEX,
+        Caches::metaNamespaceIndexFunc,
+        Caches::deletionHandlingMetaNamespaceKeyFunc,
+        storePredicate);
   }
 
   /**
@@ -61,9 +73,18 @@ public Cache(
       String indexName,
       Function<ApiType, List<String>> indexFunc,
       Function<ApiType, String> keyFunc) {
+    this(indexName, indexFunc, keyFunc, obj -> true);
+  }
+
+  public Cache(
+      String indexName,
+      Function<ApiType, List<String>> indexFunc,
+      Function<ApiType, String> keyFunc,
+      Predicate<ApiType> storePredicate) {
     this.indexers.put(indexName, indexFunc);
     this.keyFunc = keyFunc;
     this.indices.put(indexName, new HashMap<>());
+    this.storePredicate = storePredicate;
   }
 
   /**
@@ -73,6 +94,9 @@ public Cache(
    */
   @Override
   public void add(ApiType obj) {
+    if (!storePredicate.test(obj)) {
+      return;
+    }
     String key = keyFunc.apply(obj);
     synchronized (this) {
       ApiType oldObj = this.items.get(key);
@@ -91,6 +115,13 @@ public void update(ApiType obj) {
     String key = keyFunc.apply(obj);
     synchronized (this) {
       ApiType oldObj = this.items.get(key);
+      if (!storePredicate.test(obj)) {
+        if (oldObj != null) {
+          this.deleteFromIndices(oldObj, key);
+          this.items.remove(key);
+        }
+        return;
+      }
       this.items.put(key, obj);
       updateIndices(oldObj, obj, key);
     }
@@ -123,6 +154,9 @@ public void delete(ApiType obj) {
   public synchronized void replace(List<ApiType> list, String resourceVersion) {
     Map<String, ApiType> newItems = new HashMap<>();
     for (ApiType item : list) {
+      if (!storePredicate.test(item)) {
+        continue;
+      }
       String key = keyFunc.apply(item);
       newItems.put(key, item);
     }

util/src/main/java/io/kubernetes/client/informer/impl/DefaultSharedIndexInformer.java

@@ -284,11 +284,19 @@ public void handleDeltas(Deque<MutablePair<DeltaFIFO.DeltaType, KubernetesObject
           Object oldObj = this.indexer.get((ApiType) obj);
           if (oldObj != null) {
             this.indexer.update((ApiType) obj);
-            this.processor.distribute(
-                new ProcessorListener.UpdateNotification(oldObj, obj), isSync);
+            Object storedObj = this.indexer.get((ApiType) obj);
+            if (storedObj != null) {
+              this.processor.distribute(
+                  new ProcessorListener.UpdateNotification(oldObj, obj), isSync);
+            } else {
+              this.processor.distribute(new ProcessorListener.DeleteNotification(oldObj), false);
+            }
           } else {
             this.indexer.add((ApiType) obj);
-            this.processor.distribute(new ProcessorListener.AddNotification(obj), isSync);
+            Object storedObj = this.indexer.get((ApiType) obj);
+            if (storedObj != null) {
+              this.processor.distribute(new ProcessorListener.AddNotification(obj), isSync);
+            }
           }
           break;
         case Deleted:

util/src/test/java/io/kubernetes/client/informer/cache/CacheTest.java

@@ -24,6 +24,7 @@
 import java.util.List;
 import java.util.Map;
 import java.util.function.Function;
+import org.junit.jupiter.api.Test;
 import org.junit.jupiter.params.ParameterizedTest;
 import org.junit.jupiter.params.provider.MethodSource;
 
@@ -190,4 +191,52 @@ void addIndexers(KubernetesObject obj, String index) {
     List<V1Pod> nodeNameIndexedPods = podCache.byIndex(nodeIndex, "node1");
     assertThat(nodeNameIndexedPods).hasSize(1);
   }
+
+  @Test
+  void cachePredicateShouldFilterAddAndReplace() {
+    Cache<V1Pod> podCache =
+        new Cache<>(pod -> "keep".equals(pod.getMetadata().getLabels().get("scope")));
+    V1Pod kept =
+        new V1Pod()
+            .metadata(
+                new V1ObjectMeta()
+                    .namespace("ns")
+                    .name("kept")
+                    .labels(Map.of("scope", "keep")));
+    V1Pod dropped =
+        new V1Pod()
+            .metadata(
+                new V1ObjectMeta()
+                    .namespace("ns")
+                    .name("dropped")
+                    .labels(Map.of("scope", "drop")));
+
+    podCache.add(kept);
+    podCache.add(dropped);
+    assertThat(podCache.list()).containsExactly(kept);
+
+    podCache.replace(Arrays.asList(kept, dropped), "1");
+    assertThat(podCache.list()).containsExactly(kept);
+  }
+
+  @Test
+  void cachePredicateShouldEvictOnUpdateWhenObjectNoLongerMatches() {
+    Cache<V1Pod> podCache =
+        new Cache<>(pod -> "keep".equals(pod.getMetadata().getLabels().get("scope")));
+    V1Pod pod =
+        new V1Pod()
+            .metadata(
+                new V1ObjectMeta()
+                    .namespace("ns")
+                    .name("pod")
+                    .labels(new HashMap<>(Map.of("scope", "keep"))));
+
+    podCache.add(pod);
+    assertThat(podCache.list()).hasSize(1);
+
+    pod.getMetadata().setLabels(Map.of("scope", "drop"));
+    podCache.update(pod);
+
+    assertThat(podCache.list()).isEmpty();
+  }
 }