bundle: record and read deployment state via DMS with server-generated IDs

acceptance/bundle/dms/record/databricks.yml

@@ -0,0 +1,10 @@
+bundle:
+  name: dms-record
+
+experimental:
+  record_deployment_history: true
+
+resources:
+  jobs:
+    foo:
+      name: foo

acceptance/bundle/dms/record/output.txt

@@ -0,0 +1,140 @@
+
+=== Deploy: the server assigns the deployment ID, and a version + create operation are recorded
+>>> [CLI] bundle deploy
+Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/dms-record/default/files...
+Deploying resources...
+Updating deployment state...
+Deployment complete!
+
+>>> print_requests.py //api/2.0/bundle --sort
+{
+  "method": "POST",
+  "path": "/api/2.0/bundle/deployments",
+  "body": {
+    "target_name": "default"
+  }
+}
+{
+  "method": "POST",
+  "path": "/api/2.0/bundle/deployments/[UUID]/versions",
+  "q": {
+    "version_id": "1"
+  },
+  "body": {
+    "cli_version": "[CLI_VERSION]",
+    "target_name": "default",
+    "version_type": "VERSION_TYPE_DEPLOY"
+  }
+}
+{
+  "method": "POST",
+  "path": "/api/2.0/bundle/deployments/[UUID]/versions/1/complete",
+  "body": {
+    "completion_reason": "VERSION_COMPLETE_SUCCESS"
+  }
+}
+{
+  "method": "POST",
+  "path": "/api/2.0/bundle/deployments/[UUID]/versions/1/operations",
+  "q": {
+    "resource_key": "jobs.foo"
+  },
+  "body": {
+    "action_type": "OPERATION_ACTION_TYPE_CREATE",
+    "resource_id": "[NUMID]",
+    "resource_key": "jobs.foo",
+    "state": {
+      "deployment": {
+        "kind": "BUNDLE",
+        "metadata_file_path": "/Workspace/Users/[USERNAME]/.bundle/dms-record/default/state/metadata.json"
+      },
+      "edit_mode": "UI_LOCKED",
+      "format": "MULTI_TASK",
+      "max_concurrent_runs": 1,
+      "name": "foo",
+      "queue": {
+        "enabled": true
+      }
+    },
+    "status": "OPERATION_STATUS_SUCCEEDED"
+  }
+}
+
+=== The server-assigned deployment ID is persisted in the local state file
+>>> jq .deployment_id .databricks/bundle/default/resources.json
+"[UUID]"
+
+=== Redeploy after deleting the local cache: the deployment ID is recovered from remote state, the same deployment is reused, and the version increments (no new CreateDeployment)
+>>> [CLI] bundle deploy
+Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/dms-record/default/files...
+Deploying resources...
+Updating deployment state...
+Deployment complete!
+
+>>> print_requests.py //api/2.0/bundle --sort
+{
+  "method": "POST",
+  "path": "/api/2.0/bundle/deployments/[UUID]/versions",
+  "q": {
+    "version_id": "2"
+  },
+  "body": {
+    "cli_version": "[CLI_VERSION]",
+    "target_name": "default",
+    "version_type": "VERSION_TYPE_DEPLOY"
+  }
+}
+{
+  "method": "POST",
+  "path": "/api/2.0/bundle/deployments/[UUID]/versions/2/complete",
+  "body": {
+    "completion_reason": "VERSION_COMPLETE_SUCCESS"
+  }
+}
+
+=== Destroy: a destroy version and delete operation are recorded, then the deployment is deleted
+>>> [CLI] bundle destroy --auto-approve
+The following resources will be deleted:
+  delete resources.jobs.foo
+
+All files and directories at the following location will be deleted: /Workspace/Users/[USERNAME]/.bundle/dms-record/default
+
+Deleting files...
+Destroy complete!
+
+>>> print_requests.py //api/2.0/bundle --sort
+{
+  "method": "DELETE",
+  "path": "/api/2.0/bundle/deployments/[UUID]"
+}
+{
+  "method": "POST",
+  "path": "/api/2.0/bundle/deployments/[UUID]/versions",
+  "q": {
+    "version_id": "3"
+  },
+  "body": {
+    "cli_version": "[CLI_VERSION]",
+    "target_name": "default",
+    "version_type": "VERSION_TYPE_DESTROY"
+  }
+}
+{
+  "method": "POST",
+  "path": "/api/2.0/bundle/deployments/[UUID]/versions/3/complete",
+  "body": {
+    "completion_reason": "VERSION_COMPLETE_SUCCESS"
+  }
+}
+{
+  "method": "POST",
+  "path": "/api/2.0/bundle/deployments/[UUID]/versions/3/operations",
+  "q": {
+    "resource_key": "jobs.foo"
+  },
+  "body": {
+    "action_type": "OPERATION_ACTION_TYPE_DELETE",
+    "resource_key": "jobs.foo",
+    "status": "OPERATION_STATUS_SUCCEEDED"
+  }
+}

