fix(execution): Resume parent pipeline when a failed stage in a child…

… pipeline restarts (#3317) (#3340) * restart parent stage when child restarts * fix based on code review * remove unused function * better method name * improvements based on feedback * fix static analysis error * replace terminal by iscomplete * fetch the live version of the parent execution, and improve variable names
spinnaker · Dec 12, 2019 · b88f62a · b88f62a
1 parent e357a67
commit b88f62a
Show file tree

Hide file tree

Showing 3 changed files with 49 additions and 5 deletions.
diff --git a/orca-core/src/main/java/com/netflix/spinnaker/orca/pipeline/model/StageContext.java b/orca-core/src/main/java/com/netflix/spinnaker/orca/pipeline/model/StageContext.java
@@ -71,7 +71,7 @@ public Object get(@Nullable Object key) {
    * @param defaultValue default value to return if key is not present
    * @return value or null if not present
    */
-  Object getCurrentOnly(@Nullable Object key, Object defaultValue) {
+  public Object getCurrentOnly(@Nullable Object key, Object defaultValue) {
     return super.getOrDefault(key, defaultValue);
   }
   /*

diff --git a/orca-front50/src/main/groovy/com/netflix/spinnaker/orca/front50/pipeline/PipelineStage.java b/orca-front50/src/main/groovy/com/netflix/spinnaker/orca/front50/pipeline/PipelineStage.java
@@ -26,6 +26,7 @@
 import com.netflix.spinnaker.orca.pipeline.TaskNode;
 import com.netflix.spinnaker.orca.pipeline.model.Execution;
 import com.netflix.spinnaker.orca.pipeline.model.Stage;
+import com.netflix.spinnaker.orca.pipeline.model.StageContext;
 import com.netflix.spinnaker.orca.pipeline.persistence.ExecutionRepository;
 import com.netflix.spinnaker.orca.pipeline.tasks.artifacts.BindProducedArtifactsTask;
 import org.slf4j.Logger;
@@ -46,7 +47,10 @@ public class PipelineStage implements StageDefinitionBuilder, CancellableStage {
 
   @Override
   public void taskGraph(Stage stage, TaskNode.Builder builder) {
-    builder.withTask("startPipeline", StartPipelineTask.class);
+    // only start the pipeline if no execution ID already exists
+    if (stage.getContext().get("executionId") == null) {
+      builder.withTask("startPipeline", StartPipelineTask.class);
+    }
 
     if (!stage
         .getContext()
@@ -64,9 +68,19 @@ public void taskGraph(Stage stage, TaskNode.Builder builder) {
 
   @Override
   public void prepareStageForRestart(Stage stage) {
-    stage.getContext().remove("status");
-    stage.getContext().remove("executionName");
-    stage.getContext().remove("executionId");
+    StageContext context = (StageContext) stage.getContext();
+
+    context.remove("status");
+
+    boolean skipPipelineRestart = (boolean) context.getCurrentOnly("_skipPipelineRestart", false);
+    if (!skipPipelineRestart) {
+      stage.getContext().remove("executionName");
+      stage.getContext().remove("executionId");
+    } else {
+      // Keep the execution details in case the inner pipeline got restarted
+      // Clear the skip restart flag
+      stage.getContext().remove("_skipPipelineRestart");
+    }
   }
 
   @Override

diff --git a/orca-queue/src/main/kotlin/com/netflix/spinnaker/orca/q/handler/RestartStageHandler.kt b/orca-queue/src/main/kotlin/com/netflix/spinnaker/orca/q/handler/RestartStageHandler.kt
@@ -19,6 +19,7 @@ package com.netflix.spinnaker.orca.q.handler
 import com.netflix.spinnaker.orca.ExecutionStatus.NOT_STARTED
 import com.netflix.spinnaker.orca.ExecutionStatus.RUNNING
 import com.netflix.spinnaker.orca.pipeline.StageDefinitionBuilderFactory
+import com.netflix.spinnaker.orca.pipeline.model.PipelineTrigger
 import com.netflix.spinnaker.orca.pipeline.model.Stage
 import com.netflix.spinnaker.orca.pipeline.persistence.ExecutionRepository
 import com.netflix.spinnaker.orca.q.RestartStage
@@ -58,13 +59,42 @@ class RestartStageHandler(
         if (topStage.status.isComplete) {
           topStage.addRestartDetails(message.user)
           topStage.reset()
+          restartParentPipelineIfNeeded(message, topStage)
           repository.updateStatus(topStage.execution.type, topStage.execution.id, RUNNING)
           queue.push(StartStage(startMessage))
         }
       }
     }
   }
 
+  private fun restartParentPipelineIfNeeded(message: RestartStage, topStage: Stage) {
+    if (topStage.execution.trigger !is PipelineTrigger) {
+      return
+    }
+
+    val trigger = topStage.execution.trigger as PipelineTrigger
+    // We have a copy of the parent execution, not the live one. So we retrieve the live one.
+    val parentExecution = repository.retrieve(trigger.parentExecution.type, trigger.parentExecution.id)
+
+    if (!parentExecution.status.isComplete()) {
+      // only attempt to restart the parent pipeline if it's not running
+      return
+    }
+
+    val parentStage = parentExecution.stageById(trigger.parentPipelineStageId)
+    parentStage.addSkipRestart()
+    repository.storeStage(parentStage)
+
+    queue.push(RestartStage(trigger.parentExecution, parentStage.id, message.user))
+  }
+
+  /**
+   * Inform the parent stage when it restarts that the child is already running
+   */
+  private fun Stage.addSkipRestart() {
+    context["_skipPipelineRestart"] = true
+  }
+
   private fun Stage.addRestartDetails(user: String?) {
     context["restartDetails"] = mapOf(
       "restartedBy" to (user ?: "anonymous"),