refactor(pipelines): handle queueing pipelines with Keiko

orca-core/src/main/java/com/netflix/spinnaker/orca/config/OrcaConfiguration.java

@@ -30,18 +30,17 @@
 import com.netflix.spinnaker.orca.listeners.ExecutionCleanupListener;
 import com.netflix.spinnaker.orca.listeners.ExecutionListener;
 import com.netflix.spinnaker.orca.listeners.MetricsExecutionListener;
-import com.netflix.spinnaker.orca.pipeline.*;
+import com.netflix.spinnaker.orca.pipeline.DefaultStageDefinitionBuilderFactory;
+import com.netflix.spinnaker.orca.pipeline.StageDefinitionBuilder;
+import com.netflix.spinnaker.orca.pipeline.StageDefinitionBuilderFactory;
 import com.netflix.spinnaker.orca.pipeline.persistence.ExecutionRepository;
-import com.netflix.spinnaker.orca.pipeline.persistence.PipelineStack;
-import com.netflix.spinnaker.orca.pipeline.persistence.memory.InMemoryPipelineStack;
 import com.netflix.spinnaker.orca.pipeline.util.ContextFunctionConfiguration;
 import com.netflix.spinnaker.orca.pipeline.util.ContextParameterProcessor;
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
 import org.springframework.boot.context.properties.ConfigurationProperties;
 import org.springframework.boot.context.properties.EnableConfigurationProperties;
-import org.springframework.context.ApplicationContext;
 import org.springframework.context.ApplicationListener;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.ComponentScan;
@@ -78,11 +77,6 @@ public class OrcaConfiguration {
     return OrcaObjectMapper.newInstance();
   }
 
-  @Bean @ConditionalOnMissingBean(name = "pipelineStack")
-  public PipelineStack pipelineStack() {
-    return new InMemoryPipelineStack();
-  }
-
   @Bean @Order(Ordered.LOWEST_PRECEDENCE)
   public DefaultExceptionHandler defaultExceptionHandler() {
     return new DefaultExceptionHandler();
@@ -97,16 +91,6 @@ public ApplicationListener<ExecutionEvent> executionCleanupListenerAdapter(Execu
     return new ExecutionListenerAdapter(executionCleanupListener, repository);
   }
 
-  @Bean
-  public PipelineStarterListener pipelineStarterListener(ExecutionRepository executionRepository, PipelineStartTracker startTracker, ApplicationContext applicationContext) {
-    return new PipelineStarterListener(executionRepository, startTracker, applicationContext);
-  }
-
-  @Bean
-  public ApplicationListener<ExecutionEvent> pipelineStarterListenerAdapter(PipelineStarterListener pipelineStarterListener, ExecutionRepository repository) {
-    return new ExecutionListenerAdapter(pipelineStarterListener, repository);
-  }
-
   @Bean
   @ConditionalOnProperty(value = "jarDiffs.enabled", matchIfMissing = false)
   public ComparableLooseVersion comparableLooseVersion() {

orca-core/src/main/java/com/netflix/spinnaker/orca/pipeline/ExecutionLauncher.java

@@ -53,7 +53,6 @@ public class ExecutionLauncher {
   private final ExecutionRepository executionRepository;
   private final ExecutionRunner executionRunner;
   private final Clock clock;
-  private final Optional<PipelineStartTracker> startTracker;
   private final Optional<PipelineValidator> pipelineValidator;
   private final Optional<Registry> registry;
 
@@ -63,14 +62,12 @@ public ExecutionLauncher(ObjectMapper objectMapper,
                            ExecutionRunner executionRunner,
                            Clock clock,
                            Optional<PipelineValidator> pipelineValidator,
-                           Optional<PipelineStartTracker> startTracker,
                            Optional<Registry> registry) {
     this.objectMapper = objectMapper;
     this.executionRepository = executionRepository;
     this.executionRunner = executionRunner;
     this.clock = clock;
     this.pipelineValidator = pipelineValidator;
-    this.startTracker = startTracker;
     this.registry = registry;
   }
 
@@ -118,12 +115,7 @@ private void checkRunnable(Execution execution) {
   }
 
   public Execution start(Execution execution) throws Exception {
-    if (shouldQueue(execution)) {
-      log.info("Queueing {}", execution.getId());
-    } else {
-      executionRunner.start(execution);
-      onExecutionStarted(execution);
-    }
+    executionRunner.start(execution);
     return execution;
   }
 
@@ -161,26 +153,9 @@ private Execution handleStartupFailure(Execution execution, Throwable failure) {
     log.error("Failed to start {} {}", execution.getType(), execution.getId(), failure);
     executionRepository.updateStatus(execution.getId(), status);
     executionRepository.cancel(execution.getId(), canceledBy, reason);
-    if (execution.getType() == PIPELINE) {
-      startTracker.ifPresent(tracker -> {
-        if (execution.getPipelineConfigId() != null) {
-          tracker.removeFromQueue(execution.getPipelineConfigId(), execution.getId());
-        }
-        tracker.markAsFinished(execution.getPipelineConfigId(), execution.getId());
-      });
-    }
     return executionRepository.retrieve(execution.getType(), execution.getId());
   }
 
-  private void onExecutionStarted(Execution execution) {
-    if (execution.getPipelineConfigId() != null) {
-      startTracker
-        .ifPresent(tracker -> {
-          tracker.addToStarted(execution.getPipelineConfigId(), execution.getId());
-        });
-    }
-  }
-
   private Execution parse(ExecutionType type, String configJson) throws IOException {
     if (type == PIPELINE) {
       return parsePipeline(configJson);
@@ -268,19 +243,4 @@ private final <E extends Enum<E>> E getEnum(Map<String, ?> map, String key, Clas
     String value = (String) map.get(key);
     return value != null ? Enum.valueOf(type, value) : null;
   }
-
-  /**
-   * Decide if this execution should be queued or start immediately.
-   *
-   * @return true if the stage should be queued.
-   */
-  private boolean shouldQueue(Execution execution) {
-    if (execution.getPipelineConfigId() == null || !execution.isLimitConcurrent()) {
-      return false;
-    }
-    return startTracker
-      .map(tracker ->
-        tracker.queueIfNotStarted(execution.getPipelineConfigId(), execution.getId()))
-      .orElse(false);
-  }
 }