complementary index

Signed-off-by: Attila Mészáros <[email protected]>

Author: csviri

operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/informer/ComplementaryPrimaryToSecondaryIndex.java

@@ -7,9 +7,9 @@
 
 public interface ComplementaryPrimaryToSecondaryIndex<R extends HasMetadata> {
 
-  void explicitAddOrUpdate(R resource);
+  void explicitAdd(R resource);
 
-  void onCreateOrUpdateEvent(R resourceID);
+  void cleanupForResource(R resourceID);
 
   Set<ResourceID> getComplementarySecondaryResources(ResourceID primary);
 }

operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/informer/DefaultComplementaryPrimaryToSecondaryIndex.java

@@ -1,24 +1,61 @@
 package io.javaoperatorsdk.operator.processing.event.source.informer;
 
+import java.util.Collections;
+import java.util.HashSet;
 import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
 
 import io.fabric8.kubernetes.api.model.HasMetadata;
 import io.javaoperatorsdk.operator.processing.event.ResourceID;
+import io.javaoperatorsdk.operator.processing.event.source.SecondaryToPrimaryMapper;
 
 public class DefaultComplementaryPrimaryToSecondaryIndex<R extends HasMetadata>
     implements ComplementaryPrimaryToSecondaryIndex<R> {
 
   private final ConcurrentHashMap<ResourceID, Set<ResourceID>> index = new ConcurrentHashMap<>();
+  private final SecondaryToPrimaryMapper<R> secondaryToPrimaryMapper;
+
+  public DefaultComplementaryPrimaryToSecondaryIndex(
+      SecondaryToPrimaryMapper<R> secondaryToPrimaryMapper) {
+    this.secondaryToPrimaryMapper = secondaryToPrimaryMapper;
+  }
 
   @Override
-  public void explicitAddOrUpdate(R resource) {}
+  public void explicitAdd(R resource) {
+    Set<ResourceID> primaryResources = secondaryToPrimaryMapper.toPrimaryResourceIDs(resource);
+    primaryResources.forEach(
+        primaryResource -> {
+          var resourceSet =
+              index.computeIfAbsent(primaryResource, pr -> ConcurrentHashMap.newKeySet());
+          resourceSet.add(ResourceID.fromResource(resource));
+        });
+  }
 
   @Override
-  public void onCreateOrUpdateEvent(R resourceID) {}
+  public void cleanupForResource(R resource) {
+    Set<ResourceID> primaryResources = secondaryToPrimaryMapper.toPrimaryResourceIDs(resource);
+    primaryResources.forEach(
+        primaryResource -> {
+          var secondaryResources = index.get(primaryResource);
+          // this can be null in just very special cases, like when the secondaryToPrimaryMapper is
+          // changing dynamically. Like if a list of ResourceIDs mapped dynamically extended in the
+          // mapper between the onAddOrUpdate and onDelete is called.
+          if (secondaryResources != null) {
+            secondaryResources.remove(ResourceID.fromResource(resource));
+            if (secondaryResources.isEmpty()) {
+              index.remove(primaryResource);
+            }
+          }
+        });
+  }
 
   @Override
   public Set<ResourceID> getComplementarySecondaryResources(ResourceID primary) {
-    return Set.of();
+    var resourceIDs = index.get(primary);
+    if (resourceIDs == null) {
+      return Collections.emptySet();
+    } else {
+      return new HashSet<>(resourceIDs);
+    }
   }
 }

operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/informer/InformerEventSource.java

@@ -65,6 +65,7 @@ public class InformerEventSource<R extends HasMetadata, P extends HasMetadata>
   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 PrimaryToSecondaryMapper<P> primaryToSecondaryMapper;