acceptance/bundle/dms/record/script

@@ -0,0 +1,15 @@
+title "Deploy: the server assigns the deployment ID, and a version + create operation are recorded"
+trace $CLI bundle deploy
+trace print_requests.py //api/2.0/bundle --sort
+
+title "The server-assigned deployment ID is persisted in the local state file"
+trace jq .deployment_id .databricks/bundle/default/resources.json
+
+title "Redeploy after deleting the local cache: the deployment ID is recovered from remote state, the same deployment is reused, and the version increments (no new CreateDeployment)"
+rm -rf .databricks
+trace $CLI bundle deploy
+trace print_requests.py //api/2.0/bundle --sort
+
+title "Destroy: a destroy version and delete operation are recorded, then the deployment is deleted"
+trace $CLI bundle destroy --auto-approve
+trace print_requests.py //api/2.0/bundle --sort

acceptance/bundle/dms/test.toml

@@ -0,0 +1,13 @@
+Local = true
+Cloud = false
+
+# Deployment Metadata Service (DMS) recording is only meaningful in the direct
+# engine, where the deployment ID is stored in and read from the direct-engine
+# state.
+EnvMatrix.DATABRICKS_BUNDLE_ENGINE = ["direct"]
+
+RecordRequests = true
+
+Ignore = [
+    '.databricks',
+]

bundle/configsync/diff.go

