Redesign execution module

resilience-bench/operator/src/main/java/io/resiliencebench/BenchmarkController.java

@@ -9,7 +9,7 @@
 import io.javaoperatorsdk.operator.api.reconciler.ControllerConfiguration;
 import io.javaoperatorsdk.operator.api.reconciler.Reconciler;
 import io.javaoperatorsdk.operator.api.reconciler.UpdateControl;
-import io.resiliencebench.execution.ScenarioExecutor;
+import io.resiliencebench.execution.DefaultQueueExecutor;
 import io.resiliencebench.resources.ExecutionQueueFactory;
 import io.resiliencebench.resources.ScenarioFactory;
 import io.resiliencebench.resources.benchmark.Benchmark;
@@ -29,13 +29,13 @@ public class BenchmarkController implements Reconciler<Benchmark> {
   private final CustomResourceRepository<Workload> workloadRepository;
   private final CustomResourceRepository<ExecutionQueue> queueRepository;
 
-  private final ScenarioExecutor scenarioExecutor;
+  private final DefaultQueueExecutor queueExecutor;
 
-  public BenchmarkController(ScenarioExecutor scenarioExecutor,
+  public BenchmarkController(DefaultQueueExecutor queueExecutor,
                              CustomResourceRepository<Scenario> scenarioRepository,
                              CustomResourceRepository<Workload> workloadRepository,
                              CustomResourceRepository<ExecutionQueue> queueRepository) {
-    this.scenarioExecutor = scenarioExecutor;
+    this.queueExecutor = queueExecutor;
     this.scenarioRepository = scenarioRepository;
     this.workloadRepository = workloadRepository;
     this.queueRepository = queueRepository;
@@ -58,7 +58,7 @@ public UpdateControl<Benchmark> reconcile(Benchmark benchmark, Context<Benchmark
 
     var executionQueue = prepareToRunScenarios(benchmark, scenariosList);
 
-    scenarioExecutor.run(executionQueue);
+    queueExecutor.execute(executionQueue);
     logger.info("Benchmark reconciled {}. {} scenarios created",
             benchmark.getMetadata().getName(),
             scenariosList.size()

resilience-bench/operator/src/main/java/io/resiliencebench/execution/DefaultQueueExecutor.java

@@ -0,0 +1,62 @@
+
+package io.resiliencebench.execution;
+
+import static java.lang.String.format;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.stereotype.Service;
+
+import io.resiliencebench.resources.queue.ExecutionQueue;
+import io.resiliencebench.resources.scenario.Scenario;
+import io.resiliencebench.support.CustomResourceRepository;
+
+@Service
+public class DefaultQueueExecutor implements QueueExecutor {
+
+  private final static Logger logger = LoggerFactory.getLogger(DefaultQueueExecutor.class);
+
+  private final CustomResourceRepository<Scenario> scenarioRepository;
+  private final CustomResourceRepository<ExecutionQueue> executionRepository;
+
+  private final ScenarioExecutor scenarioExecutor;
+
+  public DefaultQueueExecutor(
+          CustomResourceRepository<Scenario> scenarioRepository,
+          CustomResourceRepository<ExecutionQueue> executionRepository,
+          ScenarioExecutor scenarioExecutor) {
+    this.scenarioRepository = scenarioRepository;
+    this.executionRepository = executionRepository;
+    this.scenarioExecutor = scenarioExecutor;
+  }
+
+  @Override
+  public void execute(ExecutionQueue queue) {
+    var queueToExecute = executionRepository.find(queue.getMetadata())
+            .orElseThrow(() -> new RuntimeException("Queue not found " + queue.getMetadata().getName()));
+
+    var nextItem = queueToExecute.getNextPendingItem();
+
+    if (nextItem.isPresent()) {
+      var namespace = queueToExecute.getMetadata().getNamespace();
+      if (nextItem.get().isPending()) {
+        executeScenario(namespace, nextItem.get().getScenario(), queueToExecute);
+      }
+    } else {
+      logger.info("No item available for queue: {}", queueToExecute.getMetadata().getName());
+      if (queueToExecute.isDone()) {
+        logger.info("All items finished for: {}", queueToExecute.getMetadata().getName());
+      }
+    }
+  }
+
+  private void executeScenario(String namespace, String scenarioName, ExecutionQueue executionQueue) {
+    logger.info("Running scenario: {}", scenarioName);
+    var scenario = scenarioRepository.find(namespace, scenarioName);
+    if (scenario.isPresent()) {
+      scenarioExecutor.execute(scenario.get(), executionQueue, () -> this.execute(executionQueue));
+    } else {
+      throw new RuntimeException(format("Scenario not found: %s.%s", namespace, scenarioName));
+    }
+  }
+}

resilience-bench/operator/src/main/java/io/resiliencebench/execution/DefaultScenarioExecutor.java

@@ -0,0 +1,76 @@
+package io.resiliencebench.execution;
+
+import io.fabric8.kubernetes.api.model.batch.v1.Job;
+import io.fabric8.kubernetes.client.KubernetesClient;
+import io.fabric8.kubernetes.client.Watcher;
+import io.fabric8.kubernetes.client.WatcherException;
+import io.resiliencebench.execution.steps.StepRegistry;
+import io.resiliencebench.execution.steps.k6.K6JobFactory;
+import io.resiliencebench.resources.queue.ExecutionQueue;
+import io.resiliencebench.resources.scenario.Scenario;
+import io.resiliencebench.resources.workload.Workload;
+import io.resiliencebench.support.CustomResourceRepository;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.stereotype.Service;
+
+import static java.util.Objects.nonNull;
+
+@Service
+public class DefaultScenarioExecutor implements ScenarioExecutor {
+
+  private final static Logger logger = LoggerFactory.getLogger(DefaultScenarioExecutor.class);
+  private final KubernetesClient kubernetesClient;
+  private final StepRegistry stepRegistry;
+  private final K6JobFactory k6JobFactory;
+
+  private final CustomResourceRepository<Workload> workloadRepository;
+
+  public DefaultScenarioExecutor(KubernetesClient kubernetesClient,
+                                 StepRegistry stepRegistry,
+                                 K6JobFactory k6JobFactory,
+                                 CustomResourceRepository<Workload> workloadRepository) {
+    this.kubernetesClient = kubernetesClient;
+    this.stepRegistry = stepRegistry;
+    this.k6JobFactory = k6JobFactory;
+    this.workloadRepository = workloadRepository;
+  }
+
+  @Override
+  public void execute(Scenario scenario, ExecutionQueue executionQueue, Runnable onCompletion) {
+    var ns = scenario.getMetadata().getNamespace();
+    var workloadName = scenario.getSpec().getWorkload().getWorkloadName();
+    var workload = workloadRepository.find(ns, workloadName)
+            .orElseThrow(() -> new IllegalArgumentException("Workload does not exists: %s".formatted(workloadName)));
+
+    executePreparationSteps(scenario, executionQueue);
+    var job = k6JobFactory.create(scenario, workload);
+    var jobsClient = kubernetesClient.batch().v1().jobs();
+    jobsClient.inNamespace(job.getMetadata().getNamespace()).withName(job.getMetadata().getName()).delete();
+    jobsClient.resource(job).create();
+    logger.info("Job created: {}", job.getMetadata().getName());
+    jobsClient.resource(job).watch(new Watcher<>() {
+      @Override
+      public void eventReceived(Action action, Job resource) {
+        if (action.equals(Action.MODIFIED) && nonNull(resource.getStatus().getCompletionTime())) {
+          logger.debug("Finished job: {}", resource.getMetadata().getName());
+          executePostExecutionSteps(scenario, executionQueue);
+          onCompletion.run();
+        }
+      }
+      @Override
+      public void onClose(WatcherException cause) {
+        logger.info("Job watcher closed", cause);
+      }
+    });
+  }
+
+
+  private void executePreparationSteps(Scenario scenario, ExecutionQueue queue) {
+    stepRegistry.getPreparationSteps().forEach(step -> step.execute(scenario, queue));
+  }
+
+  private void executePostExecutionSteps(Scenario scenario, ExecutionQueue queue) {
+    stepRegistry.getPostExecutionSteps().forEach(step -> step.execute(scenario, queue));
+  }
+}

resilience-bench/operator/src/main/java/io/resiliencebench/execution/QueueExecutor.java

@@ -0,0 +1,7 @@
+package io.resiliencebench.execution;
+
+import io.resiliencebench.resources.queue.ExecutionQueue;
+
+public interface QueueExecutor {
+  void execute(ExecutionQueue queue);
+}

resilience-bench/operator/src/main/java/io/resiliencebench/execution/ScenarioExecutor.java

@@ -1,142 +1,8 @@
-
 package io.resiliencebench.execution;
 
-import static io.resiliencebench.support.Annotations.SCENARIO;
-import static java.lang.String.format;
-import static java.util.Objects.isNull;
-import static java.util.Objects.nonNull;
-import java.util.Optional;
-
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-import org.springframework.stereotype.Service;
-
-import io.fabric8.kubernetes.api.model.batch.v1.Job;
-import io.fabric8.kubernetes.client.KubernetesClient;
-import io.fabric8.kubernetes.client.Watcher;
-import io.fabric8.kubernetes.client.WatcherException;
-import io.resiliencebench.execution.steps.StepRegister;
-import io.resiliencebench.execution.steps.k6.K6LoadGeneratorStep;
 import io.resiliencebench.resources.queue.ExecutionQueue;
-import io.resiliencebench.resources.queue.Item;
 import io.resiliencebench.resources.scenario.Scenario;
-import io.resiliencebench.support.Annotations;
-import io.resiliencebench.support.CustomResourceRepository;
-
-@Service
-public class ScenarioExecutor implements Watcher<Job> {
-
-  private final static Logger logger = LoggerFactory.getLogger(ScenarioExecutor.class);
-  private final KubernetesClient kubernetesClient;
-
-  private final CustomResourceRepository<Scenario> scenarioRepository;
-  private final CustomResourceRepository<ExecutionQueue> executionRepository;
-
-  private final StepRegister stepRegister;
-
-  private final K6LoadGeneratorStep k6LoadGeneratorStep;
-
-  public ScenarioExecutor(
-          KubernetesClient kubernetesClient,
-          StepRegister stepRegister,
-          CustomResourceRepository<Scenario> scenarioRepository,
-          CustomResourceRepository<ExecutionQueue> executionRepository,
-          K6LoadGeneratorStep k6LoadGeneratorStep) {
-    this.kubernetesClient = kubernetesClient;
-    this.stepRegister = stepRegister;
-    this.scenarioRepository = scenarioRepository;
-    this.executionRepository = executionRepository;
-    this.k6LoadGeneratorStep = k6LoadGeneratorStep;
-  }
-
-  private Optional<Item> getNextItem(ExecutionQueue queue) {
-    var items = queue.getSpec().getItems();
-    return items.stream().filter(Item::isPending).findFirst();
-  }
-
-  public void run(ExecutionQueue queue) {
-    var queueToExecute = executionRepository.find(queue.getMetadata())
-            .orElseThrow(() -> new RuntimeException("Queue not found " + queue.getMetadata().getName()));
-
-    var nextItem = getNextItem(queueToExecute);
-
-    if (nextItem.isPresent()) {
-      var namespace = queueToExecute.getMetadata().getNamespace();
-      if (nextItem.get().isPending()) {
-        if (!isRunning(namespace)) {
-          runScenario(namespace, nextItem.get().getScenario(), queueToExecute);
-        }
-      }
-    } else {
-      logger.info("No item available for queue: {}", queueToExecute.getMetadata().getName());
-      if (isAllFinished(queueToExecute)) {
-        logger.info("All items finished for: {}", queueToExecute.getMetadata().getName());
-      }
-    }
-  }
-
-  public boolean isAllFinished(ExecutionQueue queue) {
-    return queue.getSpec().getItems().stream().allMatch(Item::isFinished);
-  }
-
-  private boolean isRunning(String namespace) {
-    var jobs = kubernetesClient.batch().v1().jobs().inNamespace(namespace).list();
-    return jobs.getItems().stream().anyMatch(job ->
-            isNull(job.getStatus().getCompletionTime()) && job.getMetadata().getAnnotations().containsKey(SCENARIO));
-  }
-
-  private Job createLoadGenerationJob(Scenario scenario, ExecutionQueue executionQueue) {
-    return k6LoadGeneratorStep.execute(scenario, executionQueue);
-  }
-
-  private void runJob(Job job) {
-    var jobsClient = kubernetesClient.batch().v1().jobs();
-    var createdJob =
-            jobsClient.inNamespace(job.getMetadata().getNamespace()).withName(job.getMetadata().getName()).get();
-    if (createdJob != null) {
-      jobsClient.resource(createdJob).delete();
-    }
-    jobsClient.resource(job).create();
-    jobsClient.resource(job).watch(this);
-    logger.info("Job created: {}", job.getMetadata().getName());
-  }
-
-  private void runPreparationSteps(Scenario scenario, ExecutionQueue executionQueue) {
-    stepRegister.getPreparationSteps().forEach(step -> step.execute(scenario, executionQueue));
-  }
-
-  private void runScenario(String namespace, String scenarioName, ExecutionQueue executionQueue) {
-    logger.info("Running scenario: {}", scenarioName);
-    var scenario = scenarioRepository.find(namespace, scenarioName);
-    if (scenario.isPresent()) {
-      runPreparationSteps(scenario.get(), executionQueue);
-      var job = createLoadGenerationJob(scenario.get(), executionQueue);
-      runJob(job);
-    } else {
-      throw new RuntimeException(format("Scenario not found: %s.%s", namespace, scenarioName));
-    }
-  }
-
-  @Override
-  public void eventReceived(Action action, Job resource) { // TODO precisa melhorar esse método. mto emaranhado!
-    var namespace = resource.getMetadata().getNamespace();
-    if (action.equals(Action.MODIFIED)) {
-      if (nonNull(resource.getStatus().getCompletionTime())) {
-        logger.debug("Finished job: {}", resource.getMetadata().getName());
-        var scenarioName = resource.getMetadata().getAnnotations().get(SCENARIO);
-        var scenario = scenarioRepository.get(namespace, scenarioName);
-        var executionQueue = executionRepository.get(
-                namespace,
-                scenario.getMetadata().getAnnotations().get(Annotations.OWNED_BY)
-        );
-        stepRegister.getPostExecutionSteps().forEach(step -> step.execute(scenario, executionQueue));
-        run(executionQueue);
-      }
-    }
-  }
 
-  @Override
-  public void onClose(WatcherException cause) {
-    // TODO o que acontece pra cair aqui?
-  }
+public interface ScenarioExecutor {
+  void execute(Scenario scenario, ExecutionQueue executionQueue, Runnable onCompletion);
 }

resilience-bench/operator/src/main/java/io/resiliencebench/execution/steps/AbstractEnvironmentStep.java

@@ -16,7 +16,7 @@
 import java.util.List;
 import java.util.concurrent.TimeUnit;
 
-abstract class AbstractEnvironmentStep extends ExecutorStep<Deployment> {
+abstract class AbstractEnvironmentStep extends ExecutorStep {
 
   private final static Logger logger = LoggerFactory.getLogger(AbstractEnvironmentStep.class);
   protected final CustomResourceRepository<ResilientService> resilientServiceRepository;

resilience-bench/operator/src/main/java/io/resiliencebench/execution/steps/ApplicationReadinessStep.java

@@ -1,6 +1,5 @@
 package io.resiliencebench.execution.steps;
 
-import io.fabric8.kubernetes.api.model.apps.Deployment;
 import io.fabric8.kubernetes.client.KubernetesClient;
 import io.resiliencebench.resources.queue.ExecutionQueue;
 import io.resiliencebench.resources.scenario.Scenario;
@@ -27,7 +26,7 @@ protected boolean isApplicable(Scenario scenario) {
   }
 
   @Override
-  protected Deployment internalExecute(Scenario scenario, ExecutionQueue queue) {
+  protected void internalExecute(Scenario scenario, ExecutionQueue queue) {
     var ns = scenario.getMetadata().getNamespace();
     resilientServiceRepository.list(ns).forEach(resilientService -> {
       var deployment = getDeployment(scenario, resilientService);
@@ -37,6 +36,5 @@ protected Deployment internalExecute(Scenario scenario, ExecutionQueue queue) {
         waitUntilReady(deployment);
       }
     });
-    return null;
   }
 }