Skip to content

Update resource cache after update/create from Dependent Resource #953

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.

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

Closed
wants to merge 13 commits into from
Closed
Original file line number Diff line number Diff line change
Expand Up @@ -32,10 +32,11 @@ public abstract class KubernetesDependentResource<R extends HasMetadata, P exten
private static final Logger log = LoggerFactory.getLogger(KubernetesDependentResource.class);

protected KubernetesClient client;
private InformerEventSource<R, P> informerEventSource;
protected InformerEventSource<R, P> informerEventSource;
private boolean addOwnerReference;
protected ResourceMatcher resourceMatcher;
protected ResourceUpdatePreProcessor<R> resourceUpdatePreProcessor;
protected TemporalResourceCache<R> temporalResourceCache;

@Override
public void configureWith(KubernetesDependentResourceConfig config) {
Expand Down Expand Up @@ -76,6 +77,8 @@ public void configureWith(ConfigurationService configurationService,
this.informerEventSource = informerEventSource;
this.addOwnerReference = addOwnerReference;
initResourceMatcherAndUpdatePreProcessorIfNotSet(configurationService);

temporalResourceCache = new TemporalResourceCache<>(informerEventSource);
}

protected void beforeCreate(R desired, P primary) {
Expand All @@ -96,8 +99,10 @@ protected R create(R target, P primary, Context context) {
"{}, with id: {}", target.getClass(), ResourceID.fromResource(target));
beforeCreate(target, primary);
Class<R> targetClass = (Class<R>) target.getClass();
return client.resources(targetClass).inNamespace(target.getMetadata().getNamespace())
var newResource = client.resources(targetClass).inNamespace(target.getMetadata().getNamespace())
.create(target);
temporalResourceCache.putAddedResource(newResource);
return newResource;
}

@SuppressWarnings("unchecked")
Expand All @@ -107,10 +112,16 @@ protected R update(R actual, R target, P primary, Context context) {
ResourceID.fromResource(target));
Class<R> targetClass = (Class<R>) target.getClass();
var updatedActual = resourceUpdatePreProcessor.replaceSpecOnActual(actual, target);
return client.resources(targetClass).inNamespace(target.getMetadata().getNamespace())
.replace(updatedActual);
R updatedResource =
client.resources(targetClass).inNamespace(target.getMetadata().getNamespace())
.replace(updatedActual);
temporalResourceCache.putUpdatedResource(updatedResource,
actual.getMetadata().getResourceVersion());
return updatedResource;
}



