fix(queue): fix ability to cancel a zombied execution (#4473) (#4478)

* fix(queue): fix ability to cancel a zombied execution * fix(queue): undo unintentional change * fix(queue): add more logging --------- Co-authored-by: mergify[bot] <37929162+mergify[bot]@users.noreply.github.com> (cherry picked from commit 56c7206) Co-authored-by: Matt <[email protected]> Co-authored-by: mergify[bot] <37929162+mergify[bot]@users.noreply.github.com>
spinnaker · Jun 29, 2023 · 2ff457f · 2ff457f
1 parent ba9ac0d
commit 2ff457f
Show file tree

Hide file tree

Showing 8 changed files with 88 additions and 10 deletions.
diff --git a/keiko-core/src/main/kotlin/com/netflix/spinnaker/q/Queue.kt b/keiko-core/src/main/kotlin/com/netflix/spinnaker/q/Queue.kt
@@ -62,7 +62,7 @@ interface Queue {
   fun push(message: Message, delay: TemporalAmount): Unit
 
   /**
-   * Update [message] if it exists for immediate delivery.
+   * Update [message] if it exists for immediate delivery. No-op if the [message] does not exist.
    */
   fun reschedule(message: Message): Unit = reschedule(message, ZERO)
 
@@ -87,6 +87,11 @@ interface Queue {
    */
   fun retry() {}
 
+  /**
+   * Used for testing zombie executions. Wipes all messages from the queue.
+   */
+  fun clear() {}
+
   /**
    * The expired time after which un-acknowledged messages will be retried.
    */

diff --git a/keiko-mem/src/main/kotlin/com/netflix/spinnaker/q/memory/InMemoryQueue.kt b/keiko-mem/src/main/kotlin/com/netflix/spinnaker/q/memory/InMemoryQueue.kt
@@ -131,6 +131,10 @@ class InMemoryQueue(
     }
   }
 
+  override fun clear() {
+    queue.removeAll { true }
+  }
+
   override fun readState() =
     QueueState(
       depth = queue.size,

diff --git a/keiko-redis/src/main/kotlin/com/netflix/spinnaker/q/redis/RedisQueue.kt b/keiko-redis/src/main/kotlin/com/netflix/spinnaker/q/redis/RedisQueue.kt
@@ -154,8 +154,13 @@ class RedisQueue(
       log.debug("Re-scheduling message: $message, fingerprint: $fingerprint to deliver in $delay")
       val status: Long = redis.zadd(queueKey, score(delay), fingerprint, zAddParams().xx())
       if (status.toInt() == 1) {
+        log.debug("Rescheduled message: $message, fingerprint: $fingerprint to deliver in $delay")
         fire(MessageRescheduled(message))
       } else {
+        log.warn(
+          "Failed to reschedule message: $message, fingerprint: $fingerprint, not found " +
+            "on queue"
+        )
         fire(MessageNotFound(message))
       }
     }
@@ -229,6 +234,12 @@ class RedisQueue(
     }
   }
 
+  override fun clear() {
+    pool.resource.use { redis ->
+      redis.del(messagesKey)
+    }
+  }
+
   override fun readState(): QueueState =
     pool.resource.use { redis ->
       redis.multi {

diff --git a/keiko-sql/src/main/kotlin/com/netflix/spinnaker/q/sql/SqlQueue.kt b/keiko-sql/src/main/kotlin/com/netflix/spinnaker/q/sql/SqlQueue.kt
@@ -764,6 +764,14 @@ class SqlQueue(
     fire(RetryPolled)
   }
 
+  override fun clear() {
+    withPool(poolName) {
+      withRetry(WRITE) {
+        jooq.deleteFrom(messagesTable).execute()
+      }
+    }
+  }
+
   @Scheduled(fixedDelayString = "\${queue.cleanup.frequency.ms:2000}")
   fun cleanupMessages() {
     withPool(poolName) {

diff --git a/orca-queue-tck/src/main/kotlin/com/netflix/spinnaker/orca/q/ExecutionLatch.kt b/orca-queue-tck/src/main/kotlin/com/netflix/spinnaker/orca/q/ExecutionLatch.kt
@@ -46,6 +46,18 @@ class ExecutionLatch(private val predicate: Predicate<ExecutionComplete>) :
   fun await() = latch.await(10, TimeUnit.SECONDS)
 }
 
+fun ConfigurableApplicationContext.run(execution: PipelineExecution, launcher: (PipelineExecution) -> Unit) {
+  val latch = ExecutionLatch(
+    Predicate {
+      it.executionId == execution.id
+    }
+  )
+  addApplicationListener(latch)
+  launcher.invoke(execution)
+
+  Thread.sleep(500)
+}
+
 fun ConfigurableApplicationContext.runToCompletion(execution: PipelineExecution, launcher: (PipelineExecution) -> Unit, repository: ExecutionRepository) {
   val latch = ExecutionLatch(
     Predicate {

diff --git a/orca-queue-tck/src/main/kotlin/com/netflix/spinnaker/orca/q/QueueIntegrationTest.kt b/orca-queue-tck/src/main/kotlin/com/netflix/spinnaker/orca/q/QueueIntegrationTest.kt
@@ -91,6 +91,8 @@ import org.springframework.context.event.ApplicationEventMulticaster
 import org.springframework.context.event.SimpleApplicationEventMulticaster
 import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor
 import org.springframework.test.context.junit4.SpringRunner
+import java.util.concurrent.TimeUnit
+import java.util.function.Predicate
 
 @SpringBootTest(
   classes = [TestConfig::class],
@@ -849,6 +851,35 @@ abstract class QueueIntegrationTest {
       }
     }
   }
+
+  @Test
+  fun `cancelling a zombied execution sets task, stage and execution statuses to CANCELED`() {
+    val pipeline = pipeline {
+      application = "spinnaker"
+      stage {
+        refId = "1"
+        type = "dummy"
+      }
+    }
+    repository.store(pipeline)
+
+    whenever(dummyTask.execute(any())) doAnswer {
+      TaskResult.RUNNING
+    }
+
+    context.run(pipeline, runner::start)
+
+    // simulate a zombie by clearing the queue
+    queue.clear()
+
+    context.runToCompletion(pipeline, { runner.cancel(it, "anonymous", null) }, repository)
+
+    repository.retrieve(PIPELINE, pipeline.id).apply {
+      assertThat(status).isEqualTo(CANCELED)
+      assertThat(stageByRef("1").status).isEqualTo(CANCELED)
+      assertThat(stageByRef("1").tasks.all { it.status == CANCELED })
+    }
+  }
 }
 
 @Configuration

diff --git a/...-queue/src/main/kotlin/com/netflix/spinnaker/orca/q/handler/RescheduleExecutionHandler.kt b/...-queue/src/main/kotlin/com/netflix/spinnaker/orca/q/handler/RescheduleExecutionHandler.kt
@@ -23,6 +23,7 @@ import com.netflix.spinnaker.orca.q.RescheduleExecution
 import com.netflix.spinnaker.orca.q.RunTask
 import com.netflix.spinnaker.q.Queue
 import org.springframework.stereotype.Component
+import java.time.Duration
 
 @Component
 class RescheduleExecutionHandler(
@@ -43,14 +44,14 @@ class RescheduleExecutionHandler(
           stage.tasks
             .filter { it.status == ExecutionStatus.RUNNING }
             .forEach {
-              queue.reschedule(
-                RunTask(
-                  message,
-                  stage.id,
-                  it.id,
-                  taskResolver.getTaskClass(it.implementingClass)
-                )
+              val taskMessage = RunTask(
+                message,
+                stage.id,
+                it.id,
+                taskResolver.getTaskClass(it.implementingClass)
               )
+              queue.ensure(taskMessage, Duration.ZERO)
+              queue.reschedule(taskMessage)
             }
         }
     }

diff --git a/...ue/src/test/kotlin/com/netflix/spinnaker/orca/q/handler/RescheduleExecutionHandlerTest.kt b/...ue/src/test/kotlin/com/netflix/spinnaker/orca/q/handler/RescheduleExecutionHandlerTest.kt
@@ -37,6 +37,7 @@ import org.jetbrains.spek.api.dsl.describe
 import org.jetbrains.spek.api.dsl.it
 import org.jetbrains.spek.api.lifecycle.CachingMode
 import org.jetbrains.spek.subject.SubjectSpek
+import java.time.Duration
 
 object RescheduleExecutionHandlerTest : SubjectSpek<RescheduleExecutionHandler>({
 
@@ -101,8 +102,13 @@ object RescheduleExecutionHandlerTest : SubjectSpek<RescheduleExecutionHandler>(
       val task4 = stage2a.taskById("4")
       val task5 = stage2b.taskById("5")
 
-      verify(queue).reschedule(RunTask(message, stage2a.id, task4.id, Class.forName(task4.implementingClass) as Class<out Task>))
-      verify(queue).reschedule(RunTask(message, stage2b.id, task5.id, Class.forName(task5.implementingClass) as Class<out Task>))
+      val messageTask4 = RunTask(message, stage2a.id, task4.id, Class.forName(task4.implementingClass) as Class<out Task>)
+      val messageTask5 = RunTask(message, stage2b.id, task5.id, Class.forName(task5.implementingClass) as Class<out Task>)
+
+      verify(queue).ensure(messageTask4, Duration.ZERO)
+      verify(queue).reschedule(messageTask4)
+      verify(queue).ensure(messageTask5, Duration.ZERO)
+      verify(queue).reschedule(messageTask5)
       verifyNoMoreInteractions(queue)
     }
   }