diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/workflow/AbstractWorkflowExecutor.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/workflow/AbstractWorkflowExecutor.java
index 665d80063b..d757eff4e4 100644
--- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/workflow/AbstractWorkflowExecutor.java
+++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/workflow/AbstractWorkflowExecutor.java
@@ -92,8 +92,8 @@ protected boolean alreadyVisited(DependentResourceNode, P> dependentResourceNo
return getResultFlagFor(dependentResourceNode, BaseWorkflowResult.DetailBuilder::isVisited);
}
- protected boolean postDeleteConditionNotMet(DependentResourceNode, P> drn) {
- return getResultFlagFor(drn, BaseWorkflowResult.DetailBuilder::hasPostDeleteConditionNotMet);
+ protected boolean postDeleteConditionMet(DependentResourceNode, P> drn) {
+ return !getResultFlagFor(drn, BaseWorkflowResult.DetailBuilder::hasPostDeleteConditionNotMet);
}
protected boolean isMarkedForDelete(DependentResourceNode, P> drn) {
@@ -132,6 +132,20 @@ protected synchronized void handleExceptionInExecutor(
createOrGetResultFor(dependentResourceNode).withError(e);
}
+ /**
+ * Called with the monitor held, after the node's execution mark is cleared and before the
+ * workflow may observe that no execution is left. Anything scheduled from here keeps its own
+ * execution mark, so the workflow does not complete while it is still running.
+ *
+ *
Implementations must not block, as the monitor is held, and must schedule any follow up work
+ * synchronously before returning. Scheduling it from another thread would allow that work to
+ * finish before its execution mark is set, leaving behind a mark that is never cleared.
+ *
+ *
An exception thrown here is recorded as an error on the node instead of being propagated, so
+ * that the workflow waiting on the monitor is notified in any case.
+ */
+ protected void onNodeExecutionFinished(DependentResourceNode, P> dependentResourceNode) {}
+
protected boolean isReady(DependentResourceNode, P> dependentResourceNode) {
return getResultFlagFor(dependentResourceNode, BaseWorkflowResult.DetailBuilder::isReady);
}
@@ -144,8 +158,17 @@ protected synchronized void handleNodeExecutionFinish(
DependentResourceNode, P> dependentResourceNode) {
logger().trace("Finished execution for: {} primary: {}", dependentResourceNode, primaryID);
actualExecutions.remove(dependentResourceNode);
- if (noMoreExecutionsScheduled()) {
- this.notifyAll();
+ try {
+ onNodeExecutionFinished(dependentResourceNode);
+ } catch (Exception e) {
+ handleExceptionInExecutor(dependentResourceNode, e);
+ } catch (Error e) {
+ logger().error("java.lang.Error during execution finish", e);
+ throw e;
+ } finally {
+ if (noMoreExecutionsScheduled()) {
+ this.notifyAll();
+ }
}
}
diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/workflow/WorkflowCleanupExecutor.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/workflow/WorkflowCleanupExecutor.java
index 1132d09350..ccc61edc39 100644
--- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/workflow/WorkflowCleanupExecutor.java
+++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/workflow/WorkflowCleanupExecutor.java
@@ -137,7 +137,7 @@ private synchronized void handleDependentCleaned(
private boolean allDependentsCleaned(DependentResourceNode dependentResourceNode) {
List parents = dependentResourceNode.getParents();
return parents.isEmpty()
- || parents.stream().allMatch(d -> alreadyVisited(d) && !postDeleteConditionNotMet(d));
+ || parents.stream().allMatch(d -> alreadyVisited(d) && postDeleteConditionMet(d));
}
@SuppressWarnings("unchecked")
diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/workflow/WorkflowReconcileExecutor.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/workflow/WorkflowReconcileExecutor.java
index d544891aba..6917a95291 100644
--- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/workflow/WorkflowReconcileExecutor.java
+++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/workflow/WorkflowReconcileExecutor.java
@@ -18,6 +18,7 @@
import java.util.ArrayList;
import java.util.HashSet;
import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -34,6 +35,9 @@ class WorkflowReconcileExecutor extends AbstractWorkflowE
private static final String RECONCILE = "reconcile";
private static final String DELETE = "delete";
+ private final Set> conditionNotMetNodes =
+ ConcurrentHashMap.newKeySet();
+
public WorkflowReconcileExecutor(DefaultWorkflow workflow, P primary, Context
context) {
super(workflow, primary, context);
}
@@ -51,6 +55,13 @@ protected Logger logger() {
return log;
}
+ @Override
+ protected void onNodeExecutionFinished(DependentResourceNode, P> dependentResourceNode) {
+ if (conditionNotMetNodes.remove(dependentResourceNode)) {
+ handleReconcileOrActivationConditionNotMet(dependentResourceNode);
+ }
+ }
+
private synchronized void handleReconcile(DependentResourceNode dependentResourceNode) {
log.debug("Considering for reconcile: {} primaryID: {}", dependentResourceNode, primaryID);
@@ -90,20 +101,7 @@ private synchronized void handleReconcile(DependentResourceNode depend
return;
}
- boolean activationConditionMet =
- isConditionMet(dependentResourceNode.getActivationCondition(), dependentResourceNode);
- registerOrDeregisterEventSourceBasedOnActivation(activationConditionMet, dependentResourceNode);
-
- boolean reconcileConditionMet = true;
- if (activationConditionMet) {
- reconcileConditionMet =
- isConditionMet(dependentResourceNode.getReconcilePrecondition(), dependentResourceNode);
- }
- if (!reconcileConditionMet || !activationConditionMet) {
- handleReconcileOrActivationConditionNotMet(dependentResourceNode);
- } else {
- submit(dependentResourceNode, new NodeReconcileExecutor<>(dependentResourceNode), RECONCILE);
- }
+ submit(dependentResourceNode, new NodeReconcileExecutor<>(dependentResourceNode), RECONCILE);
}
private synchronized void handleDelete(DependentResourceNode dependentResourceNode) {
@@ -144,7 +142,7 @@ private boolean allDependentsDeletedAlready(DependentResourceNode, P> dependen
var dependents = dependentResourceNode.getParents();
return dependents.stream()
.allMatch(
- d -> alreadyVisited(d) && isReady(d) && !isInError(d) && !postDeleteConditionNotMet(d));
+ d -> alreadyVisited(d) && isReady(d) && !isInError(d) && postDeleteConditionMet(d));
}
private class NodeReconcileExecutor extends NodeExecutor {
@@ -155,6 +153,17 @@ private NodeReconcileExecutor(DependentResourceNode dependentResourceNode)
@Override
protected void doRun(DependentResourceNode dependentResourceNode) {
+ final var activationConditionMet =
+ isConditionMet(dependentResourceNode.getActivationCondition(), dependentResourceNode);
+ registerOrDeregisterEventSourceBasedOnActivation(
+ activationConditionMet, dependentResourceNode);
+ if (!activationConditionMet
+ || !isConditionMet(
+ dependentResourceNode.getReconcilePrecondition(), dependentResourceNode)) {
+ conditionNotMetNodes.add(dependentResourceNode);
+ return;
+ }
+
final var dependentResource = dependentResourceNode.getDependentResource();
log.debug("Reconciling for primary: {} node: {} ", primaryID, dependentResourceNode);
ReconcileResult reconcileResult = dependentResource.reconcile(primary, context);
@@ -238,6 +247,7 @@ private void handleReconcileOrActivationConditionNotMet(
Set bottomNodes = new HashSet<>();
markDependentsForDelete(dependentResourceNode, bottomNodes);
bottomNodes.forEach(this::handleDelete);
+ handleDelete(dependentResourceNode);
}
private void markDependentsForDelete(
diff --git a/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/dependent/workflow/WorkflowReconcileExecutorTest.java b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/dependent/workflow/WorkflowReconcileExecutorTest.java
index 117fe2afb8..fa3e703a2f 100644
--- a/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/dependent/workflow/WorkflowReconcileExecutorTest.java
+++ b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/dependent/workflow/WorkflowReconcileExecutorTest.java
@@ -15,13 +15,21 @@
*/
package io.javaoperatorsdk.operator.processing.dependent.workflow;
+import java.util.concurrent.BrokenBarrierException;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.CyclicBarrier;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
+import java.util.concurrent.RejectedExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicBoolean;
import org.assertj.core.api.Assertions;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.TestInfo;
+import org.junit.jupiter.api.Timeout;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -45,13 +53,12 @@ class WorkflowReconcileExecutorTest extends AbstractWorkflowExecutorTest {
Context mockContext = spy(Context.class);
ExecutorService executorService = Executors.newCachedThreadPool();
- EventSourceRetriever eventSourceRetriever = mock(EventSourceRetriever.class);
+ EventSourceRetriever eventSourceRetriever = mock();
TestDependent dr3 = new TestDependent("DR_3");
TestDependent dr4 = new TestDependent("DR_4");
@BeforeEach
- @SuppressWarnings("unchecked")
void setup(TestInfo testInfo) {
log.debug("==> Starting test {}", testInfo.getDisplayName());
when(mockContext.managedWorkflowAndDependentResourceContext())
@@ -673,6 +680,161 @@ void deletesDependentsOfNonActiveDependentButNotTheNonActive() {
assertThat(executionHistory).deleted(drDeleter, drDeleter3).notReconciled(dr1, drDeleter2);
}
+ @Test
+ @Timeout(10)
+ void deleteScheduledByUnmetPreconditionFinishesBeforeReconcileReturns() {
+ // the delete is scheduled from the completion hook, while the monitor is still held
+ var deleteFinished = new AtomicBoolean(false);
+ var slowDeleter =
+ new TestDeleterDependent("SLOW_DELETER") {
+ @Override
+ public void delete(TestCustomResource primary, Context context) {
+ try {
+ Thread.sleep(200);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ deleteFinished.set(true);
+ super.delete(primary, context);
+ }
+ };
+
+ var workflow =
+ new WorkflowBuilder()
+ .addDependentResourceAndConfigure(slowDeleter)
+ .withReconcilePrecondition(notMetCondition)
+ .build();
+
+ workflow.reconcile(new TestCustomResource(), mockContext);
+
+ assertTrue(deleteFinished.get(), "reconcile() returned while the delete was still running");
+ }
+
+ @Test
+ @Timeout(10)
+ void deletedIfReconcilePreconditionNotMetWhileAnotherDeleteCascadeIsRunning() {
+ // the shared dependent's cascade reaches this one while it is still evaluating its precondition
+ var sharedDependentDeleted = new CountDownLatch(1);
+ TestDeleterDependent drDeleter2 = new TestDeleterDependent("DR_DELETER_2");
+ var drDeleter3 =
+ new TestDeleterDependent("DR_DELETER_3") {
+ @Override
+ public void delete(TestCustomResource primary, Context context) {
+ super.delete(primary, context);
+ sharedDependentDeleted.countDown();
+ }
+ };
+ Condition, TestCustomResource> blockedNotMetCondition =
+ (dependentResource, primary, context) -> {
+ try {
+ if (!sharedDependentDeleted.await(5, TimeUnit.SECONDS)) {
+ throw new IllegalStateException("the shared dependent was never deleted");
+ }
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new IllegalStateException(e);
+ }
+ return false;
+ };
+
+ var workflow =
+ new WorkflowBuilder()
+ .addDependentResourceAndConfigure(drDeleter)
+ .withReconcilePrecondition(notMetCondition)
+ .addDependentResourceAndConfigure(drDeleter2)
+ .withReconcilePrecondition(blockedNotMetCondition)
+ .addDependentResourceAndConfigure(drDeleter3)
+ .dependsOn(drDeleter, drDeleter2)
+ .build();
+
+ var res = workflow.reconcile(new TestCustomResource(), mockContext);
+
+ assertThat(executionHistory).deleted(drDeleter3, drDeleter, drDeleter2);
+ Assertions.assertThat(res.getErroredDependents()).isEmpty();
+ }
+
+ @Test
+ @Timeout(10)
+ void completesIfSchedulingDeleteForUnmetPreconditionIsRejected() {
+ // the delete is scheduled from a hook that already runs in the node executor's finally block
+ ExecutorService rejectingSecondSubmit = mock();
+ when(rejectingSecondSubmit.submit(any(Runnable.class)))
+ .thenAnswer(invocation -> executorService.submit((Runnable) invocation.getArgument(0)))
+ .thenThrow(new RejectedExecutionException("executor is shut down"));
+ when(mockContext.getWorkflowExecutorService()).thenReturn(rejectingSecondSubmit);
+
+ var workflow =
+ new WorkflowBuilder()
+ .addDependentResourceAndConfigure(drDeleter)
+ .withReconcilePrecondition(notMetCondition)
+ .build();
+
+ var exception =
+ assertThrows(
+ AggregatedOperatorException.class,
+ () -> workflow.reconcile(new TestCustomResource(), mockContext));
+
+ Assertions.assertThat(exception.getAggregatedExceptions())
+ .containsOnlyKeys(drDeleter.name())
+ .extractingByKey(drDeleter.name())
+ .isInstanceOf(RejectedExecutionException.class);
+ }
+
+ @Test
+ @Timeout(10)
+ void activationConditionsEvaluatedConcurrently() {
+ // they can only meet at the barrier if neither of them holds the executor's monitor
+ var rendezvousCondition = rendezvousCondition(new CyclicBarrier(2));
+
+ var workflow =
+ new WorkflowBuilder()
+ .addDependentResourceAndConfigure(dr1)
+ .withActivationCondition(rendezvousCondition)
+ .addDependentResourceAndConfigure(dr2)
+ .withActivationCondition(rendezvousCondition)
+ .build();
+
+ var res = workflow.reconcile(new TestCustomResource(), mockContext);
+
+ assertThat(executionHistory).reconciled(dr1, dr2);
+ Assertions.assertThat(res.getErroredDependents()).isEmpty();
+ }
+
+ @Test
+ void activationConditionErrorAttributedToItsOwnDependent() {
+ var workflow =
+ new WorkflowBuilder()
+ .addDependentResource(dr1)
+ .addDependentResourceAndConfigure(dr2)
+ .dependsOn(dr1)
+ .withActivationCondition(throwingCondition())
+ .addDependentResourceAndConfigure(dr3)
+ .dependsOn(dr1)
+ .withThrowExceptionFurther(false)
+ .build();
+
+ var res = workflow.reconcile(new TestCustomResource(), mockContext);
+
+ Assertions.assertThat(res.getErroredDependents()).containsOnlyKeys(dr2);
+ assertThat(executionHistory).reconciled(dr1, dr3).notReconciled(dr2);
+ }
+
+ @Test
+ void activationConditionErrorOnTopLevelDependentDoesNotStopOthers() {
+ var workflow =
+ new WorkflowBuilder()
+ .addDependentResourceAndConfigure(dr1)
+ .withActivationCondition(throwingCondition())
+ .addDependentResource(dr2)
+ .withThrowExceptionFurther(false)
+ .build();
+
+ var res = workflow.reconcile(new TestCustomResource(), mockContext);
+
+ Assertions.assertThat(res.getErroredDependents()).containsOnlyKeys(dr1);
+ assertThat(executionHistory).reconciled(dr2).notReconciled(dr1);
+ }
+
@Test
@SuppressWarnings("unchecked")
void activationConditionOnlyCalledOnceOnDeleteDependents() {
@@ -695,7 +857,6 @@ void activationConditionOnlyCalledOnceOnDeleteDependents() {
}
@Test
- @SuppressWarnings("unchecked")
void activationConditionEventSourceRegistrationWithParentWithFalsePrecondition() {
var workflow =
new WorkflowBuilder()
@@ -713,7 +874,6 @@ void activationConditionEventSourceRegistrationWithParentWithFalsePrecondition()
}
@Test
- @SuppressWarnings("unchecked")
void activationConditionEventSourceRegistration() {
var workflow =
new WorkflowBuilder()
@@ -826,4 +986,24 @@ void shouldReturnEmptyIfNoConditionResultExists() {
final var reconcileResult = workflow.reconcile(new TestCustomResource(), mockContext);
assertTrue(reconcileResult.getNotReadyDependentResult(dr1, Integer.class).isEmpty());
}
+
+ private Condition, TestCustomResource> rendezvousCondition(CyclicBarrier barrier) {
+ return (dependentResource, primary, context) -> {
+ try {
+ barrier.await(5, TimeUnit.SECONDS);
+ return true;
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new IllegalStateException(e);
+ } catch (BrokenBarrierException | TimeoutException e) {
+ throw new IllegalStateException("activation conditions were serialized", e);
+ }
+ };
+ }
+
+ private Condition, TestCustomResource> throwingCondition() {
+ return (dependentResource, primary, context) -> {
+ throw new IllegalStateException("Test exception");
+ };
+ }
}