+  private final ComplementaryPrimaryToSecondaryIndex<R> complementaryPrimaryToSecondaryIndex;
   private final String id = UUID.randomUUID().toString();
 
   public InformerEventSource(
@@ -98,13 +99,18 @@ private InformerEventSource(
     // If there is a primary to secondary mapper there is no need for primary to secondary index.
     primaryToSecondaryMapper = configuration.getPrimaryToSecondaryMapper();
     if (primaryToSecondaryMapper == null) {
+      complementaryPrimaryToSecondaryIndex =
+          new DefaultComplementaryPrimaryToSecondaryIndex<>(
+              configuration.getSecondaryToPrimaryMapper());
       addIndexers(
           Map.of(
               PRIMARY_TO_SECONDARY_INDEX_NAME,
               (R r) ->
                   configuration.getSecondaryToPrimaryMapper().toPrimaryResourceIDs(r).stream()
                       .map(InformerEventSource::resourceIdToString)
                       .toList()));
+    } else {
+      complementaryPrimaryToSecondaryIndex = new NOOPComplementaryPrimaryToSecondaryIndex();
     }
 
     final var informerConfig = configuration.getInformerConfig();
@@ -123,7 +129,7 @@ public void onAdd(R newResource) {
           resourceType().getSimpleName(),
           newResource.getMetadata().getResourceVersion());
     }
-
+    complementaryPrimaryToSecondaryIndex.cleanupForResource(newResource);
     onAddOrUpdate(
         Operation.ADD, newResource, null, () -> InformerEventSource.super.onAdd(newResource));
   }
@@ -250,28 +256,33 @@ private void propagateEvent(R object) {
   public Set<R> getSecondaryResources(P primary) {
 
     if (useSecondaryToPrimaryIndex()) {
-
-      var resources =
-          byIndex(
-              PRIMARY_TO_SECONDARY_INDEX_NAME,
-              resourceIdToString(ResourceID.fromResource(primary)));
+      var primaryID = ResourceID.fromResource(primary);
+      var resources = byIndex(PRIMARY_TO_SECONDARY_INDEX_NAME, resourceIdToString(primaryID));
 
       log.debug(
           "Using informer primary to secondary index to find secondary resources for primary name:"
               + " {} namespace: {}. Found {}",
           primary.getMetadata().getName(),
           primary.getMetadata().getNamespace(),
           resources.size());
-
-      return resources.stream()
-          .map(
-              r -> {
-                Optional<R> resource =
-                    temporaryResourceCache.getResourceFromCache(ResourceID.fromResource(r));
-                return resource.orElse(r);
-              })
-          .collect(Collectors.toSet());
-
+      var complementaryIds =
+          complementaryPrimaryToSecondaryIndex.getComplementarySecondaryResources(primaryID);
+      var res =
+          resources.stream()
+              .map(
+                  r -> {
+                    var resourceId = ResourceID.fromResource(r);
+                    Optional<R> resource = temporaryResourceCache.getResourceFromCache(resourceId);
+                    complementaryIds.remove(resourceId);
+                    return resource.orElse(r);
+                  })
+              .collect(Collectors.toSet());
+      complementaryIds.forEach(
+          id -> {
+            Optional<R> resource = temporaryResourceCache.getResourceFromCache(id);
+            resource.ifPresent(res::add);
+          });
+      return res;
     } else {
       Set<ResourceID> secondaryIDs = primaryToSecondaryMapper.toSecondaryResourceIDs(primary);
       log.debug(
@@ -298,8 +309,9 @@ public synchronized void handleRecentResourceCreate(ResourceID resourceID, R res
   }
 
   private void handleRecentCreateOrUpdate(Operation operation, R newResource, R oldResource) {
-    //    todo
-    //    primaryToSecondaryIndex.onAddOrUpdate(newResource);
+    if (operation == Operation.ADD) {
+      complementaryPrimaryToSecondaryIndex.explicitAdd(newResource);
+    }
     temporaryResourceCache.putResource(
         newResource,
         Optional.ofNullable(oldResource)

operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/informer/NOOPComplementaryPrimaryToSecondaryIndex.java

@@ -0,0 +1,21 @@
+package io.javaoperatorsdk.operator.processing.event.source.informer;
+
+import java.util.Set;
+
+import io.fabric8.kubernetes.api.model.HasMetadata;
+import io.javaoperatorsdk.operator.processing.event.ResourceID;
+
+public class NOOPComplementaryPrimaryToSecondaryIndex<R extends HasMetadata>
+    implements ComplementaryPrimaryToSecondaryIndex<R> {
+
+  @Override
+  public void explicitAdd(R resource) {}
+
+  @Override
+  public void cleanupForResource(R resourceID) {}
+
+  @Override
+  public Set<ResourceID> getComplementarySecondaryResources(ResourceID primary) {
+    return Set.of();
+  }
+}