@Override
public EventSource eventSource(EventSourceContext<P> context) {
initResourceMatcherAndUpdatePreProcessorIfNotSet(context.getConfigurationService());
Expand All @@ -123,12 +134,6 @@ public EventSource eventSource(EventSourceContext<P> context) {
return informerEventSource;
}

public KubernetesDependentResource<R, P> setInformerEventSource(
InformerEventSource<R, P> informerEventSource) {
this.informerEventSource = informerEventSource;
return this;
}

@Override
public void delete(P primary, Context context) {
if (!addOwnerReference) {
Expand All @@ -144,7 +149,16 @@ protected Class<R> resourceType() {

@Override
public Optional<R> getResource(P primaryResource) {
return informerEventSource.getAssociated(primaryResource);
var associatedSecondaryResourceIdentifier =
informerEventSource.getConfiguration().getAssociatedResourceIdentifier();
var resourceId =
associatedSecondaryResourceIdentifier.associatedSecondaryID(primaryResource);
var tempCacheResource = temporalResourceCache.getResourceFromCache(resourceId);
if (tempCacheResource.isPresent()) {
return tempCacheResource;
} else {
return informerEventSource.get(resourceId);
}
}

@Override
Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,121 @@
package io.javaoperatorsdk.operator.processing.dependent.kubernetes;

import java.util.Map;
import java.util.Optional;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.locks.ReentrantLock;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import io.fabric8.kubernetes.api.model.HasMetadata;
import io.fabric8.kubernetes.client.informers.ResourceEventHandler;
import io.javaoperatorsdk.operator.processing.event.ResourceID;
import io.javaoperatorsdk.operator.processing.event.source.informer.InformerEventSource;

/**
* <p>
* Temporal cache is used to solve the problem for {@link KubernetesDependentResource} that is, when
* a create or update is executed the subsequent getResource opeeration might not return the
* up-to-date resource from informer cache, since it is not received yet by webhook.
* </p>
* <p>
* The idea of the solution is, that since an update (for create is simpler) was done successfully,
* and optimistic locking is in place, there were no other operations between reading the resource
* from the cache and the actual update. So when the new resource is stored in the temporal cache
* only if the informer still has the previous resource version, from before the update. If not,
* that means there were already updates on the cache (either by the actual update from
* DependentResource or other) so the resource does not needs to be cached. Subsequently if event
* received from the informer, it means that the cache of the informer was updated, so it already
* contains a more fresh version of the resource. (See one caveat below)
* </p>
*
* @param <T> resource to cache.
*/
public class TemporalResourceCache<T extends HasMetadata> implements ResourceEventHandler<T> {

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

private final Map<ResourceID, T> cache = new ConcurrentHashMap<>();
private final ReentrantLock lock = new ReentrantLock();
private final InformerEventSource<T, ?> informerEventSource;

public TemporalResourceCache(InformerEventSource<T, ?> informerEventSource) {
this.informerEventSource = informerEventSource;
}

@Override
public void onAdd(T t) {
removeResourceFromCache(t);
}

/**
* In theory, it can happen that an older event is received, that is received before we updated
* the resource. So that is a situation still not covered, but it happens with extremely low
* probability and since it should trigger a new reconciliation, eventually the system is
* consistent.
*
* @param t old object
* @param t1 new object
*/
@Override
public void onUpdate(T t, T t1) {
removeResourceFromCache(t1);
}

@Override
public void onDelete(T t, boolean b) {
removeResourceFromCache(t);
}

private void removeResourceFromCache(T resource) {
lock.lock();
try {
cache.remove(ResourceID.fromResource(resource));
} finally {
lock.unlock();
}
}

public void putAddedResource(T newResource) {
lock.lock();
try {
if (informerEventSource.get(ResourceID.fromResource(newResource)).isEmpty()) {
cache.put(ResourceID.fromResource(newResource), newResource);
}
} finally {
lock.unlock();
}
}

public void putUpdatedResource(T newResource, String previousResourceVersion) {
lock.lock();
try {
var resourceId = ResourceID.fromResource(newResource);
var informerCacheResource = informerEventSource.get(resourceId);
if (informerCacheResource.isEmpty()) {
log.debug("No cached value present for resource: {}", newResource);
return;
}
// if this is not true that means the cache was already updated
if (informerCacheResource.get().getMetadata().getResourceVersion()
.equals(previousResourceVersion)) {
cache.put(resourceId, newResource);
} else {
// if something is in cache it's surely obsolete now
cache.remove(resourceId);
}
} finally {
lock.unlock();
}
}

public Optional<T> getResourceFromCache(ResourceID resourceID) {
try {
lock.lock();
return Optional.ofNullable(cache.get(resourceID));
} finally {
lock.unlock();
}
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -104,4 +104,8 @@ public Optional<T> getAssociated(P resource) {
public Stream<T> list(String namespace, Predicate<T> predicate) {
return manager().list(namespace, predicate);
}

public InformerConfiguration<T, P> getConfiguration() {
return configuration;
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -139,6 +139,13 @@ public T remove(ResourceID key) {
@Override
public void put(ResourceID key, T resource) {
getSource(key.getNamespace().orElse(ANY_NAMESPACE_MAP_KEY))
.ifPresent(c -> c.put(key, resource));
.ifPresentOrElse(c -> c.put(key, resource),
() -> log.warn(
"Cannot put resource in the cache. No related cache found: {}. Resource: {}",
key, resource));
}

public void addEventHandler(ResourceEventHandler<T> eventHandler) {
sources.values().forEach(i -> i.addEventHandler(eventHandler));
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -81,4 +81,5 @@ public T remove(ResourceID key) {
public void put(ResourceID key, T resource) {
cache.put(key, resource);
}

}
Original file line number Diff line number Diff line change
Expand Up @@ -39,4 +39,5 @@ public void stop() {
super.stop();
manager().stop();
}

}
Original file line number Diff line number Diff line change
@@ -0,0 +1,93 @@
package io.javaoperatorsdk.operator.processing.dependent.kubernetes;

import java.util.Optional;

import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;

import io.fabric8.kubernetes.api.model.ConfigMap;
import io.fabric8.kubernetes.api.model.HasMetadata;
import io.fabric8.kubernetes.api.model.ObjectMeta;
import io.javaoperatorsdk.operator.api.config.informer.InformerConfiguration;
import io.javaoperatorsdk.operator.api.reconciler.Context;
import io.javaoperatorsdk.operator.processing.event.ResourceID;
import io.javaoperatorsdk.operator.processing.event.source.AssociatedSecondaryResourceIdentifier;
import io.javaoperatorsdk.operator.processing.event.source.informer.InformerEventSource;
import io.javaoperatorsdk.operator.sample.simple.TestCustomResource;

import static org.assertj.core.api.Assertions.assertThat;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.Mockito.*;

class KubernetesDependentResourceTest {

private TemporalResourceCache temporalResourceCacheMock = mock(TemporalResourceCache.class);
private InformerEventSource informerEventSourceMock = mock(InformerEventSource.class);
private AssociatedSecondaryResourceIdentifier associatedResourceIdentifierMock =
mock(AssociatedSecondaryResourceIdentifier.class);

KubernetesDependentResource<ConfigMap, TestCustomResource> kubernetesDependentResource =
new KubernetesDependentResource() {
{
this.temporalResourceCache = temporalResourceCacheMock;
this.informerEventSource = informerEventSourceMock;
}

@Override
protected Object desired(HasMetadata primary, Context context) {
return testResource();
}
};

@BeforeEach
public void setup() {
InformerConfiguration informerConfigurationMock = mock(InformerConfiguration.class);
when(informerEventSourceMock.getConfiguration()).thenReturn(informerConfigurationMock);
when(informerConfigurationMock.getAssociatedResourceIdentifier())
.thenReturn(associatedResourceIdentifierMock);
when(associatedResourceIdentifierMock.associatedSecondaryID(any()))
.thenReturn(ResourceID.fromResource(testResource()));
}

@Test
void getResourceCheckTheTemporalCacheFirst() {
when(temporalResourceCacheMock.getResourceFromCache(any()))
.thenReturn(Optional.of(testResource()));

kubernetesDependentResource.getResource(primaryResource());

verify(temporalResourceCacheMock, times(1)).getResourceFromCache(any());
verify(informerEventSourceMock, never()).get(any());
}

@Test
void getResourceGetsResourceFromInformerIfNotInTemporalCache() {
var resource = testResource();
when(temporalResourceCacheMock.getResourceFromCache(any())).thenReturn(Optional.empty());
when(informerEventSourceMock.get(any())).thenReturn(Optional.of(resource));

var res = kubernetesDependentResource.getResource(primaryResource());

verify(temporalResourceCacheMock, times(1)).getResourceFromCache(any());
verify(informerEventSourceMock, times(1)).get(any());
assertThat(res.orElseThrow()).isEqualTo(resource);
}

TestCustomResource primaryResource() {
TestCustomResource testCustomResource = new TestCustomResource();
testCustomResource.setMetadata(new ObjectMeta());
testCustomResource.getMetadata().setName("test");
testCustomResource.getMetadata().setNamespace("default");
return testCustomResource;
}

ConfigMap testResource() {
ConfigMap configMap = new ConfigMap();
configMap.setMetadata(new ObjectMeta());
configMap.getMetadata().setName("test");
configMap.getMetadata().setNamespace("default");
configMap.getMetadata().setResourceVersion("0");
return configMap;
}

}
Loading