@@ -149,7 +149,7 @@ func OpenDeploymentState(ctx context.Context, b *bundle.Bundle, engine engine.En
 
 	deployBundle := &direct.DeploymentBundle{}
 	_, statePath := b.StateFilenameConfigSnapshot(ctx)
-	if err := deployBundle.StateDB.Open(ctx, statePath, dstate.WithRecovery(true), dstate.WithWrite(false)); err != nil {
+	if err := deployBundle.StateDB.Open(ctx, statePath, dstate.WithRecovery(true), dstate.WithWrite(false), nil); err != nil {
 		return nil, fmt.Errorf("failed to open state: %w", err)
 	}
 	return deployBundle, nil

bundle/configsync/variables.go

@@ -147,7 +147,7 @@ func resourceIDLookup(ctx context.Context, b *bundle.Bundle) func(string) string
 	}
 	_, statePath := b.StateFilenameConfigSnapshot(ctx)
 	db := &dstate.DeploymentState{}
-	if err := db.Open(ctx, statePath, dstate.WithRecovery(false), dstate.WithWrite(false)); err != nil {
+	if err := db.Open(ctx, statePath, dstate.WithRecovery(false), dstate.WithWrite(false), nil); err != nil {
 		log.Debugf(ctx, "variable restoration: failed to open state DB at %s: %v", statePath, err)
 		return nil
 	}

bundle/direct/bind.go

@@ -62,7 +62,7 @@ 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.WithRecovery(true), dstate.WithWrite(false)); err == nil {
+	if err := checkStateDB.Open(ctx, statePath, dstate.WithRecovery(true), dstate.WithWrite(false), nil); err == nil {
 		existingID := checkStateDB.GetResourceID(resourceKey)
 		if _, err := checkStateDB.Finalize(ctx); err != nil {
 			log.Warnf(ctx, "failed to finalize state: %v", err)
@@ -86,7 +86,7 @@ func (b *DeploymentBundle) Bind(ctx context.Context, client *databricks.Workspac
 	}
 
 	// Open temp state
-	err := b.StateDB.Open(ctx, tmpStatePath, dstate.WithRecovery(false), dstate.WithWrite(true))
+	err := b.StateDB.Open(ctx, tmpStatePath, dstate.WithRecovery(false), dstate.WithWrite(true), nil)
 	if err != nil {
 		os.Remove(tmpStatePath)
 		return nil, err
@@ -109,7 +109,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.WithRecovery(true), dstate.WithWrite(false))
+	err = b.StateDB.Open(ctx, tmpStatePath, dstate.WithRecovery(true), dstate.WithWrite(false), nil)
 	if err != nil {
 		os.Remove(tmpStatePath)
 		return nil, err
@@ -145,7 +145,7 @@ func (b *DeploymentBundle) Bind(ctx context.Context, client *databricks.Workspac
 			}
 		}
 
-		err = b.StateDB.Open(ctx, tmpStatePath, dstate.WithRecovery(true), dstate.WithWrite(true))
+		err = b.StateDB.Open(ctx, tmpStatePath, dstate.WithRecovery(true), dstate.WithWrite(true), nil)
 		if err != nil {
 			os.Remove(tmpStatePath)
 			return nil, err
@@ -165,7 +165,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.WithRecovery(true), dstate.WithWrite(false))
+	err = b.StateDB.Open(ctx, tmpStatePath, dstate.WithRecovery(true), dstate.WithWrite(false), nil)
 	if err != nil {
 		os.Remove(tmpStatePath)
 		return nil, err
@@ -215,7 +215,7 @@ 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 {
-	err := b.StateDB.Open(ctx, statePath, dstate.WithRecovery(true), dstate.WithWrite(true))
+	err := b.StateDB.Open(ctx, statePath, dstate.WithRecovery(true), dstate.WithWrite(true), nil)
 	if err != nil {
 		return err
 	}

bundle/direct/bundle_apply.go

@@ -88,6 +88,11 @@ func (b *DeploymentBundle) Apply(ctx context.Context, client *databricks.Workspa
 				logdiag.LogError(ctx, fmt.Errorf("%s: %w", errorPrefix, err))
 				return false
 			}
+			// Record the delete with DMS. State is nil: the resource is gone.
+			if err := b.recordOperation(ctx, resourceKey, action, "", nil); err != nil {
+				logdiag.LogError(ctx, fmt.Errorf("%s: %w", errorPrefix, err))
+				return false
+			}
 			return true
 		}
 
@@ -116,6 +121,14 @@ func (b *DeploymentBundle) Apply(ctx context.Context, client *databricks.Workspa
 				logdiag.LogError(ctx, fmt.Errorf("%s: %w", errorPrefix, err))
 				return false
 			}
+
+			// Record the operation with DMS. The resource ID and applied config
+			// (sv.Value) come from the write just performed; GetResourceID reads
+			// the ID assigned by Deploy.
+			if err := b.recordOperation(ctx, resourceKey, action, b.StateDB.GetResourceID(resourceKey), sv.Value); err != nil {
+				logdiag.LogError(ctx, fmt.Errorf("%s: %w", errorPrefix, err))
+				return false
+			}
 		}
 
 		// TODO: Note, we only really need remote state if there are remote references.

bundle/direct/dstate/dms.go

@@ -0,0 +1,104 @@
+package dstate
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+
+	"github.com/databricks/databricks-sdk-go/apierr"
+	"github.com/databricks/databricks-sdk-go/service/bundledeployments"
+)
+
+// overlayDMSState replaces the file-derived resource state with the state
+// recorded in the deployment metadata service (DMS), when DMS owns this
+// deployment. Once DMS is authoritative its resource set is trusted even when
+// empty (a successful deploy with no resources); the file's resources are only
+// used when DMS has no successful version, or when the user opts out of
+// recording deployment history. The caller holds db.mu and has already
+// populated db.Data from the file, including the DeploymentID.
+func (db *DeploymentState) overlayDMSState(ctx context.Context, client bundledeployments.BundleDeploymentsInterface) error {
+	authoritative, err := deploymentHasSuccessfulVersion(ctx, client, db.Data.DeploymentID)
+	if err != nil {
+		return err
+	}
+	if !authoritative {
+		// DMS has no completed version for this deployment: a prior direct deploy
+		// that has not yet successfully recorded to DMS. Keep the file state.
+		return nil
+	}
+
+	resources, err := fetchDeploymentResources(ctx, client, db.Data.DeploymentID)
+	if err != nil {
+		return err
+	}
+
+	db.Data.State = resources
+	db.stateIDs = make(map[string]string, len(resources))
+	for key, entry := range resources {
+		db.stateIDs[key] = entry.ID
+	}
+	return nil
+}
+
+// deploymentHasSuccessfulVersion reports whether DMS holds a successfully
+// completed version for the deployment. It is the signal that DMS owns the
+// state: if the deployment was never recorded to DMS, or its initial DMS deploy
+// did not complete successfully, DMS state is absent or partial and Open keeps
+// the local file's resources instead.
+func deploymentHasSuccessfulVersion(ctx context.Context, client bundledeployments.BundleDeploymentsInterface, deploymentID string) (bool, error) {
+	// Versions are listed newest-first and fetched page by page, and we stop at
+	// the first successful one, so a deployment with a long version history does
+	// not require reading the whole list (typically just the first page).
+	it := client.ListVersions(ctx, bundledeployments.ListVersionsRequest{
+		Parent: "deployments/" + deploymentID,
+	})
+	for it.HasNext(ctx) {
+		v, err := it.Next(ctx)
+		if err != nil {
+			// A deployment that was never recorded to DMS is not an error here: it
+			// just means DMS is not (yet) the source of truth.
+			if errors.Is(err, apierr.ErrNotFound) {
+				return false, nil
+			}
+			return false, fmt.Errorf("listing versions from deployment metadata service: %w", err)
+		}
+		if v.Status == bundledeployments.VersionStatusVersionStatusCompleted &&
+			v.CompletionReason == bundledeployments.VersionCompleteVersionCompleteSuccess {
+			return true, nil
+		}
+	}
+	return false, nil
+}
+
+// fetchDeploymentResources lists every resource recorded for the deployment in
+// DMS and maps them into state entries keyed by the fully-qualified resource key.
+func fetchDeploymentResources(ctx context.Context, client bundledeployments.BundleDeploymentsInterface, deploymentID string) (map[string]ResourceEntry, error) {
+	it := client.ListResources(ctx, bundledeployments.ListResourcesRequest{
+		Parent: "deployments/" + deploymentID,
+	})
+
+	out := make(map[string]ResourceEntry)
+	for it.HasNext(ctx) {
+		res, err := it.Next(ctx)
+		if err != nil {
+			return nil, fmt.Errorf("listing resources from deployment metadata service: %w", err)
+		}
+
+		// DMS reports resource keys without the "resources." prefix (e.g.
+		// "jobs.foo"), but the state DB keys are fully qualified
+		// ("resources.jobs.foo"), so prepend it here.
+		key := "resources." + res.ResourceKey
+
+		var state json.RawMessage
+		if res.State != nil {
+			state = *res.State
+		}
+
+		out[key] = ResourceEntry{
+			ID:    res.ResourceId,
+			State: state,
+		}
+	}
+	return out, nil
+}