fix(queue): prevent echo listener from overwriting data from task sta…

…rting
spinnaker · May 11, 2017 · ba9d1bd · ba9d1bd
1 parent 3abaa9f
commit ba9d1bd
Show file tree

Hide file tree

Showing 10 changed files with 37 additions and 12 deletions.
diff --git a/orca-core/src/main/groovy/com/netflix/spinnaker/orca/pipeline/model/Stage.java b/orca-core/src/main/groovy/com/netflix/spinnaker/orca/pipeline/model/Stage.java
@@ -147,6 +147,14 @@ public Stage(T execution, String type) {
   @JsonIgnore
   private final AtomicInteger stageCounter = new AtomicInteger(0);
 
+  public Task taskById(String taskId) {
+    return tasks
+      .stream()
+      .filter(it -> it.getId().equals(taskId))
+      .findFirst()
+      .orElse(null);
+  }
+
   /**
    * Gets the last stage preceding this stage that has the specified type.
    */

diff --git a/.../src/main/groovy/com/netflix/spinnaker/orca/pipeline/persistence/ExecutionRepository.java b/.../src/main/groovy/com/netflix/spinnaker/orca/pipeline/persistence/ExecutionRepository.java
@@ -19,6 +19,8 @@ public interface ExecutionRepository {
 
   void storeStage(Stage<? extends Execution> stage);
 
+  void updateStageContext(Stage<? extends Execution> stage);
+
   void removeStage(Execution execution, String stageId);
 
   void addStage(Stage stage);

diff --git a/...ovy/com/netflix/spinnaker/orca/pipeline/persistence/jedis/JedisExecutionRepository.groovy b/...ovy/com/netflix/spinnaker/orca/pipeline/persistence/jedis/JedisExecutionRepository.groovy
@@ -246,6 +246,16 @@ class JedisExecutionRepository implements ExecutionRepository {
     }
   }
 
+  @Override
+  void updateStageContext(Stage<? extends Execution> stage) {
+    def execution = stage.execution
+    def type = execution.getClass().simpleName.toLowerCase()
+    def key = "${type}:${execution.id}"
+    withJedis(getJedisPoolForId(key)) { Jedis jedis ->
+      jedis.hset(key, "stage.${stage.id}.context", mapper.writeValueAsString(stage.context))
+    }
+  }
+
   @Override
   void removeStage(Execution execution, String stageId) {
     Class<? extends Execution> executionType = execution.getClass()

diff --git a/orca-echo/src/main/groovy/com/netflix/spinnaker/orca/echo/config/EchoConfiguration.groovy b/orca-echo/src/main/groovy/com/netflix/spinnaker/orca/echo/config/EchoConfiguration.groovy
@@ -73,8 +73,8 @@ class EchoConfiguration {
   }
 
   @Bean
-  EchoNotifyingStageListener echoNotifyingStageExecutionListener(EchoService echoService) {
-    new EchoNotifyingStageListener(echoService)
+  EchoNotifyingStageListener echoNotifyingStageExecutionListener(EchoService echoService, ExecutionRepository repository) {
+    new EchoNotifyingStageListener(echoService, repository)
   }
 
   @Bean

diff --git a/.../src/main/groovy/com/netflix/spinnaker/orca/echo/spring/EchoNotifyingStageListener.groovy b/.../src/main/groovy/com/netflix/spinnaker/orca/echo/spring/EchoNotifyingStageListener.groovy
@@ -5,6 +5,7 @@ import com.netflix.spinnaker.orca.echo.EchoService
 import com.netflix.spinnaker.orca.listeners.Persister
 import com.netflix.spinnaker.orca.listeners.StageListener
 import com.netflix.spinnaker.orca.pipeline.model.*
+import com.netflix.spinnaker.orca.pipeline.persistence.ExecutionRepository
 import groovy.transform.CompileDynamic
 import groovy.transform.CompileStatic
 import groovy.util.logging.Slf4j
@@ -21,10 +22,12 @@ import static java.lang.System.currentTimeMillis
 class EchoNotifyingStageListener implements StageListener {
 
   private final EchoService echoService
+  private final ExecutionRepository repository
 
   @Autowired
-  EchoNotifyingStageListener(EchoService echoService) {
+  EchoNotifyingStageListener(EchoService echoService, ExecutionRepository repository) {
     this.echoService = echoService
+    this.repository = repository
   }
 
   @Override
@@ -50,7 +53,7 @@ class EchoNotifyingStageListener implements StageListener {
         isSynthetic: stage.syntheticStageOwner != null
       ]
       stage.context.stageDetails = details
-      persister.save(stage)
+      repository.updateStageContext(stage)
 
       log.debug("***** $stage.execution.id Echo stage $stage.name starting v2")
       recordEvent("stage", "starting", stage)
@@ -78,7 +81,7 @@ class EchoNotifyingStageListener implements StageListener {
       if (stage.endTime) {
         stage.context.stageDetails.endTime = stage.endTime
       }
-      persister.save(stage)
+      repository.updateStageContext(stage)
 
       // STOPPED stages are "successful" because they allow the pipeline to
       // proceed but they are still failures in terms of the stage and should

diff --git a/...ho/src/test/groovy/com/netflix/spinnaker/orca/echo/spring/EchoEventIntegrationSpec.groovy b/...ho/src/test/groovy/com/netflix/spinnaker/orca/echo/spring/EchoEventIntegrationSpec.groovy
@@ -192,8 +192,8 @@ abstract class EchoEventIntegrationSpec<R extends ExecutionRunner> extends Speci
     }
 
     @Bean
-    EchoNotifyingStageListener echoNotifyingStageListener(EchoService echoService) {
-      new EchoNotifyingStageListener(echoService)
+    EchoNotifyingStageListener echoNotifyingStageListener(EchoService echoService, ExecutionRepository repository) {
+      new EchoNotifyingStageListener(echoService, repository)
     }
 
     @Bean

diff --git a/.../test/groovy/com/netflix/spinnaker/orca/echo/spring/EchoNotifyingStageListenerSpec.groovy b/.../test/groovy/com/netflix/spinnaker/orca/echo/spring/EchoNotifyingStageListenerSpec.groovy
@@ -6,6 +6,7 @@ import com.netflix.spinnaker.orca.pipeline.model.Orchestration
 import com.netflix.spinnaker.orca.pipeline.model.Pipeline
 import com.netflix.spinnaker.orca.pipeline.model.Stage
 import com.netflix.spinnaker.orca.pipeline.model.Task
+import com.netflix.spinnaker.orca.pipeline.persistence.ExecutionRepository
 import spock.lang.Shared
 import spock.lang.Specification
 import spock.lang.Subject
@@ -16,9 +17,10 @@ class EchoNotifyingStageListenerSpec extends Specification {
 
   def echoService = Mock(EchoService)
   def persister = Stub(Persister)
+  def repository = Mock(ExecutionRepository)
 
   @Subject
-  def echoListener = new EchoNotifyingStageListener(echoService)
+  def echoListener = new EchoNotifyingStageListener(echoService, repository)
 
   @Shared
   def pipelineStage = new Stage<>(new Pipeline(), "test", "test", [:])

diff --git a/orca-queue-redis/src/main/kotlin/com/netflix/spinnaker/orca/q/redis/RedisQueue.kt b/orca-queue-redis/src/main/kotlin/com/netflix/spinnaker/orca/q/redis/RedisQueue.kt
@@ -168,6 +168,7 @@ class RedisQueue(
   private fun Jedis.pop(from: String, to: String, delay: TemporalAmount = ZERO) =
     zrangeByScore(from, 0.0, score(), 0, 1)
       .takeIf {
+        // TODO: this isn't right, `it` is a set (often an empty one)
         getSet("$locksKey:$it", currentInstanceId) in listOf(null, currentInstanceId)
       }
       ?.also {

diff --git a/orca-queue/src/main/kotlin/com/netflix/spinnaker/orca/q/MessageHandler.kt b/orca-queue/src/main/kotlin/com/netflix/spinnaker/orca/q/MessageHandler.kt
@@ -42,8 +42,7 @@ interface MessageHandler<M : Message> : (Message) -> Unit {
   fun TaskLevel.withTask(block: (Stage<*>, Task) -> Unit) =
     withStage { stage ->
       stage
-        .getTasks()
-        .find { it.id == taskId }
+        .taskById(taskId)
         .let { task ->
           if (task == null) {
             queue.push(InvalidTaskId(this))
@@ -56,8 +55,7 @@ interface MessageHandler<M : Message> : (Message) -> Unit {
   fun StageLevel.withStage(block: (Stage<*>) -> Unit) =
     withExecution { execution ->
       execution
-        .getStages()
-        .find { it.getId() == stageId }
+        .stageById(stageId)
         .let { stage ->
           if (stage == null) {
             queue.push(InvalidStageId(this))

diff --git a/orca-queue/src/main/kotlin/com/netflix/spinnaker/orca/q/handler/StartTaskHandler.kt b/orca-queue/src/main/kotlin/com/netflix/spinnaker/orca/q/handler/StartTaskHandler.kt
@@ -43,6 +43,7 @@ open class StartTaskHandler
       task.status = RUNNING
       task.startTime = clock.millis()
       repository.storeStage(stage)
+
       queue.push(RunTask(message, task.id, task.type))
 
       publisher.publishEvent(TaskStarted(this, stage, task))