fix: also resolve ManagedWorkflow nodes relations, simplify

Author: metacosm

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

@@ -1,15 +1,20 @@
 package io.javaoperatorsdk.operator.processing.dependent.workflow;
 
+import java.util.HashMap;
 import java.util.HashSet;
 import java.util.List;
 import java.util.Set;
-import java.util.function.Function;
 import java.util.stream.Collectors;
 
 import io.fabric8.kubernetes.api.model.HasMetadata;
 import io.fabric8.kubernetes.client.KubernetesClient;
+import io.javaoperatorsdk.operator.api.config.ConfigurationServiceProvider;
 import io.javaoperatorsdk.operator.api.config.ControllerConfiguration;
 import io.javaoperatorsdk.operator.api.config.ExecutorServiceManager;
+import io.javaoperatorsdk.operator.api.config.dependent.DependentResourceSpec;
+import io.javaoperatorsdk.operator.api.reconciler.dependent.DependentResource;
+import io.javaoperatorsdk.operator.api.reconciler.dependent.EventSourceReferencer;
+import io.javaoperatorsdk.operator.api.reconciler.dependent.managed.KubernetesClientAware;
 
 import static io.javaoperatorsdk.operator.processing.dependent.workflow.Workflow.THROW_EXCEPTION_AUTOMATICALLY_DEFAULT;
 
@@ -18,24 +23,23 @@ public class DefaultManagedWorkflow<P extends HasMetadata> implements ManagedWor
 
   private final Set<String> topLevelResources;
   private final Set<String> bottomLevelResources;
-  private final Set<SpecDependentResourceNode> nodes;
+  private final List<DependentResourceSpec<?, ?>> specs;
   private final boolean hasCleaner;
 
