wip on cleanup

Author: csviri

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

@@ -1,5 +1,11 @@
 package io.javaoperatorsdk.operator.processing.dependent.workflow;
 
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.Future;
+
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -10,6 +16,15 @@ public class WorkflowCleanupExecutor<P extends HasMetadata> {
 
   private static final Logger log = LoggerFactory.getLogger(WorkflowReconcileExecutor.class);
 
+  private final Map<DependentResourceNode<?, ?>, Future<?>> actualExecutions =
+      new HashMap<>();
+  private final Map<DependentResourceNode<?, ?>, Exception> exceptionsDuringExecution =
+      new HashMap<>();
+  private final Set<DependentResourceNode<?, ?>> alreadyReconciled = new HashSet<>();
+  private final Set<DependentResourceNode<?, ?>> notReady = new HashSet<>();
+  private final Set<DependentResourceNode<?, ?>> ownOrAncestorReconcileConditionConditionNotMet =
+      new HashSet<>();
+
   private final Workflow<P> workflow;
   private final P primary;
   private final Context<P> context;
@@ -23,13 +38,104 @@ public WorkflowCleanupExecutor(Workflow<P> workflow, P primary, Context<P> conte
 
   public synchronized WorkflowCleanupResult cleanup() {
     for (DependentResourceNode<?, P> dependentResourceNode : workflow
-        .getTopLevelDependentResources()) {
+        .getBottomLevelResource()) {
       handleCleanup(dependentResourceNode, false);
     }
-    return null;
+    while (true) {
+      try {
+        this.wait();
+        if (noMoreExecutionsScheduled()) {
+          break;
+        } else {
+          log.warn("Notified but still resources under execution. This should not happen.");
+        }
+      } catch (InterruptedException e) {
+        log.warn("Thread interrupted", e);
+        Thread.currentThread().interrupt();
+      }
+    }
+    return createCleanupResult();
+  }
+
+  private WorkflowCleanupResult createCleanupResult() {
+    return new WorkflowCleanupResult();
+  }
+
+  private synchronized boolean noMoreExecutionsScheduled() {
+    return actualExecutions.isEmpty();
   }
 
   private void handleCleanup(DependentResourceNode<?, P> dependentResourceNode, boolean b) {
+    log.debug("Submitting for cleanup: {}", dependentResourceNode);
+
+    if (alreadyVisited(dependentResourceNode)
+        || isCleaningNow(dependentResourceNode)
+        || !allParentsCleaned(dependentResourceNode)
+        || hasErroredParent(dependentResourceNode)) {
+      log.debug("Skipping submit of: {}, ", dependentResourceNode);
+      return;
+    }
+
+  }
+
+  private class NodeExecutor implements Runnable {
+
+    private final DependentResourceNode<?, P> dependentResourceNode;
+    private final boolean onlyReconcileForPossibleDelete;
+
+    private NodeExecutor(DependentResourceNode<?, P> dependentResourceNode,
+        boolean onlyReconcileForDelete) {
+      this.dependentResourceNode = dependentResourceNode;
+      this.onlyReconcileForPossibleDelete = onlyReconcileForDelete;
+    }
+
+    @Override
+    @SuppressWarnings("unchecked")
+    public void run() {
+      try {
+
+      } catch (RuntimeException e) {
+        handleExceptionInExecutor(dependentResourceNode, e);
+      } finally {
+        handleNodeExecutionFinish(dependentResourceNode);
+      }
+    }
+  }
+
+  private synchronized void handleExceptionInExecutor(DependentResourceNode dependentResourceNode,
+      RuntimeException e) {
+    exceptionsDuringExecution.put(dependentResourceNode, e);
+  }
+
+  private synchronized void handleNodeExecutionFinish(DependentResourceNode dependentResourceNode) {
+    log.debug("Finished execution for: {}", dependentResourceNode);
+    actualExecutions.remove(dependentResourceNode);
+    if (actualExecutions.isEmpty()) {
+      this.notifyAll();
+    }
+  }
+
+  private boolean isCleaningNow(DependentResourceNode<?, ?> dependentResourceNode) {
+    return actualExecutions.containsKey(dependentResourceNode);
+  }
+
+
+  private boolean alreadyVisited(
+      DependentResourceNode<?, ?> dependentResourceNode) {
+    return alreadyReconciled.contains(dependentResourceNode);
+  }
+
+  private boolean allParentsCleaned(
+      DependentResourceNode<?, ?> dependentResourceNode) {
+    return dependentResourceNode.getDependsOn().isEmpty()
+        || dependentResourceNode.getDependsOn().stream()
+            .allMatch(d -> alreadyVisited(d) && !notReady.contains(d));
+  }
 
+  private boolean hasErroredParent(
+      DependentResourceNode<?, ?> dependentResourceNode) {
+    return !dependentResourceNode.getDependsOn().isEmpty()
+        && dependentResourceNode.getDependsOn().stream()
+            .anyMatch(exceptionsDuringExecution::containsKey);
   }
 }

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

@@ -38,6 +38,8 @@ public WorkflowReconcileExecutor(Workflow<P> workflow, P primary, Context<P> con
     this.workflow = workflow;
   }
 
+  // todo reverse delete order
+
   // todo add reconcile results
   // - reconciled in results should contain only truly reconciled
   public synchronized WorkflowExecutionResult reconcile() {
@@ -106,7 +108,6 @@ private synchronized void handleNodeExecutionFinish(DependentResourceNode depend
 
   // needs to be synced
   private synchronized void updateStatusForNotReady(ReadyCondition<?, P> readyCondition) {
-    // todo think through this, since more can't be not ready
     readyCondition.addNotReadyStatusInfo(primary);
   }