fix: condition for bulk resources (#1688)

Author: csviri

docs/documentation/dependent-resources.md

@@ -382,6 +382,8 @@ interface.
 Various examples are provided as [integration tests](https://github.com/java-operator-sdk/java-operator-sdk/tree/f5ffcfb6f546d79b4bab04ea503c8bad9d6acce6/operator-framework/src/test/java/io/javaoperatorsdk/operator/bulkdependent)
 .
 
+To see how to add conditions on a bulk dependent resource [see integration test](https://github.com/java-operator-sdk/java-operator-sdk/blob/ad8759856cc59380ea4b0973478daa1140ec93e7/operator-framework/src/test/java/io/javaoperatorsdk/operator/bulkdependent/BulkDependentWithConditionIT.java).
+
 ## Dependent Resources with Explicit State
 
 For cases when an external (non-Kubernetes) resource generates an ID during creation and from that point

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

@@ -14,6 +14,7 @@
 import io.fabric8.kubernetes.api.model.HasMetadata;
 import io.javaoperatorsdk.operator.api.reconciler.Context;
 import io.javaoperatorsdk.operator.api.reconciler.dependent.DependentResource;
+import io.javaoperatorsdk.operator.processing.dependent.BulkDependentResource;
 
 @SuppressWarnings("rawtypes")
 public abstract class AbstractWorkflowExecutor<P extends HasMetadata> {
@@ -101,9 +102,15 @@ protected synchronized void handleNodeExecutionFinish(
 
   protected <R> boolean isConditionMet(Optional<Condition<R, P>> condition,
       DependentResource<R, P> dependentResource) {
+    if (condition.isEmpty()) {
+      return true;
+    }
+    var resources = dependentResource instanceof BulkDependentResource
+        ? ((BulkDependentResource) dependentResource).getSecondaryResources(primary, context)
+        : dependentResource.getSecondaryResource(primary, context).orElse(null);
+
     return condition.map(c -> c.isMet(primary,
-        dependentResource.getSecondaryResource(primary, context).orElse(null),
-        context))
+        (R) resources, context))
         .orElse(true);
   }
 }

operator-framework/src/test/java/io/javaoperatorsdk/operator/bulkdependent/BulkDependentWithConditionIT.java

@@ -0,0 +1,46 @@
+package io.javaoperatorsdk.operator.bulkdependent;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
+
+import io.javaoperatorsdk.operator.junit.LocallyRunOperatorExtension;
+import io.javaoperatorsdk.operator.sample.bulkdependent.BulkDependentTestCustomResource;
+import io.javaoperatorsdk.operator.sample.bulkdependent.ManagedBulkDependentWithReadyConditionReconciler;
+
+import static io.javaoperatorsdk.operator.bulkdependent.BulkDependentTestBase.INITIAL_NUMBER_OF_CONFIG_MAPS;
+import static io.javaoperatorsdk.operator.bulkdependent.BulkDependentTestBase.testResource;
+import static io.javaoperatorsdk.operator.sample.bulkdependent.ConfigMapDeleterBulkDependentResource.LABEL_KEY;
+import static io.javaoperatorsdk.operator.sample.bulkdependent.ConfigMapDeleterBulkDependentResource.LABEL_VALUE;
+import static org.assertj.core.api.Assertions.*;
+import static org.awaitility.Awaitility.await;
+
+class BulkDependentWithConditionIT {
+
+  @RegisterExtension
+  LocallyRunOperatorExtension extension =
+      LocallyRunOperatorExtension.builder()
+          .withReconciler(new ManagedBulkDependentWithReadyConditionReconciler())
+          .build();
+
+  @Test
+  void handlesBulkDependentWithPrecondition() {
+    var resource = testResource();
+    extension.create(resource);
+
+    await().untilAsserted(() -> {
+      var res = extension.get(BulkDependentTestCustomResource.class,
+          testResource().getMetadata().getName());
+      assertThat(res.getStatus()).isNotNull();
+      assertThat(res.getStatus().getReady()).isTrue();
+
+      var cms = extension.getKubernetesClient().configMaps().inNamespace(extension.getNamespace())
+          .withLabel(LABEL_KEY, LABEL_VALUE)
+          .list().getItems();
+      assertThat(cms).hasSize(INITIAL_NUMBER_OF_CONFIG_MAPS);
+
+    });
+  }
+
+
+
+}

operator-framework/src/test/java/io/javaoperatorsdk/operator/sample/bulkdependent/BulkDependentTestCustomResource.java

@@ -10,6 +10,6 @@
 @Version("v1")
 @ShortNames("sbd")
 public class BulkDependentTestCustomResource
-    extends CustomResource<BulkDependentTestSpec, Void>
+    extends CustomResource<BulkDependentTestSpec, BulkDependentTestStatus>
     implements Namespaced {
 }

operator-framework/src/test/java/io/javaoperatorsdk/operator/sample/bulkdependent/BulkDependentTestStatus.java

@@ -0,0 +1,15 @@
+package io.javaoperatorsdk.operator.sample.bulkdependent;
+
+public class BulkDependentTestStatus {
+
+  private Boolean ready;
+
+  public Boolean getReady() {
+    return ready;
+  }
+
+  public BulkDependentTestStatus setReady(boolean ready) {
+    this.ready = ready;
+    return this;
+  }
+}

operator-framework/src/test/java/io/javaoperatorsdk/operator/sample/bulkdependent/ManagedBulkDependentWithReadyConditionReconciler.java

@@ -0,0 +1,36 @@
+package io.javaoperatorsdk.operator.sample.bulkdependent;
+
+import java.util.concurrent.atomic.AtomicInteger;
+
+import io.javaoperatorsdk.operator.api.reconciler.Context;
+import io.javaoperatorsdk.operator.api.reconciler.ControllerConfiguration;
+import io.javaoperatorsdk.operator.api.reconciler.Reconciler;
+import io.javaoperatorsdk.operator.api.reconciler.UpdateControl;
+import io.javaoperatorsdk.operator.api.reconciler.dependent.Dependent;
+
+@ControllerConfiguration(dependents = @Dependent(readyPostcondition = SampleBulkCondition.class,
+    type = CRUDConfigMapBulkDependentResource.class))
+public class ManagedBulkDependentWithReadyConditionReconciler
+    implements Reconciler<BulkDependentTestCustomResource> {
+
+  private final AtomicInteger numberOfExecutions = new AtomicInteger(0);
+
+  @Override
+  public UpdateControl<BulkDependentTestCustomResource> reconcile(
+      BulkDependentTestCustomResource resource,
+      Context<BulkDependentTestCustomResource> context) throws Exception {
+    numberOfExecutions.incrementAndGet();
+
+    var ready = context.managedDependentResourceContext().getWorkflowReconcileResult()
+        .map(res -> res.allDependentResourcesReady()).orElseThrow();
+
+    resource.setStatus(new BulkDependentTestStatus());
+    resource.getStatus().setReady(ready);
+
+    return UpdateControl.patchStatus(resource);
+  }
+
+  public int getNumberOfExecutions() {
+    return numberOfExecutions.get();
+  }
+}

operator-framework/src/test/java/io/javaoperatorsdk/operator/sample/bulkdependent/SampleBulkCondition.java

@@ -0,0 +1,23 @@
+package io.javaoperatorsdk.operator.sample.bulkdependent;
+
+import java.util.Map;
+
+import io.fabric8.kubernetes.api.model.ConfigMap;
+import io.javaoperatorsdk.operator.api.reconciler.Context;
+import io.javaoperatorsdk.operator.processing.dependent.workflow.Condition;
+
+public class SampleBulkCondition
+    implements Condition<Map<String, ConfigMap>, BulkDependentTestCustomResource> {
+
+  // We use ConfigMaps here just to show how to check some properties of resources managed by a
+  // BulkDependentResource. In real life example this would be rather based on some status of those
+  // resources, like Pods.
+
+  @Override
+  public boolean isMet(BulkDependentTestCustomResource primary, Map<String, ConfigMap> secondary,
+      Context<BulkDependentTestCustomResource> context) {
+
+    return secondary.values().stream().allMatch(cm -> !cm.getData().isEmpty());
+
+  }
+}