chore: concurrent image build with context cancellation

internal/pkg/cli/deploy/workload.go

@@ -5,12 +5,15 @@ package deploy
 
 import (
 	"bytes"
+	"context"
 	"errors"
 	"fmt"
 	"io"
 	"os"
 	"path/filepath"
 	"strings"
+	"sync"
+	"time"
 
 	"github.com/aws/aws-sdk-go/aws"
 	"github.com/aws/aws-sdk-go/aws/session"
@@ -35,11 +38,14 @@ import (
 	"github.com/aws/copilot-cli/internal/pkg/template/artifactpath"
 	"github.com/aws/copilot-cli/internal/pkg/template/diff"
 	"github.com/aws/copilot-cli/internal/pkg/term/color"
+	"github.com/aws/copilot-cli/internal/pkg/term/cursor"
 	"github.com/aws/copilot-cli/internal/pkg/term/log"
 	termprogress "github.com/aws/copilot-cli/internal/pkg/term/progress"
+	"github.com/aws/copilot-cli/internal/pkg/term/syncbuffer"
 	"github.com/aws/copilot-cli/internal/pkg/version"
 	"github.com/aws/copilot-cli/internal/pkg/workspace"
 	"github.com/spf13/afero"
+	"golang.org/x/sync/errgroup"
 )
 
 const (
@@ -56,6 +62,11 @@ const (
 	labelForVersion       = "com.aws.copilot.image.version"
 	labelForContainerName = "com.aws.copilot.image.container.name"
 )
+const (
+	paddingInSpacesForBuildAndPush = 5
+	pollIntervalForBuildAndPush    = 60 * time.Millisecond
+	defaultNumLinesForBuildAndPush = 5
+)
 
 // ActionRecommender contains methods that output action recommendation.
 type ActionRecommender interface {
@@ -70,7 +81,7 @@ func (noopActionRecommender) RecommendedActions() []string {
 
 type repositoryService interface {
 	Login() (string, error)
-	BuildAndPush(args *dockerengine.BuildArguments) (string, error)
+	BuildAndPush(ctx context.Context, args *dockerengine.BuildArguments, w io.Writer) (string, error)
 }
 
 type templater interface {
@@ -105,6 +116,11 @@ type spinner interface {
 	Stop(label string)
 }
 
+type labeledTermPrinter interface {
+	IsDone() bool
+	Print() error
+}
+
 // StackRuntimeConfiguration contains runtime configuration for a workload CloudFormation stack.
 type StackRuntimeConfiguration struct {
 	ImageDigests              map[string]ContainerImageIdentifier // Container name to image.
@@ -150,18 +166,19 @@ type workloadDeployer struct {
 	workspacePath string
 
 	// Dependencies.
-	fs               afero.Fs
-	s3Client         uploader
-	addons           stackBuilder
-	repository       repositoryService
-	deployer         serviceDeployer
-	tmplGetter       deployedTemplateGetter
-	endpointGetter   endpointGetter
-	spinner          spinner
-	templateFS       template.Reader
-	envVersionGetter versionGetter
-	overrider        Overrider
-	customResources  customResourcesFunc
+	fs                 afero.Fs
+	s3Client           uploader
+	addons             stackBuilder
+	repository         repositoryService
+	deployer           serviceDeployer
+	tmplGetter         deployedTemplateGetter
+	endpointGetter     endpointGetter
+	spinner            spinner
+	templateFS         template.Reader
+	envVersionGetter   versionGetter
+	overrider          Overrider
+	customResources    customResourcesFunc
+	labeledTermPrinter func(fw syncbuffer.FileWriter, bufs []*syncbuffer.LabeledSyncBuffer, opts ...syncbuffer.LabeledTermPrinterOption) labeledTermPrinter
 
 	// Cached variables.
 	defaultSess              *session.Session
@@ -251,6 +268,9 @@ func newWorkloadDeployer(in *WorkloadDeployerInput) (*workloadDeployer, error) {
 
 	cfn := cloudformation.New(envSession, cloudformation.WithProgressTracker(os.Stderr))
 
+	labeledTermPrinter := func(fw syncbuffer.FileWriter, bufs []*syncbuffer.LabeledSyncBuffer, opts ...syncbuffer.LabeledTermPrinterOption) labeledTermPrinter {
+		return syncbuffer.NewLabeledTermPrinter(fw, bufs, opts...)
+	}
 	return &workloadDeployer{
 		name:                     in.Name,
 		app:                      in.App,
@@ -275,6 +295,7 @@ func newWorkloadDeployer(in *WorkloadDeployerInput) (*workloadDeployer, error) {
 		envSess:                  envSession,
 		store:                    store,
 		envConfig:                envConfig,
+		labeledTermPrinter:       labeledTermPrinter,
 
 		mft:    in.Mft,
 		rawMft: in.RawMft,
@@ -367,24 +388,66 @@ func (d *workloadDeployer) uploadContainerImages(out *UploadArtifactsOutput) err
 	if err != nil {
 		return fmt.Errorf("login to image repository: %w", err)
 	}
+
+	var digestsMu sync.Mutex
 	out.ImageDigests = make(map[string]ContainerImageIdentifier, len(buildArgsPerContainer))
+	var labeledBuffers []*syncbuffer.LabeledSyncBuffer
+	g, ctx := errgroup.WithContext(context.Background())
+	cursor := cursor.New()
+	cursor.Hide()
 	for name, buildArgs := range buildArgsPerContainer {
+		// create a copy of loop variables to avoid data race.
+		name := name
+		buildArgs := buildArgs
+
 		buildArgs.URI = uri
 		buildArgsList, err := buildArgs.GenerateDockerBuildArgs(dockerengine.New(exec.NewCmd()))
 		if err != nil {
-			return fmt.Errorf("generate docker build args: %w", err)
-		}
-		// TODO(adi) handle this log in syncBuffer's Print function
-		log.Infof("Building your container image: docker %s\n", strings.Join(buildArgsList, " "))
-		digest, err := d.repository.BuildAndPush(buildArgs)
-		if err != nil {
-			return fmt.Errorf("build and push image: %w", err)
+			return fmt.Errorf("generate docker build args for %q: %w", name, err)
 		}
-		out.ImageDigests[name] = ContainerImageIdentifier{
-			Digest:            digest,
-			CustomTag:         d.image.CustomTag,
-			GitShortCommitTag: d.image.GitShortCommitTag,
+		buf := syncbuffer.New()
+		labeledBuffers = append(labeledBuffers, buf.WithLabel(fmt.Sprintf("Building your container image %q: docker %s", name, strings.Join(buildArgsList, " "))))
+		pr, pw := io.Pipe()
+		g.Go(func() error {
+			defer pw.Close()
+			digest, err := d.repository.BuildAndPush(ctx, buildArgs, pw)
+			if err != nil {
+				return fmt.Errorf("build and push the image %q: %w", name, err)
+			}
+			digestsMu.Lock()
+			defer digestsMu.Unlock()
+			out.ImageDigests[name] = ContainerImageIdentifier{
+				Digest:            digest,
+				CustomTag:         d.image.CustomTag,
+				GitShortCommitTag: d.image.GitShortCommitTag,
+			}
+			return nil
+		})
+		g.Go(func() error {
+			if err := buf.Copy(pr); err != nil {
+				return fmt.Errorf("copy build and push output for %q: %w", name, err)
+			}
+			return nil
+		})
+	}
+	opts := []syncbuffer.LabeledTermPrinterOption{syncbuffer.WithPadding(paddingInSpacesForBuildAndPush)}
+	if os.Getenv("CI") != "true" {
+		opts = append(opts, syncbuffer.WithNumLines(defaultNumLinesForBuildAndPush))
+	}
+	ltp := d.labeledTermPrinter(os.Stderr, labeledBuffers, opts...)
+	g.Go(func() error {
+		for {
+			if err := ltp.Print(); err != nil {
+				return fmt.Errorf("print logs: %w", err)
+			}
+			if ltp.IsDone() {
+				return nil
+			}
+			time.Sleep(pollIntervalForBuildAndPush)
 		}
+	})
+	if err := g.Wait(); err != nil {
+		return err
 	}
 	return nil
 }