Skip to content

Add cache-level predicate support for informers #4896

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
brendandburns wants to merge 1 commit into
base: master
Choose a base branch
from
Open

Add cache-level predicate support for informers #4896

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
60 changes: 60 additions & 0 deletions e2e/src/test/java/io/kubernetes/client/e2e/informer/NamespaceInformerTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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 {
Expand Down Expand Up @@ -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();
}
}
}
74 changes: 71 additions & 3 deletions util/src/main/java/io/kubernetes/client/informer/SharedInformerFactory.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand Down Expand Up @@ -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) {
Expand All @@ -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;
Expand All @@ -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.
*
Expand All @@ -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);
}

/**
Expand All @@ -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>
Expand Down
49 changes: 48 additions & 1 deletion util/src/main/java/io/kubernetes/client/informer/cache/Cache.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -31,6 +34,8 @@
// TODO(yue9944882): Cache is very similar to a Map, replace/inherit w/ Map interface
public class Cache<ApiType extends KubernetesObject> implements Indexer<ApiType> {

private static final Logger log = LoggerFactory.getLogger(Cache.class);

/** keyFunc defines how to map objects into indices */
private Function<ApiType, String> keyFunc;

Expand All @@ -43,11 +48,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);
}

/**
Expand All @@ -61,9 +77,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;
}

/**
Expand All @@ -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);
Expand All @@ -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);
}
Expand Down Expand Up @@ -123,6 +158,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 (!matchesStorePredicate(item)) {
continue;
}
String key = keyFunc.apply(item);
newItems.put(key, item);
}
Expand All @@ -135,6 +173,15 @@ public synchronized void replace(List<ApiType> 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() {
Expand Down
14 changes: 11 additions & 3 deletions util/src/main/java/io/kubernetes/client/informer/impl/DefaultSharedIndexInformer.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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:
Expand Down
Loading
Loading