Show file sync progress in bundle deploy

bundle/deploy/files/upload.go

@@ -8,9 +8,12 @@ import (
 	"github.com/databricks/cli/libs/cmdio"
 	"github.com/databricks/cli/libs/diag"
 	"github.com/databricks/cli/libs/log"
+	"github.com/databricks/cli/libs/sync"
 )
 
-type upload struct{}
+type upload struct {
+	outpuHandler sync.OutputHandler
+}
 
 func (m *upload) Name() string {
 	return "files.Upload"
@@ -18,7 +21,13 @@ func (m *upload) Name() string {
 
 func (m *upload) Apply(ctx context.Context, b *bundle.Bundle) diag.Diagnostics {
 	cmdio.LogString(ctx, fmt.Sprintf("Uploading bundle files to %s...", b.Config.Workspace.FilePath))
-	sync, err := GetSync(ctx, bundle.ReadOnly(b))
+	opts, err := GetSyncOptions(ctx, bundle.ReadOnly(b))
+	if err != nil {
+		return diag.FromErr(err)
+	}
+
+	opts.OutputHandler = m.outpuHandler
+	sync, err := sync.New(ctx, *opts)
 	if err != nil {
 		return diag.FromErr(err)
 	}
@@ -32,6 +41,6 @@ func (m *upload) Apply(ctx context.Context, b *bundle.Bundle) diag.Diagnostics {
 	return nil
 }
 
-func Upload() bundle.Mutator {
-	return &upload{}
+func Upload(outputHandler sync.OutputHandler) bundle.Mutator {
+	return &upload{outputHandler}
 }

bundle/phases/deploy.go

@@ -18,6 +18,7 @@ import (
 	"github.com/databricks/cli/bundle/python"
 	"github.com/databricks/cli/bundle/scripts"
 	"github.com/databricks/cli/libs/cmdio"
+	"github.com/databricks/cli/libs/sync"
 	terraformlib "github.com/databricks/cli/libs/terraform"
 )
 
@@ -88,7 +89,7 @@ func approvalForUcSchemaDelete(ctx context.Context, b *bundle.Bundle) (bool, err
 }
 
 // The deploy phase deploys artifacts and resources.
-func Deploy() bundle.Mutator {
+func Deploy(outputHandler sync.OutputHandler) bundle.Mutator {
 	// Core mutators that CRUD resources and modify deployment state. These
 	// mutators need informed consent if they are potentially destructive.
 	deployCore := bundle.Defer(
@@ -117,7 +118,7 @@ func Deploy() bundle.Mutator {
 				libraries.ExpandGlobReferences(),
 				libraries.Upload(),
 				python.TransformWheelTask(),
-				files.Upload(),
+				files.Upload(outputHandler),
 				deploy.StateUpdate(),
 				deploy.StatePush(),
 				permissions.ApplyWorkspaceRootPermissions(),

cmd/bundle/deploy.go

@@ -10,6 +10,7 @@ import (
 	"github.com/databricks/cli/cmd/bundle/utils"
 	"github.com/databricks/cli/cmd/root"
 	"github.com/databricks/cli/libs/diag"
+	"github.com/databricks/cli/libs/sync"
 	"github.com/spf13/cobra"
 )
 
@@ -25,11 +26,14 @@ func newDeployCommand() *cobra.Command {
 	var failOnActiveRuns bool
 	var computeID string
 	var autoApprove bool
+	var showSyncProgress bool
 	cmd.Flags().BoolVar(&force, "force", false, "Force-override Git branch validation.")
 	cmd.Flags().BoolVar(&forceLock, "force-lock", false, "Force acquisition of deployment lock.")
 	cmd.Flags().BoolVar(&failOnActiveRuns, "fail-on-active-runs", false, "Fail if there are running jobs or pipelines in the deployment.")
 	cmd.Flags().StringVarP(&computeID, "compute-id", "c", "", "Override compute in the deployment with the given compute ID.")
 	cmd.Flags().BoolVar(&autoApprove, "auto-approve", false, "Skip interactive approvals that might be required for deployment.")
+	cmd.Flags().BoolVar(&showSyncProgress, "sync-progress", false, "Show file synchronisation progress.")
+	cmd.Flags().MarkHidden("sync-progress")
 
 	cmd.RunE = func(cmd *cobra.Command, args []string) error {
 		ctx := cmd.Context()
@@ -51,11 +55,18 @@ func newDeployCommand() *cobra.Command {
 				return nil
 			})
 
+			var outputHandler sync.OutputHandler
+			if showSyncProgress {
+				outputHandler = func(ctx context.Context, c <-chan sync.Event) {
+					sync.TextOutput(ctx, c, cmd.OutOrStdout())
+				}
+			}
+
 			diags = diags.Extend(
 				bundle.Apply(ctx, b, bundle.Seq(
 					phases.Initialize(),
 					phases.Build(),
-					phases.Deploy(),
+					phases.Deploy(outputHandler),
 				)),
 			)
 		}

cmd/sync/sync.go

@@ -6,7 +6,6 @@ import (
 	"fmt"
 	"io"
 	"path/filepath"
-	stdsync "sync"
 	"time"
 
 	"github.com/databricks/cli/bundle"
@@ -46,6 +45,21 @@ func (f *syncFlags) syncOptionsFromArgs(cmd *cobra.Command, args []string) (*syn
 		return nil, flag.ErrHelp
 	}
 
+	var outputFunc func(context.Context, <-chan sync.Event, io.Writer)
+	switch f.output {
+	case flags.OutputText:
+		outputFunc = sync.TextOutput
+	case flags.OutputJSON:
+		outputFunc = sync.JsonOutput
+	}
+
+	var outputHandler sync.OutputHandler
+	if outputFunc != nil {
+		outputHandler = func(ctx context.Context, events <-chan sync.Event) {
+			outputFunc(ctx, events, cmd.OutOrStdout())
+		}
+	}
+
 	opts := sync.SyncOptions{
 		LocalRoot: vfs.MustNew(args[0]),
 		Paths:     []string{"."},
@@ -62,6 +76,8 @@ func (f *syncFlags) syncOptionsFromArgs(cmd *cobra.Command, args []string) (*syn
 		// exist and add it to the `.gitignore` file in the root.
 		SnapshotBasePath: filepath.Join(args[0], ".databricks"),
 		WorkspaceClient:  root.WorkspaceClient(cmd.Context()),
+
+		OutputHandler: outputHandler,
 	}
 	return &opts, nil
 }
@@ -119,23 +135,6 @@ func New() *cobra.Command {
 			return err
 		}
 
-		var outputFunc func(context.Context, <-chan sync.Event, io.Writer)
-		switch f.output {
-		case flags.OutputText:
-			outputFunc = textOutput
-		case flags.OutputJSON:
-			outputFunc = jsonOutput
-		}
-
-		var wg stdsync.WaitGroup
-		if outputFunc != nil {
-			wg.Add(1)
-			go func() {
-				defer wg.Done()
-				outputFunc(ctx, s.Events(), cmd.OutOrStdout())
-			}()
-		}
-
 		if f.watch {
 			err = s.RunContinuous(ctx)
 		} else {
@@ -143,7 +142,6 @@ func New() *cobra.Command {
 		}
 
 		s.Close()
-		wg.Wait()
 		return err
 	}
 

libs/sync/output.go

@@ -5,12 +5,10 @@ import (
 	"context"
 	"encoding/json"
 	"io"
-
-	"github.com/databricks/cli/libs/sync"
 )
 
 // Read synchronization events and write them as JSON to the specified writer (typically stdout).
-func jsonOutput(ctx context.Context, ch <-chan sync.Event, w io.Writer) {
+func JsonOutput(ctx context.Context, ch <-chan Event, w io.Writer) {
 	enc := json.NewEncoder(w)
 	for {
 		select {
@@ -31,7 +29,7 @@ func jsonOutput(ctx context.Context, ch <-chan sync.Event, w io.Writer) {
 }
 
 // Read synchronization events and write them as text to the specified writer (typically stdout).
-func textOutput(ctx context.Context, ch <-chan sync.Event, w io.Writer) {
+func TextOutput(ctx context.Context, ch <-chan Event, w io.Writer) {
 	bw := bufio.NewWriter(w)
 
 	for {

libs/sync/sync.go

@@ -3,6 +3,7 @@ package sync
 import (
 	"context"
 	"fmt"
+	stdsync "sync"
 	"time"
 
 	"github.com/databricks/cli/libs/filer"
@@ -15,6 +16,8 @@ import (
 	"github.com/databricks/databricks-sdk-go/service/iam"
 )
 
+type OutputHandler func(context.Context, <-chan Event)
+
 type SyncOptions struct {
 	LocalRoot vfs.Path
 	Paths     []string
@@ -34,6 +37,8 @@ type SyncOptions struct {
 	CurrentUser *iam.User
 
 	Host string
+
+	OutputHandler OutputHandler
 }
 
 type Sync struct {
@@ -47,8 +52,9 @@ type Sync struct {
 	filer    filer.Filer
 
 	// Synchronization progress events are sent to this event notifier.
-	notifier EventNotifier
-	seq      int
+	notifier   EventNotifier
+	notifierWg *stdsync.WaitGroup
+	seq        int
 }
 
 // New initializes and returns a new [Sync] instance.
@@ -106,6 +112,20 @@ func New(ctx context.Context, opts SyncOptions) (*Sync, error) {
 		return nil, err
 	}
 
+	var notifier EventNotifier
+	var notifierWg = &stdsync.WaitGroup{}
+	if opts.OutputHandler != nil {
+		ch := make(chan Event, MaxRequestsInFlight)
+		notifier = &ChannelNotifier{ch}
+		notifierWg.Add(1)
+		go func() {
+			defer notifierWg.Done()
+			opts.OutputHandler(ctx, ch)
+		}()
+	} else {
+		notifier = &NopNotifier{}
+	}
+
 	return &Sync{
 		SyncOptions: &opts,
 
@@ -114,7 +134,8 @@ func New(ctx context.Context, opts SyncOptions) (*Sync, error) {
 		excludeFileSet: excludeFileSet,
 		snapshot:       snapshot,
 		filer:          filer,
-		notifier:       &NopNotifier{},
+		notifier:       notifier,
+		notifierWg:     notifierWg,
 		seq:            0,
 	}, nil
 }
@@ -126,6 +147,7 @@ func (s *Sync) Events() <-chan Event {
 }
 
 func (s *Sync) Close() {
+	s.notifierWg.Wait()
 	if s.notifier == nil {
 		return
 	}