Skip to content

Commit 7c8cd35

Browse files
brendandburnsbrendandburns
committed
Add cache-level predicate support for informers
1 parent 7601bd2 commit 7c8cd35

6 files changed

Lines changed: 334 additions & 7 deletions

File tree

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

Lines changed: 60 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -19,10 +19,13 @@
1919
import io.kubernetes.client.informer.SharedInformerFactory;
2020
import io.kubernetes.client.informer.cache.Lister;
2121
import io.kubernetes.client.openapi.ApiClient;
22+
import io.kubernetes.client.openapi.apis.CoreV1Api;
2223
import io.kubernetes.client.openapi.models.V1Namespace;
2324
import io.kubernetes.client.openapi.models.V1NamespaceList;
25+
import io.kubernetes.client.openapi.models.V1ObjectMeta;
2426
import io.kubernetes.client.util.ClientBuilder;
2527
import io.kubernetes.client.util.generic.GenericKubernetesApi;
28+
import java.util.stream.Collectors;
2629
import org.junit.jupiter.api.Test;
2730

2831
class NamespaceInformerTest {
@@ -57,4 +60,61 @@ void listWatchingNamespaces() throws Exception {
5760
informerFactory.stopAllRegisteredInformers(true);
5861
}
5962
}
63+
64+
@Test
65+
void listWatchingNamespacesWithCachePredicate() throws Exception {
66+
ApiClient client = ClientBuilder.defaultClient();
67+
CoreV1Api coreV1Api = new CoreV1Api(client);
68+
SharedInformerFactory informerFactory = new SharedInformerFactory(client);
69+
String selectedNamespace = "e2e-cache-selected";
70+
String ignoredNamespace = "e2e-cache-ignored";
71+
72+
coreV1Api
73+
.createNamespace(
74+
new V1Namespace()
75+
.metadata(
76+
new V1ObjectMeta()
77+
.name(selectedNamespace)
78+
.labels(java.util.Map.of("cache-filter", "keep"))))
79+
.execute();
80+
coreV1Api
81+
.createNamespace(
82+
new V1Namespace()
83+
.metadata(
84+
new V1ObjectMeta()
85+
.name(ignoredNamespace)
86+
.labels(java.util.Map.of("cache-filter", "drop"))))
87+
.execute();
88+
89+
GenericKubernetesApi<V1Namespace, V1NamespaceList> api =
90+
new GenericKubernetesApi<>(V1Namespace.class, V1NamespaceList.class, "", "v1", "namespaces", client);
91+
SharedIndexInformer<V1Namespace> nsInformer =
92+
informerFactory.sharedIndexInformerFor(
93+
api,
94+
V1Namespace.class,
95+
0,
96+
ns ->
97+
ns.getMetadata() != null
98+
&& ns.getMetadata().getLabels() != null
99+
&& "keep".equals(ns.getMetadata().getLabels().get("cache-filter")));
100+
101+
try {
102+
informerFactory.startAllRegisteredInformers();
103+
await().untilAsserted(() -> assertThat(nsInformer.hasSynced()).isTrue());
104+
await()
105+
.untilAsserted(
106+
() -> {
107+
java.util.List<String> cachedNamespaceNames =
108+
nsInformer.getIndexer().list().stream()
109+
.map(ns -> ns.getMetadata().getName())
110+
.collect(Collectors.toList());
111+
assertThat(cachedNamespaceNames).contains(selectedNamespace);
112+
assertThat(cachedNamespaceNames).doesNotContain(ignoredNamespace);
113+
});
114+
} finally {
115+
informerFactory.stopAllRegisteredInformers(true);
116+
coreV1Api.deleteNamespace(selectedNamespace).execute();
117+
coreV1Api.deleteNamespace(ignoredNamespace).execute();
118+
}
119+
}
60120
}

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

Lines changed: 71 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -37,6 +37,7 @@
3737
import java.util.concurrent.Executors;
3838
import java.util.concurrent.Future;
3939
import java.util.function.BiConsumer;
40+
import java.util.function.Predicate;
4041

