Added flag to gh-trigger-workflow tool to wait for pre-existing runs (#24696)

* Added flag to `gh-trigger-workflow tool` to wait for pre-existing runs

* Addressed PR comments

* workflowId -> workflowID

* Id -> ID

.drone.yml

@@ -6603,7 +6603,7 @@ steps:
   commands:
   - cd "/go/src/github.com/gravitational/teleport/build.assets/tooling"
   - 'go run ./cmd/gh-trigger-workflow -owner ${DRONE_REPO_OWNER} -repo teleport.e
-    -timeout 12h0m0s -workflow deploy-packages.yaml -workflow-ref=refs/heads/master
+    -tag-workflow -series-run -timeout 12h0m0s -workflow deploy-packages.yaml -workflow-ref=refs/heads/master
     -input "artifact-tag=${DRONE_TAG}" -input "environment=$(cat "/go/vars/release-environment.txt")"
     -input "package-name-filter=$($DRONE_REPO_PRIVATE && echo "*ent*" || echo "")"
     -input "package-to-test=teleport-ent" -input "release-channel=stable" -input "repo-type=apt"
@@ -6671,7 +6671,7 @@ steps:
   commands:
   - cd "/go/src/github.com/gravitational/teleport/build.assets/tooling"
   - 'go run ./cmd/gh-trigger-workflow -owner ${DRONE_REPO_OWNER} -repo teleport.e
-    -timeout 12h0m0s -workflow deploy-packages.yaml -workflow-ref=refs/heads/master
+    -tag-workflow -series-run -timeout 12h0m0s -workflow deploy-packages.yaml -workflow-ref=refs/heads/master
     -input "artifact-tag=${DRONE_TAG}" -input "environment=$(cat "/go/vars/release-environment.txt")"
     -input "package-name-filter=$($DRONE_REPO_PRIVATE && echo "*ent*" || echo "")"
     -input "package-to-test=teleport-ent" -input "release-channel=stable" -input "repo-type=yum"
@@ -6739,7 +6739,7 @@ steps:
   commands:
   - cd "/go/src/github.com/gravitational/teleport/build.assets/tooling"
   - 'go run ./cmd/gh-trigger-workflow -owner ${DRONE_REPO_OWNER} -repo teleport.e
-    -timeout 12h0m0s -workflow deploy-packages.yaml -workflow-ref=refs/heads/master
+    -tag-workflow -series-run -timeout 12h0m0s -workflow deploy-packages.yaml -workflow-ref=refs/heads/master
     -input "artifact-tag=${DRONE_TAG}" -input "environment=$(cat "/go/vars/release-environment.txt")"
     -input "package-name-filter=teleport-ent-updater*" -input "release-channel=stable"
     -input "repo-type=apt" -input "version-channel=cloud" '
@@ -6806,7 +6806,7 @@ steps:
   commands:
   - cd "/go/src/github.com/gravitational/teleport/build.assets/tooling"
   - 'go run ./cmd/gh-trigger-workflow -owner ${DRONE_REPO_OWNER} -repo teleport.e
-    -timeout 12h0m0s -workflow deploy-packages.yaml -workflow-ref=refs/heads/master
+    -tag-workflow -series-run -timeout 12h0m0s -workflow deploy-packages.yaml -workflow-ref=refs/heads/master
     -input "artifact-tag=${DRONE_TAG}" -input "environment=$(cat "/go/vars/release-environment.txt")"
     -input "package-name-filter=teleport-ent-updater*" -input "release-channel=stable"
     -input "repo-type=yum" -input "version-channel=cloud" '
@@ -20232,6 +20232,6 @@ image_pull_secrets:
 - DOCKERHUB_CREDENTIALS
 ---
 kind: signature
-hmac: 2d735b23ccff9dfe334b4defcfcd95a03e2a6ccec8d40b4aa0b1a9b3d2a94b25
+hmac: 88c1a612809e190c5b9ebfe10e6df274532105a408af9d6583823a530598fddd
 
 ...

build.assets/tooling/cmd/gh-trigger-workflow/args.go

@@ -74,6 +74,7 @@ type args struct {
 	workflow       string
 	workflowRef    string
 	useWorkflowTag bool
+	seriesRun      bool
 	timeout        time.Duration
 	inputs         inputMap
 }
@@ -93,6 +94,7 @@ func parseCommandLine() (args, error) {
 	flag.StringVar(&cliArgs.workflow, "workflow", "", "Path to workflow")
 	flag.StringVar(&cliArgs.workflowRef, "workflow-ref", cliArgs.workflowRef, "Revision reference")
 	flag.BoolVar(&cliArgs.useWorkflowTag, "tag-workflow", false, "Use a workflow input to tag and ID workflows spawned by the event")
+	flag.BoolVar(&cliArgs.seriesRun, "series-run", false, "Attempts to wait for any workflows scheduled but not completed before starting this one")
 	flag.DurationVar(&cliArgs.timeout, "timeout", time.Duration(0), "Timeout. If not specified, waits forever.")
 	flag.Var(cliArgs.inputs, "input", "Input to target workflow")
 

build.assets/tooling/cmd/gh-trigger-workflow/main.go

@@ -91,6 +91,13 @@ func main() {
 
 	gh := ghapi.NewClient(&http.Client{Transport: tx})
 
+	if args.seriesRun {
+		err := waitForActiveWorkflowRuns(ctx, gh, args)
+		if err != nil {
+			log.Fatalf("Failed to wait for existing workflow runs: %s", err)
+		}
+	}
+
 	dispatchCtx, cancelDispatch := context.WithTimeout(ctx, 1*time.Minute)
 	defer cancelDispatch()
 
@@ -99,7 +106,7 @@ func main() {
 	// our dispatch event. Note that we pick a time slightly in the past to handle
 	// any clock skew.
 	baselineTime := time.Now().Add(-2 * time.Minute)
-	oldRuns, err := github.ListWorkflowRuns(dispatchCtx, gh.Actions, args.owner, args.repo, args.workflow, getBranchForRef(args.workflowRef), baselineTime)
+	oldRuns, err := github.ListWorkflowRunIDs(dispatchCtx, gh.Actions, args.owner, args.repo, args.workflow, getBranchForRef(args.workflowRef), baselineTime)
 	if err != nil {
 		log.Fatalf("Failed to fetch initial task list: %s", err)
 	}
@@ -130,7 +137,7 @@ func main() {
 	}
 	log.Printf("Workflow run: %s", run.GetHTMLURL())
 
-	conclusion, err := github.WaitForRun(ctx, gh.Actions, args.owner, args.repo, args.workflow, args.workflowRef, run.GetID())
+	conclusion, err := github.WaitForRun(ctx, gh.Actions, args.owner, args.repo, args.workflow, run.GetID())
 	if err != nil {
 		log.Fatalf("Failed to wait for run to exit %s", err)
 	}
@@ -172,6 +179,48 @@ func lookupInstallationID(ctx context.Context, args args) (int64, error) {
 	return installationID, nil
 }
 
+// Returns the first incomplete matching workflow run found. If none are found, returns nil.
+func getIncompleteWorkflowRunID(ctx context.Context, gh *ghapi.Client, args args) (*ghapi.WorkflowRun, error) {
+	// If there are runs lasting longer than one hour then there is a probably a much larger problem at play
+	recentRuns, err := github.ListWorkflowRuns(ctx, gh.Actions, args.owner, args.repo, args.workflow, "", time.Now().Add(-time.Hour))
+	if err != nil {
+		return nil, trace.Wrap(err, "failed to get a list of current workflow runs")
+	}
+
+	for _, recentRun := range recentRuns {
+		runStatus := recentRun.GetStatus()
+		if runStatus == "" {
+			return nil, trace.Errorf("failed to get status for run ID %q", recentRun.GetID())
+		}
+
+		if runStatus != "completed" {
+			return recentRun, nil
+		}
+	}
+
+	return nil, nil
+}
+
+func waitForActiveWorkflowRuns(ctx context.Context, gh *ghapi.Client, args args) error {
+	for {
+		incompleteWorkflowRun, err := getIncompleteWorkflowRunID(ctx, gh, args)
+		if err != nil {
+			return trace.Wrap(err, "failed to check if workflow has pending runs")
+		}
+
+		if incompleteWorkflowRun == nil {
+			return nil
+		}
+
+		workflowID := incompleteWorkflowRun.GetID()
+		log.Printf("Waiting on pre-existing incomplete run: %s", incompleteWorkflowRun.GetHTMLURL())
+		_, err = github.WaitForRun(ctx, gh.Actions, args.owner, args.repo, args.workflow, workflowID)
+		if err != nil {
+			return trace.Wrap(err, "failed to wait for workflow run %d to complete", workflowID)
+		}
+	}
+}
+
 func waitForNewWorkflowRun(ctx context.Context, gh *ghapi.Client, args args, tag string, baselineTime time.Time, existingRuns github.RunIDSet) (*ghapi.WorkflowRun, error) {
 	// Now we need to wait and see if a new workflow is spawned
 	ticker := time.NewTicker(1 * time.Second)
@@ -183,7 +232,7 @@ func waitForNewWorkflowRun(ctx context.Context, gh *ghapi.Client, args args, tag
 			log.Fatal("Timed out waiting for workflow run to start")
 
 		case <-ticker.C:
-			newRuns, err := github.ListWorkflowRuns(ctx, gh.Actions, args.owner, args.repo, args.workflow, getBranchForRef(args.workflowRef), baselineTime)
+			newRuns, err := github.ListWorkflowRunIDs(ctx, gh.Actions, args.owner, args.repo, args.workflow, getBranchForRef(args.workflowRef), baselineTime)
 			if err != nil {
 				return nil, trace.Wrap(err, "Failed polling for new workflow runs")
 			}

build.assets/tooling/lib/github/workflows.go

@@ -62,9 +62,8 @@ type WorkflowRuns interface {
 	ListWorkflowRunsByFileName(ctx context.Context, owner, repo, workflowFileName string, opts *github.ListWorkflowRunsOptions) (*github.WorkflowRuns, *github.Response, error)
 }
 
-// ListWorkflowRuns returns a set of RunIDs, representing the set of all for
-// workflow runs created since the supplied start time.
-func ListWorkflowRuns(ctx context.Context, actions WorkflowRuns, owner, repo, path, branch string, since time.Time) (RunIDSet, error) {
+// Returns information about all matched runs started after `since`.
+func ListWorkflowRuns(ctx context.Context, actions WorkflowRuns, owner, repo, path, branch string, since time.Time) ([]*github.WorkflowRun, error) {
 	listOptions := github.ListWorkflowRunsOptions{
 		ListOptions: github.ListOptions{
 			PerPage: 100,
@@ -73,7 +72,7 @@ func ListWorkflowRuns(ctx context.Context, actions WorkflowRuns, owner, repo, pa
 		Created: ">" + since.Format(time.RFC3339),
 	}
 
-	runIDs := make(RunIDSet)
+	allRuns := make([]*github.WorkflowRun, 0)
 
 	for {
 		runs, resp, err := actions.ListWorkflowRunsByFileName(ctx, owner, repo, path, &listOptions)
@@ -81,9 +80,7 @@ func ListWorkflowRuns(ctx context.Context, actions WorkflowRuns, owner, repo, pa
 			return nil, trace.Wrap(err, "Failed to fetch runs")
 		}
 
-		for _, r := range runs.WorkflowRuns {
-			runIDs.Insert(r.GetID())
-		}
+		allRuns = append(allRuns, runs.WorkflowRuns...)
 
 		if resp.NextPage == 0 {
 			break
@@ -92,6 +89,22 @@ func ListWorkflowRuns(ctx context.Context, actions WorkflowRuns, owner, repo, pa
 		listOptions.Page = resp.NextPage
 	}
 
+	return allRuns, nil
+}
+
+// ListWorkflowRunIDs returns a set of RunIDs, representing the set of all for
+// workflow runs created since the supplied start time.
+func ListWorkflowRunIDs(ctx context.Context, actions WorkflowRuns, owner, repo, path, branch string, since time.Time) (RunIDSet, error) {
+	workflowRuns, err := ListWorkflowRuns(ctx, actions, owner, repo, path, branch, since)
+	if err != nil {
+		return nil, trace.Wrap(err, "failed to get a list of workflow runs")
+	}
+
+	runIDs := make(RunIDSet, len(workflowRuns))
+	for _, workflowRun := range workflowRuns {
+		runIDs.Insert(workflowRun.GetID())
+	}
+
 	return runIDs, nil
 }
 
@@ -126,7 +139,7 @@ func ListWorkflowJobs(ctx context.Context, lister WorkflowJobLister, owner, repo
 
 // WaitForRun blocks until the specified workflow run completes, and returns the overall
 // workflow status.
-func WaitForRun(ctx context.Context, actions WorkflowRuns, owner, repo, path, ref string, runID int64) (string, error) {
+func WaitForRun(ctx context.Context, actions WorkflowRuns, owner, repo, path string, runID int64) (string, error) {
 	ticker := time.NewTicker(30 * time.Second)
 	defer ticker.Stop()
 

dronegen/gha.go

@@ -34,6 +34,7 @@ type ghaBuildType struct {
 	slackOnError      bool
 	dependsOn         []string
 	shouldTagWorkflow bool
+	seriesRun         bool
 	inputs            map[string]string
 }
 
@@ -52,6 +53,10 @@ func ghaBuildPipeline(b ghaBuildType) pipeline {
 		cmd.WriteString(`-tag-workflow `)
 	}
 
+	if b.seriesRun {
+		cmd.WriteString(`-series-run `)
+	}
+
 	fmt.Fprintf(&cmd, `-timeout %s `, b.timeout.String())
 	fmt.Fprintf(&cmd, `-workflow %s `, b.ghaWorkflow)
 	fmt.Fprintf(&cmd, `-workflow-ref=%s `, b.workflowRef)

dronegen/os_repos.go

@@ -68,7 +68,8 @@ func buildPromoteOsPackagePipeline(repoType, versionChannel, packageNameFilter,
 		ghaWorkflow:       "deploy-packages.yaml",
 		timeout:           12 * time.Hour, // DR takes a long time
 		workflowRef:       "refs/heads/master",
-		shouldTagWorkflow: false,
+		shouldTagWorkflow: true,
+		seriesRun:         true,
 		inputs:            inputs,
 	})