All Resource Event Sources can handle multiple seconday resources for a primary resources (#1169)

Author: csviri

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

@@ -1,7 +1,5 @@
 package io.javaoperatorsdk.operator.api.reconciler;
 
-import java.util.Collections;
-import java.util.List;
 import java.util.Optional;
 import java.util.Set;
 import java.util.stream.Collectors;
@@ -10,7 +8,6 @@
 import io.javaoperatorsdk.operator.api.config.ControllerConfiguration;
 import io.javaoperatorsdk.operator.api.reconciler.dependent.managed.ManagedDependentResourceContext;
 import io.javaoperatorsdk.operator.processing.Controller;
-import io.javaoperatorsdk.operator.processing.MultiResourceOwner;
 
 public class DefaultContext<P extends HasMetadata> implements Context<P> {
 
@@ -37,17 +34,8 @@ public Optional<RetryInfo> getRetryInfo() {
   @SuppressWarnings("unchecked")
   public <T> Set<T> getSecondaryResources(Class<T> expectedType) {
     return controller.getEventSourceManager().getEventSourcesFor(expectedType).stream()
-        .map(
-            es -> {
-              if (es instanceof MultiResourceOwner) {
-                return ((MultiResourceOwner<T, P>) es).getSecondaryResources(primaryResource);
-              } else {
-                return es.getSecondaryResource(primaryResource)
-                    .map(List::of)
-                    .orElse(Collections.emptyList());
-              }
-            })
-        .flatMap(List::stream)
+        .map(es -> es.getSecondaryResources(primaryResource))
+        .flatMap(Set::stream)
         .collect(Collectors.toSet());
   }
 

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

@@ -6,5 +6,5 @@ public interface RecentOperationCacheFiller<R> {
 
   void handleRecentResourceCreate(ResourceID resourceID, R resource);
 
-  void handleRecentResourceUpdate(ResourceID resourceID, R resource, R previousResourceVersion);
+  void handleRecentResourceUpdate(ResourceID resourceID, R resource, R previousVersionOfResource);
 }

operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/MultiResourceOwner.java

@@ -4,7 +4,7 @@
 
 import io.fabric8.kubernetes.api.model.HasMetadata;
 import io.javaoperatorsdk.operator.processing.dependent.AbstractEventSourceHolderDependentResource;
-import io.javaoperatorsdk.operator.processing.event.ExternalResourceCachingEventSource;
+import io.javaoperatorsdk.operator.processing.event.source.ExternalResourceCachingEventSource;
 
 public abstract class AbstractCachingDependentResource<R, P extends HasMetadata>
     extends
@@ -15,8 +15,6 @@ protected AbstractCachingDependentResource(Class<R> resourceType) {
     this.resourceType = resourceType;
   }
 
-  public abstract Optional<R> fetchResource(P primaryResource);
-
   @Override
   public Class<R> resourceType() {
     return resourceType;

operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/external/AbstractCachingDependentResource.java

@@ -1,9 +1,10 @@
 package io.javaoperatorsdk.operator.processing.dependent.external;
 
 import io.fabric8.kubernetes.api.model.HasMetadata;
+import io.javaoperatorsdk.operator.processing.event.source.CacheKeyMapper;
 
 public abstract class AbstractPollingDependentResource<R, P extends HasMetadata>
-    extends AbstractCachingDependentResource<R, P> {
+    extends AbstractCachingDependentResource<R, P> implements CacheKeyMapper<R> {
 
   public static final int DEFAULT_POLLING_PERIOD = 5000;
   private long pollingPeriod;
@@ -24,4 +25,10 @@ public void setPollingPeriod(long pollingPeriod) {
   public long getPollingPeriod() {
     return pollingPeriod;
   }
+
+  // for now dependent resources support event sources only with one owned resource.
+  @Override
+  public String keyFor(R resource) {
+    return CacheKeyMapper.singleResourceCacheKeyMapper().keyFor(resource);
+  }
 }

operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/external/AbstractPollingDependentResource.java

@@ -2,12 +2,14 @@
 
 import io.fabric8.kubernetes.api.model.HasMetadata;
 import io.javaoperatorsdk.operator.api.reconciler.EventSourceContext;
-import io.javaoperatorsdk.operator.processing.event.ExternalResourceCachingEventSource;
+import io.javaoperatorsdk.operator.processing.event.source.ExternalResourceCachingEventSource;
 import io.javaoperatorsdk.operator.processing.event.source.polling.PerResourcePollingEventSource;
 
 public abstract class PerResourcePollingDependentResource<R, P extends HasMetadata>
     extends AbstractPollingDependentResource<R, P>
     implements PerResourcePollingEventSource.ResourceFetcher<R, P> {
+
+
   public PerResourcePollingDependentResource(Class<R> resourceType) {
     super(resourceType);
   }
@@ -20,6 +22,7 @@ public PerResourcePollingDependentResource(Class<R> resourceType, long pollingPe
   protected ExternalResourceCachingEventSource<R, P> createEventSource(
       EventSourceContext<P> context) {
     return new PerResourcePollingEventSource<>(this, context.getPrimaryCache(),
-        getPollingPeriod(), resourceType());
+        getPollingPeriod(), resourceType(), this);
   }
+
 }

operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/external/PerResourcePollingDependentResource.java

@@ -1,28 +1,32 @@
 package io.javaoperatorsdk.operator.processing.dependent.external;
 
-import java.util.Map;
-import java.util.function.Supplier;
-
 import io.fabric8.kubernetes.api.model.HasMetadata;
 import io.javaoperatorsdk.operator.api.reconciler.EventSourceContext;
-import io.javaoperatorsdk.operator.processing.event.ExternalResourceCachingEventSource;
-import io.javaoperatorsdk.operator.processing.event.ResourceID;
+import io.javaoperatorsdk.operator.processing.event.source.CacheKeyMapper;
+import io.javaoperatorsdk.operator.processing.event.source.ExternalResourceCachingEventSource;
 import io.javaoperatorsdk.operator.processing.event.source.polling.PollingEventSource;
 
 public abstract class PollingDependentResource<R, P extends HasMetadata>
-    extends AbstractPollingDependentResource<R, P> implements Supplier<Map<ResourceID, R>> {
+    extends AbstractPollingDependentResource<R, P>
+    implements PollingEventSource.GenericResourceFetcher<R> {
+
+  private final CacheKeyMapper<R> cacheKeyMapper;
 
-  public PollingDependentResource(Class<R> resourceType) {
+  public PollingDependentResource(Class<R> resourceType, CacheKeyMapper<R> cacheKeyMapper) {
     super(resourceType);
+    this.cacheKeyMapper = cacheKeyMapper;
   }
 
-  public PollingDependentResource(Class<R> resourceType, long pollingPeriod) {
+  public PollingDependentResource(Class<R> resourceType, long pollingPeriod,
+      CacheKeyMapper<R> cacheKeyMapper) {
     super(resourceType, pollingPeriod);
+    this.cacheKeyMapper = cacheKeyMapper;
   }
 
   @Override
   protected ExternalResourceCachingEventSource<R, P> createEventSource(
       EventSourceContext<P> context) {
-    return new PollingEventSource<>(this, getPollingPeriod(), resourceType());
+    return new PollingEventSource<>(this, getPollingPeriod(), resourceType(), cacheKeyMapper);
   }
+
 }

operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/external/PollingDependentResource.java

@@ -4,7 +4,6 @@
 import io.javaoperatorsdk.operator.processing.event.EventHandler;
 
 public abstract class AbstractEventSource implements EventSource {
-
   private EventHandler handler;
   private volatile boolean running = false;
 

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

@@ -2,7 +2,7 @@
 
 import io.fabric8.kubernetes.api.model.HasMetadata;
 
-public abstract class AbstractResourceEventSource<P extends HasMetadata, R>
+public abstract class AbstractResourceEventSource<R, P extends HasMetadata>
     extends AbstractEventSource
     implements ResourceEventSource<R, P> {
   private final Class<R> resourceClass;

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

@@ -0,0 +1,18 @@
+package io.javaoperatorsdk.operator.processing.event.source;
+
+public interface CacheKeyMapper<R> {
+
+  String keyFor(R resource);
+
+  /**
+   * Used if a polling event source handles only single secondary resource. See also docs for:
+   * {@link ExternalResourceCachingEventSource}
+   *
+   * @return static id mapper, all resources are mapped for same id.
+   * @param <T> secondary resource type
+   */
+  static <T> CacheKeyMapper<T> singleResourceCacheKeyMapper() {
+    return r -> "id";
+  }
+
+}

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

@@ -14,7 +14,7 @@
  * @param <R> represents the type of resources (usually external non-kubernetes ones) being handled.
  */
 public abstract class CachingEventSource<R, P extends HasMetadata>
-    extends AbstractResourceEventSource<P, R> implements Cache<R> {
+    extends AbstractResourceEventSource<R, P> implements Cache<R> {
 
   protected UpdatableCache<R> cache;
 
@@ -43,12 +43,9 @@ public Stream<R> list(Predicate<R> predicate) {
     return cache.list(predicate);
   }
 
-  protected UpdatableCache<R> initCache() {
-    return new ConcurrentHashMapCache<>();
-  }
-
   public Optional<R> getCachedValue(ResourceID resourceID) {
     return cache.get(resourceID);
   }
 
+  protected abstract UpdatableCache<R> initCache();
 }