4142
/**
4243
* SharedInformerFactory class constructs and caches informers for api types.
@@ -200,6 +201,17 @@ SharedIndexInformer<ApiType> sharedIndexInformerFor(
200201
Class<ApiType> apiTypeClass,
201202
long resyncPeriodInMillis,
202203
BiConsumer<Class<ApiType>, Throwable> exceptionHandler) {
204+
return sharedIndexInformerFor(
205+
listerWatcher, apiTypeClass, resyncPeriodInMillis, exceptionHandler, obj -> true);
206+
}
207+
208+
public synchronized <ApiType extends KubernetesObject, ApiListType extends KubernetesListObject>
209+
SharedIndexInformer<ApiType> sharedIndexInformerFor(
210+
ListerWatcher<ApiType, ApiListType> listerWatcher,
211+
Class<ApiType> apiTypeClass,
212+
long resyncPeriodInMillis,
213+
BiConsumer<Class<ApiType>, Throwable> exceptionHandler,
214+
Predicate<ApiType> cachePredicate) {
203215
Type apiType = TypeToken.get(apiTypeClass).getType();
204216

205217
if(informers.containsKey(apiType) && reuseExistingCachedInformer) {
@@ -208,7 +220,11 @@ SharedIndexInformer<ApiType> sharedIndexInformerFor(
208220

209221
SharedIndexInformer<ApiType> informer =
210222
new DefaultSharedIndexInformer<>(
211-
apiTypeClass, listerWatcher, resyncPeriodInMillis, new Cache<>(), exceptionHandler);
223+
apiTypeClass,
224+
listerWatcher,
225+
resyncPeriodInMillis,
226+
new Cache<>(cachePredicate),
227+
exceptionHandler);
212228

213229
this.informers.putIfAbsent(apiType, informer);
214230
return informer;
@@ -233,6 +249,21 @@ SharedIndexInformer<ApiType> sharedIndexInformerFor(
233249
genericKubernetesApi, apiTypeClass, resyncPeriodInMillis, Namespaces.NAMESPACE_ALL);
234250
}
235251

252+
public synchronized <ApiType extends KubernetesObject, ApiListType extends KubernetesListObject>
253+
SharedIndexInformer<ApiType> sharedIndexInformerFor(
254+
GenericKubernetesApi<ApiType, ApiListType> genericKubernetesApi,
255+
Class<ApiType> apiTypeClass,
256+
long resyncPeriodInMillis,
257+
Predicate<ApiType> cachePredicate) {
258+
return sharedIndexInformerFor(
259+
genericKubernetesApi,
260+
apiTypeClass,
261+
resyncPeriodInMillis,
262+
Namespaces.NAMESPACE_ALL,
263+
null,
264+
cachePredicate);
265+
}
266+
236267
/**
237268
* Working the same as {@link SharedInformerFactory#sharedIndexInformerFor} above.
238269
*
@@ -253,7 +284,27 @@ SharedIndexInformer<ApiType> sharedIndexInformerFor(
253284
long resyncPeriodInMillis,
254285
String namespace) {
255286
return sharedIndexInformerFor(
256-
genericKubernetesApi, apiTypeClass, resyncPeriodInMillis, namespace, null);
287+
genericKubernetesApi,
288+
apiTypeClass,
289+
resyncPeriodInMillis,
290+
namespace,
291+
(BiConsumer<Class<ApiType>, Throwable>) null);
292+
}
293+
294+
public synchronized <ApiType extends KubernetesObject, ApiListType extends KubernetesListObject>
295+
SharedIndexInformer<ApiType> sharedIndexInformerFor(
296+
GenericKubernetesApi<ApiType, ApiListType> genericKubernetesApi,
297+
Class<ApiType> apiTypeClass,
298+
long resyncPeriodInMillis,
299+
String namespace,
300+
Predicate<ApiType> cachePredicate) {
301+
return sharedIndexInformerFor(
302+
genericKubernetesApi,
303+
apiTypeClass,
304+
resyncPeriodInMillis,
305+
namespace,
306+
null,
307+
cachePredicate);
257308
}
258309

259310
/**
@@ -277,10 +328,27 @@ SharedIndexInformer<ApiType> sharedIndexInformerFor(
277328
long resyncPeriodInMillis,
278329
String namespace,
279330
BiConsumer<Class<ApiType>, Throwable> exceptionHandler) {
331+
return sharedIndexInformerFor(
332+
genericKubernetesApi,
333+
apiTypeClass,
334+
resyncPeriodInMillis,
335+
namespace,
336+
exceptionHandler,
337+
obj -> true);
338+
}
339+
340+
public synchronized <ApiType extends KubernetesObject, ApiListType extends KubernetesListObject>
341+
SharedIndexInformer<ApiType> sharedIndexInformerFor(
342+
GenericKubernetesApi<ApiType, ApiListType> genericKubernetesApi,
343+
Class<ApiType> apiTypeClass,
344+
long resyncPeriodInMillis,
345+
String namespace,
346+
BiConsumer<Class<ApiType>, Throwable> exceptionHandler,
347+
Predicate<ApiType> cachePredicate) {
280348
ListerWatcher<ApiType, ApiListType> listerWatcher =
281349
listerWatcherFor(genericKubernetesApi, namespace);
282350
return sharedIndexInformerFor(
283-
listerWatcher, apiTypeClass, resyncPeriodInMillis, exceptionHandler);
351+
listerWatcher, apiTypeClass, resyncPeriodInMillis, exceptionHandler, cachePredicate);
284352
}
285353

286354
private <ApiType extends KubernetesObject, ApiListType extends KubernetesListObject>

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

Lines changed: 48 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -21,8 +21,11 @@
2121
import java.util.Map;
2222
import java.util.Set;
2323
import java.util.function.Function;
24+
import java.util.function.Predicate;
2425
import org.apache.commons.collections4.CollectionUtils;
2526
import org.apache.commons.collections4.MapUtils;
27+
import org.slf4j.Logger;
28+
import org.slf4j.LoggerFactory;
2629

2730
/**
2831
* Cache is a java port of k/client-go's ThreadSafeStore. It basically saves and indexes all the
@@ -31,6 +34,8 @@
3134
// TODO(yue9944882): Cache is very similar to a Map, replace/inherit w/ Map interface
3235
public class Cache<ApiType extends KubernetesObject> implements Indexer<ApiType> {
3336

37+
private static final Logger log = LoggerFactory.getLogger(Cache.class);
38+
3439
/** keyFunc defines how to map objects into indices */
3540
private Function<ApiType, String> keyFunc;
3641

