fix(queue): After trying to start a queued execution that is CANCELED…

…, check the pipeline queue for additional waiting executions (#3015)
spinnaker · Jun 26, 2019 · 2a8492b · 2a8492b
1 parent 0b42263
commit 2a8492b
Show file tree

Hide file tree

Showing 2 changed files with 33 additions and 1 deletion.
diff --git a/orca-queue/src/main/kotlin/com/netflix/spinnaker/orca/q/handler/StartExecutionHandler.kt b/orca-queue/src/main/kotlin/com/netflix/spinnaker/orca/q/handler/StartExecutionHandler.kt
@@ -28,6 +28,7 @@ import com.netflix.spinnaker.orca.pipeline.persistence.ExecutionRepository
 import com.netflix.spinnaker.orca.q.CancelExecution
 import com.netflix.spinnaker.orca.q.StartExecution
 import com.netflix.spinnaker.orca.q.StartStage
+import com.netflix.spinnaker.orca.q.StartWaitingExecutions
 import com.netflix.spinnaker.orca.q.pending.PendingExecutionService
 import com.netflix.spinnaker.q.Queue
 import net.logstash.logback.argument.StructuredArguments.value
@@ -99,6 +100,9 @@ class StartExecutionHandler(
   private fun terminate(execution: Execution) {
     if (execution.status == CANCELED || execution.isCanceled) {
       publisher.publishEvent(ExecutionComplete(this, execution.type, execution.id, execution.status))
+      execution.pipelineConfigId?.let {
+        queue.push(StartWaitingExecutions(it, purgeQueue = !execution.isKeepWaitingPipelines))
+      }
     } else {
       log.warn("Execution (type: ${execution.type}, id: {}, status: ${execution.status}, application: {})" +
         " cannot be started unless state is NOT_STARTED. Ignoring StartExecution message.",

diff --git a/orca-queue/src/test/kotlin/com/netflix/spinnaker/orca/q/handler/StartExecutionHandlerTest.kt b/orca-queue/src/test/kotlin/com/netflix/spinnaker/orca/q/handler/StartExecutionHandlerTest.kt
@@ -29,6 +29,7 @@ import com.netflix.spinnaker.orca.pipeline.persistence.ExecutionRepository
 import com.netflix.spinnaker.orca.q.CancelExecution
 import com.netflix.spinnaker.orca.q.StartExecution
 import com.netflix.spinnaker.orca.q.StartStage
+import com.netflix.spinnaker.orca.q.StartWaitingExecutions
 import com.netflix.spinnaker.orca.q.pending.PendingExecutionService
 import com.netflix.spinnaker.orca.q.singleTaskStage
 import com.netflix.spinnaker.q.Queue
@@ -140,7 +141,7 @@ object StartExecutionHandlerTest : SubjectSpek<StartExecutionHandler>({
       }
     }
 
-    given("a pipeline that was previously canceled and status is NOT_STARTED") {
+    given("a pipeline with no pipelineConfigId that was previously canceled and status is NOT_STARTED") {
       val pipeline = pipeline {
         stage {
           type = singleTaskStage.type
@@ -173,6 +174,33 @@ object StartExecutionHandlerTest : SubjectSpek<StartExecutionHandler>({
       }
     }
 
+    given("a pipeline with a pipelineConfigId that was previously canceled and status is NOT_STARTED") {
+      val pipeline = pipeline {
+        pipelineConfigId = "aaaaa-12345-bbbbb-67890"
+        isKeepWaitingPipelines = false
+        stage {
+          type = singleTaskStage.type
+        }
+        isCanceled = true
+      }
+
+      val message = StartExecution(pipeline)
+
+      beforeGroup {
+        whenever(repository.retrieve(message.executionType, message.executionId)) doReturn pipeline
+      }
+
+      afterGroup(::resetMocks)
+
+      on("receiving a message") {
+        subject.handle(message)
+      }
+
+      it("starts waiting executions for the pipelineConfigId") {
+        verify(queue).push(StartWaitingExecutions("aaaaa-12345-bbbbb-67890", true))
+      }
+    }
+
     given("a pipeline with multiple initial stages") {
       val pipeline = pipeline {
         stage {