Show current worker goroutine status below progress bar

CHANGELOG.md

@@ -20,6 +20,7 @@ All notable changes to `src-cli` are documented in this file.
 - Error reporting by `src campaign [preview|apply]` has been improved and now includes more information about which step failed in which repository. [#325](https://github.com/sourcegraph/src-cli/pull/325)
 - The default behaviour of `src campaigns [preview|apply]` has been changed to retain downloaded archives of repositories for better performance across re-runs of the command. To use the old behaviour and delete the archives use the `-clean-archives` flag. Repository archives are also not stored in the directory for temp data (see `-tmp` flag) anymore but in the cache directory, which can be configured with the `-cache` flag. To manually delete archives between runs, delete the `*.zip` files in the `-cache` directory (see `src campaigns -help` for its default location).
 - `src campaign [preview|apply]` now check whether `git` and `docker` are available before trying to execute a campaign spec's steps. [#326](https://github.com/sourcegraph/src-cli/pull/326)
+- The progress bar displayed by `src campaign [preview|apply]` has been extended by status bars that show which steps are currently being executed for each repository. [#338](https://github.com/sourcegraph/src-cli/pull/338)
 
 ### Fixed
 

cmd/src/campaign_progress_printer.go

@@ -0,0 +1,188 @@
+package main
+
+import (
+	"fmt"
+	"strings"
+
+	"github.com/sourcegraph/go-diff/diff"
+	"github.com/sourcegraph/src-cli/internal/campaigns"
+	"github.com/sourcegraph/src-cli/internal/output"
+)
+
+func newCampaignProgressPrinter(out *output.Output, numParallelism int) *campaignProgressPrinter {
+	return &campaignProgressPrinter{
+		out: out,
+
+		numParallelism: numParallelism,
+
+		completedTasks: map[string]bool{},
+		runningTasks:   map[string]*campaigns.TaskStatus{},
+
+		repoStatusBar: map[string]int{},
+		statusBarRepo: map[int]string{},
+	}
+}
+
+type campaignProgressPrinter struct {
+	out      *output.Output
+	progress output.ProgressWithStatusBars
+
+	maxRepoName    int
+	numParallelism int
+
+	completedTasks map[string]bool
+	runningTasks   map[string]*campaigns.TaskStatus
+
+	repoStatusBar map[string]int
+	statusBarRepo map[int]string
+}
+
+func (p *campaignProgressPrinter) initProgressBar(statuses []*campaigns.TaskStatus) {
+	statusBars := []*output.StatusBar{}
+	for i := 0; i < p.numParallelism; i++ {
+		statusBars = append(statusBars, output.NewStatusBarWithLabel("Starting worker..."))
+	}
+
+	p.progress = p.out.ProgressWithStatusBars([]output.ProgressBar{{
+		Label: fmt.Sprintf("Executing steps in %d repositories", len(statuses)),
+		Max:   float64(len(statuses)),
+	}}, statusBars, nil)
+}
+
+func (p *campaignProgressPrinter) Complete() {
+	if p.progress != nil {
+		p.progress.Complete()
+	}
+}
+
+func (p *campaignProgressPrinter) PrintStatuses(statuses []*campaigns.TaskStatus) {
+	if p.progress == nil {
+		p.initProgressBar(statuses)
+	}
+
+	newlyCompleted := []*campaigns.TaskStatus{}
+	currentlyRunning := []*campaigns.TaskStatus{}
+
+	for _, ts := range statuses {
+		if len(ts.RepoName) > p.maxRepoName {
+			p.maxRepoName = len(ts.RepoName)
+		}
+
+		if ts.IsCompleted() {
+			if !p.completedTasks[ts.RepoName] {
+				p.completedTasks[ts.RepoName] = true
+				newlyCompleted = append(newlyCompleted, ts)
+			}
+
+			if _, ok := p.runningTasks[ts.RepoName]; ok {
+				delete(p.runningTasks, ts.RepoName)
+
+				// Free slot
+				idx := p.repoStatusBar[ts.RepoName]
+				delete(p.statusBarRepo, idx)
+			}
+		}
+
+		if ts.IsRunning() {
+			currentlyRunning = append(currentlyRunning, ts)
+		}
+
+	}
+
+	p.progress.SetValue(0, float64(len(p.completedTasks)))
+
+	newlyStarted := map[string]*campaigns.TaskStatus{}
+	statusBarIndex := 0
+	for _, ts := range currentlyRunning {
+		if _, ok := p.runningTasks[ts.RepoName]; ok {
+			continue
+		}
+
+		newlyStarted[ts.RepoName] = ts
+		p.runningTasks[ts.RepoName] = ts
+
+		// Find free slot
+		_, ok := p.statusBarRepo[statusBarIndex]
+		for ok {
+			statusBarIndex += 1
+			_, ok = p.statusBarRepo[statusBarIndex]
+		}
+
+		p.statusBarRepo[statusBarIndex] = ts.RepoName
+		p.repoStatusBar[ts.RepoName] = statusBarIndex
+	}
+
+	for _, ts := range newlyCompleted {
+		statusText, err := taskStatusText(ts)
+		if err != nil {
+			p.progress.Verbosef("%-*s failed to display status: %s", p.maxRepoName, ts.RepoName, err)
+			continue
+		}
+
+		p.progress.Verbosef("%-*s %s", p.maxRepoName, ts.RepoName, statusText)
+
+		if idx, ok := p.repoStatusBar[ts.RepoName]; ok {
+			// Log that this task completed, but only if there is no
+			// currently executing one in this bar, to avoid flicker.
+			if _, ok := p.statusBarRepo[idx]; !ok {
+				p.progress.StatusBarCompletef(idx, statusText)
+			}
+			delete(p.repoStatusBar, ts.RepoName)
+		}
+	}
+
+	for statusBar, repo := range p.statusBarRepo {
+		ts, ok := p.runningTasks[repo]
+		if !ok {
+			// This should not happen
+			continue
+		}
+
+		statusText, err := taskStatusText(ts)
+		if err != nil {
+			p.progress.Verbosef("%-*s failed to display status: %s", p.maxRepoName, ts.RepoName, err)
+			continue
+		}
+
+		if _, ok := newlyStarted[repo]; ok {
+			p.progress.StatusBarResetf(statusBar, ts.RepoName, statusText)
+		} else {
+			p.progress.StatusBarUpdatef(statusBar, statusText)
+		}
+	}
+}
+
+func taskStatusText(ts *campaigns.TaskStatus) (string, error) {
+	var statusText string
+
+	if ts.IsCompleted() {
+		if ts.ChangesetSpec == nil {
+			statusText = "No changes"
+		} else {
+			fileDiffs, err := diff.ParseMultiFileDiff([]byte(ts.ChangesetSpec.Commits[0].Diff))
+			if err != nil {
+				return "", err
+			}
+
+			statusText = diffStatDescription(fileDiffs) + " " + diffStatDiagram(sumDiffStats(fileDiffs))
+		}
+
+		if ts.Cached {
+			statusText += " (cached)"
+		}
+	} else if ts.IsRunning() {
+		if ts.CurrentlyExecuting != "" {
+			lines := strings.Split(ts.CurrentlyExecuting, "\n")
+			escapedLine := strings.ReplaceAll(lines[0], "%", "%%")
+			if len(lines) > 1 {
+				statusText = fmt.Sprintf("%s ...", escapedLine)
+			} else {
+				statusText = fmt.Sprintf("%s", escapedLine)
+			}
+		} else {
+			statusText = fmt.Sprintf("...")
+		}
+	}
+
+	return statusText, nil
+}

cmd/src/campaigns_common.go

@@ -240,12 +240,12 @@ func campaignsExecute(ctx context.Context, out *output.Output, svc *campaigns.Se
 		campaignsCompletePending(pending, "Resolved repositories")
 	}
 
-	execProgress, execProgressComplete := executeCampaignSpecProgress(out)
-	specs, err := svc.ExecuteCampaignSpec(ctx, repos, executor, campaignSpec, execProgress)
+	p := newCampaignProgressPrinter(out, opts.Parallelism)
+	specs, err := svc.ExecuteCampaignSpec(ctx, repos, executor, campaignSpec, p.PrintStatuses)
 	if err != nil {
 		return "", "", err
 	}
-	execProgressComplete()
+	p.Complete()
 
 	if logFiles := executor.LogFiles(); len(logFiles) > 0 && flags.keepLogs {
 		func() {
@@ -372,9 +372,10 @@ func diffStatDiagram(stat diff.Stat) string {
 		added *= x
 		deleted *= x
 	}
-	return fmt.Sprintf("%s%s%s%s",
+	return fmt.Sprintf("%s%s%s%s%s",
 		output.StyleLinesAdded, strings.Repeat("+", int(added)),
 		output.StyleLinesDeleted, strings.Repeat("-", int(deleted)),
+		output.StyleReset,
 	)
 }
 
@@ -409,78 +410,3 @@ func contextCancelOnInterrupt(parent context.Context) (context.Context, func())
 		ctxCancel()
 	}
 }
-
-// executeCampaignSpecProgress returns a function that can be passed to
-// (*Service).ExecuteCampaignSpec as the "progress" function.
-//
-// It prints a progress bar and, if verbose mode is activated, diff stats of
-// the produced diffs.
-//
-// The second return value is the "complete" function that completes the
-// progress bar and should be called after ExecuteCampaignSpec returns
-// successfully.
-func executeCampaignSpecProgress(out *output.Output) (func(statuses []*campaigns.TaskStatus), func()) {
-	var (
-		progress       output.Progress
-		maxRepoName    int
-		completedTasks = map[string]bool{}
-	)
-
-	complete := func() {
-		if progress != nil {
-			progress.Complete()
-		}
-	}
-
-	progressFunc := func(statuses []*campaigns.TaskStatus) {
-		if progress == nil {
-			progress = out.Progress([]output.ProgressBar{{
-				Label: fmt.Sprintf("Executing steps in %d repositories", len(statuses)),
-				Max:   float64(len(statuses)),
-			}}, nil)
-		}
-
-		unloggedCompleted := []*campaigns.TaskStatus{}
-
-		for _, ts := range statuses {
-			if len(ts.RepoName) > maxRepoName {
-				maxRepoName = len(ts.RepoName)
-			}
-
-			if ts.FinishedAt.IsZero() {
-				continue
-			}
-
-			if !completedTasks[ts.RepoName] {
-				completedTasks[ts.RepoName] = true
-				unloggedCompleted = append(unloggedCompleted, ts)
-			}
-
-		}
-
-		progress.SetValue(0, float64(len(completedTasks)))
-
-		for _, ts := range unloggedCompleted {
-			var statusText string
-
-			if ts.ChangesetSpec == nil {
-				statusText = "No changes"
-			} else {
-				fileDiffs, err := diff.ParseMultiFileDiff([]byte(ts.ChangesetSpec.Commits[0].Diff))
-				if err != nil {
-					panic(err)
-				}
-
-				statusText = diffStatDescription(fileDiffs) + " " + diffStatDiagram(sumDiffStats(fileDiffs))
-			}
-
-			if ts.Cached {
-				statusText += " (cached)"
-			}
-
-			progress.Verbosef("%-*s %s", maxRepoName, ts.RepoName, statusText)
-		}
-	}
-
-	return progressFunc, complete
-}

internal/campaigns/executor.go

@@ -60,12 +60,21 @@ type TaskStatus struct {
 	FinishedAt time.Time
 
 	// TODO: add current step and progress fields.
+	CurrentlyExecuting string
 
 	// Result fields.
 	ChangesetSpec *ChangesetSpec
 	Err           error
 }
 
+func (ts *TaskStatus) IsRunning() bool {
+	return !ts.StartedAt.IsZero() && ts.FinishedAt.IsZero()
+}
+
+func (ts *TaskStatus) IsCompleted() bool {
+	return !ts.StartedAt.IsZero() && !ts.FinishedAt.IsZero()
+}
+
 type executor struct {
 	ExecutorOpts
 
@@ -156,6 +165,7 @@ func (x *executor) do(ctx context.Context, task *Task) (err error) {
 	// Ensure that the status is updated when we're done.
 	defer func() {
 		status.FinishedAt = time.Now()
+		status.CurrentlyExecuting = ""
 		status.Err = err
 		x.updateTaskStatus(task, status)
 	}()
@@ -231,7 +241,10 @@ func (x *executor) do(ctx context.Context, task *Task) (err error) {
 	defer cancel()
 
 	// Actually execute the steps.
-	diff, err := runSteps(runCtx, x.creator, task.Repository, task.Steps, log, x.tempDir)
+	diff, err := runSteps(runCtx, x.creator, task.Repository, task.Steps, log, x.tempDir, func(currentlyExecuting string) {
+		status.CurrentlyExecuting = currentlyExecuting
+		x.updateTaskStatus(task, status)
+	})
 	if err != nil {
 		if reachedTimeout(runCtx, err) {
 			err = &errTimeoutReached{timeout: x.Timeout}

internal/campaigns/run_steps.go

@@ -17,7 +17,9 @@ import (
 	"github.com/sourcegraph/src-cli/internal/campaigns/graphql"
 )
 
-func runSteps(ctx context.Context, wc *WorkspaceCreator, repo *graphql.Repository, steps []Step, logger *TaskLogger, tempDir string) ([]byte, error) {
+func runSteps(ctx context.Context, wc *WorkspaceCreator, repo *graphql.Repository, steps []Step, logger *TaskLogger, tempDir string, reportProgress func(string)) ([]byte, error) {
+	reportProgress("Downloading archive")
+
 	volumeDir, err := wc.Create(ctx, repo)
 	if err != nil {
 		return nil, errors.Wrap(err, "creating workspace")
@@ -34,6 +36,7 @@ func runSteps(ctx context.Context, wc *WorkspaceCreator, repo *graphql.Repositor
 		return out, nil
 	}
 
+	reportProgress("Initializing workspace")
 	if _, err := runGitCmd("init"); err != nil {
 		return nil, errors.Wrap(err, "git init failed")
 	}
@@ -59,6 +62,7 @@ func runSteps(ctx context.Context, wc *WorkspaceCreator, repo *graphql.Repositor
 
 	for i, step := range steps {
 		logger.Logf("[Step %d] docker run %s %q", i+1, step.Container, step.Run)
+		reportProgress(step.Run)
 
 		cidFile, err := ioutil.TempFile(tempDir, repo.Slug()+"-container-id")
 		if err != nil {
@@ -143,6 +147,7 @@ func runSteps(ctx context.Context, wc *WorkspaceCreator, repo *graphql.Repositor
 		return nil, errors.Wrap(err, "git add failed")
 	}
 
+	reportProgress("Calculating diff")
 	// As of Sourcegraph 3.14 we only support unified diff format.
 	// That means we need to strip away the `a/` and `/b` prefixes with `--no-prefix`.
 	// See: https://github.com/sourcegraph/sourcegraph/blob/82d5e7e1562fef6be5c0b17f18631040fd330835/enterprise/internal/campaigns/service.go#L324-L329