Added bundle deployment bind and unbind command (#1131)

## Changes
Added `bundle deployment bind` and `unbind` command.

This command allows to bind bundle-defined resources to existing
resources in Databricks workspace so they become DABs-managed.

## Tests
Manually + added E2E test

bundle/config/resources.go

@@ -1,9 +1,11 @@
 package config
 
 import (
+	"context"
 	"fmt"
 
 	"github.com/databricks/cli/bundle/config/resources"
+	"github.com/databricks/databricks-sdk-go"
 )
 
 // Resources defines Databricks resources associated with the bundle.
@@ -168,3 +170,36 @@ func (r *Resources) Merge() error {
 	}
 	return nil
 }
+
+type ConfigResource interface {
+	Exists(ctx context.Context, w *databricks.WorkspaceClient, id string) (bool, error)
+	TerraformResourceName() string
+}
+
+func (r *Resources) FindResourceByConfigKey(key string) (ConfigResource, error) {
+	found := make([]ConfigResource, 0)
+	for k := range r.Jobs {
+		if k == key {
+			found = append(found, r.Jobs[k])
+		}
+	}
+	for k := range r.Pipelines {
+		if k == key {
+			found = append(found, r.Pipelines[k])
+		}
+	}
+
+	if len(found) == 0 {
+		return nil, fmt.Errorf("no such resource: %s", key)
+	}
+
+	if len(found) > 1 {
+		keys := make([]string, 0, len(found))
+		for _, r := range found {
+			keys = append(keys, fmt.Sprintf("%s:%s", r.TerraformResourceName(), key))
+		}
+		return nil, fmt.Errorf("ambiguous: %s (can resolve to all of %s)", key, keys)
+	}
+
+	return found[0], nil
+}

bundle/config/resources/job.go

