Merge pull request #4053 from hashicorp/bug/pipeline_run/fix-complete…

…-pipeline Bug: Fix pipeline run complete state machine
hashicorp · Oct 17, 2022 · 5f63cd6 · 5f63cd6
2 parents 86ecafc + f50812b
commit 5f63cd6
Show file tree

Hide file tree

Showing 6 changed files with 39 additions and 12 deletions.
diff --git a/.changelog/4053.txt b/.changelog/4053.txt
@@ -0,0 +1,3 @@
+```release-note:bug
+pipelines: Properly mark a pipeline run as complete
+```
diff --git a/internal/server/boltdbstate/job.go b/internal/server/boltdbstate/job.go
@@ -1772,11 +1772,25 @@ func (s *State) pipelineComplete(jobId string) error {
 	if job.State == pb.Job_ERROR {
 		run.State = pb.PipelineRun_ERROR
 	} else if job.State == pb.Job_SUCCESS {
-		// If job Id matches last job queued by pipeline.
-		// We will have to change this in the future when pipeline steps run in parallel,
-		// and the last job queued may not be the last job to complete in the pipeline
-		// TODO:XX figure out how ^
-		if job.Id == run.Jobs[len(run.Jobs)-1].Id {
+		// Look at all job ids in a run and check if any are not SUCCESS
+		runComplete := true
+		for _, j := range run.Jobs {
+			if j.Id == job.Id {
+				continue
+			}
+
+			rj, err := s.JobById(j.Id, nil)
+			if err != nil {
+				return err
+			}
+
+			if rj.State != pb.Job_SUCCESS {
+				runComplete = false
+				break
+			}
+		}
+
+		if runComplete {
 			run.State = pb.PipelineRun_SUCCESS
 			s.log.Trace("pipeline run is complete", "job", job.Id, "pipeline", job.Pipeline.PipelineId, "run", run.Sequence)
 		}

diff --git a/pkg/server/gen/server.pb.go b/pkg/server/gen/server.pb.go
diff --git a/pkg/server/gen/server.swagger.json b/pkg/server/gen/server.swagger.json
@@ -9477,16 +9477,19 @@
           "description": "The sequence number for this pipeline run."
         },
         "pipeline": {
-          "$ref": "#/definitions/hashicorp.waypoint.Ref.Pipeline"
+          "$ref": "#/definitions/hashicorp.waypoint.Ref.Pipeline",
+          "description": "The pipeline associated with this run."
         },
         "jobs": {
           "type": "array",
           "items": {
             "$ref": "#/definitions/hashicorp.waypoint.Ref.Job"
-          }
+          },
+          "description": "The full list of jobs that are associated with this run."
         },
         "state": {
-          "$ref": "#/definitions/hashicorp.waypoint.PipelineRun.State"
+          "$ref": "#/definitions/hashicorp.waypoint.PipelineRun.State",
+          "title": "The current state of this pipeline run,"
         }
       }
     },

diff --git a/pkg/server/proto/server.proto b/pkg/server/proto/server.proto
@@ -4870,10 +4870,13 @@ message PipelineRun {
   // The sequence number for this pipeline run.
   uint64 sequence = 2;
 
+  // The pipeline associated with this run.
   Ref.Pipeline pipeline = 3;
 
+  // The full list of jobs that are associated with this run.
   repeated Ref.Job jobs = 4;
 
+  // The current state of this pipeline run,
   State state = 5;
 
   enum State {

diff --git a/pkg/server/singleprocess/service_pipeline.go b/pkg/server/singleprocess/service_pipeline.go
@@ -512,6 +512,7 @@ func (s *Service) buildStepJobs(
 			}
 		}
 
+		// Include a list of all associated jobs for this specific run
 		pipelineRun.Jobs = append(pipelineRun.Jobs, &pb.Ref_Job{Id: job.Id})
 		stepJobs = append(stepJobs, &pb.QueueJobRequest{
 			Job: job,