-  @SuppressWarnings("unchecked")
-  DefaultManagedWorkflow(Set<SpecDependentResourceNode> dependentResourceSpecs,
+  DefaultManagedWorkflow(List<DependentResourceSpec<?, ?>> dependentResourceSpecs,
       boolean hasCleaner) {
     this.hasCleaner = hasCleaner;
     topLevelResources = new HashSet<>(dependentResourceSpecs.size());
-    bottomLevelResources =
-        dependentResourceSpecs.stream().map(SpecDependentResourceNode::getName).collect(
-            Collectors.toSet());
-    nodes = dependentResourceSpecs;
-    dependentResourceSpecs.forEach(drn -> {
+    bottomLevelResources = dependentResourceSpecs.stream()
+        .map(DependentResourceSpec::getName)
+        .collect(Collectors.toSet());
+    specs = dependentResourceSpecs;
+    dependentResourceSpecs.forEach(spec -> {
       // add cycle detection?
-      if (drn.getDependsOn().isEmpty()) {
-        topLevelResources.add(drn.getName());
+      if (spec.getDependsOn().isEmpty()) {
+        topLevelResources.add(spec.getName());
       } else {
-        for (String dependsOn : (List<String>) drn.getDependsOn()) {
+        for (String dependsOn : spec.getDependsOn()) {
           bottomLevelResources.remove(dependsOn);
         }
       }
@@ -50,8 +54,8 @@ Set<String> getBottomLevelResources() {
     return bottomLevelResources;
   }
 
-  Set<SpecDependentResourceNode> nodes() {
-    return nodes;
+  List<String> nodeNames() {
+    return specs.stream().map(DependentResourceSpec::getName).collect(Collectors.toList());
   }
 
   @Override
@@ -61,26 +65,54 @@ public boolean hasCleaner() {
 
   @Override
   public boolean isEmpty() {
-    return nodes.isEmpty();
+    return specs.isEmpty();
   }
 
   @Override
+  @SuppressWarnings("unchecked")
   public Workflow<P> resolve(KubernetesClient client,
       ControllerConfiguration<P> configuration) {
-    final var map = nodes.stream()
-        .map(spec -> createDRN(client, configuration, spec))
-        .collect(Collectors.toMap(DependentResourceNode::getName, Function.identity()));
-    final var bottom = bottomLevelResources.stream().map(map::get).collect(Collectors.toSet());
-    final var top = topLevelResources.stream().map(map::get).collect(Collectors.toSet());
-    return new Workflow<>(map, bottom, top,
+    final var alreadyResolved = new HashMap<String, DependentResourceNode>(specs.size());
+    for (DependentResourceSpec spec : specs) {
+      final var node = new DependentResourceNode(resolve(spec, client, configuration));
+      alreadyResolved.put(node.getName(), node);
+      spec.getDependsOn()
+          .forEach(depend -> node.addDependsOnRelation(alreadyResolved.get((String) depend)));
+    }
+
+    final var bottom =
+        bottomLevelResources.stream().map(alreadyResolved::get).collect(Collectors.toSet());
+    final var top =
+        topLevelResources.stream().map(alreadyResolved::get).collect(Collectors.toSet());
+    return new Workflow<>(alreadyResolved, bottom, top,
         ExecutorServiceManager.instance().workflowExecutorService(),
         THROW_EXCEPTION_AUTOMATICALLY_DEFAULT, hasCleaner);
   }
 
-  @SuppressWarnings("unchecked")
-  private DependentResourceNode createDRN(
-      KubernetesClient client, ControllerConfiguration<P> configuration,
-      SpecDependentResourceNode spec) {
-    return new DependentResourceNode(spec.resolve(client, configuration));
+  @SuppressWarnings({"rawtypes", "unchecked"})
+  private <R> DependentResource<R, P> resolve(DependentResourceSpec<R, P> spec,
+      KubernetesClient client,
+      ControllerConfiguration<P> configuration) {
+    final DependentResource<R, P> dependentResource =
+        ConfigurationServiceProvider.instance().dependentResourceFactory()
+            .createFrom(spec, configuration);
+
+    if (dependentResource instanceof KubernetesClientAware) {
+      ((KubernetesClientAware) dependentResource).setKubernetesClient(client);
+    }
+
+    spec.getUseEventSourceWithName()
+        .ifPresent(esName -> {
+          if (dependentResource instanceof EventSourceReferencer) {
+            ((EventSourceReferencer) dependentResource).useEventSourceWithName(esName);
+          } else {
+            throw new IllegalStateException(
+                "DependentResource " + spec + " wants to use EventSource named " + esName
+                    + " but doesn't implement support for this feature by implementing "
+                    + EventSourceReferencer.class.getSimpleName());
+          }
+        });
+
+    return dependentResource;
   }
 }

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

@@ -54,47 +54,32 @@ public <P extends HasMetadata> ManagedWorkflow<P> createWorkflow(
 
   <P extends HasMetadata> DefaultManagedWorkflow<P> createAsDefault(
       List<DependentResourceSpec> dependentResourceSpecs) {
-    var orderedResourceSpecs = orderAndDetectCycles(dependentResourceSpecs);
-    final var alreadyCreated =
-        new ArrayList<SpecDependentResourceNode>(orderedResourceSpecs.size());
     final boolean[] cleanerHolder = {false};
-    final var nodes = orderedResourceSpecs.stream()
-        .map(spec -> createFrom(spec, alreadyCreated, cleanerHolder))
-        .collect(Collectors.toSet());
-    return new DefaultManagedWorkflow<>(nodes, cleanerHolder[0]);
-  }
-
-  private SpecDependentResourceNode createFrom(DependentResourceSpec spec,
-      List<SpecDependentResourceNode> alreadyCreated, boolean[] cleanerHolder) {
-    final var node = new SpecDependentResourceNode<>(spec);
-    alreadyCreated.add(node);
-    // if any previously checked dependent was a cleaner, no need to check further
-    cleanerHolder[0] = cleanerHolder[0] || Workflow.isDeletable(spec.getDependentResourceClass());
-    spec.getDependsOn().forEach(depend -> {
-      final SpecDependentResourceNode dependsOn = alreadyCreated.stream()
-          .filter(drn -> depend.equals(drn.getName())).findFirst()
-          .orElseThrow();
-      node.addDependsOnRelation(dependsOn);
-    });
-    return node;
+    var orderedResourceSpecs = orderAndDetectCycles(dependentResourceSpecs, cleanerHolder);
+    return new DefaultManagedWorkflow<>(orderedResourceSpecs, cleanerHolder[0]);
   }
 
   /**
    * @param dependentResourceSpecs list of specs
    * @return top-bottom ordered resources that can be added safely to workflow
    * @throws OperatorException if there is a cycle in the dependencies
    */
-  public List<DependentResourceSpec> orderAndDetectCycles(
-      List<DependentResourceSpec> dependentResourceSpecs) {
+  private List<DependentResourceSpec<?, ?>> orderAndDetectCycles(
+      List<DependentResourceSpec> dependentResourceSpecs, boolean[] cleanerHolder) {
 
     final var drInfosByName = createDRInfos(dependentResourceSpecs);
-    final var orderedSpecs = new ArrayList<DependentResourceSpec>(dependentResourceSpecs.size());
+    final var orderedSpecs =
+        new ArrayList<DependentResourceSpec<?, ?>>(dependentResourceSpecs.size());
     final var alreadyVisited = new HashSet<String>();
     var toVisit = getTopDependentResources(dependentResourceSpecs);
 
     while (!toVisit.isEmpty()) {
       final var toVisitNext = new HashSet<DependentResourceSpec>();
       toVisit.forEach(dr -> {
+        if (cleanerHolder != null) {
+          cleanerHolder[0] =
+              cleanerHolder[0] || Workflow.isDeletable(dr.getDependentResourceClass());
+        }
         final var name = dr.getName();
         var drInfo = drInfosByName.get(name);
         if (drInfo != null) {
@@ -118,6 +103,16 @@ public List<DependentResourceSpec> orderAndDetectCycles(
     return orderedSpecs;
   }
 
+  /**
+   * @param dependentResourceSpecs list of specs
+   * @return top-bottom ordered resources that can be added safely to workflow
+   * @throws OperatorException if there is a cycle in the dependencies
+   */
+  public List<DependentResourceSpec<?, ?>> orderAndDetectCycles(
+      List<DependentResourceSpec> dependentResourceSpecs) {
+    return orderAndDetectCycles(dependentResourceSpecs, null);
+  }
+
   private static class DRInfo {
 
     private final DependentResourceSpec spec;

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

@@ -140,7 +140,7 @@ void createsWorkflow() {
 
     var workflow = managedWorkflowSupport.createAsDefault(specs);
 
-    assertThat(workflow.nodes()).map(SpecDependentResourceNode::getName)
+    assertThat(workflow.nodeNames())
         .containsExactlyInAnyOrder(NAME_1, NAME_2, NAME_3, NAME_4);
     assertThat(workflow.getTopLevelResources()).containsExactly(NAME_1);
     assertThat(workflow.getBottomLevelResources()).containsExactly(NAME_4);