feat: support indexers in informer related event sources (#1131)

* feat: support indexers in informer related event sources

* fix: simplification

* fix: add primary indexer

* fix: naming of IT

* fix: added IT

* fix: remove smell

* fix: code improvements

* feat: add convenient addIndexer method

* chore: add equivalent managed dependent implementation and test

Co-authored-by: Chris Laprun <metacosm@gmail.com>

Author: csviri

operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/informer/InformerConfiguration.java

@@ -64,13 +64,13 @@ private InformerConfigurationBuilder(Class<R> resourceClass) {
       this.resourceClass = resourceClass;
     }
 
-    public InformerConfigurationBuilder<R, P> withPrimaryResourcesRetriever(
+    public InformerConfigurationBuilder<R, P> withSecondaryToPrimaryMapper(
         SecondaryToPrimaryMapper<R> secondaryToPrimaryMapper) {
       this.secondaryToPrimaryResourcesIdSet = secondaryToPrimaryMapper;
       return this;
     }
 
-    public InformerConfigurationBuilder<R, P> withAssociatedSecondaryResourceIdentifier(
+    public InformerConfigurationBuilder<R, P> withPrimaryToSecondaryMapper(
         PrimaryToSecondaryMapper<P> associatedWith) {
       this.associatedWith = associatedWith;
       return this;
@@ -115,8 +115,8 @@ static <R extends HasMetadata, P extends HasMetadata> InformerConfigurationBuild
     return new InformerConfigurationBuilder<R, P>(configuration.getResourceClass())
         .withNamespaces(configuration.getNamespaces())
         .withLabelSelector(configuration.getLabelSelector())
-        .withAssociatedSecondaryResourceIdentifier(
+        .withPrimaryToSecondaryMapper(
             configuration.getPrimaryToSecondaryMapper())
-        .withPrimaryResourcesRetriever(configuration.getSecondaryToPrimaryMapper());
+        .withSecondaryToPrimaryMapper(configuration.getSecondaryToPrimaryMapper());
   }
 }

operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/reconciler/EventSourceContext.java

@@ -4,7 +4,7 @@
 import io.fabric8.kubernetes.client.KubernetesClient;
 import io.javaoperatorsdk.operator.api.config.ControllerConfiguration;
 import io.javaoperatorsdk.operator.processing.event.source.EventSource;
-import io.javaoperatorsdk.operator.processing.event.source.ResourceCache;
+import io.javaoperatorsdk.operator.processing.event.source.IndexerResourceCache;
 
 /**
  * Contextual information made available to event sources.
@@ -13,11 +13,11 @@
  */
 public class EventSourceContext<P extends HasMetadata> {
 
-  private final ResourceCache<P> primaryCache;
+  private final IndexerResourceCache<P> primaryCache;
   private final ControllerConfiguration<P> controllerConfiguration;
   private final KubernetesClient client;
 
-  public EventSourceContext(ResourceCache<P> primaryCache,
+  public EventSourceContext(IndexerResourceCache<P> primaryCache,
       ControllerConfiguration<P> controllerConfiguration,
       KubernetesClient client) {
     this.primaryCache = primaryCache;
@@ -30,7 +30,7 @@ public EventSourceContext(ResourceCache<P> primaryCache,
    *
    * @return the primary resource cache
    */
-  public ResourceCache<P> getPrimaryCache() {
+  public IndexerResourceCache<P> getPrimaryCache() {
     return primaryCache;
   }
 

operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/kubernetes/KubernetesDependentResource.java

@@ -66,8 +66,8 @@ private void configureWith(String labelSelector, Set<String> namespaces) {
         InformerConfiguration.from(resourceType())
             .withLabelSelector(labelSelector)
             .withNamespaces(namespaces)
-            .withPrimaryResourcesRetriever(primaryResourcesRetriever)
-            .withAssociatedSecondaryResourceIdentifier(secondaryResourceIdentifier)
+            .withSecondaryToPrimaryMapper(primaryResourcesRetriever)
+            .withPrimaryToSecondaryMapper(secondaryResourceIdentifier)
             .build();
     configureWith(new InformerEventSource<>(ic, client));
   }

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

@@ -0,0 +1,19 @@
+package io.javaoperatorsdk.operator.processing.event.source;
+
+import java.util.List;
+import java.util.Map;
+import java.util.function.Function;
+
+import io.fabric8.kubernetes.api.model.HasMetadata;
+
+public interface IndexerResourceCache<T extends HasMetadata> extends ResourceCache<T> {
+
+  void addIndexers(Map<String, Function<T, List<String>>> indexers);
+
+  default void addIndexer(String name, Function<T, List<String>> indexer) {
+    addIndexers(Map.of(name, indexer));
+  }
+
+  List<T> byIndex(String indexName, String indexKey);
+
+}

operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/controller/ControllerResourceCache.java

@@ -14,7 +14,6 @@
 import io.javaoperatorsdk.operator.processing.event.Event;
 import io.javaoperatorsdk.operator.processing.event.EventHandler;
 import io.javaoperatorsdk.operator.processing.event.ResourceID;
-import io.javaoperatorsdk.operator.processing.event.source.ResourceCache;
 
 /**
  * <p>
@@ -64,7 +63,7 @@
  */
 public class InformerEventSource<R extends HasMetadata, P extends HasMetadata>
     extends ManagedInformerEventSource<R, P, InformerConfiguration<R, P>>
-    implements ResourceCache<R>, ResourceEventHandler<R>, RecentOperationEventFilter<R> {
+    implements ResourceEventHandler<R>, RecentOperationEventFilter<R> {
 
   private static final Logger log = LoggerFactory.getLogger(InformerEventSource.class);
 

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

@@ -1,9 +1,10 @@
 package io.javaoperatorsdk.operator.processing.event.source.informer;
 
-import java.util.Map;
-import java.util.Optional;
+import java.util.*;
 import java.util.concurrent.ConcurrentHashMap;
+import java.util.function.Function;
 import java.util.function.Predicate;
+import java.util.stream.Collectors;
 import java.util.stream.Stream;
 
 import org.slf4j.Logger;
@@ -22,11 +23,11 @@
 import io.javaoperatorsdk.operator.processing.LifecycleAware;
 import io.javaoperatorsdk.operator.processing.event.ResourceID;
 import io.javaoperatorsdk.operator.processing.event.source.Cache;
-import io.javaoperatorsdk.operator.processing.event.source.ResourceCache;
+import io.javaoperatorsdk.operator.processing.event.source.IndexerResourceCache;
 import io.javaoperatorsdk.operator.processing.event.source.UpdatableCache;
 
 public class InformerManager<T extends HasMetadata, C extends ResourceConfiguration<T>>
-    implements LifecycleAware, ResourceCache<T>, UpdatableCache<T> {
+    implements LifecycleAware, IndexerResourceCache<T>, UpdatableCache<T> {
 
   private static final String ANY_NAMESPACE_MAP_KEY = "anyNamespace";
   private static final Logger log = LoggerFactory.getLogger(InformerManager.class);
@@ -89,7 +90,7 @@ public void stop() {
   @Override
   public Stream<T> list(Predicate<T> predicate) {
     if (predicate == null) {
-      return sources.values().stream().flatMap(ResourceCache::list);
+      return sources.values().stream().flatMap(IndexerResourceCache::list);
     }
     return sources.values().stream().flatMap(i -> i.list(predicate));
   }
@@ -144,4 +145,14 @@ public void put(ResourceID key, T resource) {
                 key, resource));
   }
 
+  @Override
+  public void addIndexers(Map<String, Function<T, List<String>>> indexers) {
+    sources.values().forEach(s -> s.addIndexers(indexers));
+  }
+
+  @Override
+  public List<T> byIndex(String indexName, String indexKey) {
+    return sources.values().stream().map(s -> s.byIndex(indexName, indexKey))
+        .flatMap(List::stream).collect(Collectors.toList());
+  }
 }