Added deployment state for bundles

bundle/deploy/filer.go

@@ -0,0 +1,14 @@
+package deploy
+
+import (
+	"github.com/databricks/cli/bundle"
+	"github.com/databricks/cli/libs/filer"
+)
+
+// FilerFactory is a function that returns a filer.Filer.
+type FilerFactory func(b *bundle.Bundle) (filer.Filer, error)
+
+// StateFiler returns a filer.Filer that can be used to read/write state files.
+func StateFiler(b *bundle.Bundle) (filer.Filer, error) {
+	return filer.NewWorkspaceFilesClient(b.WorkspaceClient(), b.Config.Workspace.StatePath)
+}

bundle/deploy/files/delete.go

@@ -45,7 +45,7 @@ func (m *delete) Apply(ctx context.Context, b *bundle.Bundle) error {
 	}
 
 	// Clean up sync snapshot file
-	sync, err := getSync(ctx, b)
+	sync, err := GetSync(ctx, b)
 	if err != nil {
 		return err
 	}

bundle/deploy/files/sync.go

@@ -8,7 +8,15 @@ import (
 	"github.com/databricks/cli/libs/sync"
 )
 
-func getSync(ctx context.Context, b *bundle.Bundle) (*sync.Sync, error) {
+func GetSync(ctx context.Context, b *bundle.Bundle) (*sync.Sync, error) {
+	opts, err := GetSyncOptions(ctx, b)
+	if err != nil {
+		return nil, fmt.Errorf("cannot get sync options: %w", err)
+	}
+	return sync.New(ctx, *opts)
+}
+
+func GetSyncOptions(ctx context.Context, b *bundle.Bundle) (*sync.SyncOptions, error) {
 	cacheDir, err := b.CacheDir(ctx)
 	if err != nil {
 		return nil, fmt.Errorf("cannot get bundle cache directory: %w", err)
@@ -19,17 +27,22 @@ func getSync(ctx context.Context, b *bundle.Bundle) (*sync.Sync, error) {
 		return nil, fmt.Errorf("cannot get list of sync includes: %w", err)
 	}
 
-	opts := sync.SyncOptions{
+	opts := &sync.SyncOptions{
 		LocalPath:  b.Config.Path,
 		RemotePath: b.Config.Workspace.FilePath,
 		Include:    includes,
 		Exclude:    b.Config.Sync.Exclude,
+		Host:       b.WorkspaceClient().Config.Host,
 
-		Full:        false,
-		CurrentUser: b.Config.Workspace.CurrentUser.User,
+		Full: false,
 
 		SnapshotBasePath: cacheDir,
 		WorkspaceClient:  b.WorkspaceClient(),
 	}
-	return sync.New(ctx, opts)
+
+	if b.Config.Workspace.CurrentUser != nil {
+		opts.CurrentUser = b.Config.Workspace.CurrentUser.User
+	}
+
+	return opts, nil
 }

bundle/deploy/files/upload.go

@@ -17,7 +17,7 @@ func (m *upload) Name() string {
 
 func (m *upload) Apply(ctx context.Context, b *bundle.Bundle) error {
 	cmdio.LogString(ctx, fmt.Sprintf("Uploading bundle files to %s...", b.Config.Workspace.FilePath))
-	sync, err := getSync(ctx, b)
+	sync, err := GetSync(ctx, b)
 	if err != nil {
 		return err
 	}

bundle/deploy/state.go

@@ -0,0 +1,168 @@
+package deploy
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"io"
+	"io/fs"
+	"os"
+	"path/filepath"
+	"time"
+
+	"github.com/databricks/cli/bundle"
+	"github.com/databricks/cli/libs/fileset"
+)
+
+const DeploymentStateFileName = "deployment.json"
+const DeploymentStateVersion = 1
+
+type File struct {
+	Path       string `json:"path"`
+	IsNotebook bool   `json:"is_notebook"`
+}
+
+type Filelist []File
+
+type DeploymentState struct {
+	// Version is the version of the deployment state.
+	Version int64 `json:"version"`
+
+	// Seq is the sequence number of the deployment state.
+	// This number is incremented on every deployment.
+	// It is used to detect if the deployment state is stale.
+	Seq int64 `json:"seq"`
+
+	// CliVersion is the version of the CLI which created the deployment state.
+	CliVersion string `json:"cli_version"`
+
+	// Timestamp is the time when the deployment state was created.
+	Timestamp time.Time `json:"timestamp"`
+
+	// Files is a list of files which has been deployed as part of this deployment.
+	Files Filelist `json:"files"`
+}
+
+// We use this entry type as a proxy to fs.DirEntry.
+// When we construct sync snapshot from deployment state,
+// we use a fileset.File which embeds fs.DirEntry as the DirEntry field.
+// Because we can't marshal/unmarshal fs.DirEntry directly, instead when we unmarshal
+// the deployment state, we use this entry type to represent the fs.DirEntry in fileset.File instance.
+type entry struct {
+	path string
+	info fs.FileInfo
+}
+
+func newEntry(path string) *entry {
+	info, err := os.Stat(path)
+	if err != nil {
+		return &entry{path, nil}
+	}
+
+	return &entry{path, info}
+}
+
+func (e *entry) Name() string {
+	return filepath.Base(e.path)
+}
+
+func (e *entry) IsDir() bool {
+	// If the entry is nil, it is a non-existent file so return false.
+	if e.info == nil {
+		return false
+	}
+	return e.info.IsDir()
+}
+
+func (e *entry) Type() fs.FileMode {
+	// If the entry is nil, it is a non-existent file so return 0.
+	if e.info == nil {
+		return 0
+	}
+	return e.info.Mode()
+}
+
+func (e *entry) Info() (fs.FileInfo, error) {
+	if e.info == nil {
+		return nil, fmt.Errorf("no info available")
+	}
+	return e.info, nil
+}
+
+func FromSlice(files []fileset.File) (Filelist, error) {
+	var f Filelist
+	for _, file := range files {
+		isNotebook, err := file.IsNotebook()
+		if err != nil {
+			return nil, err
+		}
+		f = append(f, File{
+			Path:       file.Relative,
+			IsNotebook: isNotebook,
+		})
+	}
+	return f, nil
+}
+
+func (f Filelist) ToSlice(basePath string) []fileset.File {
+	var files []fileset.File
+	for _, file := range f {
+		absPath := filepath.Join(basePath, file.Path)
+		if file.IsNotebook {
+			files = append(files, fileset.NewNotebookFile(newEntry(absPath), absPath, file.Path))
+		} else {
+			files = append(files, fileset.NewSourceFile(newEntry(absPath), absPath, file.Path))
+		}
+	}
+	return files
+}
+
+func isLocalStateStale(local io.Reader, remote io.Reader) bool {
+	localState, err := loadState(local)
+	if err != nil {
+		return true
+	}
+
+	remoteState, err := loadState(remote)
+	if err != nil {
+		return false
+	}
+
+	return localState.Seq < remoteState.Seq
+}
+
+func validateRemoteStateCompatibility(remote io.Reader) error {
+	state, err := loadState(remote)
+	if err != nil {
+		return err
+	}
+
+	// If the remote state version is greater than the CLI version, we can't proceed.
+	if state.Version > DeploymentStateVersion {
+		return fmt.Errorf("remote deployment state is incompatible with current version of CLI, please upgarde to at least %s", state.CliVersion)
+	}
+
+	return nil
+}
+
+func loadState(r io.Reader) (*DeploymentState, error) {
+	content, err := io.ReadAll(r)
+	if err != nil {
+		return nil, err
+	}
+	var s DeploymentState
+	err = json.Unmarshal(content, &s)
+	if err != nil {
+		return nil, err
+	}
+
+	return &s, nil
+}
+
+func getPathToStateFile(ctx context.Context, b *bundle.Bundle) (string, error) {
+	cacheDir, err := b.CacheDir(ctx)
+	if err != nil {
+		return "", fmt.Errorf("cannot get bundle cache directory: %w", err)
+	}
+	return filepath.Join(cacheDir, DeploymentStateFileName), nil
+}

bundle/deploy/state_pull.go

@@ -0,0 +1,146 @@
+package deploy
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"errors"
+	"io"
+	"io/fs"
+	"os"
+	"time"
+
+	"github.com/databricks/cli/bundle"
+	"github.com/databricks/cli/bundle/deploy/files"
+	"github.com/databricks/cli/libs/filer"
+	"github.com/databricks/cli/libs/log"
+	"github.com/databricks/cli/libs/sync"
+)
+
+type statePull struct {
+	filerFactory FilerFactory
+}
+
+func (s *statePull) Apply(ctx context.Context, b *bundle.Bundle) error {
+	f, err := s.filerFactory(b)
+	if err != nil {
+		return err
+	}
+
+	// Download deployment state file from filer to local cache directory.
+	log.Infof(ctx, "Opening remote deployment state file")
+	remote, err := s.remoteState(ctx, f)
+	if err != nil {
+		log.Infof(ctx, "Unable to open remote deployment state file: %s", err)
+		return err
+	}
+	if remote == nil {
+		log.Infof(ctx, "Remote deployment state file does not exist")
+		return nil
+	}
+
+	statePath, err := getPathToStateFile(ctx, b)
+	if err != nil {
+		return err
+	}
+
+	local, err := os.OpenFile(statePath, os.O_CREATE|os.O_RDWR, 0600)
+	if err != nil {
+		return err
+	}
+	defer local.Close()
+
+	data := remote.Bytes()
+	err = validateRemoteStateCompatibility(bytes.NewReader(data))
+	if err != nil {
+		return err
+	}
+
+	if !isLocalStateStale(local, bytes.NewReader(data)) {
+		log.Infof(ctx, "Local deployment state is the same or newer, ignoring remote state")
+		return nil
+	}
+
+	// Truncating the file before writing
+	local.Truncate(0)
+	local.Seek(0, 0)
+
+	// Write file to disk.
+	log.Infof(ctx, "Writing remote deployment state file to local cache directory")
+	_, err = io.Copy(local, bytes.NewReader(data))
+	if err != nil {
+		return err
+	}
+
+	opts, err := files.GetSyncOptions(ctx, b)
+	if err != nil {
+		return err
+	}
+
+	snapshotPath, err := sync.SnapshotPath(opts)
+	if err != nil {
+		return err
+	}
+
+	var state DeploymentState
+	err = json.Unmarshal(data, &state)
+	if err != nil {
+		return err
+	}
+
+	// Create a new snapshot based on the deployment state file.
+	log.Infof(ctx, "Creating new snapshot")
+	snapshotState, err := sync.NewSnapshotState(state.Files.ToSlice(b.Config.Path))
+	if err != nil {
+		return err
+	}
+
+	// Reset last modified times to 0 to make sure all files are synced
+	for k := range snapshotState.LastModifiedTimes {
+		snapshotState.LastModifiedTimes[k] = time.Unix(0, 0)
+	}
+
+	snapshot := &sync.Snapshot{
+		SnapshotPath:  snapshotPath,
+		New:           true,
+		Version:       sync.LatestSnapshotVersion,
+		Host:          opts.Host,
+		RemotePath:    opts.RemotePath,
+		SnapshotState: snapshotState,
+	}
+
+	// Persist the snapshot to disk.
+	log.Infof(ctx, "Persisting snapshot to disk")
+	return snapshot.Save(ctx)
+}
+
+func (s *statePull) remoteState(ctx context.Context, f filer.Filer) (*bytes.Buffer, error) {
+	// Download deployment state file from filer to local cache directory.
+	remote, err := f.Read(ctx, DeploymentStateFileName)
+	if err != nil {
+		// On first deploy this file doesn't yet exist.
+		if errors.Is(err, fs.ErrNotExist) {
+			return nil, nil
+		}
+		return nil, err
+	}
+
+	defer remote.Close()
+
+	var buf bytes.Buffer
+	_, err = io.Copy(&buf, remote)
+	if err != nil {
+		return nil, err
+	}
+
+	return &buf, nil
+}
+
+func (s *statePull) Name() string {
+	return "deploy:state-pull"
+}
+
+// StatePull returns a mutator that pulls the deployment state from the Databricks workspace
+func StatePull() bundle.Mutator {
+	return &statePull{StateFiler}
+}