@@ -43,11 +48,22 @@ public class Cache<ApiType extends KubernetesObject> implements Indexer<ApiType>
4348
/** indices stores objects' keys by their indices */
4449
private Map<String, Map<String, Set<String>>> indices = new HashMap<>();
4550

51+
private Predicate<ApiType> storePredicate;
52+
4653
public Cache() {
4754
this(
4855
Caches.NAMESPACE_INDEX,
4956
Caches::metaNamespaceIndexFunc,
50-
Caches::deletionHandlingMetaNamespaceKeyFunc);
57+
Caches::deletionHandlingMetaNamespaceKeyFunc,
58+
obj -> true);
59+
}
60+
61+
public Cache(Predicate<ApiType> storePredicate) {
62+
this(
63+
Caches.NAMESPACE_INDEX,
64+
Caches::metaNamespaceIndexFunc,
65+
Caches::deletionHandlingMetaNamespaceKeyFunc,
66+
storePredicate);
5167
}
5268

5369
/**
@@ -61,9 +77,18 @@ public Cache(
6177
String indexName,
6278
Function<ApiType, List<String>> indexFunc,
6379
Function<ApiType, String> keyFunc) {
80+
this(indexName, indexFunc, keyFunc, obj -> true);
81+
}
82+
83+
public Cache(
84+
String indexName,
85+
Function<ApiType, List<String>> indexFunc,
86+
Function<ApiType, String> keyFunc,
87+
Predicate<ApiType> storePredicate) {
6488
this.indexers.put(indexName, indexFunc);
6589
this.keyFunc = keyFunc;
6690
this.indices.put(indexName, new HashMap<>());
91+
this.storePredicate = storePredicate;
6792
}
6893

6994
/**
@@ -73,6 +98,9 @@ public Cache(
7398
*/
7499
@Override
75100
public void add(ApiType obj) {
101+
if (!matchesStorePredicate(obj)) {
102+
return;
103+
}
76104
String key = keyFunc.apply(obj);
77105
synchronized (this) {
78106
ApiType oldObj = this.items.get(key);
@@ -91,6 +119,13 @@ public void update(ApiType obj) {
91119
String key = keyFunc.apply(obj);
92120
synchronized (this) {
93121
ApiType oldObj = this.items.get(key);
122+
if (!matchesStorePredicate(obj)) {
123+
if (oldObj != null) {
124+
this.deleteFromIndices(oldObj, key);
125+
this.items.remove(key);
126+
}
127+
return;
128+
}
94129
this.items.put(key, obj);
95130
updateIndices(oldObj, obj, key);
96131
}
@@ -123,6 +158,9 @@ public void delete(ApiType obj) {
123158
public synchronized void replace(List<ApiType> list, String resourceVersion) {
124159
Map<String, ApiType> newItems = new HashMap<>();
125160
for (ApiType item : list) {
161+
if (!matchesStorePredicate(item)) {
162+
continue;
163+
}
126164
String key = keyFunc.apply(item);
127165
newItems.put(key, item);
128166
}
@@ -135,6 +173,15 @@ public synchronized void replace(List<ApiType> list, String resourceVersion) {
135173
}
136174
}
137175

176+
private boolean matchesStorePredicate(ApiType obj) {
177+
try {
178+
return storePredicate.test(obj);
179+
} catch (RuntimeException e) {
180+
log.warn("Cache predicate threw an exception; excluding object from informer cache", e);
181+
return false;
182+
}
183+
}
184+
138185
/** Resync. */
139186
@Override
140187
public void resync() {

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

Lines changed: 11 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -284,11 +284,19 @@ public void handleDeltas(Deque<MutablePair<DeltaFIFO.DeltaType, KubernetesObject
284284
Object oldObj = this.indexer.get((ApiType) obj);
285285
if (oldObj != null) {
286286
this.indexer.update((ApiType) obj);
287-
this.processor.distribute(
288-
new ProcessorListener.UpdateNotification(oldObj, obj), isSync);
287+
Object storedObj = this.indexer.get((ApiType) obj);
288+
if (storedObj != null) {
289+
this.processor.distribute(
290+
new ProcessorListener.UpdateNotification(oldObj, obj), isSync);
291+
} else {
292+
this.processor.distribute(new ProcessorListener.DeleteNotification(oldObj), false);
293+
}
289294
} else {
290295
this.indexer.add((ApiType) obj);
291-
this.processor.distribute(new ProcessorListener.AddNotification(obj), isSync);
296+
Object storedObj = this.indexer.get((ApiType) obj);
297+
if (storedObj != null) {
298+
this.processor.distribute(new ProcessorListener.AddNotification(obj), isSync);
299+
}
292300
}
293301
break;
294302
case Deleted:

0 commit comments

Comments
 (0)