Skip to content

bundle: look up the deployment-history node alongside the remote state reads #7086

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.

Sign up for GitHub

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

Open
denik wants to merge 1 commit into from
Open

bundle: look up the deployment-history node alongside the remote state reads #7086

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
43 changes: 38 additions & 5 deletions bundle/statemgmt/state_pull.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,8 +17,10 @@ import (
"github.com/databricks/cli/bundle"
"github.com/databricks/cli/bundle/config/engine"
"github.com/databricks/cli/bundle/deploy"
bundleenv "github.com/databricks/cli/bundle/env"
"github.com/databricks/cli/libs/atomicfile"
"github.com/databricks/cli/libs/diag"
"github.com/databricks/cli/libs/dms"
"github.com/databricks/cli/libs/filer"
"github.com/databricks/cli/libs/log"
"github.com/databricks/cli/libs/logdiag"
Expand All @@ -43,6 +45,16 @@ type StateDesc struct {
IsLocal bool `json:"-"`

AllStates []*StateDesc

// Deployment is the deployment-history lookup made alongside the remote state reads; nil when
// none was made (state not pulled, or deployment history not configured).
Deployment *DeploymentLookup `json:"-"`
}

// DeploymentLookup is the result of dms.LookupDeploymentID.
type DeploymentLookup struct {
ID string
Err error
}

func (s *StateDesc) String() string {
Expand Down Expand Up @@ -151,7 +163,7 @@ func PullResourcesState(ctx context.Context, b *bundle.Bundle, alwaysPull Always
_, localPathDirect := b.StateFilenameDirect(ctx)
_, localPathTerraform := b.StateFilenameTerraform(ctx)

states := readStates(ctx, b, alwaysPull)
states, deployment := readStates(ctx, b, alwaysPull)

if logdiag.HasError(ctx) {
return nil
Expand All @@ -170,6 +182,7 @@ func PullResourcesState(ctx context.Context, b *bundle.Bundle, alwaysPull Always
winner = states[len(states)-1]
winner.AllStates = states
}
winner.Deployment = deployment

log.Infof(ctx, "Available resource state files (from least to most preferred): %v", states)

Expand Down Expand Up @@ -206,14 +219,15 @@ func PullResourcesState(ctx context.Context, b *bundle.Bundle, alwaysPull Always
return winner
}

func readStates(ctx context.Context, b *bundle.Bundle, alwaysPull AlwaysPull) []*StateDesc {
func readStates(ctx context.Context, b *bundle.Bundle, alwaysPull AlwaysPull) ([]*StateDesc, *DeploymentLookup) {
var states []*StateDesc
var deployment *DeploymentLookup

remotePathDirect, localPathDirect := b.StateFilenameDirect(ctx)
remotePathTerraform, localPathTerraform := b.StateFilenameTerraform(ctx)

if logdiag.HasError(ctx) {
return nil
return nil, nil
}

directLocalState := localRead(ctx, localPathDirect, engine.EngineDirect)
Expand All @@ -223,7 +237,7 @@ func readStates(ctx context.Context, b *bundle.Bundle, alwaysPull AlwaysPull) []
f, err := deploy.StateFiler(ctx, b)
if err != nil {
logdiag.LogError(ctx, err)
return nil
return nil, nil
}

var wg sync.WaitGroup
Expand All @@ -237,6 +251,16 @@ func readStates(ctx context.Context, b *bundle.Bundle, alwaysPull AlwaysPull) []
terraformRemoteState = filerRead(ctx, f, remotePathTerraform, engine.EngineTerraform)
})

// Whether the state records deployment history is only known once it is read, but the
// lookup is needed right after; going by the config overlaps it with the reads. A state
// that disagrees with the config is either rejected or looked up again by the caller.
if ConfiguresDeploymentHistory(ctx, b) {
wg.Go(func() {
id, err := dms.LookupDeploymentID(ctx, b.WorkspaceClient(ctx), b.Config.Workspace.StatePath)
deployment = &DeploymentLookup{ID: id, Err: err}
})
}

wg.Wait()

// find highest serial across all state files
Expand All @@ -250,7 +274,16 @@ func readStates(ctx context.Context, b *bundle.Bundle, alwaysPull AlwaysPull) []
return a.Serial - b.Serial
})

return states
return states, deployment
}

// ConfiguresDeploymentHistory reports whether the bundle config (or env) turns deployment history on.
func ConfiguresDeploymentHistory(ctx context.Context, b *bundle.Bundle) bool {
var configured *bool
if b.Config.Experimental != nil {
configured = b.Config.Experimental.DeploymentHistory
}
return bundleenv.RecordsDeploymentHistory(ctx, configured)
}

func validateStates(states []*StateDesc) error {
Expand Down
39 changes: 15 additions & 24 deletions cmd/bundle/utils/process.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4,7 +4,6 @@ import (
"context"
"errors"
"fmt"
"path"
"path/filepath"
"strconv"
"time"
Expand All @@ -18,7 +17,6 @@ import (
"github.com/databricks/cli/bundle/deployplan"
"github.com/databricks/cli/bundle/direct"
"github.com/databricks/cli/bundle/direct/dstate"
bundleenv "github.com/databricks/cli/bundle/env"
"github.com/databricks/cli/bundle/phases"
"github.com/databricks/cli/bundle/scripts"
"github.com/databricks/cli/bundle/statemgmt"
Expand All @@ -35,7 +33,6 @@ import (
"github.com/databricks/cli/libs/telemetry"
"github.com/databricks/cli/libs/telemetry/protos"
"github.com/databricks/databricks-sdk-go"
"github.com/databricks/databricks-sdk-go/apierr"
"github.com/databricks/databricks-sdk-go/service/bundledeployments"
"github.com/databricks/databricks-sdk-go/useragent"
"github.com/spf13/cobra"
Expand Down Expand Up @@ -372,7 +369,7 @@ func ProcessBundleRet(cmd *cobra.Command, opts ProcessOptions) (b *bundle.Bundle

if needDirectState && stateDesc.IsDMS() {
var err error
dmsDeploymentID, dmsDeployment, err = fetchDeploymentFromStatePath(ctx, b.WorkspaceClient(ctx), b.Config.Workspace.StatePath)
dmsDeploymentID, dmsDeployment, err = fetchDeploymentFromStatePath(ctx, b.WorkspaceClient(ctx), b.Config.Workspace.StatePath, stateDesc)
if err != nil {
logdiag.LogError(ctx, err)
return b, stateDesc, root.ErrAlreadyPrinted
Expand Down Expand Up @@ -586,6 +583,8 @@ func ValidateEngineSetting(ctx context.Context, b *bundle.Bundle) error {

// Lookup and return the deployment object from ${workspace.state_path}/resources.deployment.json
//
// The id comes from stateDesc.Deployment when the state pull already looked it up.
//
// TODO: a deployment is only usable when both the node and the service's record exist, and a
// half-created one - node present, record missing - blocks the bundle here even though
// CreateDeployment already recovers from it. Move this behind a dms.ReadDeployment(ctx, statePath)
Expand All @@ -594,17 +593,17 @@ func ValidateEngineSetting(ctx context.Context, b *bundle.Bundle) error {
//
// TODO: ask the service for a lookup by state path, so this is one round trip rather than two - a
// workspace lookup to turn the node into an id, then a get by that id.
func fetchDeploymentFromStatePath(ctx context.Context, w *databricks.WorkspaceClient, statePath string) (string, *bundledeployments.Deployment, error) {
nodePath := path.Join(statePath, dms.DeploymentNodeName)

obj, err := w.Workspace.GetStatusByPath(ctx, nodePath)
if errors.Is(err, apierr.ErrNotFound) || errors.Is(err, apierr.ErrResourceDoesNotExist) {
return "", nil, nil
func fetchDeploymentFromStatePath(ctx context.Context, w *databricks.WorkspaceClient, statePath string, stateDesc *statemgmt.StateDesc) (string, *bundledeployments.Deployment, error) {
var deploymentID string
var err error
if stateDesc.Deployment != nil {
deploymentID, err = stateDesc.Deployment.ID, stateDesc.Deployment.Err
} else {
deploymentID, err = dms.LookupDeploymentID(ctx, w, statePath)
}
if err != nil {
return "", nil, fmt.Errorf("looking up deployment at %s: %w", nodePath, err)
if err != nil || deploymentID == "" {
return "", nil, err
}
deploymentID := strconv.FormatInt(obj.ObjectId, 10)
deployment, err := w.BundleDeployments.GetDeployment(ctx, bundledeployments.GetDeploymentRequest{
Name: dms.DeploymentName(deploymentID),
})
Expand Down Expand Up @@ -639,7 +638,7 @@ func OpenDirectStateForRead(ctx context.Context, b *bundle.Bundle, stateDesc *st
return enforceDeploymentHistorySetting(ctx, b, stateDesc, false)
}

dmsDeploymentID, dmsDeployment, err := fetchDeploymentFromStatePath(ctx, b.WorkspaceClient(ctx), b.Config.Workspace.StatePath)
dmsDeploymentID, dmsDeployment, err := fetchDeploymentFromStatePath(ctx, b.WorkspaceClient(ctx), b.Config.Workspace.StatePath, stateDesc)
if err != nil {
return err
}
Expand All @@ -658,7 +657,7 @@ func OpenDirectStateForRead(ctx context.Context, b *bundle.Bundle, stateDesc *st
}

func resolveDeploymentHistory(ctx context.Context, b *bundle.Bundle, stateDesc *statemgmt.StateDesc) bool {
configured := configuresDeploymentHistory(ctx, b)
configured := statemgmt.ConfiguresDeploymentHistory(ctx, b)
if stateDesc.SourcePath == "" {
if configured {
stateDesc.Features = map[string]struct{}{dstate.FeatureDeploymentHistory: {}}
Expand All @@ -673,7 +672,7 @@ func enforceDeploymentHistorySetting(ctx context.Context, b *bundle.Bundle, stat
if stateDesc.SourcePath == "" {
return nil
}
configured := configuresDeploymentHistory(ctx, b)
configured := statemgmt.ConfiguresDeploymentHistory(ctx, b)
recorded := stateDesc.IsDMS()
if configured == recorded {
return nil
Expand All @@ -692,14 +691,6 @@ Run "databricks bundle destroy" first, then deploy again with deployment history
return nil
}

func configuresDeploymentHistory(ctx context.Context, b *bundle.Bundle) bool {
var configured *bool
if b.Config.Experimental != nil {
configured = b.Config.Experimental.DeploymentHistory
}
return bundleenv.RecordsDeploymentHistory(ctx, configured)
}

// isNewerVersion reports whether the state's recorded CLI version is strictly
// newer than the running build. Both are bare versions without a leading "v".
// An empty stateVersion (state not written by any CLI yet) or an unparseable
Expand Down
19 changes: 19 additions & 0 deletions libs/dms/metadata.go
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,11 +1,16 @@
package dms

import (
"context"
"errors"
"fmt"
"path"
"strconv"
"strings"

"github.com/databricks/cli/libs/structs/structdiff"
"github.com/databricks/databricks-sdk-go"
"github.com/databricks/databricks-sdk-go/apierr"
"github.com/databricks/databricks-sdk-go/service/bundledeployments"
)

Expand Down Expand Up @@ -85,6 +90,20 @@ func NextVersion(lastVersionID string) (int, error) {
// Keep it equal to the service's DEPLOYMENT_NODE_NAME.
const DeploymentNodeName = "resources.deployment.json"

// LookupDeploymentID returns the id of the deployment whose node sits in statePath, the node's
// object id. It returns "" when there is no node, i.e. nothing was deployed with history yet.
func LookupDeploymentID(ctx context.Context, w *databricks.WorkspaceClient, statePath string) (string, error) {
nodePath := path.Join(statePath, DeploymentNodeName)
obj, err := w.Workspace.GetStatusByPath(ctx, nodePath)
if errors.Is(err, apierr.ErrNotFound) || errors.Is(err, apierr.ErrResourceDoesNotExist) {
return "", nil
}
if err != nil {
return "", fmt.Errorf("looking up deployment at %s: %w", nodePath, err)
}
return strconv.FormatInt(obj.ObjectId, 10), nil
}

// DeploymentName, VersionName and OperationName build the resource names the
// service uses, so callers pass only ids.
func DeploymentName(deploymentID string) string {
Expand Down
Loading