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, 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, 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, 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, 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, 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, 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, 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, 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.