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

With deployment history configured, plan/deploy/destroy read resources.json and only then
get-status resources.deployment.json to find the deployment id. readStates now runs that
lookup concurrently with the state reads and passes the result on StateDesc, saving one
sequential round trip.

Co-authored-by: Isaac <no-reply@databricks.com>

Author: denik

bundle/statemgmt/state_pull.go

@@ -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"
@@ -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 {
@@ -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
@@ -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)
 
@@ -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)
@@ -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
@@ -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
@@ -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 {

cmd/bundle/utils/process.go

@@ -4,7 +4,6 @@ import (
 	"context"
 	"errors"
 	"fmt"
-	"path"
 	"path/filepath"
 	"strconv"
 	"time"
@@ -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"
@@ -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"
@@ -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
@@ -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)
@@ -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),
 	})
@@ -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
 	}
@@ -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: {}}
@@ -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
@@ -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

libs/dms/metadata.go

@@ -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"
 )
 
@@ -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 {