WIP: feat: depends on wait condition design

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

@@ -3,14 +3,14 @@
 import io.fabric8.kubernetes.api.model.HasMetadata;
 
 @SuppressWarnings("rawtypes")
-public class UpdateControl<T extends HasMetadata> extends BaseControl<UpdateControl<T>> {
+public class UpdateControl<P extends HasMetadata> extends BaseControl<UpdateControl<P>> {
 
-  private final T resource;
+  private final P resource;
   private final boolean updateStatus;
   private final boolean updateResource;
 
   private UpdateControl(
-      T resource, boolean updateStatus, boolean updateResource) {
+      P resource, boolean updateStatus, boolean updateResource) {
     if ((updateResource || updateStatus) && resource == null) {
       throw new IllegalArgumentException("CustomResource cannot be null in case of update");
     }
@@ -54,7 +54,7 @@ public static <T extends HasMetadata> UpdateControl<T> noUpdate() {
     return new UpdateControl<>(null, false, false);
   }
 
-  public T getResource() {
+  public P getResource() {
     return resource;
   }
 

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

@@ -131,8 +131,9 @@ public EventSource initEventSource(EventSourceContext<P> context) {
     if (informerEventSource == null) {
       configureWith(context.getConfigurationService(), null, null,
           KubernetesDependent.ADD_OWNER_REFERENCE_DEFAULT);
-      log.warn("Using default configuration for " + resourceType().getSimpleName()
-          + " KubernetesDependentResource, call configureWith to provide configuration");
+      log.warn(
+          "Using default configuration for {} KubernetesDependentResource, call configureWith to provide configuration",
+          resourceType().getSimpleName());
     }
     return informerEventSource;
   }

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

@@ -0,0 +1,8 @@
+package io.javaoperatorsdk.operator.processing.dependent.waitfor;
+
+@FunctionalInterface
+public interface Condition<R> {
+
+  boolean isFulfilled(R resource);
+
+}

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

@@ -0,0 +1,111 @@
+package io.javaoperatorsdk.operator.processing.dependent.waitfor;
+
+import java.time.Duration;
+import java.time.Instant;
+import java.time.temporal.ChronoUnit;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.function.Supplier;
+
+import io.fabric8.kubernetes.api.model.HasMetadata;
+import io.javaoperatorsdk.operator.api.reconciler.UpdateControl;
+import io.javaoperatorsdk.operator.api.reconciler.dependent.DependentResource;
+
+import static java.lang.Thread.sleep;
+
+public class ConditionChecker<R> {
+
+  public static final Duration DEFAULT_POLLING_INTERVAL = Duration.ofSeconds(1);
+  public static final Duration DEFAULT_TIMEOUT = Duration.ZERO;
+  private static final int MIN_THREAD_SLEEP = 25;
+
+  private Duration pollingInterval;
+  private Duration timeout;
+  private UnfulfillmentHandler<?> unfulfillmentHandler;
+  private Condition<R> condition;
+
+  public static <T> ConditionChecker<T> checker() {
+    return new ConditionChecker<>();
+  }
+
+  public ConditionChecker() {
+    this(DEFAULT_POLLING_INTERVAL, DEFAULT_TIMEOUT, UpdateControl::noUpdate);
+  }
+
+  public ConditionChecker(Duration pollingInterval, Duration timeout,
+      UnfulfillmentHandler unfulfillmentHandler) {
+    this.pollingInterval = pollingInterval;
+    this.timeout = timeout;
+    this.unfulfillmentHandler = unfulfillmentHandler;
+  }
+
+  public <P extends HasMetadata> void check(DependentResource<R, P> resource, P primary) {
+    check(() -> resource.getResource(primary));
+  }
+
+  public void check(Supplier<Optional<R>> supplier) {
+    checkSetup();
+    Optional<R> resource = supplier.get();
+    if (timeout.isNegative() || timeout.isZero()) {
+      if (resource.isPresent() && condition.isFulfilled(resource.get())) {
+        return;
+      } else {
+        handleConditionNotMet();
+      }
+    }
+    var deadline = Instant.now().plus(timeout.toMillis(), ChronoUnit.MILLIS);
+    while (Instant.now().isBefore(deadline)) {
+      resource = supplier.get();
+      if (resource.isPresent() && condition.isFulfilled(resource.get())) {
+        return;
+      } else {
+        var timeLeft = Duration.between(Instant.now(), deadline);
+        if (timeLeft.isZero() || timeLeft.isNegative()) {
+          handleConditionNotMet();
+        } else {
+          sleepUntilNextPoll(timeLeft);
+        }
+      }
+    }
+    handleConditionNotMet();
+  }
+
+  private void checkSetup() {
+    Objects.requireNonNull(unfulfillmentHandler, "ConditionNotFulfilledHandler is not set");
+    Objects.requireNonNull(condition, "Condition is not set");
+  }
+
+  private void sleepUntilNextPoll(Duration timeLeft) {
+    try {
+      sleep(Math.max(MIN_THREAD_SLEEP, Math.min(pollingInterval.toMillis(), timeLeft.toMillis())));
+    } catch (InterruptedException e) {
+      Thread.currentThread().interrupt();
+      throw new IllegalStateException("Thread interrupted.", e);
+    }
+  }
+
+  private void handleConditionNotMet() {
+    throw new ConditionUnfulfilledException(unfulfillmentHandler);
+  }
+
+  public ConditionChecker<R> withPollingInterval(Duration pollingInterval) {
+    this.pollingInterval = pollingInterval;
+    return this;
+  }
+
+  public ConditionChecker<R> withTimeout(Duration timeout) {
+    this.timeout = timeout;
+    return this;
+  }
+
+  public ConditionChecker<R> withUnfulfilledHandler(
+      UnfulfillmentHandler unFulfillmentHandler) {
+    this.unfulfillmentHandler = unFulfillmentHandler;
+    return this;
+  }
+
+  public ConditionChecker<R> withCondition(Condition<R> condition) {
+    this.condition = condition;
+    return this;
+  }
+}

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

@@ -0,0 +1,22 @@
+package io.javaoperatorsdk.operator.processing.dependent.waitfor;
+
+import io.javaoperatorsdk.operator.OperatorException;
+
+public class ConditionUnfulfilledException extends OperatorException {
+
+  private final UnfulfillmentHandler unfulfillmentHandler;
+
+  public ConditionUnfulfilledException(UnfulfillmentHandler unfulfillmentHandler) {
+    this.unfulfillmentHandler = unfulfillmentHandler;
+  }
+
+  public ConditionUnfulfilledException(String message,
+      UnfulfillmentHandler unfulfillmentHandler) {
+    super(message);
+    this.unfulfillmentHandler = unfulfillmentHandler;
+  }
+
+  public UnfulfillmentHandler getUnfulfillmentHandler() {
+    return unfulfillmentHandler;
+  }
+}

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

@@ -0,0 +1,9 @@
+package io.javaoperatorsdk.operator.processing.dependent.waitfor;
+
+import io.javaoperatorsdk.operator.api.reconciler.BaseControl;
+
+public interface UnfulfillmentHandler<P extends BaseControl<P>> {
+
+  BaseControl<P> control();
+
+}

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

@@ -20,6 +20,7 @@
 import io.javaoperatorsdk.operator.api.reconciler.RetryInfo;
 import io.javaoperatorsdk.operator.api.reconciler.UpdateControl;
 import io.javaoperatorsdk.operator.processing.Controller;
+import io.javaoperatorsdk.operator.processing.dependent.waitfor.ConditionUnfulfilledException;
 
 import static io.javaoperatorsdk.operator.processing.KubernetesResourceUtils.getName;
 import static io.javaoperatorsdk.operator.processing.KubernetesResourceUtils.getUID;
@@ -139,8 +140,12 @@ private PostExecutionControl<R> reconcileExecution(ExecutionScope<R> executionSc
         getName(resourceForExecution),
         getVersion(resourceForExecution),
         executionScope);
-
-    UpdateControl<R> updateControl = controller.reconcile(resourceForExecution, context);
+    UpdateControl<R> updateControl;
+    try {
+      updateControl = controller.reconcile(resourceForExecution, context);
+    } catch (ConditionUnfulfilledException ex) {
+      updateControl = (UpdateControl<R>) ex.getUnfulfillmentHandler().control();
+    }
     R updatedCustomResource = null;
     if (updateControl.isUpdateResourceAndStatus()) {
       updatedCustomResource = updateCustomResource(updateControl.getResource());
@@ -248,8 +253,12 @@ private PostExecutionControl<R> handleCleanup(R resource, Context context) {
         "Executing delete for resource: {} with version: {}",
         getName(resource),
         getVersion(resource));
-
-    DeleteControl deleteControl = controller.cleanup(resource, context);
+    DeleteControl deleteControl;
+    try {
+      deleteControl = controller.cleanup(resource, context);
+    } catch (ConditionUnfulfilledException ex) {
+      deleteControl = (DeleteControl) ex.getUnfulfillmentHandler().control();
+    }
     final var useFinalizer = configuration().useFinalizer();
     if (useFinalizer) {
       // note that we don't reschedule here even if instructed. Removing finalizer means that

operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/dependent/waitfor/ConditionCheckerTest.java

@@ -0,0 +1,77 @@
+package io.javaoperatorsdk.operator.processing.dependent.waitfor;
+
+import java.time.Duration;
+import java.util.Optional;
+import java.util.concurrent.TimeUnit;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.Timeout;
+
+import io.javaoperatorsdk.operator.sample.simple.TestCustomResource;
+
+import static io.javaoperatorsdk.operator.processing.dependent.waitfor.ConditionChecker.checker;
+
+class ConditionCheckerTest {
+
+  @Test
+  void returnsIfDurationIsZeroAndConditionMet() {
+    checker()
+        .withTimeout(Duration.ZERO)
+        .withCondition(r -> true)
+        .check(() -> Optional.of(new TestCustomResource()));
+  }
+
+  @Test
+  void throwsExceptionIfDurationZeroConditionNotFulfilled() {
+    Assertions.assertThrows(
+        ConditionUnfulfilledException.class,
+        () -> new ConditionChecker<TestCustomResource>()
+            .withTimeout(Duration.ZERO)
+            .withCondition(r -> false)
+            .check(() -> Optional.of(new TestCustomResource())));
+  }
+
+  @Test
+  void throwsExceptionNoDurationIfResourceNotPresentWithinTimeout() {
+    Assertions.assertThrows(
+        ConditionUnfulfilledException.class,
+        () -> new ConditionChecker<TestCustomResource>()
+            .withTimeout(Duration.ZERO)
+            .withCondition(r -> true)
+            .check(Optional::empty));
+  }
+
+  @Test
+  @Timeout(value = 100, unit = TimeUnit.MILLISECONDS)
+  void waitsForTheConditionToFulfill() {
+    new ConditionChecker<TestCustomResource>()
+        .withTimeout(Duration.ofMillis(200))
+        .withPollingInterval(Duration.ofMillis(50))
+        .withCondition(r -> true)
+        .check(() -> Optional.of(new TestCustomResource()));
+  }
+
+  @Test
+  @Timeout(value = 230, unit = TimeUnit.MILLISECONDS)
+  void throwsExceptionIfConditionNotFulfilledWithinTimeout() {
+    Assertions.assertThrows(
+        ConditionUnfulfilledException.class, () -> new ConditionChecker<TestCustomResource>()
+            .withTimeout(Duration.ofMillis(200))
+            .withPollingInterval(Duration.ofMillis(50))
+            .withCondition(r -> false)
+            .check(() -> Optional.of(new TestCustomResource())));
+
+  }
+
+  @Test
+  @Timeout(value = 230, unit = TimeUnit.MILLISECONDS)
+  void throwsExceptionIfResourceNotPresentWithinTimeout() {
+    Assertions.assertThrows(
+        ConditionUnfulfilledException.class, () -> new ConditionChecker<TestCustomResource>()
+            .withTimeout(Duration.ofMillis(200))
+            .withPollingInterval(Duration.ofMillis(50))
+            .withCondition(r -> true)
+            .check(() -> Optional.empty()));
+  }
+}

operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/ReconciliationDispatcherTest.java

@@ -27,6 +27,8 @@
 import io.javaoperatorsdk.operator.api.reconciler.RetryInfo;
 import io.javaoperatorsdk.operator.api.reconciler.UpdateControl;
 import io.javaoperatorsdk.operator.processing.Controller;
+import io.javaoperatorsdk.operator.processing.dependent.waitfor.ConditionUnfulfilledException;
+import io.javaoperatorsdk.operator.processing.dependent.waitfor.UnfulfillmentHandler;
 import io.javaoperatorsdk.operator.processing.event.ReconciliationDispatcher.CustomResourceFacade;
 import io.javaoperatorsdk.operator.sample.observedgeneration.ObservedGenCustomResource;
 import io.javaoperatorsdk.operator.sample.simple.TestCustomResource;
@@ -470,6 +472,41 @@ void canSkipSchedulingMaxDelayIf() {
     assertThat(control.getReScheduleDelay()).isNotPresent();
   }
 
+  @Test
+  void handlesConditionUnfulfilledExceptionOnReconcile() {
+    testCustomResource.addFinalizer(DEFAULT_FINALIZER);
+    var uc = UpdateControl.<TestCustomResource>noUpdate()
+        .rescheduleAfter(1000, TimeUnit.MILLISECONDS);
+    UnfulfillmentHandler<UpdateControl<TestCustomResource>> unfulfillmentHandler = () -> uc;
+
+    when(reconciler.reconcile(eq(testCustomResource), any()))
+        .thenThrow(new ConditionUnfulfilledException(unfulfillmentHandler));
+
+    PostExecutionControl control =
+        reconciliationDispatcher.handleExecution(executionScopeWithCREvent(testCustomResource));
+
+    assertThat(control.getReScheduleDelay()).isPresent();
+    assertThat(control.getReScheduleDelay().get()).isEqualTo(1000L);
+    assertThat(control.getUpdatedCustomResource()).isEmpty();
+  }
+
+  @Test
+  void handlesConditionUnfulfilledExceptionOnCleanup() {
+    testCustomResource.addFinalizer(DEFAULT_FINALIZER);
+    markForDeletion(testCustomResource);
+    var uc = DeleteControl.noFinalizerRemoval().rescheduleAfter(1000);
+    UnfulfillmentHandler<DeleteControl> unfulfillmentHandler = () -> uc;
+    when(reconciler.cleanup(eq(testCustomResource), any()))
+        .thenThrow(new ConditionUnfulfilledException(unfulfillmentHandler));
+
+    PostExecutionControl control =
+        reconciliationDispatcher.handleExecution(executionScopeWithCREvent(testCustomResource));
+
+    verify(reconciler, times(1)).cleanup(eq(testCustomResource), any());
+    assertThat(control.getReScheduleDelay()).isPresent();
+    assertThat(control.getReScheduleDelay().get()).isEqualTo(1000L);
+  }
+
   private ObservedGenCustomResource createObservedGenCustomResource() {
     ObservedGenCustomResource observedGenCustomResource = new ObservedGenCustomResource();
     observedGenCustomResource.setMetadata(new ObjectMeta());