Skip to content
Draft
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
19 changes: 13 additions & 6 deletions bundle/direct/bind.go
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,7 @@ package direct

import (
"context"
"errors"
"fmt"
"os"
"strings"
Expand Down Expand Up @@ -63,13 +64,16 @@ type BindResult struct {
func (b *DeploymentBundle) Bind(ctx context.Context, client *databricks.WorkspaceClient, configRoot *config.Root, statePath, resourceKey, resourceID string) (*BindResult, error) {
// Check if the resource is already managed (bound to a different ID)
var checkStateDB dstate.DeploymentState
if err := checkStateDB.Open(ctx, statePath, dstate.ReadingFile, dstate.WithRecovery(true), dstate.OpenDmsArgs{}); err != nil {
if err := checkStateDB.Open(ctx, statePath, dstate.WithRecovery(true)); err != nil {
// State that cannot be read is not the same as state without a binding:
// the resource may well be managed already, and binding on top of it
// would take over a resource whose ownership was never checked. A state
// file that does not exist yet opens successfully as an empty one.
return nil, fmt.Errorf("cannot check whether %s is already bound: %w", resourceKey, err)
}
if checkStateDB.IsDeploymentMetadataService() {
return nil, errors.New("bind is not supported for a bundle target that records deployment history")
}
existingID := checkStateDB.GetResourceID(resourceKey)
if _, err := checkStateDB.Finalize(ctx); err != nil {
log.Warnf(ctx, "failed to finalize state: %v", err)
Expand All @@ -92,7 +96,7 @@ func (b *DeploymentBundle) Bind(ctx context.Context, client *databricks.Workspac
}

// Open temp state
err := b.StateDB.Open(ctx, tmpStatePath, dstate.ReadingFile, dstate.WithRecovery(false), dstate.OpenDmsArgs{})
err := b.StateDB.Open(ctx, tmpStatePath, dstate.WithRecovery(false))
if err != nil {
os.Remove(tmpStatePath)
return nil, err
Expand All @@ -119,7 +123,7 @@ func (b *DeploymentBundle) Bind(ctx context.Context, client *databricks.Workspac
log.Infof(ctx, "Bound %s to id=%s (in temp state)", resourceKey, resourceID)

// First plan + update: populate state with resolved config
err = b.StateDB.Open(ctx, tmpStatePath, dstate.ReadingFile, dstate.WithRecovery(true), dstate.OpenDmsArgs{})
err = b.StateDB.Open(ctx, tmpStatePath, dstate.WithRecovery(true))
if err != nil {
os.Remove(tmpStatePath)
return nil, err
Expand Down Expand Up @@ -168,7 +172,7 @@ func (b *DeploymentBundle) Bind(ctx context.Context, client *databricks.Workspac
return nil, fmt.Errorf("compacting state: %w", err)
}

err = b.StateDB.Open(ctx, tmpStatePath, dstate.ReadingFile, dstate.WithRecovery(true), dstate.OpenDmsArgs{})
err = b.StateDB.Open(ctx, tmpStatePath, dstate.WithRecovery(true))
if err != nil {
os.Remove(tmpStatePath)
return nil, err
Expand All @@ -192,7 +196,7 @@ func (b *DeploymentBundle) Bind(ctx context.Context, client *databricks.Workspac
}

// Second plan: this is the plan to present to the user (change between remote resource and config)
err = b.StateDB.Open(ctx, tmpStatePath, dstate.ReadingFile, dstate.WithRecovery(true), dstate.OpenDmsArgs{})
err = b.StateDB.Open(ctx, tmpStatePath, dstate.WithRecovery(true))
if err != nil {
os.Remove(tmpStatePath)
return nil, err
Expand Down Expand Up @@ -242,9 +246,12 @@ func (result *BindResult) Cancel() {
// Unbind removes a resource from direct engine state without deleting
// the workspace resource. Also removes associated permissions/grants entries.
func (b *DeploymentBundle) Unbind(ctx context.Context, statePath, resourceKey string) error {
if err := b.StateDB.Open(ctx, statePath, dstate.ReadingFile, dstate.WithRecovery(true), dstate.OpenDmsArgs{}); err != nil {
if err := b.StateDB.Open(ctx, statePath, dstate.WithRecovery(true)); err != nil {
return err
}
if b.StateDB.IsDeploymentMetadataService() {
return errors.New("unbind is not supported for a bundle target that records deployment history")
}
if err := b.StateDB.UpgradeToWrite(); err != nil {
return err
}
Expand Down
38 changes: 38 additions & 0 deletions bundle/direct/bind_test.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,38 @@
package direct

import (
"encoding/json"
"os"
"path/filepath"
"testing"

"github.com/databricks/cli/bundle/direct/dstate"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)

func TestBindAndUnbindRejectDMSAuthoritativeState(t *testing.T) {
for _, operation := range []string{"bind", "unbind"} {
t.Run(operation, func(t *testing.T) {
path := filepath.Join(t.TempDir(), "resources.json")
state := dstate.NewDatabase("lineage", 1)
state.Features = map[string]struct{}{dstate.FeatureDeploymentHistory: {}}
data, err := json.Marshal(state)
require.NoError(t, err)
require.NoError(t, os.WriteFile(path, data, 0o600))

var db DeploymentBundle
// Reject file-only operations before they need a resource API client.
if operation == "bind" {
_, err = db.Bind(t.Context(), nil, nil, path, "resources.jobs.foo", "123")
} else {
err = db.Unbind(t.Context(), path, "resources.jobs.foo")
}
require.ErrorContains(t, err, operation+" is not supported")
assert.NoFileExists(t, path+".wal")
persisted, err := os.ReadFile(path)
require.NoError(t, err)
assert.Equal(t, data, persisted)
})
}
}
41 changes: 41 additions & 0 deletions bundle/direct/dstate/dms.go
Original file line number Diff line number Diff line change
@@ -1,11 +1,15 @@
package dstate

import (
"context"
"encoding/json"
"fmt"

"github.com/databricks/cli/bundle/deployplan"
"github.com/databricks/cli/libs/cmdctx"
"github.com/databricks/cli/libs/dms"
"github.com/databricks/cli/libs/retry"
"github.com/databricks/databricks-sdk-go/service/bundledeployments"
)

// RecordedState is what the CLI serializes into the DMS operation's state field. It wraps the
Expand All @@ -16,6 +20,43 @@ type RecordedState struct {
DependsOn []deployplan.DependsOnEntry `json:"depends_on,omitempty"`
}

// LoadDMSState loads authoritative resources into an already-open read-only checkpoint.
// The service version must be at least as recent as the file's cached serial.
func (db *DeploymentState) LoadDMSState(ctx context.Context, deploymentID string, lastVersionID int) error {
db.mu.Lock()
defer db.mu.Unlock()
db.AssertOpenedForRead()

if db.Data.Serial > lastVersionID {
return fmt.Errorf("inconsistent deployment state: resource state serial (%d) is ahead of deployment history version (%d)", db.Data.Serial, lastVersionID)
}
if deploymentID != "" {
w := cmdctx.WorkspaceClient(ctx)
recorded, err := retry.With(ctx, retry.IsTransientRead, func() ([]bundledeployments.Resource, error) {
return w.BundleDeployments.ListResourcesAll(ctx, bundledeployments.ListResourcesRequest{
Parent: dms.DeploymentName(deploymentID),
})
})
if err != nil {
return fmt.Errorf("listing resources from the deployment history service: %w", err)
}
resources := make([]dms.Resource, len(recorded))
for i, res := range recorded {
// DMS keys drop the "resources." prefix; add it back for local state.
resources[i] = dms.Resource{
Key: dms.StatePrefix + res.ResourceKey,
ID: res.ResourceId,
State: res.State,
}
}
if err := db.applyDMSState(resources); err != nil {
return err
}
}
db.DeploymentID, db.VersionID = deploymentID, lastVersionID
return nil
}

// applyDMSState fills in the resource state from what DMS recorded. For a deployment that uses DMS
// resources.json does not track state. The service is queried for state for resources and then they
// are filled in-place in DeploymentState.Data.State:
Expand Down
Loading
Loading