Add verbose flag to the "bundle deploy" command (#1774)

## Changes
- Extract sync output logic from `cmd/sync` into `lib/sync`
- Add hidden `verbose` flag to the `bundle deploy` command, it's false
by default and hidden from the `--help` output
- Pass output handler to the `deploy/files/upload` mutator if the
verbose option is true

The was an idea to use in-place output overriding each past file sync
event in the output, bit that wont work for the extension, since it
doesn't display deploy logs in the terminal.

Example output:
```
~/tmp/defpy: ~/cli/cli bundle deploy --sync-progress
Building defpy...
Uploading defpy-0.0.1+20240917.112755-py3-none-any.whl...
Uploading bundle files to /Users/ilia.babanov@databricks.com/.bundle/defpy/dev/files...
Action: PUT: requirements-dev.txt, resources/defpy_pipeline.yml, pytest.ini, src/defpy/main.py, src/defpy/__init__.py, src/dlt_pipeline.ipynb, tests/main_test.py, src/notebook.ipynb, setup.py, resources/defpy_job.yml, .vscode/extensions.json, .vscode/settings.json, fixtures/.gitkeep, .vscode/__builtins__.pyi, README.md, .gitignore, databricks.yml
Uploaded tests
Uploaded resources
Uploaded fixtures
Uploaded .vscode
Uploaded src/defpy
Uploaded requirements-dev.txt
Uploaded .gitignore
Uploaded fixtures/.gitkeep
Uploaded src/defpy/__init__.py
Uploaded databricks.yml
Uploaded README.md
Uploaded setup.py
Uploaded .vscode/__builtins__.pyi
Uploaded .vscode/extensions.json
Uploaded src/dlt_pipeline.ipynb
Uploaded .vscode/settings.json
Uploaded resources/defpy_job.yml
Uploaded pytest.ini
Uploaded src/defpy/main.py
Uploaded tests/main_test.py
Uploaded resources/defpy_pipeline.yml
Uploaded src/notebook.ipynb
Initial Sync Complete
Deploying resources...
Updating deployment state...
Deployment complete!
```

Output example in the extension:
<img width="1843" alt="Screenshot 2024-09-19 at 11 07 48"
src="https://github.com/user-attachments/assets/0fafd095-cdc6-44b8-b482-27a38ada0330">


## Tests
Manually for the `sync` and `bundle deploy` commands + vscode extension
sync and deploy flows

bundle/deploy/files/upload.go

@@ -8,20 +8,30 @@ 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 {
+	outputHandler sync.OutputHandler
+}
 
 func (m *upload) Name() string {
 	return "files.Upload"
 }
 
 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.outputHandler
+	sync, err := sync.New(ctx, *opts)
 	if err != nil {
 		return diag.FromErr(err)
 	}
+	defer sync.Close()
 
 	b.Files, err = sync.RunOnce(ctx)
 	if err != nil {
@@ -32,6 +42,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"
 	tfjson "github.com/hashicorp/terraform-json"
 )
@@ -128,7 +129,7 @@ properties such as the 'catalog' or 'storage' are changed:`
 }
 
 // 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(
@@ -157,7 +158,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,15 @@ func newDeployCommand() *cobra.Command {
 	var failOnActiveRuns bool
 	var computeID string
 	var autoApprove bool
+	var verbose 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(&verbose, "verbose", false, "Enable verbose output.")
+	// Verbose flag currently only affects file sync output, it's used by the vscode extension
+	cmd.Flags().MarkHidden("verbose")
 
 	cmd.RunE = func(cmd *cobra.Command, args []string) error {
 		ctx := cmd.Context()
@@ -51,11 +56,18 @@ func newDeployCommand() *cobra.Command {
 				return nil
 			})
 
+			var outputHandler sync.OutputHandler
+			if verbose {
+				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
 }
@@ -118,32 +134,14 @@ func New() *cobra.Command {
 		if err != nil {
 			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())
-			}()
-		}
+		defer s.Close()
 
 		if f.watch {
 			err = s.RunContinuous(ctx)
 		} else {
 			_, err = s.RunOnce(ctx)
 		}
 
-		s.Close()
-		wg.Wait()
 		return err
 	}
 

cmd/sync/output.go → 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 {
@@ -49,6 +54,10 @@ type Sync struct {
 	// Synchronization progress events are sent to this event notifier.
 	notifier EventNotifier
 	seq      int
+
+	// WaitGroup is automatically created when an output handler is provided in the SyncOptions.
+	// Close call is required to ensure the output handler goroutine handles all events in time.
+	outputWaitGroup *stdsync.WaitGroup
 }
 
 // New initializes and returns a new [Sync] instance.
@@ -106,31 +115,41 @@ func New(ctx context.Context, opts SyncOptions) (*Sync, error) {
 		return nil, err
 	}
 
+	var notifier EventNotifier
+	var outputWaitGroup = &stdsync.WaitGroup{}
+	if opts.OutputHandler != nil {
+		ch := make(chan Event, MaxRequestsInFlight)
+		notifier = &ChannelNotifier{ch}
+		outputWaitGroup.Add(1)
+		go func() {
+			defer outputWaitGroup.Done()
+			opts.OutputHandler(ctx, ch)
+		}()
+	} else {
+		notifier = &NopNotifier{}
+	}
+
 	return &Sync{
 		SyncOptions: &opts,
 
-		fileSet:        fileSet,
-		includeFileSet: includeFileSet,
-		excludeFileSet: excludeFileSet,
-		snapshot:       snapshot,
-		filer:          filer,
-		notifier:       &NopNotifier{},
-		seq:            0,
+		fileSet:         fileSet,
+		includeFileSet:  includeFileSet,
+		excludeFileSet:  excludeFileSet,
+		snapshot:        snapshot,
+		filer:           filer,
+		notifier:        notifier,
+		outputWaitGroup: outputWaitGroup,
+		seq:             0,
 	}, nil
 }
 
-func (s *Sync) Events() <-chan Event {
-	ch := make(chan Event, MaxRequestsInFlight)
-	s.notifier = &ChannelNotifier{ch}
-	return ch
-}
-
 func (s *Sync) Close() {
 	if s.notifier == nil {
 		return
 	}
 	s.notifier.Close()
 	s.notifier = nil
+	s.outputWaitGroup.Wait()
 }
 
 func (s *Sync) notifyStart(ctx context.Context, d diff) {