feat: bulk dependent resources (#1448)

Author: csviri

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

@@ -7,5 +7,4 @@
 public interface ResourceDiscriminator<R, P extends HasMetadata> {
 
   Optional<R> distinguish(Class<R> resource, P primary, Context<P> context);
-
 }

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

@@ -1,14 +1,14 @@
 package io.javaoperatorsdk.operator.api.reconciler.dependent;
 
-import java.util.Optional;
+import java.util.*;
+import java.util.stream.Collectors;
 
 import io.fabric8.kubernetes.api.model.HasMetadata;
 import io.javaoperatorsdk.operator.processing.event.ResourceID;
 
 public class ReconcileResult<R> {
 
-  private final R resource;
-  private final Operation operation;
+  private final Map<R, Operation> resourceOperations;
 
   public static <T> ReconcileResult<T> resourceCreated(T resource) {
     return new ReconcileResult<>(resource, Operation.CREATED);
@@ -22,25 +22,49 @@ public static <T> ReconcileResult<T> noOperation(T resource) {
     return new ReconcileResult<>(resource, Operation.NONE);
   }
 
+  @SafeVarargs
+  public static <T> ReconcileResult<T> aggregatedResult(ReconcileResult<T>... results) {
+    if (results == null) {
+      throw new IllegalArgumentException("Should provide results to aggregate");
+    }
+    if (results.length == 1) {
+      return results[0];
+    }
+    final Map<T, Operation> operations = new HashMap<>(results.length);
+    for (ReconcileResult<T> res : results) {
+      res.getSingleResource().ifPresent(r -> operations.put(r, res.getSingleOperation()));
+    }
+    return new ReconcileResult<>(operations);
+  }
+
   @Override
   public String toString() {
-    return getResource()
-        .map(r -> r instanceof HasMetadata ? ResourceID.fromResource((HasMetadata) r) : r)
-        .orElse("no resource")
-        + " -> " + operation;
+    return resourceOperations.entrySet().stream().collect(Collectors.toMap(
+        e -> e instanceof HasMetadata ? ResourceID.fromResource((HasMetadata) e) : e,
+        Map.Entry::getValue))
+        .toString();
   }
 
   private ReconcileResult(R resource, Operation operation) {
-    this.resource = resource;
-    this.operation = operation;
+    resourceOperations = resource != null ? Map.of(resource, operation) : Collections.emptyMap();
+  }
+
+  private ReconcileResult(Map<R, Operation> operations) {
+    resourceOperations = Collections.unmodifiableMap(operations);
+  }
+
+  public Optional<R> getSingleResource() {
+    return resourceOperations.entrySet().stream().findFirst().map(Map.Entry::getKey);
   }
 
-  public Optional<R> getResource() {
-    return Optional.ofNullable(resource);
+  public Operation getSingleOperation() {
+    return resourceOperations.entrySet().stream().findFirst().map(Map.Entry::getValue)
+        .orElseThrow();
   }
 
-  public Operation getOperation() {
-    return operation;
+  @SuppressWarnings("unused")
+  public Map<R, Operation> getResourceOperations() {
+    return resourceOperations;
   }
 
   public enum Operation {

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

@@ -1,5 +1,7 @@
 package io.javaoperatorsdk.operator.processing.dependent;
 
+import java.util.ArrayList;
+import java.util.List;
 import java.util.Optional;
 
 import org.slf4j.Logger;
@@ -20,25 +22,73 @@ public abstract class AbstractDependentResource<R, P extends HasMetadata>
 
   protected final boolean creatable = this instanceof Creator;
   protected final boolean updatable = this instanceof Updater;
+  protected final boolean bulk = this instanceof BulkDependentResource;
 
   protected Creator<R, P> creator;
   protected Updater<R, P> updater;
+  protected BulkDependentResource<R, P> bulkDependentResource;
 
-  private ResourceDiscriminator<R, P> resourceDiscriminator;
+  private final List<ResourceDiscriminator<R, P>> resourceDiscriminator = new ArrayList<>(1);
 
   @SuppressWarnings("unchecked")
   public AbstractDependentResource() {
     creator = creatable ? (Creator<R, P>) this : null;
     updater = updatable ? (Updater<R, P>) this : null;
+
+    bulkDependentResource = bulk ? (BulkDependentResource<R, P>) this : null;
   }
 
   @Override
   public ReconcileResult<R> reconcile(P primary, Context<P> context) {
-    Optional<R> maybeActual = getSecondaryResource(primary, context);
+    if (bulk) {
+      final var count = bulkDependentResource.count(primary, context);
+      deleteBulkResourcesIfRequired(count, lastKnownBulkSize(), primary, context);
+      adjustDiscriminators(count);
+      @SuppressWarnings("unchecked")
+      final ReconcileResult<R>[] results = new ReconcileResult[count];
+      for (int i = 0; i < count; i++) {
+        results[i] = reconcileIndexAware(primary, i, context);
+      }
+      return ReconcileResult.aggregatedResult(results);
+    } else {
+      return reconcileIndexAware(primary, 0, context);
+    }
+  }
+
+  protected void deleteBulkResourcesIfRequired(int targetCount, int actualCount, P primary,
+      Context<P> context) {
+    if (targetCount >= actualCount) {
+      return;
+    }
+    for (int i = targetCount; i < actualCount; i++) {
+      var resource = getSecondaryResourceIndexAware(primary, i, context);
+      var index = i;
+      resource.ifPresent(
+          r -> bulkDependentResource.deleteBulkResourceWithIndex(primary, r, index, context));
+    }
+  }
+
+  private void adjustDiscriminators(int count) {
+    if (resourceDiscriminator.size() == count) {
+      return;
+    }
+    if (resourceDiscriminator.size() < count) {
+      for (int i = resourceDiscriminator.size(); i < count; i++) {
+        resourceDiscriminator.add(bulkDependentResource.getResourceDiscriminator(i));
+      }
+    }
+    if (resourceDiscriminator.size() > count) {
+      resourceDiscriminator.subList(count, resourceDiscriminator.size()).clear();
+    }
+  }
+
+  protected ReconcileResult<R> reconcileIndexAware(P primary, int i, Context<P> context) {
+    Optional<R> maybeActual = bulk ? getSecondaryResourceIndexAware(primary, i, context)
+        : getSecondaryResource(primary, context);
     if (creatable || updatable) {
       if (maybeActual.isEmpty()) {
         if (creatable) {
-          var desired = desired(primary, context);
+          var desired = desiredIndexAware(primary, i, context);
           throwIfNull(desired, primary, "Desired");
           logForOperation("Creating", primary, desired);
           var createdResource = handleCreate(desired, primary, context);
@@ -47,9 +97,15 @@ public ReconcileResult<R> reconcile(P primary, Context<P> context) {
       } else {
         final var actual = maybeActual.get();
         if (updatable) {
-          final var match = updater.match(actual, primary, context);
+          final Matcher.Result<R> match;
+          if (bulk) {
+            match = updater.match(actual, primary, i, context);
+          } else {
+            match = updater.match(actual, primary, context);
+          }
           if (!match.matched()) {
-            final var desired = match.computedDesired().orElseGet(() -> desired(primary, context));
+            final var desired =
+                match.computedDesired().orElse(desiredIndexAware(primary, i, context));
             throwIfNull(desired, primary, "Desired");
             logForOperation("Updating", primary, desired);
             var updatedResource = handleUpdate(actual, desired, primary, context);
@@ -67,9 +123,18 @@ public ReconcileResult<R> reconcile(P primary, Context<P> context) {
     return ReconcileResult.noOperation(maybeActual.orElse(null));
   }
 
+  private R desiredIndexAware(P primary, int i, Context<P> context) {
+    return bulk ? desired(primary, i, context)
+        : desired(primary, context);
+  }
+
   public Optional<R> getSecondaryResource(P primary, Context<P> context) {
-    return resourceDiscriminator == null ? context.getSecondaryResource(resourceType())
-        : resourceDiscriminator.distinguish(resourceType(), primary, context);
+    return resourceDiscriminator.isEmpty() ? context.getSecondaryResource(resourceType())
+        : resourceDiscriminator.get(0).distinguish(resourceType(), primary, context);
+  }
+
+  protected Optional<R> getSecondaryResourceIndexAware(P primary, int index, Context<P> context) {
+    return context.getSecondaryResource(resourceType(), resourceDiscriminator.get(index));
   }
 
   private void throwIfNull(R desired, P primary, String descriptor) {
@@ -97,7 +162,7 @@ protected R handleCreate(R desired, P primary, Context<P> context) {
   }
 
   /**
-   * Allows sub-classes to perform additional processing (e.g. caching) on the created resource if
+   * Allows subclasses to perform additional processing (e.g. caching) on the created resource if
    * needed.
    *
    * @param primaryResourceId the {@link ResourceID} of the primary resource associated with the
@@ -129,12 +194,29 @@ protected R desired(P primary, Context<P> context) {
         "desired method must be implemented if this DependentResource can be created and/or updated");
   }
 
-  public void setResourceDiscriminator(
+  protected R desired(P primary, int index, Context<P> context) {
+    throw new IllegalStateException(
+        "Must be implemented for bulk DependentResource creation");
+  }
+
+  public AbstractDependentResource<R, P> setResourceDiscriminator(
       ResourceDiscriminator<R, P> resourceDiscriminator) {
-    this.resourceDiscriminator = resourceDiscriminator;
+    if (resourceDiscriminator != null) {
+      this.resourceDiscriminator.add(resourceDiscriminator);
+    }
+    return this;
   }
 
   public ResourceDiscriminator<R, P> getResourceDiscriminator() {
-    return resourceDiscriminator;
+    if (this.resourceDiscriminator.isEmpty()) {
+      return null;
+    } else {
+      return this.resourceDiscriminator.get(0);
+    }
   }
+
+  protected int lastKnownBulkSize() {
+    return resourceDiscriminator.size();
+  }
+
 }

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

@@ -0,0 +1,35 @@
+package io.javaoperatorsdk.operator.processing.dependent;
+
+import io.fabric8.kubernetes.api.model.HasMetadata;
+import io.javaoperatorsdk.operator.api.reconciler.Context;
+import io.javaoperatorsdk.operator.api.reconciler.ResourceDiscriminator;
+import io.javaoperatorsdk.operator.api.reconciler.dependent.Deleter;
+
+/**
+ * Manages dynamic number of resources created for a primary resource. Since the point of a bulk
+ * dependent resource is to manage the number of secondary resources dynamically it implement
+ * {@link Creator} and {@link Deleter} interfaces out of the box. A concrete dependent resource can
+ * implement additionally also {@link Updater}.
+ */
+public interface BulkDependentResource<R, P extends HasMetadata> extends Creator<R, P>, Deleter<P> {
+
+  /**
+   * @return number of resources to create
+   */
+  int count(P primary, Context<P> context);
+
+  R desired(P primary, int index, Context<P> context);
+
+  /**
+   * Used to delete resource if the desired count is lower than the actual count of a resource.
+   *
+   * @param primary resource
+   * @param resource actual resource from the cache for the index
+   * @param i index of the resource
+   * @param context actual context
+   */
+  void deleteBulkResourceWithIndex(P primary, R resource, int i, Context<P> context);
+
+  ResourceDiscriminator<R, P> getResourceDiscriminator(int index);
+
+}

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

@@ -0,0 +1,20 @@
+package io.javaoperatorsdk.operator.processing.dependent;
+
+import io.fabric8.kubernetes.api.model.HasMetadata;
+import io.javaoperatorsdk.operator.api.reconciler.Context;
+
+/**
+ * Helper for the Bulk Dependent Resources to make it more explicit that bulk needs to only
+ * implement the index aware match method.
+ *
+ * @param <R> secondary resource type
+ * @param <P> primary resource type
+ */
+public interface BulkUpdater<R, P extends HasMetadata> extends Updater<R, P> {
+
+  default Matcher.Result<R> match(R actualResource, P primary, Context<P> context) {
+    throw new IllegalStateException();
+  }
+
+  Matcher.Result<R> match(R actualResource, P primary, int index, Context<P> context);
+}

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

@@ -16,4 +16,10 @@ public Result<R> match(R actualResource, P primary, Context<P> context) {
     var desired = abstractDependentResource.desired(primary, context);
     return Result.computed(actualResource.equals(desired), desired);
   }
+
+  @Override
+  public Result<R> match(R actualResource, P primary, int index, Context<P> context) {
+    var desired = abstractDependentResource.desired(primary, index, context);
+    return Result.computed(actualResource.equals(desired), desired);
+  }
 }

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

@@ -95,4 +95,19 @@ public Optional<T> computedDesired() {
    *         {@link Result#computed(boolean, Object)})
    */
   Result<R> match(R actualResource, P primary, Context<P> context);
+
+  /**
+   * Determines whether the specified secondary resource matches the desired state with target index
+   * of a bulk resource as defined from the specified primary resource, given the specified
+   * {@link Context}.
+   *
+   * @param actualResource the resource we want to determine whether it's matching the desired state
+   * @param primary the primary resource from which the desired state is inferred
+   * @param context the context in which the resource is being matched
+   * @return a {@link Result} encapsulating whether the resource matched its desired state and this
+   *         associated state if it was computed as part of the matching process. Use the static
+   *         convenience methods ({@link Result#nonComputed(boolean)} and
+   *         {@link Result#computed(boolean, Object)})
+   */
+  Result<R> match(R actualResource, P primary, int index, Context<P> context);
 }

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

@@ -8,4 +8,8 @@ public interface Updater<R, P extends HasMetadata> {
   R update(R actual, R desired, P primary, Context<P> context);
 
   Result<R> match(R actualResource, P primary, Context<P> context);
+
+  default Result<R> match(R actualResource, P primary, int index, Context<P> context) {
+    throw new IllegalStateException("Implement this for bulk matching");
+  }
 }