feat: expectations

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

@@ -3,10 +3,17 @@
 import java.time.Duration;
 import java.util.Optional;
 import java.util.concurrent.TimeUnit;
+import java.util.function.BiPredicate;
 
-public abstract class BaseControl<T extends BaseControl<T>> {
+import io.fabric8.kubernetes.api.model.HasMetadata;
+import io.javaoperatorsdk.operator.api.reconciler.expectation.Expectation;
+import io.javaoperatorsdk.operator.api.reconciler.expectation.ExpectationAdapter;
+import io.javaoperatorsdk.operator.api.reconciler.expectation.ExpectationContext;
+
+public abstract class BaseControl<T extends BaseControl<T, P>, P extends HasMetadata> {
 
   private Long scheduleDelay = null;
+  private Expectation<P> expectation;
 
   public T rescheduleAfter(long delay) {
     rescheduleAfter(Duration.ofMillis(delay));
@@ -25,4 +32,16 @@ public T rescheduleAfter(long delay, TimeUnit timeUnit) {
   public Optional<Long> getScheduleDelay() {
     return Optional.ofNullable(scheduleDelay);
   }
+
+  public void expect(Expectation<P> expectation) {
+    this.expectation = expectation;
+  }
+
+  public void expect(BiPredicate<P, ExpectationContext<P>> expectation, Duration timeout) {
+    this.expectation = new ExpectationAdapter<>(expectation, timeout);
+  }
+
+  public Optional<Expectation<P>> getExpectation() {
+    return Optional.ofNullable(expectation);
+  }
 }

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

@@ -0,0 +1,44 @@
+package io.javaoperatorsdk.operator.api.reconciler;
+
+import java.util.Optional;
+import java.util.Set;
+import java.util.stream.Stream;
+
+import io.fabric8.kubernetes.api.model.HasMetadata;
+import io.javaoperatorsdk.operator.api.config.ControllerConfiguration;
+import io.javaoperatorsdk.operator.processing.event.EventSourceRetriever;
+import io.javaoperatorsdk.operator.processing.event.source.IndexerResourceCache;
+
+public interface CacheAware<P extends HasMetadata> {
+  default <R> Optional<R> getSecondaryResource(Class<R> expectedType) {
+    return getSecondaryResource(expectedType, null);
+  }
+
+  <R> Set<R> getSecondaryResources(Class<R> expectedType);
+
+  default <R> Stream<R> getSecondaryResourcesAsStream(Class<R> expectedType) {
+    return getSecondaryResources(expectedType).stream();
+  }
+
+  <R> Optional<R> getSecondaryResource(Class<R> expectedType, String eventSourceName);
+
+  ControllerConfiguration<P> getControllerConfiguration();
+
+  /**
+   * Retrieves the primary resource.
+   *
+   * @return the primary resource associated with the current reconciliation
+   */
+  P getPrimaryResource();
+
+  /**
+   * Retrieves the primary resource cache.
+   *
+   * @return the {@link IndexerResourceCache} associated with the associated {@link Reconciler} for
+   *     this context
+   */
+  @SuppressWarnings("unused")
+  IndexedResourceCache<P> getPrimaryCache();
+
+  EventSourceRetriever<P> eventSourceRetriever();
+}

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

@@ -1,35 +1,18 @@
 package io.javaoperatorsdk.operator.api.reconciler;
 
 import java.util.Optional;
-import java.util.Set;
 import java.util.concurrent.ExecutorService;
-import java.util.stream.Stream;
 
 import io.fabric8.kubernetes.api.model.HasMetadata;
 import io.fabric8.kubernetes.client.KubernetesClient;
-import io.javaoperatorsdk.operator.api.config.ControllerConfiguration;
 import io.javaoperatorsdk.operator.api.reconciler.dependent.managed.ManagedWorkflowAndDependentResourceContext;
-import io.javaoperatorsdk.operator.processing.event.EventSourceRetriever;
+import io.javaoperatorsdk.operator.api.reconciler.expectation.ExpectationResult;
 import io.javaoperatorsdk.operator.processing.event.source.IndexerResourceCache;
 
-public interface Context<P extends HasMetadata> {
+public interface Context<P extends HasMetadata> extends CacheAware<P> {
 
   Optional<RetryInfo> getRetryInfo();
 
-  default <R> Optional<R> getSecondaryResource(Class<R> expectedType) {
-    return getSecondaryResource(expectedType, null);
-  }
-
-  <R> Set<R> getSecondaryResources(Class<R> expectedType);
-
-  default <R> Stream<R> getSecondaryResourcesAsStream(Class<R> expectedType) {
-    return getSecondaryResources(expectedType).stream();
-  }
-
-  <R> Optional<R> getSecondaryResource(Class<R> expectedType, String eventSourceName);
-
-  ControllerConfiguration<P> getControllerConfiguration();
-
   /**
    * Retrieve the {@link ManagedWorkflowAndDependentResourceContext} used to interact with {@link
    * io.javaoperatorsdk.operator.api.reconciler.dependent.DependentResource}s and associated {@link
@@ -39,8 +22,6 @@ default <R> Stream<R> getSecondaryResourcesAsStream(Class<R> expectedType) {
    */
   ManagedWorkflowAndDependentResourceContext managedWorkflowAndDependentResourceContext();
 
-  EventSourceRetriever<P> eventSourceRetriever();
-
   KubernetesClient getClient();
 
   /** ExecutorService initialized by framework for workflows. Used for workflow standalone mode. */
@@ -72,4 +53,6 @@ default <R> Stream<R> getSecondaryResourcesAsStream(Class<R> expectedType) {
    * @return {@code true} is another reconciliation is already scheduled, {@code false} otherwise
    */
   boolean isNextReconciliationImminent();
+
+  Optional<ExpectationResult<P>> expectationResult();
 }

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

@@ -0,0 +1,79 @@
+package io.javaoperatorsdk.operator.api.reconciler;
+
+import java.util.Optional;
+import java.util.Set;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+import io.fabric8.kubernetes.api.model.HasMetadata;
+import io.javaoperatorsdk.operator.api.config.ControllerConfiguration;
+import io.javaoperatorsdk.operator.processing.Controller;
+import io.javaoperatorsdk.operator.processing.event.EventSourceRetriever;
+import io.javaoperatorsdk.operator.processing.event.NoEventSourceForClassException;
+
+public class DefaultCacheAware<P extends HasMetadata> implements CacheAware<P> {
+
+  protected final Controller<P> controller;
+  protected final P primaryResource;
+
+  public DefaultCacheAware(Controller<P> controller, P primaryResource) {
+    this.controller = controller;
+    this.primaryResource = primaryResource;
+  }
+
+  @Override
+  public <T> Set<T> getSecondaryResources(Class<T> expectedType) {
+    return getSecondaryResourcesAsStream(expectedType).collect(Collectors.toSet());
+  }
+
+  @Override
+  public <R> Stream<R> getSecondaryResourcesAsStream(Class<R> expectedType) {
+    return controller.getEventSourceManager().getEventSourcesFor(expectedType).stream()
+        .map(es -> es.getSecondaryResources(primaryResource))
+        .flatMap(Set::stream);
+  }
+
+  @Override
+  public <T> Optional<T> getSecondaryResource(Class<T> expectedType, String eventSourceName) {
+    try {
+      return controller
+          .getEventSourceManager()
+          .getEventSourceFor(expectedType, eventSourceName)
+          .getSecondaryResource(primaryResource);
+    } catch (NoEventSourceForClassException e) {
+      /*
+       * If a workflow has an activation condition there can be event sources which are only
+       * registered if the activation condition holds, but to provide a consistent API we return an
+       * Optional instead of throwing an exception.
+       *
+       * Note that not only the resource which has an activation condition might not be registered
+       * but dependents which depend on it.
+       */
+      if (eventSourceName == null && controller.workflowContainsDependentForType(expectedType)) {
+        return Optional.empty();
+      } else {
+        throw e;
+      }
+    }
+  }
+
+  @Override
+  public EventSourceRetriever<P> eventSourceRetriever() {
+    return controller.getEventSourceManager();
+  }
+
+  @Override
+  public ControllerConfiguration<P> getControllerConfiguration() {
+    return controller.getConfiguration();
+  }
+
+  @Override
+  public P getPrimaryResource() {
+    return primaryResource;
+  }
+
+  @Override
+  public IndexedResourceCache<P> getPrimaryCache() {
+    return controller.getEventSourceManager().getControllerEventSource();
+  }
+}

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

@@ -1,95 +1,48 @@
 package io.javaoperatorsdk.operator.api.reconciler;
 
 import java.util.Optional;
-import java.util.Set;
 import java.util.concurrent.ExecutorService;
-import java.util.stream.Collectors;
-import java.util.stream.Stream;
 
 import io.fabric8.kubernetes.api.model.HasMetadata;
 import io.fabric8.kubernetes.client.KubernetesClient;
-import io.javaoperatorsdk.operator.api.config.ControllerConfiguration;
 import io.javaoperatorsdk.operator.api.reconciler.dependent.managed.DefaultManagedWorkflowAndDependentResourceContext;
 import io.javaoperatorsdk.operator.api.reconciler.dependent.managed.ManagedWorkflowAndDependentResourceContext;
+import io.javaoperatorsdk.operator.api.reconciler.expectation.ExpectationResult;
 import io.javaoperatorsdk.operator.processing.Controller;
-import io.javaoperatorsdk.operator.processing.event.EventSourceRetriever;
-import io.javaoperatorsdk.operator.processing.event.NoEventSourceForClassException;
 import io.javaoperatorsdk.operator.processing.event.ResourceID;
 
-public class DefaultContext<P extends HasMetadata> implements Context<P> {
+public class DefaultContext<P extends HasMetadata> extends DefaultCacheAware<P>
+    implements Context<P> {
 
   private RetryInfo retryInfo;
-  private final Controller<P> controller;
-  private final P primaryResource;
-  private final ControllerConfiguration<P> controllerConfiguration;
+
   private final DefaultManagedWorkflowAndDependentResourceContext<P>
       defaultManagedDependentResourceContext;
 
-  public DefaultContext(RetryInfo retryInfo, Controller<P> controller, P primaryResource) {
+  private final ExpectationResult<P> expectationResult;
+
+  public DefaultContext(
+      RetryInfo retryInfo,
+      Controller<P> controller,
+      P primaryResource,
+      ExpectationResult<P> expectationResult) {
+    super(controller, primaryResource);
     this.retryInfo = retryInfo;
-    this.controller = controller;
-    this.primaryResource = primaryResource;
-    this.controllerConfiguration = controller.getConfiguration();
     this.defaultManagedDependentResourceContext =
         new DefaultManagedWorkflowAndDependentResourceContext<>(controller, primaryResource, this);
+    this.expectationResult = expectationResult;
   }
 
   @Override
   public Optional<RetryInfo> getRetryInfo() {
     return Optional.ofNullable(retryInfo);
   }
 
-  @Override
-  public <T> Set<T> getSecondaryResources(Class<T> expectedType) {
-    return getSecondaryResourcesAsStream(expectedType).collect(Collectors.toSet());
-  }
-
-  @Override
-  public <R> Stream<R> getSecondaryResourcesAsStream(Class<R> expectedType) {
-    return controller.getEventSourceManager().getEventSourcesFor(expectedType).stream()
-        .map(es -> es.getSecondaryResources(primaryResource))
-        .flatMap(Set::stream);
-  }
-
-  @Override
-  public <T> Optional<T> getSecondaryResource(Class<T> expectedType, String eventSourceName) {
-    try {
-      return controller
-          .getEventSourceManager()
-          .getEventSourceFor(expectedType, eventSourceName)
-          .getSecondaryResource(primaryResource);
-    } catch (NoEventSourceForClassException e) {
-      /*
-       * If a workflow has an activation condition there can be event sources which are only
-       * registered if the activation condition holds, but to provide a consistent API we return an
-       * Optional instead of throwing an exception.
-       *
-       * Note that not only the resource which has an activation condition might not be registered
-       * but dependents which depend on it.
-       */
-      if (eventSourceName == null && controller.workflowContainsDependentForType(expectedType)) {
-        return Optional.empty();
-      } else {
-        throw e;
-      }
-    }
-  }
-
-  @Override
-  public ControllerConfiguration<P> getControllerConfiguration() {
-    return controllerConfiguration;
-  }
-
   @Override
   public ManagedWorkflowAndDependentResourceContext managedWorkflowAndDependentResourceContext() {
     return defaultManagedDependentResourceContext;
   }
 
-  @Override
-  public EventSourceRetriever<P> eventSourceRetriever() {
-    return controller.getEventSourceManager();
-  }
-
   @Override
   public KubernetesClient getClient() {
     return controller.getClient();
@@ -102,23 +55,18 @@ public ExecutorService getWorkflowExecutorService() {
     return controller.getExecutorServiceManager().workflowExecutorService();
   }
 
-  @Override
-  public P getPrimaryResource() {
-    return primaryResource;
-  }
-
-  @Override
-  public IndexedResourceCache<P> getPrimaryCache() {
-    return controller.getEventSourceManager().getControllerEventSource();
-  }
-
   @Override
   public boolean isNextReconciliationImminent() {
     return controller
         .getEventProcessor()
         .isNextReconciliationImminent(ResourceID.fromResource(primaryResource));
   }
 
+  @Override
+  public Optional<ExpectationResult<P>> expectationResult() {
+    return Optional.ofNullable(expectationResult);
+  }
+
   public DefaultContext<P> setRetryInfo(RetryInfo retryInfo) {
     this.retryInfo = retryInfo;
     return this;

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

@@ -1,6 +1,8 @@
 package io.javaoperatorsdk.operator.api.reconciler;
 
-public class DeleteControl extends BaseControl<DeleteControl> {
+import io.fabric8.kubernetes.api.model.HasMetadata;
+
+public class DeleteControl<P extends HasMetadata> extends BaseControl<DeleteControl<P>, P> {
 
   private final boolean removeFinalizer;
 
@@ -27,7 +29,7 @@ public static DeleteControl defaultDelete() {
    *
    * @return delete control that will not remove finalizer.
    */
-  public static DeleteControl noFinalizerRemoval() {
+  public static <T extends HasMetadata> DeleteControl<T> noFinalizerRemoval() {
     return new DeleteControl(false);
   }
 
@@ -36,7 +38,7 @@ public boolean isRemoveFinalizer() {
   }
 
   @Override
-  public DeleteControl rescheduleAfter(long delay) {
+  public DeleteControl<P> rescheduleAfter(long delay) {
     if (removeFinalizer) {
       throw new IllegalStateException("Cannot reschedule cleanup if removing finalizer");
     }

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

@@ -6,7 +6,7 @@
 import io.fabric8.kubernetes.api.model.HasMetadata;
 
 public class ErrorStatusUpdateControl<P extends HasMetadata>
-    extends BaseControl<ErrorStatusUpdateControl<P>> {
+    extends BaseControl<ErrorStatusUpdateControl<P>, P> {
 
   private final P resource;
   private boolean noRetry = false;

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

@@ -5,7 +5,7 @@
 import io.fabric8.kubernetes.api.model.HasMetadata;
 import io.fabric8.kubernetes.client.CustomResource;
 
-public class UpdateControl<P extends HasMetadata> extends BaseControl<UpdateControl<P>> {
+public class UpdateControl<P extends HasMetadata> extends BaseControl<UpdateControl<P>, P> {
 
   private final P resource;
   private final boolean patchResource;