From 7c8cd35b1a268983cce1d5e4b13924bdc840a888 Mon Sep 17 00:00:00 2001 From: Brendan Burns <5751682+brendandburns@users.noreply.github.com> Date: Tue, 1 Sep 2026 19:21:11 +0000 Subject: [PATCH] Add cache-level predicate support for informers --- .../e2e/informer/NamespaceInformerTest.java | 60 ++++++++++++++ .../informer/SharedInformerFactory.java | 74 ++++++++++++++++- .../client/informer/cache/Cache.java | 49 +++++++++++- .../impl/DefaultSharedIndexInformer.java | 14 +++- .../client/informer/cache/CacheTest.java | 80 +++++++++++++++++++ ...SharedIndexInformerCachePredicateTest.java | 64 +++++++++++++++ 6 files changed, 334 insertions(+), 7 deletions(-) create mode 100644 util/src/test/java/io/kubernetes/client/informer/impl/DefaultSharedIndexInformerCachePredicateTest.java diff --git a/e2e/src/test/java/io/kubernetes/client/e2e/informer/NamespaceInformerTest.java b/e2e/src/test/java/io/kubernetes/client/e2e/informer/NamespaceInformerTest.java index bbf92916b2..c4feef351d 100644 --- a/e2e/src/test/java/io/kubernetes/client/e2e/informer/NamespaceInformerTest.java +++ b/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 api = + new GenericKubernetesApi<>(V1Namespace.class, V1NamespaceList.class, "", "v1", "namespaces", client); + SharedIndexInformer 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 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(); + } + } } diff --git a/util/src/main/java/io/kubernetes/client/informer/SharedInformerFactory.java b/util/src/main/java/io/kubernetes/client/informer/SharedInformerFactory.java index f4fcac6e52..7c92ae3506 100644 --- a/util/src/main/java/io/kubernetes/client/informer/SharedInformerFactory.java +++ b/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 sharedIndexInformerFor( Class apiTypeClass, long resyncPeriodInMillis, BiConsumer, Throwable> exceptionHandler) { + return sharedIndexInformerFor( + listerWatcher, apiTypeClass, resyncPeriodInMillis, exceptionHandler, obj -> true); + } + + public synchronized + SharedIndexInformer sharedIndexInformerFor( + ListerWatcher listerWatcher, + Class apiTypeClass, + long resyncPeriodInMillis, + BiConsumer, Throwable> exceptionHandler, + Predicate cachePredicate) { Type apiType = TypeToken.get(apiTypeClass).getType(); if(informers.containsKey(apiType) && reuseExistingCachedInformer) { @@ -208,7 +220,11 @@ SharedIndexInformer sharedIndexInformerFor( SharedIndexInformer 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 sharedIndexInformerFor( genericKubernetesApi, apiTypeClass, resyncPeriodInMillis, Namespaces.NAMESPACE_ALL); } + public synchronized + SharedIndexInformer sharedIndexInformerFor( + GenericKubernetesApi genericKubernetesApi, + Class apiTypeClass, + long resyncPeriodInMillis, + Predicate cachePredicate) { + return sharedIndexInformerFor( + genericKubernetesApi, + apiTypeClass, + resyncPeriodInMillis, + Namespaces.NAMESPACE_ALL, + null, + cachePredicate); + } + /** * Working the same as {@link SharedInformerFactory#sharedIndexInformerFor} above. * @@ -253,7 +284,27 @@ SharedIndexInformer sharedIndexInformerFor( long resyncPeriodInMillis, String namespace) { return sharedIndexInformerFor( - genericKubernetesApi, apiTypeClass, resyncPeriodInMillis, namespace, null); + genericKubernetesApi, + apiTypeClass, + resyncPeriodInMillis, + namespace, + (BiConsumer, Throwable>) null); + } + + public synchronized + SharedIndexInformer sharedIndexInformerFor( + GenericKubernetesApi genericKubernetesApi, + Class apiTypeClass, + long resyncPeriodInMillis, + String namespace, + Predicate cachePredicate) { + return sharedIndexInformerFor( + genericKubernetesApi, + apiTypeClass, + resyncPeriodInMillis, + namespace, + null, + cachePredicate); } /** @@ -277,10 +328,27 @@ SharedIndexInformer sharedIndexInformerFor( long resyncPeriodInMillis, String namespace, BiConsumer, Throwable> exceptionHandler) { + return sharedIndexInformerFor( + genericKubernetesApi, + apiTypeClass, + resyncPeriodInMillis, + namespace, + exceptionHandler, + obj -> true); + } + + public synchronized + SharedIndexInformer sharedIndexInformerFor( + GenericKubernetesApi genericKubernetesApi, + Class apiTypeClass, + long resyncPeriodInMillis, + String namespace, + BiConsumer, Throwable> exceptionHandler, + Predicate cachePredicate) { ListerWatcher listerWatcher = listerWatcherFor(genericKubernetesApi, namespace); return sharedIndexInformerFor( - listerWatcher, apiTypeClass, resyncPeriodInMillis, exceptionHandler); + listerWatcher, apiTypeClass, resyncPeriodInMillis, exceptionHandler, cachePredicate); } private diff --git a/util/src/main/java/io/kubernetes/client/informer/cache/Cache.java b/util/src/main/java/io/kubernetes/client/informer/cache/Cache.java index 9b2f5ae96a..7943cddd25 100644 --- a/util/src/main/java/io/kubernetes/client/informer/cache/Cache.java +++ b/util/src/main/java/io/kubernetes/client/informer/cache/Cache.java @@ -21,8 +21,11 @@ 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; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; /** * Cache is a java port of k/client-go's ThreadSafeStore. It basically saves and indexes all the @@ -31,6 +34,8 @@ // TODO(yue9944882): Cache is very similar to a Map, replace/inherit w/ Map interface public class Cache implements Indexer { + private static final Logger log = LoggerFactory.getLogger(Cache.class); + /** keyFunc defines how to map objects into indices */ private Function keyFunc; @@ -43,11 +48,22 @@ public class Cache implements Indexer /** indices stores objects' keys by their indices */ private Map>> indices = new HashMap<>(); + private Predicate storePredicate; + public Cache() { this( Caches.NAMESPACE_INDEX, Caches::metaNamespaceIndexFunc, - Caches::deletionHandlingMetaNamespaceKeyFunc); + Caches::deletionHandlingMetaNamespaceKeyFunc, + obj -> true); + } + + public Cache(Predicate storePredicate) { + this( + Caches.NAMESPACE_INDEX, + Caches::metaNamespaceIndexFunc, + Caches::deletionHandlingMetaNamespaceKeyFunc, + storePredicate); } /** @@ -61,9 +77,18 @@ public Cache( String indexName, Function> indexFunc, Function keyFunc) { + this(indexName, indexFunc, keyFunc, obj -> true); + } + + public Cache( + String indexName, + Function> indexFunc, + Function keyFunc, + Predicate storePredicate) { this.indexers.put(indexName, indexFunc); this.keyFunc = keyFunc; this.indices.put(indexName, new HashMap<>()); + this.storePredicate = storePredicate; } /** @@ -73,6 +98,9 @@ public Cache( */ @Override public void add(ApiType obj) { + if (!matchesStorePredicate(obj)) { + return; + } String key = keyFunc.apply(obj); synchronized (this) { ApiType oldObj = this.items.get(key); @@ -91,6 +119,13 @@ public void update(ApiType obj) { String key = keyFunc.apply(obj); synchronized (this) { ApiType oldObj = this.items.get(key); + if (!matchesStorePredicate(obj)) { + if (oldObj != null) { + this.deleteFromIndices(oldObj, key); + this.items.remove(key); + } + return; + } this.items.put(key, obj); updateIndices(oldObj, obj, key); } @@ -123,6 +158,9 @@ public void delete(ApiType obj) { public synchronized void replace(List list, String resourceVersion) { Map newItems = new HashMap<>(); for (ApiType item : list) { + if (!matchesStorePredicate(item)) { + continue; + } String key = keyFunc.apply(item); newItems.put(key, item); } @@ -135,6 +173,15 @@ public synchronized void replace(List list, String resourceVersion) { } } + private boolean matchesStorePredicate(ApiType obj) { + try { + return storePredicate.test(obj); + } catch (RuntimeException e) { + log.warn("Cache predicate threw an exception; excluding object from informer cache", e); + return false; + } + } + /** Resync. */ @Override public void resync() { diff --git a/util/src/main/java/io/kubernetes/client/informer/impl/DefaultSharedIndexInformer.java b/util/src/main/java/io/kubernetes/client/informer/impl/DefaultSharedIndexInformer.java index 4fd1db93a3..93a94dac1b 100644 --- a/util/src/main/java/io/kubernetes/client/informer/impl/DefaultSharedIndexInformer.java +++ b/util/src/main/java/io/kubernetes/client/informer/impl/DefaultSharedIndexInformer.java @@ -284,11 +284,19 @@ public void handleDeltas(Deque nodeNameIndexedPods = podCache.byIndex(nodeIndex, "node1"); assertThat(nodeNameIndexedPods).hasSize(1); } + + @Test + void cachePredicateShouldFilterAddAndReplace() { + Cache 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 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(); + } + + @Test + void cachePredicateExceptionShouldExcludeObject() { + Cache podCache = + new Cache<>( + pod -> { + if ("fail".equals(pod.getMetadata().getLabels().get("predicate"))) { + throw new IllegalStateException("predicate failure"); + } + return true; + }); + V1Pod pod = + new V1Pod() + .metadata( + new V1ObjectMeta() + .namespace("ns") + .name("excluded") + .labels(new HashMap<>())); + + podCache.add(pod); + assertThat(podCache.list()).containsExactly(pod); + + pod.getMetadata().getLabels().put("predicate", "fail"); + podCache.update(pod); + assertThat(podCache.list()).isEmpty(); + + podCache.add(pod); + podCache.replace(List.of(pod), "1"); + + assertThat(podCache.list()).isEmpty(); + } } diff --git a/util/src/test/java/io/kubernetes/client/informer/impl/DefaultSharedIndexInformerCachePredicateTest.java b/util/src/test/java/io/kubernetes/client/informer/impl/DefaultSharedIndexInformerCachePredicateTest.java new file mode 100644 index 0000000000..041b73e29b --- /dev/null +++ b/util/src/test/java/io/kubernetes/client/informer/impl/DefaultSharedIndexInformerCachePredicateTest.java @@ -0,0 +1,64 @@ +/* +Copyright 2026 The Kubernetes 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.kubernetes.client.informer.impl; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.Mockito.mock; + +import io.kubernetes.client.common.KubernetesObject; +import io.kubernetes.client.informer.ListerWatcher; +import io.kubernetes.client.informer.cache.Cache; +import io.kubernetes.client.informer.cache.DeltaFIFO; +import io.kubernetes.client.openapi.models.V1ObjectMeta; +import io.kubernetes.client.openapi.models.V1Pod; +import io.kubernetes.client.openapi.models.V1PodList; +import java.util.ArrayDeque; +import java.util.Deque; +import java.util.Map; +import org.apache.commons.lang3.tuple.MutablePair; +import org.junit.jupiter.api.Test; + +class DefaultSharedIndexInformerCachePredicateTest { + + private static V1Pod pod(String name, String scope) { + return new V1Pod() + .metadata( + new V1ObjectMeta().namespace("default").name(name).labels(Map.of("scope", scope))); + } + + @Test + void handleDeltasShouldRespectCachePredicateForAddAndUpdate() { + Cache cache = + new Cache<>(p -> "keep".equals(p.getMetadata().getLabels().get("scope"))); + DefaultSharedIndexInformer informer = + new DefaultSharedIndexInformer<>( + V1Pod.class, + mock(ListerWatcher.class), + 0, + cache); + + Deque> deltas = new ArrayDeque<>(); + deltas.add(MutablePair.of(DeltaFIFO.DeltaType.Added, pod("dropped", "drop"))); + deltas.add(MutablePair.of(DeltaFIFO.DeltaType.Added, pod("kept", "keep"))); + informer.handleDeltas(deltas); + + assertThat(cache.list()).extracting(p -> p.getMetadata().getName()).containsExactly("kept"); + + V1Pod becomesDropped = pod("kept", "drop"); + Deque> update = new ArrayDeque<>(); + update.add(MutablePair.of(DeltaFIFO.DeltaType.Updated, becomesDropped)); + informer.handleDeltas(update); + + assertThat(cache.list()).isEmpty(); + } +}