Skip to content

fix: startup resource cache access #2876

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

Open
wants to merge 18 commits into
base: main
Choose a base branch
from
Original file line number Diff line number Diff line change
Expand Up @@ -7,17 +7,19 @@
import io.javaoperatorsdk.operator.processing.event.ResourceID;
import io.javaoperatorsdk.operator.processing.event.source.SecondaryToPrimaryMapper;

class DefaultPrimaryToSecondaryIndex<R extends HasMetadata> implements PrimaryToSecondaryIndex<R> {
class DefaultTemporalPrimaryToSecondaryIndex<R extends HasMetadata>
implements TemporalPrimaryToSecondaryIndex<R> {

private final SecondaryToPrimaryMapper<R> secondaryToPrimaryMapper;
private final Map<ResourceID, Set<ResourceID>> index = new HashMap<>();

public DefaultPrimaryToSecondaryIndex(SecondaryToPrimaryMapper<R> secondaryToPrimaryMapper) {
public DefaultTemporalPrimaryToSecondaryIndex(
SecondaryToPrimaryMapper<R> secondaryToPrimaryMapper) {
this.secondaryToPrimaryMapper = secondaryToPrimaryMapper;
}

@Override
public synchronized void onAddOrUpdate(R resource) {
public synchronized void explicitAddOrUpdate(R resource) {
Set<ResourceID> primaryResources = secondaryToPrimaryMapper.toPrimaryResourceIDs(resource);
primaryResources.forEach(
primaryResource -> {
Expand All @@ -28,7 +30,7 @@ public synchronized void onAddOrUpdate(R resource) {
}

@Override
public synchronized void onDelete(R resource) {
public synchronized void cleanupForResource(R resource) {
Set<ResourceID> primaryResources = secondaryToPrimaryMapper.toPrimaryResourceIDs(resource);
primaryResources.forEach(
primaryResource -> {
Expand All @@ -51,7 +53,7 @@ public synchronized Set<ResourceID> getSecondaryResources(ResourceID primary) {
if (resourceIDs == null) {
return Collections.emptySet();
} else {
return Collections.unmodifiableSet(resourceIDs);
return new HashSet<>(resourceIDs);
}
}
}
Original file line number Diff line number Diff line change
@@ -1,5 +1,6 @@
package io.javaoperatorsdk.operator.processing.event.source.informer;

import java.util.Map;
import java.util.Optional;
import java.util.Set;
import java.util.UUID;
Expand Down Expand Up @@ -58,11 +59,13 @@ public class InformerEventSource<R extends HasMetadata, P extends HasMetadata>
extends ManagedInformerEventSource<R, P, InformerEventSourceConfiguration<R>>
implements ResourceEventHandler<R> {

public static final String PRIMARY_TO_SECONDARY_INDEX_NAME = "primaryToSecondary";

public static final String PREVIOUS_ANNOTATION_KEY = "javaoperatorsdk.io/previous";
private static final Logger log = LoggerFactory.getLogger(InformerEventSource.class);
// we need direct control for the indexer to propagate the just update resource also to the index
private final PrimaryToSecondaryIndex<R> primaryToSecondaryIndex;
private final PrimaryToSecondaryMapper<P> primaryToSecondaryMapper;
private final TemporalPrimaryToSecondaryIndex<R> temporalPrimaryToSecondaryIndex;
private final String id = UUID.randomUUID().toString();

public InformerEventSource(
Expand Down Expand Up @@ -95,12 +98,18 @@ private InformerEventSource(
parseResourceVersions);
// If there is a primary to secondary mapper there is no need for primary to secondary index.
primaryToSecondaryMapper = configuration.getPrimaryToSecondaryMapper();
if (primaryToSecondaryMapper == null) {
primaryToSecondaryIndex =
// The index uses the secondary to primary mapper (always present) to build the index
new DefaultPrimaryToSecondaryIndex<>(configuration.getSecondaryToPrimaryMapper());
if (useSecondaryToPrimaryIndex()) {
temporalPrimaryToSecondaryIndex =
new DefaultTemporalPrimaryToSecondaryIndex<>(configuration.getSecondaryToPrimaryMapper());
addIndexers(
Map.of(
PRIMARY_TO_SECONDARY_INDEX_NAME,
(R r) ->
configuration.getSecondaryToPrimaryMapper().toPrimaryResourceIDs(r).stream()
.map(InformerEventSource::resourceIdToString)
.toList()));
} else {
primaryToSecondaryIndex = NOOPPrimaryToSecondaryIndex.getInstance();
temporalPrimaryToSecondaryIndex = NOOPTemporalPrimaryToSecondaryIndex.getInstance();
}

final var informerConfig = configuration.getInformerConfig();
Expand All @@ -119,7 +128,6 @@ public void onAdd(R newResource) {
resourceType().getSimpleName(),
newResource.getMetadata().getResourceVersion());
}
primaryToSecondaryIndex.onAddOrUpdate(newResource);
onAddOrUpdate(
Operation.ADD, newResource, null, () -> InformerEventSource.super.onAdd(newResource));
}
Expand All @@ -134,7 +142,7 @@ public void onUpdate(R oldObject, R newObject) {
newObject.getMetadata().getResourceVersion(),
oldObject.getMetadata().getResourceVersion());
}
primaryToSecondaryIndex.onAddOrUpdate(newObject);

onAddOrUpdate(
Operation.UPDATE,
newObject,
Expand All @@ -150,7 +158,7 @@ public void onDelete(R resource, boolean b) {
ResourceID.fromResource(resource),
resourceType().getSimpleName());
}
primaryToSecondaryIndex.onDelete(resource);
temporalPrimaryToSecondaryIndex.cleanupForResource(resource);
super.onDelete(resource, b);
if (acceptedByDeleteFilters(resource, b)) {
propagateEvent(resource);
Expand All @@ -160,7 +168,7 @@ public void onDelete(R resource, boolean b) {
private synchronized void onAddOrUpdate(
Operation operation, R newObject, R oldObject, Runnable superOnOp) {
var resourceID = ResourceID.fromResource(newObject);

temporalPrimaryToSecondaryIndex.cleanupForResource(newObject);
if (canSkipEvent(newObject, oldObject, resourceID)) {
log.debug(
"Skipping event propagation for {}, since was a result of a reconcile action. Resource"
Expand Down Expand Up @@ -244,42 +252,68 @@ private void propagateEvent(R object) {

@Override
public Set<R> getSecondaryResources(P primary) {
Set<ResourceID> secondaryIDs;

if (useSecondaryToPrimaryIndex()) {
var primaryResourceID = ResourceID.fromResource(primary);
secondaryIDs = primaryToSecondaryIndex.getSecondaryResources(primaryResourceID);
var primaryID = ResourceID.fromResource(primary);
// Note that the order matter is these lines. This method is not synchronized
// because of performance reasons. If it was in reverse order, it could happen
// that we did not receive yet an event in the informer so the index would not
// be updated. However, before reading it from temp IDs the event arrives and erases
// the temp index. So in case of Add not id would be found.
var temporalIds = temporalPrimaryToSecondaryIndex.getSecondaryResources(primaryID);
var resources = byIndex(PRIMARY_TO_SECONDARY_INDEX_NAME, resourceIdToString(primaryID));

log.debug(
"Using PrimaryToSecondaryIndex to find secondary resources for primary: {}. Found"
+ " secondary ids: {} ",
primaryResourceID,
secondaryIDs);
"Using informer primary to secondary index to find secondary resources for primary name:"
+ " {} namespace: {}. Found number {}",
primary.getMetadata().getName(),
primary.getMetadata().getNamespace(),
resources.size());

log.debug("Complementary ids: {}", temporalIds);
var res =
resources.stream()
.map(
r -> {
var resourceId = ResourceID.fromResource(r);
Optional<R> resource = temporaryResourceCache.getResourceFromCache(resourceId);
temporalIds.remove(resourceId);
return resource.orElse(r);
})
.collect(Collectors.toSet());
temporalIds.forEach(
id -> {
Optional<R> resource = get(id);
resource.ifPresentOrElse(res::add, () -> log.warn("Resource not found: {}", id));
});
return res;
} else {
secondaryIDs = primaryToSecondaryMapper.toSecondaryResourceIDs(primary);
Set<ResourceID> secondaryIDs = primaryToSecondaryMapper.toSecondaryResourceIDs(primary);
log.debug(
"Using PrimaryToSecondaryMapper to find secondary resources for primary: {}. Found"
+ " secondary ids: {} ",
primary,
secondaryIDs);
return secondaryIDs.stream()
.map(this::get)
.flatMap(Optional::stream)
.collect(Collectors.toSet());
}
return secondaryIDs.stream()
.map(this::get)
.flatMap(Optional::stream)
.collect(Collectors.toSet());
}

@Override
public synchronized void handleRecentResourceUpdate(
ResourceID resourceID, R resource, R previousVersionOfResource) {
handleRecentCreateOrUpdate(Operation.UPDATE, resource, previousVersionOfResource);
handleRecentCreateOrUpdate(resource, previousVersionOfResource);
}

@Override
public synchronized void handleRecentResourceCreate(ResourceID resourceID, R resource) {
handleRecentCreateOrUpdate(Operation.ADD, resource, null);
handleRecentCreateOrUpdate(resource, null);
}

private void handleRecentCreateOrUpdate(Operation operation, R newResource, R oldResource) {
primaryToSecondaryIndex.onAddOrUpdate(newResource);
private void handleRecentCreateOrUpdate(R newResource, R oldResource) {
temporalPrimaryToSecondaryIndex.explicitAddOrUpdate(newResource);
temporaryResourceCache.putResource(
newResource,
Optional.ofNullable(oldResource)
Expand Down Expand Up @@ -332,4 +366,8 @@ private enum Operation {
ADD,
UPDATE
}

private static String resourceIdToString(ResourceID resourceID) {
return resourceID.getName() + "#" + resourceID.getNamespace().orElse("$na");
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -5,25 +5,27 @@
import io.fabric8.kubernetes.api.model.HasMetadata;
import io.javaoperatorsdk.operator.processing.event.ResourceID;

class NOOPPrimaryToSecondaryIndex<R extends HasMetadata> implements PrimaryToSecondaryIndex<R> {
class NOOPTemporalPrimaryToSecondaryIndex<R extends HasMetadata>
implements TemporalPrimaryToSecondaryIndex<R> {

@SuppressWarnings("rawtypes")
private static final NOOPPrimaryToSecondaryIndex instance = new NOOPPrimaryToSecondaryIndex();
private static final NOOPTemporalPrimaryToSecondaryIndex instance =
new NOOPTemporalPrimaryToSecondaryIndex();

@SuppressWarnings("unchecked")
public static <T extends HasMetadata> NOOPPrimaryToSecondaryIndex<T> getInstance() {
public static <T extends HasMetadata> NOOPTemporalPrimaryToSecondaryIndex<T> getInstance() {
return instance;
}

private NOOPPrimaryToSecondaryIndex() {}
private NOOPTemporalPrimaryToSecondaryIndex() {}

@Override
public void onAddOrUpdate(R resource) {
public void explicitAddOrUpdate(R resource) {
// empty method because of noop implementation
}

@Override
public void onDelete(R resource) {
public void cleanupForResource(R resource) {
// empty method because of noop implementation
}

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -5,11 +5,11 @@
import io.fabric8.kubernetes.api.model.HasMetadata;
import io.javaoperatorsdk.operator.processing.event.ResourceID;

public interface PrimaryToSecondaryIndex<R extends HasMetadata> {
public interface TemporalPrimaryToSecondaryIndex<R extends HasMetadata> {

void onAddOrUpdate(R resource);
void explicitAddOrUpdate(R resource);

void onDelete(R resource);
void cleanupForResource(R resource);

Set<ResourceID> getSecondaryResources(ResourceID primary);
}
Original file line number Diff line number Diff line change
Expand Up @@ -15,14 +15,14 @@
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;

class PrimaryToSecondaryIndexTest {
class TemporalPrimaryToSecondaryIndexTest {

@SuppressWarnings("unchecked")
private final SecondaryToPrimaryMapper<ConfigMap> secondaryToPrimaryMapperMock =
mock(SecondaryToPrimaryMapper.class);

private final PrimaryToSecondaryIndex<ConfigMap> primaryToSecondaryIndex =
new DefaultPrimaryToSecondaryIndex<>(secondaryToPrimaryMapperMock);
private final TemporalPrimaryToSecondaryIndex<ConfigMap> temporalPrimaryToSecondaryIndex =
new DefaultTemporalPrimaryToSecondaryIndex<>(secondaryToPrimaryMapperMock);

private final ResourceID primaryID1 = new ResourceID("id1", "default");
private final ResourceID primaryID2 = new ResourceID("id2", "default");
Expand All @@ -37,28 +37,29 @@ void setup() {

@Test
void returnsEmptySetOnEmptyIndex() {
var res = primaryToSecondaryIndex.getSecondaryResources(ResourceID.fromResource(secondary1));
var res =
temporalPrimaryToSecondaryIndex.getSecondaryResources(ResourceID.fromResource(secondary1));
assertThat(res).isEmpty();
}

@Test
void indexesNewResources() {
primaryToSecondaryIndex.onAddOrUpdate(secondary1);
temporalPrimaryToSecondaryIndex.explicitAddOrUpdate(secondary1);

var secondaryResources1 = primaryToSecondaryIndex.getSecondaryResources(primaryID1);
var secondaryResources2 = primaryToSecondaryIndex.getSecondaryResources(primaryID2);
var secondaryResources1 = temporalPrimaryToSecondaryIndex.getSecondaryResources(primaryID1);
var secondaryResources2 = temporalPrimaryToSecondaryIndex.getSecondaryResources(primaryID2);

assertThat(secondaryResources1).containsOnly(ResourceID.fromResource(secondary1));
assertThat(secondaryResources2).containsOnly(ResourceID.fromResource(secondary1));
}

@Test
void indexesAdditionalResources() {
primaryToSecondaryIndex.onAddOrUpdate(secondary1);
primaryToSecondaryIndex.onAddOrUpdate(secondary2);
temporalPrimaryToSecondaryIndex.explicitAddOrUpdate(secondary1);
temporalPrimaryToSecondaryIndex.explicitAddOrUpdate(secondary2);

var secondaryResources1 = primaryToSecondaryIndex.getSecondaryResources(primaryID1);
var secondaryResources2 = primaryToSecondaryIndex.getSecondaryResources(primaryID2);
var secondaryResources1 = temporalPrimaryToSecondaryIndex.getSecondaryResources(primaryID1);
var secondaryResources2 = temporalPrimaryToSecondaryIndex.getSecondaryResources(primaryID2);

assertThat(secondaryResources1)
.containsOnly(ResourceID.fromResource(secondary1), ResourceID.fromResource(secondary2));
Expand All @@ -68,20 +69,20 @@ void indexesAdditionalResources() {

@Test
void removingResourceFromIndex() {
primaryToSecondaryIndex.onAddOrUpdate(secondary1);
primaryToSecondaryIndex.onAddOrUpdate(secondary2);
primaryToSecondaryIndex.onDelete(secondary1);
temporalPrimaryToSecondaryIndex.explicitAddOrUpdate(secondary1);
temporalPrimaryToSecondaryIndex.explicitAddOrUpdate(secondary2);
temporalPrimaryToSecondaryIndex.cleanupForResource(secondary1);

var secondaryResources1 = primaryToSecondaryIndex.getSecondaryResources(primaryID1);
var secondaryResources2 = primaryToSecondaryIndex.getSecondaryResources(primaryID2);
var secondaryResources1 = temporalPrimaryToSecondaryIndex.getSecondaryResources(primaryID1);
var secondaryResources2 = temporalPrimaryToSecondaryIndex.getSecondaryResources(primaryID2);

assertThat(secondaryResources1).containsOnly(ResourceID.fromResource(secondary2));
assertThat(secondaryResources2).containsOnly(ResourceID.fromResource(secondary2));

primaryToSecondaryIndex.onDelete(secondary2);
temporalPrimaryToSecondaryIndex.cleanupForResource(secondary2);

secondaryResources1 = primaryToSecondaryIndex.getSecondaryResources(primaryID1);
secondaryResources2 = primaryToSecondaryIndex.getSecondaryResources(primaryID2);
secondaryResources1 = temporalPrimaryToSecondaryIndex.getSecondaryResources(primaryID1);
secondaryResources2 = temporalPrimaryToSecondaryIndex.getSecondaryResources(primaryID2);

assertThat(secondaryResources1).isEmpty();
assertThat(secondaryResources2).isEmpty();
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -111,6 +111,10 @@ public <T extends HasMetadata> T create(T resource) {
return kubernetesClient.resource(resource).inNamespace(namespace).create();
}

public <T extends HasMetadata> T serverSideApply(T resource) {
return kubernetesClient.resource(resource).inNamespace(namespace).serverSideApply();
}

public <T extends HasMetadata> T replace(T resource) {
return kubernetesClient.resource(resource).inNamespace(namespace).replace();
}
Expand Down
Loading