From 6c57683dc6077282dd95e03b19396f602dd5d635 Mon Sep 17 00:00:00 2001 From: "Lennart Kats (databricks)" Date: Sat, 21 Sep 2024 08:36:47 +0200 Subject: [PATCH 1/3] Reduce time until the prompt is shown for bundle run (#1727) ## Summary Makes the `databricks bundle run` command use local state before showing the menu prompt, which makes it show more quickly. For large/busy workspaces this means the prompt can show 2-3 seconds earlier. --- cmd/bundle/run.go | 18 +++++++++++------- 1 file changed, 11 insertions(+), 7 deletions(-) diff --git a/cmd/bundle/run.go b/cmd/bundle/run.go index 63458f85c..9ef5eb8ff 100644 --- a/cmd/bundle/run.go +++ b/cmd/bundle/run.go @@ -55,13 +55,7 @@ task or a Python wheel task, the second example applies. return diags.Error() } - diags = bundle.Apply(ctx, b, bundle.Seq( - phases.Initialize(), - terraform.Interpolate(), - terraform.Write(), - terraform.StatePull(), - terraform.Load(terraform.ErrorOnEmptyState), - )) + diags = bundle.Apply(ctx, b, phases.Initialize()) if err := diags.Error(); err != nil { return err } @@ -84,6 +78,16 @@ task or a Python wheel task, the second example applies. return fmt.Errorf("expected a KEY of the resource to run") } + diags = bundle.Apply(ctx, b, bundle.Seq( + terraform.Interpolate(), + terraform.Write(), + terraform.StatePull(), + terraform.Load(terraform.ErrorOnEmptyState), + )) + if err := diags.Error(); err != nil { + return err + } + runner, err := run.Find(b, args[0]) if err != nil { return err From 7665c639bd34392f5d95c177b520d48b9ffa40f4 Mon Sep 17 00:00:00 2001 From: "Lennart Kats (databricks)" Date: Mon, 23 Sep 2024 11:52:04 +0200 Subject: [PATCH 2/3] Use Unity Catalog for pipelines in the default-python template (#1766) ## Summary Enables Unity Catalog for pipelines in the default template. Pipelines will default to non-Unity Catalog pipelines if a catalog is not specified. *Small caveat*: there are cases where admins lock down the default catalog of a workspace and don't allow the creation of a new schema there. If that happens, the pipeline would fail at runtime with a clear error indicating what happened. ("PERMISSION_DENIED: User does not have CREATE SCHEMA on Catalog 'main'."). I've seen this with an internal Databricks workspace, where creating new non-UC schemas wasn't locked down, but creation in the `main` was. ## Testing - Validated on a non-UC + UC workspace. The catalog selection logic here is the same as applied for the SQL templates. --- .../resources/{{.project_name}}_pipeline.yml.tmpl | 6 ++++++ 1 file changed, 6 insertions(+) diff --git a/libs/template/templates/default-python/template/{{.project_name}}/resources/{{.project_name}}_pipeline.yml.tmpl b/libs/template/templates/default-python/template/{{.project_name}}/resources/{{.project_name}}_pipeline.yml.tmpl index 4b8f74d17..bf4690461 100644 --- a/libs/template/templates/default-python/template/{{.project_name}}/resources/{{.project_name}}_pipeline.yml.tmpl +++ b/libs/template/templates/default-python/template/{{.project_name}}/resources/{{.project_name}}_pipeline.yml.tmpl @@ -3,6 +3,12 @@ resources: pipelines: {{.project_name}}_pipeline: name: {{.project_name}}_pipeline + {{- if eq default_catalog ""}} + ## Specify the 'catalog' field to configure this pipeline to make use of Unity Catalog: + # catalog: catalog_name + {{- else}} + catalog: {{default_catalog}} + {{- end}} target: {{.project_name}}_${bundle.environment} libraries: - notebook: From ac80d3dfcb648c26c131ad21aec45449ac5c3b31 Mon Sep 17 00:00:00 2001 From: Ilia Babanov Date: Mon, 23 Sep 2024 12:09:11 +0200 Subject: [PATCH 3/3] 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: Screenshot 2024-09-19 at 11 07 48 ## Tests Manually for the `sync` and `bundle deploy` commands + vscode extension sync and deploy flows --- bundle/deploy/files/upload.go | 18 ++++++++++---- bundle/phases/deploy.go | 5 ++-- cmd/bundle/deploy.go | 14 ++++++++++- cmd/sync/sync.go | 38 ++++++++++++++--------------- {cmd => libs}/sync/output.go | 6 ++--- libs/sync/sync.go | 45 +++++++++++++++++++++++++---------- 6 files changed, 82 insertions(+), 44 deletions(-) rename {cmd => libs}/sync/output.go (83%) diff --git a/bundle/deploy/files/upload.go b/bundle/deploy/files/upload.go index 2c126623e..77b83611b 100644 --- a/bundle/deploy/files/upload.go +++ b/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 { + outputHandler sync.OutputHandler +} func (m *upload) Name() string { return "files.Upload" @@ -18,11 +21,18 @@ 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.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 { return diag.FromErr(err) @@ -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} } diff --git a/bundle/phases/deploy.go b/bundle/phases/deploy.go index 49544227e..097c561eb 100644 --- a/bundle/phases/deploy.go +++ b/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(), diff --git a/cmd/bundle/deploy.go b/cmd/bundle/deploy.go index 1166875ab..492317347 100644 --- a/cmd/bundle/deploy.go +++ b/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), )), ) } diff --git a/cmd/sync/sync.go b/cmd/sync/sync.go index 23a4c018f..2092d9e33 100644 --- a/cmd/sync/sync.go +++ b/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,23 +134,7 @@ 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) @@ -142,8 +142,6 @@ func New() *cobra.Command { _, err = s.RunOnce(ctx) } - s.Close() - wg.Wait() return err } diff --git a/cmd/sync/output.go b/libs/sync/output.go similarity index 83% rename from cmd/sync/output.go rename to libs/sync/output.go index 2785343f9..c01b25ef6 100644 --- a/cmd/sync/output.go +++ b/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 { diff --git a/libs/sync/sync.go b/libs/sync/sync.go index 9eaebf2ad..cc9c73944 100644 --- a/libs/sync/sync.go +++ b/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) {