improve: ResourceIDMapper and ResourceIDProvider for external resources

Will check if this could be elegantly extended to bulk resources

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

Author: csviri

operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/CacheKeyMapper.java renamed to operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/ResourceIDMapper.java

@@ -13,13 +13,13 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package io.javaoperatorsdk.operator.processing.event.source;
+package io.javaoperatorsdk.operator.processing;
 
-import io.javaoperatorsdk.operator.processing.ResourceIDProvider;
+import io.javaoperatorsdk.operator.processing.event.source.ExternalResourceCachingEventSource;
 
-public interface CacheKeyMapper<R, ID> {
+public interface ResourceIDMapper<R, ID> {
 
-  ID keyFor(R resource);
+  ID idFor(R resource);
 
   /**
    * Used if a polling event source handles only single secondary resource and the id is String. See
@@ -28,11 +28,11 @@ public interface CacheKeyMapper<R, ID> {
    * @return static id mapper, all resources are mapped for same id.
    * @param <T> secondary resource type
    */
-  static <T> CacheKeyMapper<T, String> singleResourceCacheKeyMapper() {
+  static <T> ResourceIDMapper<T, String> singleResourceCacheKeyMapper() {
     return r -> "id";
   }
 
-  static <T, ID> CacheKeyMapper<T, ID> resourceIdProviderMapper() {
+  static <T, ID> ResourceIDMapper<T, ID> resourceIdProviderMapper() {
     return r -> {
       if (r instanceof ResourceIDProvider resourceIDProvider) {
         return (ID) resourceIDProvider.resourceId();

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

@@ -19,8 +19,7 @@
  * Provides the identifier for an object that represents resource. This ID is used to select target
  * external resource for a dependent resource from the resources returned by `{@link
  * io.javaoperatorsdk.operator.api.reconciler.Context#getSecondaryResources(Class)}`. But also for
- * {@link io.javaoperatorsdk.operator.processing.event.source.CacheKeyMapper} for event sources in
- * external resources
+ * {@link ResourceIDMapper} for event sources in external resources
  *
  * @param <T>
  */

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

@@ -22,20 +22,22 @@
 import io.fabric8.kubernetes.api.model.HasMetadata;
 import io.javaoperatorsdk.operator.api.reconciler.Context;
 import io.javaoperatorsdk.operator.api.reconciler.dependent.RecentOperationCacheFiller;
-import io.javaoperatorsdk.operator.processing.ResourceIDProvider;
+import io.javaoperatorsdk.operator.processing.ResourceIDMapper;
 import io.javaoperatorsdk.operator.processing.event.EventSourceRetriever;
 import io.javaoperatorsdk.operator.processing.event.ResourceID;
 import io.javaoperatorsdk.operator.processing.event.source.EventSource;
 import io.javaoperatorsdk.operator.processing.event.source.informer.InformerEventSource;
 
 public abstract class AbstractExternalDependentResource<
-        R, P extends HasMetadata, T extends EventSource<R, P>>
+        R, P extends HasMetadata, T extends EventSource<R, P>, ID>
     extends AbstractEventSourceHolderDependentResource<R, P, T> {
 
   private final boolean isDependentResourceWithExplicitState =
       this instanceof DependentResourceWithExplicitState;
   private final boolean isBulkDependentResource = this instanceof BulkDependentResource;
 
+  protected ResourceIDMapper<R, ID> resourceIDMapper = ResourceIDMapper.resourceIdProviderMapper();
+
   @SuppressWarnings("rawtypes")
   private DependentResourceWithExplicitState dependentResourceWithExplicitState;
 
@@ -132,21 +134,23 @@ protected Optional<R> selectTargetSecondaryResource(
       Set<R> secondaryResources, P primary, Context<P> context) {
     R desired = desired(primary, context);
     List<R> targetResources;
-    if (desired instanceof ResourceIDProvider<?> desiredWithId) {
-      targetResources =
-          secondaryResources.stream()
-              .filter(
-                  r -> ((ResourceIDProvider<?>) r).resourceId().equals(desiredWithId.resourceId()))
-              .toList();
-    } else {
-      throw new IllegalStateException(
-          "Either implement ExternalDependentIDProvider or override this "
-              + " (selectTargetSecondaryResource) method.");
-    }
+    var desiredID = resourceIDMapper.idFor(desired);
+    targetResources =
+        secondaryResources.stream()
+            .filter(r -> resourceIDMapper.idFor(r).equals(desiredID))
+            .toList();
     if (targetResources.size() > 1) {
       throw new IllegalStateException(
           "More than one secondary resource related to primary: " + targetResources);
     }
     return targetResources.isEmpty() ? Optional.empty() : Optional.of(targetResources.get(0));
   }
+
+  public ResourceIDMapper<R, ID> getResourceIDMapper() {
+    return resourceIDMapper;
+  }
+
+  public void setResourceIDMapper(ResourceIDMapper<R, ID> resourceIDMapper) {
+    this.resourceIDMapper = resourceIDMapper;
+  }
 }

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

@@ -20,13 +20,12 @@
 import io.fabric8.kubernetes.api.model.HasMetadata;
 import io.javaoperatorsdk.operator.api.reconciler.Ignore;
 import io.javaoperatorsdk.operator.processing.dependent.AbstractExternalDependentResource;
-import io.javaoperatorsdk.operator.processing.event.source.CacheKeyMapper;
 import io.javaoperatorsdk.operator.processing.event.source.ExternalResourceCachingEventSource;
 
 @Ignore
 public abstract class AbstractPollingDependentResource<R, P extends HasMetadata, ID>
-    extends AbstractExternalDependentResource<R, P, ExternalResourceCachingEventSource<R, P, ID>>
-    implements CacheKeyMapper<R, ID> {
+    extends AbstractExternalDependentResource<
+        R, P, ExternalResourceCachingEventSource<R, P, ID>, ID> {
 
   public static final Duration DEFAULT_POLLING_PERIOD = Duration.ofMillis(5000);
   private Duration pollingPeriod;
@@ -49,10 +48,4 @@ public void setPollingPeriod(Duration pollingPeriod) {
   public Duration getPollingPeriod() {
     return pollingPeriod;
   }
-
-  // for now dependent resources support event sources only with one owned resource.
-  @Override
-  public ID keyFor(R resource) {
-    return CacheKeyMapper.<R, ID>resourceIdProviderMapper().keyFor(resource);
-  }
 }

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

@@ -47,7 +47,7 @@ protected ExternalResourceCachingEventSource<R, P, ID> createEventSource(
         resourceType(),
         context,
         new PerResourcePollingConfigurationBuilder<R, P, ID>(this, getPollingPeriod())
-            .withCacheKeyMapper(this)
+            .withResourceIDMapper(getResourceIDMapper())
             .withName(name())
             .build());
   }

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

@@ -20,7 +20,7 @@
 import io.fabric8.kubernetes.api.model.HasMetadata;
 import io.javaoperatorsdk.operator.api.reconciler.EventSourceContext;
 import io.javaoperatorsdk.operator.api.reconciler.Ignore;
-import io.javaoperatorsdk.operator.processing.event.source.CacheKeyMapper;
+import io.javaoperatorsdk.operator.processing.ResourceIDMapper;
 import io.javaoperatorsdk.operator.processing.event.source.ExternalResourceCachingEventSource;
 import io.javaoperatorsdk.operator.processing.event.source.polling.PollingConfiguration;
 import io.javaoperatorsdk.operator.processing.event.source.polling.PollingEventSource;
@@ -30,24 +30,24 @@ public abstract class PollingDependentResource<R, P extends HasMetadata, ID>
     extends AbstractPollingDependentResource<R, P, ID>
     implements PollingEventSource.GenericResourceFetcher<R> {
 
-  private final CacheKeyMapper<R, ID> cacheKeyMapper;
+  private final ResourceIDMapper<R, ID> resourceIDMapper;
 
-  public PollingDependentResource(Class<R> resourceType, CacheKeyMapper<R, ID> cacheKeyMapper) {
+  public PollingDependentResource(Class<R> resourceType, ResourceIDMapper<R, ID> resourceIDMapper) {
     super(resourceType);
-    this.cacheKeyMapper = cacheKeyMapper;
+    this.resourceIDMapper = resourceIDMapper;
   }
 
   public PollingDependentResource(
-      Class<R> resourceType, Duration pollingPeriod, CacheKeyMapper<R, ID> cacheKeyMapper) {
+      Class<R> resourceType, Duration pollingPeriod, ResourceIDMapper<R, ID> resourceIDMapper) {
     super(resourceType, pollingPeriod);
-    this.cacheKeyMapper = cacheKeyMapper;
+    this.resourceIDMapper = resourceIDMapper;
   }
 
   @Override
   protected ExternalResourceCachingEventSource<R, P, ID> createEventSource(
       EventSourceContext<P> context) {
     return new PollingEventSource<>(
         resourceType(),
-        new PollingConfiguration<>(name(), this, getPollingPeriod(), cacheKeyMapper));
+        new PollingConfiguration<>(name(), this, getPollingPeriod(), resourceIDMapper));
   }
 }

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

@@ -32,6 +32,7 @@
 
 import io.fabric8.kubernetes.api.model.HasMetadata;
 import io.javaoperatorsdk.operator.api.reconciler.dependent.RecentOperationCacheFiller;
+import io.javaoperatorsdk.operator.processing.ResourceIDMapper;
 import io.javaoperatorsdk.operator.processing.ResourceIDProvider;
 import io.javaoperatorsdk.operator.processing.event.Event;
 import io.javaoperatorsdk.operator.processing.event.ResourceID;
@@ -61,24 +62,24 @@ public abstract class ExternalResourceCachingEventSource<R, P extends HasMetadat
   private static final Logger log =
       LoggerFactory.getLogger(ExternalResourceCachingEventSource.class);
 
-  protected final CacheKeyMapper<R, ID> cacheKeyMapper;
+  protected final ResourceIDMapper<R, ID> resourceIDMapper;
 
   protected Map<ResourceID, Map<ID, R>> cache = new ConcurrentHashMap<>();
 
   protected ExternalResourceCachingEventSource(
-      Class<R> resourceClass, CacheKeyMapper<R, ID> cacheKeyMapper) {
-    this(null, resourceClass, cacheKeyMapper);
+      Class<R> resourceClass, ResourceIDMapper<R, ID> resourceIDMapper) {
+    this(null, resourceClass, resourceIDMapper);
   }
 
   protected ExternalResourceCachingEventSource(
-      String name, Class<R> resourceClass, CacheKeyMapper<R, ID> cacheKeyMapper) {
+      String name, Class<R> resourceClass, ResourceIDMapper<R, ID> resourceIDMapper) {
     super(resourceClass, name);
-    if (cacheKeyMapper == CacheKeyMapper.resourceIdProviderMapper()
+    if (resourceIDMapper == ResourceIDMapper.resourceIdProviderMapper()
         && !ResourceIDProvider.class.isAssignableFrom(resourceClass)) {
       throw new IllegalArgumentException(
           "resource class is not a " + ResourceIDProvider.class.getSimpleName());
     }
-    this.cacheKeyMapper = cacheKeyMapper;
+    this.resourceIDMapper = resourceIDMapper;
   }
 
   protected synchronized void handleDelete(ResourceID primaryID) {
@@ -90,11 +91,11 @@ protected synchronized void handleDelete(ResourceID primaryID) {
 
   protected synchronized void handleDeletes(ResourceID primaryID, Set<R> resource) {
     handleDelete(
-        primaryID, resource.stream().map(cacheKeyMapper::keyFor).collect(Collectors.toSet()));
+        primaryID, resource.stream().map(resourceIDMapper::idFor).collect(Collectors.toSet()));
   }
 
   protected synchronized void handleDelete(ResourceID primaryID, R resource) {
-    handleDelete(primaryID, Set.of(cacheKeyMapper.keyFor(resource)));
+    handleDelete(primaryID, Set.of(resourceIDMapper.idFor(resource)));
   }
 
   protected synchronized void handleDelete(ResourceID primaryID, Set<ID> resourceIDs) {
@@ -143,7 +144,7 @@ protected synchronized void handleResources(
       cachedResources = Collections.emptyMap();
     }
     var newResourcesMap =
-        newResources.stream().collect(Collectors.toMap(cacheKeyMapper::keyFor, r -> r));
+        newResources.stream().collect(Collectors.toMap(resourceIDMapper::idFor, r -> r));
     cache.put(primaryID, newResourcesMap);
     if (propagateEvent
         && !newResourcesMap.equals(cachedResources)
@@ -205,7 +206,7 @@ private boolean acceptedByGenericFiler(R resource) {
   @Override
   public synchronized void handleRecentResourceCreate(ResourceID primaryID, R resource) {
     var actualValues = cache.get(primaryID);
-    var resourceId = cacheKeyMapper.keyFor(resource);
+    var resourceId = resourceIDMapper.idFor(resource);
     if (actualValues == null) {
       actualValues = new HashMap<>();
       cache.put(primaryID, actualValues);
@@ -220,7 +221,7 @@ public synchronized void handleRecentResourceUpdate(
       ResourceID primaryID, R resource, R previousVersionOfResource) {
     var actualValues = cache.get(primaryID);
     if (actualValues != null) {
-      var resourceId = cacheKeyMapper.keyFor(resource);
+      var resourceId = resourceIDMapper.idFor(resource);
       R actualResource = actualValues.get(resourceId);
       if (actualResource.equals(previousVersionOfResource)) {
         actualValues.put(resourceId, resource);

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

@@ -21,8 +21,8 @@
 import java.util.concurrent.ConcurrentHashMap;
 
 import io.fabric8.kubernetes.api.model.HasMetadata;
+import io.javaoperatorsdk.operator.processing.ResourceIDMapper;
 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.ResourceEventAware;
 
@@ -35,8 +35,8 @@ public class CachingInboundEventSource<R, P extends HasMetadata, ID>
   public CachingInboundEventSource(
       ResourceFetcher<R, P> resourceFetcher,
       Class<R> resourceClass,
-      CacheKeyMapper<R, ID> cacheKeyMapper) {
-    super(resourceClass, cacheKeyMapper);
+      ResourceIDMapper<R, ID> resourceIDMapper) {
+    super(resourceClass, resourceIDMapper);
     this.resourceFetcher = resourceFetcher;
   }
 

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

@@ -22,12 +22,12 @@
 import java.util.function.Predicate;
 
 import io.fabric8.kubernetes.api.model.HasMetadata;
-import io.javaoperatorsdk.operator.processing.event.source.CacheKeyMapper;
+import io.javaoperatorsdk.operator.processing.ResourceIDMapper;
 
 public record PerResourcePollingConfiguration<R, P extends HasMetadata, ID>(
     String name,
     ScheduledExecutorService executorService,
-    CacheKeyMapper<R, ID> cacheKeyMapper,
+    ResourceIDMapper<R, ID> resourceIDMapper,
     PerResourcePollingEventSource.ResourceFetcher<R, P> resourceFetcher,
     Predicate<P> registerPredicate,
     Duration defaultPollingPeriod) {
@@ -37,7 +37,7 @@ public record PerResourcePollingConfiguration<R, P extends HasMetadata, ID>(
   public PerResourcePollingConfiguration(
       String name,
       ScheduledExecutorService executorService,
-      CacheKeyMapper<R, ID> cacheKeyMapper,
+      ResourceIDMapper<R, ID> resourceIDMapper,
       PerResourcePollingEventSource.ResourceFetcher<R, P> resourceFetcher,
       Predicate<P> registerPredicate,
       Duration defaultPollingPeriod) {
@@ -46,8 +46,8 @@ public PerResourcePollingConfiguration(
         executorService == null
             ? new ScheduledThreadPoolExecutor(DEFAULT_EXECUTOR_THREAD_NUMBER)
             : executorService;
-    this.cacheKeyMapper =
-        cacheKeyMapper == null ? CacheKeyMapper.resourceIdProviderMapper() : cacheKeyMapper;
+    this.resourceIDMapper =
+        resourceIDMapper == null ? ResourceIDMapper.resourceIdProviderMapper() : resourceIDMapper;
     this.resourceFetcher = Objects.requireNonNull(resourceFetcher);
     this.registerPredicate = registerPredicate;
     this.defaultPollingPeriod = defaultPollingPeriod;

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

@@ -20,7 +20,7 @@
 import java.util.function.Predicate;
 
 import io.fabric8.kubernetes.api.model.HasMetadata;
-import io.javaoperatorsdk.operator.processing.event.source.CacheKeyMapper;
+import io.javaoperatorsdk.operator.processing.ResourceIDMapper;
 
 public final class PerResourcePollingConfigurationBuilder<R, P extends HasMetadata, ID> {
 
@@ -30,7 +30,7 @@ public final class PerResourcePollingConfigurationBuilder<R, P extends HasMetada
   private String name;
   private Predicate<P> registerPredicate;
   private ScheduledExecutorService executorService;
-  private CacheKeyMapper<R, ID> cacheKeyMapper;
+  private ResourceIDMapper<R, ID> resourceIDMapper;
 
   public PerResourcePollingConfigurationBuilder(
       PerResourcePollingEventSource.ResourceFetcher<R, P> resourceFetcher,
@@ -52,9 +52,9 @@ public PerResourcePollingConfigurationBuilder<R, P, ID> withRegisterPredicate(
     return this;
   }
 
-  public PerResourcePollingConfigurationBuilder<R, P, ID> withCacheKeyMapper(
-      CacheKeyMapper<R, ID> cacheKeyMapper) {
-    this.cacheKeyMapper = cacheKeyMapper;
+  public PerResourcePollingConfigurationBuilder<R, P, ID> withResourceIDMapper(
+      ResourceIDMapper<R, ID> resourceIDMapper) {
+    this.resourceIDMapper = resourceIDMapper;
     return this;
   }
 
@@ -67,7 +67,7 @@ public PerResourcePollingConfiguration<R, P, ID> build() {
     return new PerResourcePollingConfiguration<>(
         name,
         executorService,
-        cacheKeyMapper,
+        resourceIDMapper,
         resourceFetcher,
         registerPredicate,
         defaultPollingPeriod);