@@ -1,7 +1,12 @@
 package resources
 
 import (
+	"context"
+	"strconv"
+
 	"github.com/databricks/cli/bundle/config/paths"
+	"github.com/databricks/cli/libs/log"
+	"github.com/databricks/databricks-sdk-go"
 	"github.com/databricks/databricks-sdk-go/marshal"
 	"github.com/databricks/databricks-sdk-go/service/jobs"
 	"github.com/imdario/mergo"
@@ -90,3 +95,22 @@ func (j *Job) MergeTasks() error {
 	j.Tasks = tasks
 	return nil
 }
+
+func (j *Job) Exists(ctx context.Context, w *databricks.WorkspaceClient, id string) (bool, error) {
+	jobId, err := strconv.Atoi(id)
+	if err != nil {
+		return false, err
+	}
+	_, err = w.Jobs.Get(ctx, jobs.GetJobRequest{
+		JobId: int64(jobId),
+	})
+	if err != nil {
+		log.Debugf(ctx, "job %s does not exist", id)
+		return false, err
+	}
+	return true, nil
+}
+
+func (j *Job) TerraformResourceName() string {
+	return "databricks_job"
+}

bundle/config/resources/pipeline.go

@@ -1,9 +1,12 @@
 package resources
 
 import (
+	"context"
 	"strings"
 
 	"github.com/databricks/cli/bundle/config/paths"
+	"github.com/databricks/cli/libs/log"
+	"github.com/databricks/databricks-sdk-go"
 	"github.com/databricks/databricks-sdk-go/marshal"
 	"github.com/databricks/databricks-sdk-go/service/pipelines"
 	"github.com/imdario/mergo"
@@ -73,3 +76,18 @@ func (p *Pipeline) MergeClusters() error {
 	p.Clusters = output
 	return nil
 }
+
+func (p *Pipeline) Exists(ctx context.Context, w *databricks.WorkspaceClient, id string) (bool, error) {
+	_, err := w.Pipelines.Get(ctx, pipelines.GetPipelineRequest{
+		PipelineId: id,
+	})
+	if err != nil {
+		log.Debugf(ctx, "pipeline %s does not exist", id)
+		return false, err
+	}
+	return true, nil
+}
+
+func (p *Pipeline) TerraformResourceName() string {
+	return "databricks_pipeline"
+}

bundle/deploy/lock/release.go

@@ -12,6 +12,8 @@ import (
 type Goal string
 
 const (
+	GoalBind    = Goal("bind")
+	GoalUnbind  = Goal("unbind")
 	GoalDeploy  = Goal("deploy")
 	GoalDestroy = Goal("destroy")
 )
@@ -46,6 +48,8 @@ func (m *release) Apply(ctx context.Context, b *bundle.Bundle) error {
 	switch m.goal {
 	case GoalDeploy:
 		return b.Locker.Unlock(ctx)
+	case GoalBind, GoalUnbind:
+		return b.Locker.Unlock(ctx)
 	case GoalDestroy:
 		return b.Locker.Unlock(ctx, locker.AllowLockFileNotExist)
 	default:

bundle/deploy/terraform/import.go

@@ -0,0 +1,108 @@
+package terraform
+
+import (
+	"bytes"
+	"context"
+	"fmt"
+	"io"
+	"os"
+	"path/filepath"
+
+	"github.com/databricks/cli/bundle"
+	"github.com/databricks/cli/libs/cmdio"
+	"github.com/hashicorp/terraform-exec/tfexec"
+)
+
+type BindOptions struct {
+	AutoApprove  bool
+	ResourceType string
+	ResourceKey  string
+	ResourceId   string
+}
+
+type importResource struct {
+	opts *BindOptions
+}
+
+// Apply implements bundle.Mutator.
+func (m *importResource) Apply(ctx context.Context, b *bundle.Bundle) error {
+	dir, err := Dir(ctx, b)
+	if err != nil {
+		return err
+	}
+
+	tf := b.Terraform
+	if tf == nil {
+		return fmt.Errorf("terraform not initialized")
+	}
+
+	err = tf.Init(ctx, tfexec.Upgrade(true))
+	if err != nil {
+		return fmt.Errorf("terraform init: %w", err)
+	}
+	tmpDir, err := os.MkdirTemp("", "state-*")
+	if err != nil {
+		return fmt.Errorf("terraform init: %w", err)
+	}
+	tmpState := filepath.Join(tmpDir, TerraformStateFileName)
+
+	importAddress := fmt.Sprintf("%s.%s", m.opts.ResourceType, m.opts.ResourceKey)
+	err = tf.Import(ctx, importAddress, m.opts.ResourceId, tfexec.StateOut(tmpState))
+	if err != nil {
+		return fmt.Errorf("terraform import: %w", err)
+	}
+
+	buf := bytes.NewBuffer(nil)
+	tf.SetStdout(buf)
+
+	//lint:ignore SA1019 We use legacy -state flag for now to plan the import changes based on temporary state file
+	changed, err := tf.Plan(ctx, tfexec.State(tmpState), tfexec.Target(importAddress))
+	if err != nil {
+		return fmt.Errorf("terraform plan: %w", err)
+	}
+
+	defer os.RemoveAll(tmpDir)
+
+	if changed && !m.opts.AutoApprove {
+		output := buf.String()
+		// Remove output starting from Warning until end of output
+		output = output[:bytes.Index([]byte(output), []byte("Warning:"))]
+		cmdio.LogString(ctx, output)
+		ans, err := cmdio.AskYesOrNo(ctx, "Confirm import changes? Changes will be remotely applied only after running 'bundle deploy'.")
+		if err != nil {
+			return err
+		}
+		if !ans {
+			return fmt.Errorf("import aborted")
+		}
+	}
+
+	// If user confirmed changes, move the state file from temp dir to state location
+	f, err := os.Create(filepath.Join(dir, TerraformStateFileName))
+	if err != nil {
+		return err
+	}
+	defer f.Close()
+
+	tmpF, err := os.Open(tmpState)
+	if err != nil {
+		return err
+	}
+	defer tmpF.Close()
+
+	_, err = io.Copy(f, tmpF)
+	if err != nil {
+		return err
+	}
+
+	return nil
+}
+
+// Name implements bundle.Mutator.
+func (*importResource) Name() string {
+	return "terraform.Import"
+}
+
+func Import(opts *BindOptions) bundle.Mutator {
+	return &importResource{opts: opts}
+}

bundle/deploy/terraform/unbind.go

@@ -0,0 +1,41 @@
+package terraform
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/databricks/cli/bundle"
+	"github.com/hashicorp/terraform-exec/tfexec"
+)
+
+type unbind struct {
+	resourceType string
+	resourceKey  string
+}
+
+func (m *unbind) Apply(ctx context.Context, b *bundle.Bundle) error {
+	tf := b.Terraform
+	if tf == nil {
+		return fmt.Errorf("terraform not initialized")
+	}
+
+	err := tf.Init(ctx, tfexec.Upgrade(true))
+	if err != nil {
+		return fmt.Errorf("terraform init: %w", err)
+	}
+
+	err = tf.StateRm(ctx, fmt.Sprintf("%s.%s", m.resourceType, m.resourceKey))
+	if err != nil {
+		return fmt.Errorf("terraform state rm: %w", err)
+	}
+
+	return nil
+}
+
+func (*unbind) Name() string {
+	return "terraform.Unbind"
+}
+
+func Unbind(resourceType string, resourceKey string) bundle.Mutator {
+	return &unbind{resourceType: resourceType, resourceKey: resourceKey}
+}

bundle/phases/bind.go

@@ -0,0 +1,45 @@
+package phases
+
+import (
+	"github.com/databricks/cli/bundle"
+	"github.com/databricks/cli/bundle/deploy/lock"
+	"github.com/databricks/cli/bundle/deploy/terraform"
+)
+
+func Bind(opts *terraform.BindOptions) bundle.Mutator {
+	return newPhase(
+		"bind",
+		[]bundle.Mutator{
+			lock.Acquire(),
+			bundle.Defer(
+				bundle.Seq(
+					terraform.StatePull(),
+					terraform.Interpolate(),
+					terraform.Write(),
+					terraform.Import(opts),
+					terraform.StatePush(),
+				),
+				lock.Release(lock.GoalBind),
+			),
+		},
+	)
+}
+
+func Unbind(resourceType string, resourceKey string) bundle.Mutator {
+	return newPhase(
+		"unbind",
+		[]bundle.Mutator{
+			lock.Acquire(),
+			bundle.Defer(
+				bundle.Seq(
+					terraform.StatePull(),
+					terraform.Interpolate(),
+					terraform.Write(),
+					terraform.Unbind(resourceType, resourceKey),
+					terraform.StatePush(),
+				),
+				lock.Release(lock.GoalUnbind),
+			),
+		},
+	)
+}

bundle/phases/destroy.go

@@ -14,9 +14,9 @@ func Destroy() bundle.Mutator {
 		lock.Acquire(),
 		bundle.Defer(
 			bundle.Seq(
+				terraform.StatePull(),
 				terraform.Interpolate(),
 				terraform.Write(),
-				terraform.StatePull(),
 				terraform.Plan(terraform.PlanGoal("destroy")),
 				terraform.Destroy(),
 				terraform.StatePush(),

cmd/bundle/bundle.go

@@ -1,6 +1,7 @@
 package bundle
 
 import (
+	"github.com/databricks/cli/cmd/bundle/deployment"
 	"github.com/spf13/cobra"
 )
 
@@ -24,5 +25,6 @@ func New() *cobra.Command {
 	cmd.AddCommand(newInitCommand())
 	cmd.AddCommand(newSummaryCommand())
 	cmd.AddCommand(newGenerateCommand())
+	cmd.AddCommand(deployment.NewDeploymentCommand())
 	return cmd
 }

cmd/bundle/deploy.go

@@ -3,14 +3,15 @@ package bundle
 import (
 	"github.com/databricks/cli/bundle"
 	"github.com/databricks/cli/bundle/phases"
+	"github.com/databricks/cli/cmd/bundle/utils"
 	"github.com/spf13/cobra"
 )
 
 func newDeployCommand() *cobra.Command {
 	cmd := &cobra.Command{
 		Use:     "deploy",
 		Short:   "Deploy bundle",
-		PreRunE: ConfigureBundleWithVariables,
+		PreRunE: utils.ConfigureBundleWithVariables,
 	}
 
 	var force bool