Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Added deployment state for bundles #1267

Merged
merged 9 commits into from
Mar 18, 2024
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
24 changes: 9 additions & 15 deletions bundle/config/mutator/process_root_includes_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -4,7 +4,6 @@ import (
"context"
"os"
"path"
"path/filepath"
"runtime"
"strings"
"testing"
Expand All @@ -13,16 +12,11 @@ import (
"github.com/databricks/cli/bundle/config"
"github.com/databricks/cli/bundle/config/mutator"
"github.com/databricks/cli/bundle/env"
"github.com/databricks/cli/internal/testutil"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)

func touch(t *testing.T, path, file string) {
f, err := os.Create(filepath.Join(path, file))
require.NoError(t, err)
f.Close()
}

func TestProcessRootIncludesEmpty(t *testing.T) {
b := &bundle.Bundle{
Config: config.Root{
Expand Down Expand Up @@ -64,9 +58,9 @@ func TestProcessRootIncludesSingleGlob(t *testing.T) {
},
}

touch(t, b.Config.Path, "databricks.yml")
touch(t, b.Config.Path, "a.yml")
touch(t, b.Config.Path, "b.yml")
testutil.Touch(t, b.Config.Path, "databricks.yml")
testutil.Touch(t, b.Config.Path, "a.yml")
testutil.Touch(t, b.Config.Path, "b.yml")

err := bundle.Apply(context.Background(), b, mutator.ProcessRootIncludes())
require.NoError(t, err)
Expand All @@ -85,8 +79,8 @@ func TestProcessRootIncludesMultiGlob(t *testing.T) {
},
}

touch(t, b.Config.Path, "a1.yml")
touch(t, b.Config.Path, "b1.yml")
testutil.Touch(t, b.Config.Path, "a1.yml")
testutil.Touch(t, b.Config.Path, "b1.yml")

err := bundle.Apply(context.Background(), b, mutator.ProcessRootIncludes())
require.NoError(t, err)
Expand All @@ -105,7 +99,7 @@ func TestProcessRootIncludesRemoveDups(t *testing.T) {
},
}

touch(t, b.Config.Path, "a.yml")
testutil.Touch(t, b.Config.Path, "a.yml")

err := bundle.Apply(context.Background(), b, mutator.ProcessRootIncludes())
require.NoError(t, err)
Expand All @@ -129,7 +123,7 @@ func TestProcessRootIncludesNotExists(t *testing.T) {
func TestProcessRootIncludesExtrasFromEnvVar(t *testing.T) {
rootPath := t.TempDir()
testYamlName := "extra_include_path.yml"
touch(t, rootPath, testYamlName)
testutil.Touch(t, rootPath, testYamlName)
t.Setenv(env.IncludesVariable, path.Join(rootPath, testYamlName))

b := &bundle.Bundle{
Expand All @@ -146,7 +140,7 @@ func TestProcessRootIncludesExtrasFromEnvVar(t *testing.T) {
func TestProcessRootIncludesDedupExtrasFromEnvVar(t *testing.T) {
rootPath := t.TempDir()
testYamlName := "extra_include_path.yml"
touch(t, rootPath, testYamlName)
testutil.Touch(t, rootPath, testYamlName)
t.Setenv(env.IncludesVariable, strings.Join(
[]string{
path.Join(rootPath, testYamlName),
Expand Down
14 changes: 14 additions & 0 deletions bundle/deploy/filer.go
Original file line number Diff line number Diff line change
@@ -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)
}
2 changes: 1 addition & 1 deletion bundle/deploy/files/delete.go
Original file line number Diff line number Diff line change
Expand Up @@ -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
}
Expand Down
23 changes: 18 additions & 5 deletions bundle/deploy/files/sync.go
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand All @@ -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,
andrewnester marked this conversation as resolved.
Show resolved Hide resolved

SnapshotBasePath: cacheDir,
WorkspaceClient: b.WorkspaceClient(),
}
return sync.New(ctx, opts)

if b.Config.Workspace.CurrentUser != nil {
opts.CurrentUser = b.Config.Workspace.CurrentUser.User
}
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Curious: when is this nil?

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Good point, I don't think realistically it can, just did the check based on type and it can be potentially nil.


return opts, nil
}
2 changes: 1 addition & 1 deletion bundle/deploy/files/upload.go
Original file line number Diff line number Diff line change
Expand Up @@ -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
}
Expand Down
174 changes: 174 additions & 0 deletions bundle/deploy/state.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,174 @@
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 {
LocalPath string `json:"local_path"`

// If true, this file is a notebook.
// This property must be persisted because notebooks are stripped of their extension.
// If the local file is no longer present, we need to know what to remove on the workspace side.
IsNotebook bool `json:"is_notebook"`
}

type Filelist []File

type DeploymentState struct {
// Version is the version of the deployment state.
andrewnester marked this conversation as resolved.
Show resolved Hide resolved
// To be incremented when the schema changes.
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 {
andrewnester marked this conversation as resolved.
Show resolved Hide resolved
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 k := range files {
file := &files[k]
isNotebook, err := file.IsNotebook()
if err != nil {
return nil, err
}
f = append(f, File{
LocalPath: 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.LocalPath)
if file.IsNotebook {
files = append(files, fileset.NewNotebookFile(newEntry(absPath), absPath, file.LocalPath))
} else {
files = append(files, fileset.NewSourceFile(newEntry(absPath), absPath, file.LocalPath))
}
}
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 the current version of the CLI, please upgrade 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
}
Loading
Loading