From 52e9daf29182d8cd61344d656e86d759f87d86dd Mon Sep 17 00:00:00 2001 From: Shreyas Goenka Date: Tue, 6 Oct 2026 17:14:06 +0000 Subject: [PATCH 1/2] bundle: migrate Direct deployments to DMS --- .../bundle/dms/existing-state/output.txt | 59 ++---- acceptance/bundle/dms/existing-state/script | 32 +--- .../dms/migration-resume/databricks.yml.tmpl | 9 + .../bundle/dms/migration-resume/out.test.toml | 2 + .../bundle/dms/migration-resume/output.txt | 56 ++++++ acceptance/bundle/dms/migration-resume/script | 18 ++ .../bundle/dms/migration-resume/test.toml | 2 + .../help/bundle-deployment-migrate/output.txt | 9 +- .../migrate/auto/dms-migration/output.txt | 5 +- bundle/deploy/lock/release.go | 3 +- bundle/direct/dstate/state.go | 37 +++- bundle/direct/dstate/state_test.go | 26 +++ bundle/phases/dms.go | 59 +++--- bundle/phases/migrate_to_dms.go | 176 ++++++++++++++++++ bundle/phases/migrate_to_dms_test.go | 109 +++++++++++ bundle/statemgmt/state_push.go | 24 +++ cmd/bundle/deployment/migrate.go | 50 ++++- cmd/bundle/utils/process.go | 61 ++---- libs/dms/client.go | 4 + libs/dms/deployment.go | 37 ++++ 20 files changed, 639 insertions(+), 139 deletions(-) create mode 100644 acceptance/bundle/dms/migration-resume/databricks.yml.tmpl create mode 100644 acceptance/bundle/dms/migration-resume/out.test.toml create mode 100644 acceptance/bundle/dms/migration-resume/output.txt create mode 100644 acceptance/bundle/dms/migration-resume/script create mode 100644 acceptance/bundle/dms/migration-resume/test.toml create mode 100644 bundle/phases/migrate_to_dms.go create mode 100644 bundle/phases/migrate_to_dms_test.go create mode 100644 libs/dms/deployment.go diff --git a/acceptance/bundle/dms/existing-state/output.txt b/acceptance/bundle/dms/existing-state/output.txt index 8dc33d924fe..4d11e4ad28f 100644 --- a/acceptance/bundle/dms/existing-state/output.txt +++ b/acceptance/bundle/dms/existing-state/output.txt @@ -25,49 +25,18 @@ Plan: 0 to add, 0 to change, 0 to delete, 1 unchanged "features": null } -=== Enabling recording is refused because the existing deployment is not recorded ->>> DATABRICKS_BUNDLE_DEPLOYMENT_HISTORY=true [CLI] bundle deploy -Error: deployment history setting (true) does not match the existing state (false) +=== Migrate the existing deployment to DMS without applying resource changes +>>> DATABRICKS_BUNDLE_DEPLOYMENT_HISTORY=true [CLI] bundle deployment migrate --enable-history +Migrate: 1 resources moved to deployment history -Update experimental.deployment_history to match the existing deployment, or run "databricks bundle destroy" to start over - - ->>> print_requests.py --dms //api/2.0/bundle --oneline - -=== Wiping the local cache still refuses after pulling the non-DMS state from the workspace ->>> DATABRICKS_BUNDLE_DEPLOYMENT_HISTORY=true [CLI] bundle deploy -Error: deployment history setting (true) does not match the existing state (false) - -Update experimental.deployment_history to match the existing deployment, or run "databricks bundle destroy" to start over - - -=== A leftover local WAL with the wrong lineage is still refused ->>> DATABRICKS_BUNDLE_DEPLOYMENT_HISTORY=true [CLI] bundle deploy -Error: reading state from [TEST_TMP_DIR]/.databricks/bundle/default/resources.json: WAL recovery failed: WAL lineage ("stale") does not match state lineage ("[UUID]") - - -=== Destroy clears the deployment. With nothing recorded remotely the bundle is treated as new, so recording can be enabled once the stale WAL is gone ->>> DATABRICKS_BUNDLE_DEPLOYMENT_HISTORY=true [CLI] bundle destroy --auto-approve -The following resources will be deleted: - delete resources.jobs.one - -All files and directories at the following location will be deleted: /Workspace/Users/[USERNAME]/.bundle/dms-existing-state-[UNIQUE_NAME]/default - -Destroy: 1 deleted - ->>> DATABRICKS_BUNDLE_DEPLOYMENT_HISTORY=true [CLI] bundle deploy -Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/dms-existing-state-[UNIQUE_NAME]/default/files... -Created jobs.one -Files: 2 uploaded, 0 deleted -Resources: 1 created, 0 changed, 0 deleted, 0 unchanged - ->>> print_requests.py --dms //api/2.0/bundle --oneline +>>> print_requests.py --dms --keep //api/2.0/bundle --oneline {"method": "POST", "path": "/api/2.0/bundle/deployments", "body": {"display_name": "dms-existing-state-[UNIQUE_NAME]", "initial_parent_path": "/Workspace/Users/[USERNAME]/.bundle/dms-existing-state-[UNIQUE_NAME]/default/state", "target_name": "default", "workspace_info": {"file_path": "/Workspace/Users/[USERNAME]/.bundle/dms-existing-state-[UNIQUE_NAME]/default/files", "root_path": "/Workspace/Users/[USERNAME]/.bundle/dms-existing-state-[UNIQUE_NAME]/default"}}} -{"method": "POST", "path": "/api/2.0/bundle/deployments/[NUMID]/versions", "q": {"version_id": "1"}, "body": {"cli_version": "[CLI_VERSION]", "version_type": "VERSION_TYPE_DEPLOY", "operations": [{"resource_key": "jobs.one", "action_type": "OPERATION_ACTION_TYPE_CREATE"}]}} -{"method": "PATCH", "path": "/api/2.0/bundle/deployments/[NUMID]/versions/1/operations/jobs.one", "q": {"update_mask": "state,error_message,resource_id,status"}, "body": {"error_message": "", "resource_id": "[NUMID]", "sequence_id": "0", "state": "{\"state\":{\"deployment\":{\"deployment_id\":\"[NUMID]\",\"kind\":\"BUNDLE\",\"metadata_file_path\":\"/Workspace/Users/[USERNAME]/.bundle/dms-existing-state-[UNIQUE_NAME]/default/state/metadata.json\",\"version_id\":\"1\"},\"edit_mode\":\"UI_LOCKED\",\"format\":\"MULTI_TASK\",\"max_concurrent_runs\":1,\"name\":\"one\",\"queue\":{\"enabled\":true}}}", "status": "OPERATION_STATUS_SUCCEEDED"}} -{"method": "POST", "path": "/api/2.0/bundle/deployments/[NUMID]/versions/1/complete", "body": {"completion_reason": "VERSION_COMPLETE_SUCCESS"}} +{"method": "POST", "path": "/api/2.0/bundle/deployments/[NUMID]/versions", "q": {"version_id": "2"}, "body": {"cli_version": "[CLI_VERSION]", "version_type": "VERSION_TYPE_DEPLOY", "operations": [{"resource_key": "jobs.one", "action_type": "OPERATION_ACTION_TYPE_MIGRATE"}]}} +{"method": "PATCH", "path": "/api/2.0/bundle/deployments/[NUMID]/versions/2/operations/jobs.one", "q": {"update_mask": "state,error_message,resource_id,status"}, "body": {"error_message": "", "resource_id": "[NUMID]", "sequence_id": "0", "state": "{\"state\":{\"deployment\":{\"kind\":\"BUNDLE\",\"metadata_file_path\":\"/Workspace/Users/[USERNAME]/.bundle/dms-existing-state-[UNIQUE_NAME]/default/state/metadata.json\"},\"edit_mode\":\"UI_LOCKED\",\"format\":\"MULTI_TASK\",\"max_concurrent_runs\":1,\"name\":\"one\",\"queue\":{\"enabled\":true}}}", "status": "OPERATION_STATUS_SUCCEEDED"}} +{"method": "POST", "path": "/api/2.0/bundle/deployments/[NUMID]/versions/2/complete", "body": {"completion_reason": "VERSION_COMPLETE_SUCCESS"}} ->>> find.py resources[.]json[.]wal --expect 0 +>>> print_requests.py --keep --del-field raw_body //resources.json.backup --oneline +{"method": "POST", "path": "/api/2.0/workspace-files/import-file/Workspace/Users/[USERNAME]/.bundle/dms-existing-state-[UNIQUE_NAME]/default/state/resources.json.backup", "q": {"overwrite": "true"}, "body": {"state_version": 3, "cli_version": "[CLI_VERSION]", "lineage": "[UUID]", "serial": 1, "state": {"resources.jobs.one": {"__id__": "[NUMID]", "state": {"deployment": {"kind": "BUNDLE", "metadata_file_path": "/Workspace/Users/[USERNAME]/.bundle/dms-existing-state-[UNIQUE_NAME]/default/state/metadata.json"}, "edit_mode": "UI_LOCKED", "format": "MULTI_TASK", "max_concurrent_runs": 1, "name": "one", "queue": {"enabled": true}}}}}} >>> jq {state_version, features} .databricks/bundle/default/resources.json { @@ -76,3 +45,13 @@ Resources: 1 created, 0 changed, 0 deleted, 0 unchanged "deployment_history": {} } } + +=== The next plan adds deployment metadata without recreating resources +>>> DATABRICKS_BUNDLE_DEPLOYMENT_HISTORY=true [CLI] bundle plan +update jobs.one + +Plan: 0 to add, 1 to change, 0 to delete, 0 unchanged + +=== Migration is idempotent after DMS becomes authoritative +>>> DATABRICKS_BUNDLE_DEPLOYMENT_HISTORY=true [CLI] bundle deployment migrate --enable-history +Deployment already records deployment history. diff --git a/acceptance/bundle/dms/existing-state/script b/acceptance/bundle/dms/existing-state/script index 7ec44e4a729..8496efc1c2a 100644 --- a/acceptance/bundle/dms/existing-state/script +++ b/acceptance/bundle/dms/existing-state/script @@ -6,30 +6,18 @@ dms_plan trace print_requests.py --dms //api/2.0/bundle --oneline trace jq '{state_version, features}' .databricks/bundle/default/resources.json -title "Enabling recording is refused because the existing deployment is not recorded" -musterr trace DATABRICKS_BUNDLE_DEPLOYMENT_HISTORY=true $CLI bundle deploy -trace print_requests.py --dms //api/2.0/bundle --oneline - -title "Wiping the local cache still refuses after pulling the non-DMS state from the workspace" -rm -rf .databricks -musterr trace DATABRICKS_BUNDLE_DEPLOYMENT_HISTORY=true $CLI bundle deploy +title "Migrate the existing deployment to DMS without applying resource changes" +trace DATABRICKS_BUNDLE_DEPLOYMENT_HISTORY=true $CLI bundle deployment migrate --enable-history +trace print_requests.py --dms --keep //api/2.0/bundle --oneline +trace print_requests.py --keep --del-field raw_body //resources.json.backup --oneline +trace jq '{state_version, features}' .databricks/bundle/default/resources.json -title "A leftover local WAL with the wrong lineage is still refused" -mkdir -p .databricks/bundle/default -printf '{"lineage":"stale","serial":99,"state_version":2,"features":{"deployment_history":{}}}\n' > .databricks/bundle/default/resources.json.wal -musterr trace DATABRICKS_BUNDLE_DEPLOYMENT_HISTORY=true $CLI bundle deploy +title "The next plan adds deployment metadata without recreating resources" +trace DATABRICKS_BUNDLE_DEPLOYMENT_HISTORY=true $CLI bundle plan -title "Destroy clears the deployment. With nothing recorded remotely the bundle is treated as new, so recording can be enabled once the stale WAL is gone" -# Clear the stale WAL first: it was refused, not deleted, and destroy runs without recording, so it -# would otherwise try to replay it and fail on the lineage mismatch. -rm -f .databricks/bundle/default/resources.json.wal -# Keep the mismatched setting here to verify that destroy ignores configuration and follows state. -trace DATABRICKS_BUNDLE_DEPLOYMENT_HISTORY=true $CLI bundle destroy --auto-approve -trace DATABRICKS_BUNDLE_DEPLOYMENT_HISTORY=true $CLI bundle deploy -trace print_requests.py --dms //api/2.0/bundle --oneline -trace find.py 'resources[.]json[.]wal' --expect 0 -trace jq '{state_version, features}' .databricks/bundle/default/resources.json +title "Migration is idempotent after DMS becomes authoritative" +trace DATABRICKS_BUNDLE_DEPLOYMENT_HISTORY=true $CLI bundle deployment migrate --enable-history # The recorded requests are asserted above via print_requests.py; drop whatever the run added after -# that (telemetry, and the state probes of the non-recording variant) so nothing is left to compare. +# that (telemetry and read-only probes) so nothing is left to compare. rm -f "$OUT_REQUESTS" diff --git a/acceptance/bundle/dms/migration-resume/databricks.yml.tmpl b/acceptance/bundle/dms/migration-resume/databricks.yml.tmpl new file mode 100644 index 00000000000..5c5ca852160 --- /dev/null +++ b/acceptance/bundle/dms/migration-resume/databricks.yml.tmpl @@ -0,0 +1,9 @@ +bundle: + name: dms-migration-resume-$UNIQUE_NAME + +resources: + jobs: + one: + name: one + two: + name: two diff --git a/acceptance/bundle/dms/migration-resume/out.test.toml b/acceptance/bundle/dms/migration-resume/out.test.toml new file mode 100644 index 00000000000..f68b4a89c36 --- /dev/null +++ b/acceptance/bundle/dms/migration-resume/out.test.toml @@ -0,0 +1,2 @@ +Cloud = true +EnvMatrix.DMS = [""] diff --git a/acceptance/bundle/dms/migration-resume/output.txt b/acceptance/bundle/dms/migration-resume/output.txt new file mode 100644 index 00000000000..b48e4882c20 --- /dev/null +++ b/acceptance/bundle/dms/migration-resume/output.txt @@ -0,0 +1,56 @@ + +=== Create a direct deployment with two resources +>>> [CLI] bundle deploy +Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/dms-migration-resume-[UNIQUE_NAME]/default/files... +Created jobs.one +Created jobs.two +Files: 2 uploaded, 0 deleted +Resources: 2 created, 0 changed, 0 deleted, 0 unchanged + +=== A failed operation leaves resources.json authoritative +>>> fault.py PATCH /api/2.0/bundle/* 500 0 1 + +>>> DATABRICKS_BUNDLE_DEPLOYMENT_HISTORY=true [CLI] bundle deployment migrate --enable-history +Error: recording migration resources: recording operation for resources.jobs.one: Fault injected by test. (500 INJECTED) + +Endpoint: PATCH [DATABRICKS_URL]/api/2.0/bundle/deployments/[NUMID]/versions/2/operations/jobs.one?update_mask=state%2Cerror_message%2Cresource_id%2Cstatus +HTTP Status: 500 Internal Server Error +API error_code: INJECTED +API message: Fault injected by test. + + +>>> jq {serial, features, resources: (.state | keys)} .databricks/bundle/default/resources.json +{ + "serial": 1, + "features": null, + "resources": [ + "resources.jobs.one", + "resources.jobs.two" + ] +} + +=== Retry migrates only the resource missing from DMS +>>> DATABRICKS_BUNDLE_DEPLOYMENT_HISTORY=true [CLI] bundle deployment migrate --enable-history +Migrate: 2 resources moved to deployment history + +>>> print_requests.py --dms --keep //api/2.0/bundle --oneline +{"method": "POST", "path": "/api/2.0/bundle/deployments", "body": {"display_name": "dms-migration-resume-[UNIQUE_NAME]", "initial_parent_path": "/Workspace/Users/[USERNAME]/.bundle/dms-migration-resume-[UNIQUE_NAME]/default/state", "target_name": "default", "workspace_info": {"file_path": "/Workspace/Users/[USERNAME]/.bundle/dms-migration-resume-[UNIQUE_NAME]/default/files", "root_path": "/Workspace/Users/[USERNAME]/.bundle/dms-migration-resume-[UNIQUE_NAME]/default"}}} +{"method": "POST", "path": "/api/2.0/bundle/deployments/[NUMID]/versions", "q": {"version_id": "2"}, "body": {"cli_version": "[CLI_VERSION]", "version_type": "VERSION_TYPE_DEPLOY", "operations": [{"resource_key": "jobs.one", "action_type": "OPERATION_ACTION_TYPE_MIGRATE"}, {"resource_key": "jobs.two", "action_type": "OPERATION_ACTION_TYPE_MIGRATE"}]}} +{"method": "PATCH", "path": "/api/2.0/bundle/deployments/[NUMID]/versions/2/operations/jobs.one", "q": {"update_mask": "state,error_message,resource_id,status"}, "body": {"error_message": "", "resource_id": "[NUMID]", "sequence_id": "0", "state": "{\"state\":{\"deployment\":{\"kind\":\"BUNDLE\",\"metadata_file_path\":\"/Workspace/Users/[USERNAME]/.bundle/dms-migration-resume-[UNIQUE_NAME]/default/state/metadata.json\"},\"edit_mode\":\"UI_LOCKED\",\"format\":\"MULTI_TASK\",\"max_concurrent_runs\":1,\"name\":\"one\",\"queue\":{\"enabled\":true}}}", "status": "OPERATION_STATUS_SUCCEEDED"}} +{"method": "PATCH", "path": "/api/2.0/bundle/deployments/[NUMID]/versions/2/operations/jobs.two", "q": {"update_mask": "state,error_message,resource_id,status"}, "body": {"error_message": "", "resource_id": "[NUMID]", "sequence_id": "0", "state": "{\"state\":{\"deployment\":{\"kind\":\"BUNDLE\",\"metadata_file_path\":\"/Workspace/Users/[USERNAME]/.bundle/dms-migration-resume-[UNIQUE_NAME]/default/state/metadata.json\"},\"edit_mode\":\"UI_LOCKED\",\"format\":\"MULTI_TASK\",\"max_concurrent_runs\":1,\"name\":\"two\",\"queue\":{\"enabled\":true}}}", "status": "OPERATION_STATUS_SUCCEEDED"}} +{"method": "POST", "path": "/api/2.0/bundle/deployments/[NUMID]/versions/2/complete", "body": {"completion_reason": "VERSION_COMPLETE_FAILURE"}} +{"method": "POST", "path": "/api/2.0/bundle/deployments/[NUMID]/versions", "q": {"version_id": "3"}, "body": {"cli_version": "[CLI_VERSION]", "version_type": "VERSION_TYPE_DEPLOY", "previous_version_id": "2", "operations": [{"resource_key": "jobs.one", "action_type": "OPERATION_ACTION_TYPE_MIGRATE"}]}} +{"method": "PATCH", "path": "/api/2.0/bundle/deployments/[NUMID]/versions/3/operations/jobs.one", "q": {"update_mask": "state,error_message,resource_id,status"}, "body": {"error_message": "", "resource_id": "[NUMID]", "sequence_id": "0", "state": "{\"state\":{\"deployment\":{\"kind\":\"BUNDLE\",\"metadata_file_path\":\"/Workspace/Users/[USERNAME]/.bundle/dms-migration-resume-[UNIQUE_NAME]/default/state/metadata.json\"},\"edit_mode\":\"UI_LOCKED\",\"format\":\"MULTI_TASK\",\"max_concurrent_runs\":1,\"name\":\"one\",\"queue\":{\"enabled\":true}}}", "status": "OPERATION_STATUS_SUCCEEDED"}} +{"method": "POST", "path": "/api/2.0/bundle/deployments/[NUMID]/versions/3/complete", "body": {"completion_reason": "VERSION_COMPLETE_SUCCESS"}} + +>>> print_requests.py --keep --del-field raw_body //resources.json.backup --oneline +{"method": "POST", "path": "/api/2.0/workspace-files/import-file/Workspace/Users/[USERNAME]/.bundle/dms-migration-resume-[UNIQUE_NAME]/default/state/resources.json.backup", "q": {"overwrite": "true"}, "body": {"state_version": 3, "cli_version": "[CLI_VERSION]", "lineage": "[UUID]", "serial": 1, "state": {"resources.jobs.one": {"__id__": "[NUMID]", "state": {"deployment": {"kind": "BUNDLE", "metadata_file_path": "/Workspace/Users/[USERNAME]/.bundle/dms-migration-resume-[UNIQUE_NAME]/default/state/metadata.json"}, "edit_mode": "UI_LOCKED", "format": "MULTI_TASK", "max_concurrent_runs": 1, "name": "one", "queue": {"enabled": true}}}, "resources.jobs.two": {"__id__": "[NUMID]", "state": {"deployment": {"kind": "BUNDLE", "metadata_file_path": "/Workspace/Users/[USERNAME]/.bundle/dms-migration-resume-[UNIQUE_NAME]/default/state/metadata.json"}, "edit_mode": "UI_LOCKED", "format": "MULTI_TASK", "max_concurrent_runs": 1, "name": "two", "queue": {"enabled": true}}}}}} + +>>> jq {serial, features, resources: (.state | keys)} .databricks/bundle/default/resources.json +{ + "serial": null, + "features": { + "deployment_history": {} + }, + "resources": [] +} diff --git a/acceptance/bundle/dms/migration-resume/script b/acceptance/bundle/dms/migration-resume/script new file mode 100644 index 00000000000..39812ace033 --- /dev/null +++ b/acceptance/bundle/dms/migration-resume/script @@ -0,0 +1,18 @@ +envsubst < databricks.yml.tmpl > databricks.yml + +title "Create a direct deployment with two resources" +trace $CLI bundle deploy +rm -f "$OUT_REQUESTS" + +title "A failed operation leaves resources.json authoritative" +trace fault.py "PATCH /api/2.0/bundle/*" 500 0 1 +musterr trace DATABRICKS_BUNDLE_DEPLOYMENT_HISTORY=true $CLI bundle deployment migrate --enable-history +trace jq '{serial, features, resources: (.state | keys)}' .databricks/bundle/default/resources.json + +title "Retry migrates only the resource missing from DMS" +trace DATABRICKS_BUNDLE_DEPLOYMENT_HISTORY=true $CLI bundle deployment migrate --enable-history +trace print_requests.py --dms --keep //api/2.0/bundle --oneline +trace print_requests.py --keep --del-field raw_body //resources.json.backup --oneline +trace jq '{serial, features, resources: (.state | keys)}' .databricks/bundle/default/resources.json + +rm -f "$OUT_REQUESTS" diff --git a/acceptance/bundle/dms/migration-resume/test.toml b/acceptance/bundle/dms/migration-resume/test.toml new file mode 100644 index 00000000000..ba6d1dbbe63 --- /dev/null +++ b/acceptance/bundle/dms/migration-resume/test.toml @@ -0,0 +1,2 @@ +# This test creates a direct state first and enables DMS only for the migration command. +EnvMatrix.DMS = [""] diff --git a/acceptance/bundle/help/bundle-deployment-migrate/output.txt b/acceptance/bundle/help/bundle-deployment-migrate/output.txt index 9312fb85ffc..cba47b94cbb 100644 --- a/acceptance/bundle/help/bundle-deployment-migrate/output.txt +++ b/acceptance/bundle/help/bundle-deployment-migrate/output.txt @@ -8,12 +8,17 @@ lineage and incremented serial number. Note, the migration is performed locally only. To finalize it, run 'bundle deploy'. This will synchronize the state file to the workspace so that subsequent deploys of this bundle use direct deployment engine as well. +With --enable-history, this command instead migrates an existing Direct deployment +to the deployment metadata service. The migration completes before any later deploy, +and resources.json is backed up before DMS becomes authoritative. + Usage: databricks bundle deployment migrate [flags] Flags: - -h, --help help for migrate - --noplancheck No-op (kept for compatibility). + --enable-history Migrate an existing Direct deployment to deployment history. + -h, --help help for migrate + --noplancheck No-op (kept for compatibility). Global Flags: --debug enable debug logging diff --git a/acceptance/bundle/migrate/auto/dms-migration/output.txt b/acceptance/bundle/migrate/auto/dms-migration/output.txt index 980c30afa4f..c861168fb11 100644 --- a/acceptance/bundle/migrate/auto/dms-migration/output.txt +++ b/acceptance/bundle/migrate/auto/dms-migration/output.txt @@ -4,10 +4,9 @@ === Direct + deployment-history requested against a terraform state is rejected before commit >>> DATABRICKS_BUNDLE_ENGINE=direct DATABRICKS_BUNDLE_DEPLOYMENT_HISTORY=true errcode [CLI] bundle deploy -Notice: migrating your bundle to direct deployment engine (https://github.com/databricks/cli/issues/6765). -Error: deployment history setting (true) does not match the existing state (false) +Error: cannot migrate Terraform state directly to deployment history -Update experimental.deployment_history to match the existing deployment, or run "databricks bundle destroy" to start over +First disable experimental.deployment_history and run "databricks bundle deploy" to migrate to Direct state. Then re-enable deployment history and run "databricks bundle deployment migrate --enable-history" Exit code: 1 diff --git a/bundle/deploy/lock/release.go b/bundle/deploy/lock/release.go index 7c8eef10e99..ab488dd917e 100644 --- a/bundle/deploy/lock/release.go +++ b/bundle/deploy/lock/release.go @@ -14,6 +14,7 @@ type Goal string const ( GoalBind = Goal("bind") GoalUnbind = Goal("unbind") + GoalMigrate = Goal("migrate") GoalDeploy = Goal("deploy") GoalDestroy = Goal("destroy") ) @@ -48,7 +49,7 @@ func (m *release) Apply(ctx context.Context, b *bundle.Bundle) diag.Diagnostics switch m.goal { case GoalDeploy: return diag.FromErr(b.Locker.Unlock(ctx)) - case GoalBind, GoalUnbind: + case GoalBind, GoalUnbind, GoalMigrate: return diag.FromErr(b.Locker.Unlock(ctx)) case GoalDestroy: // Destroy may have proceeded without acquiring a lock (see lock.Acquire: diff --git a/bundle/direct/dstate/state.go b/bundle/direct/dstate/state.go index 9f347750659..78b86555aea 100644 --- a/bundle/direct/dstate/state.go +++ b/bundle/direct/dstate/state.go @@ -572,9 +572,9 @@ func (db *DeploymentState) unlockedOpen(ctx context.Context, path string, withRe db.Data.Features[FeatureDeploymentHistory] = struct{}{} recorded = true case recording && !recorded: - return errors.New(`enabling experimental.deployment_history for an existing deployment is not supported + return errors.New(`deployment history is enabled for an existing Direct deployment that has not been migrated -Run "databricks bundle destroy" first, then deploy again with deployment history enabled`) +Run "databricks bundle deployment migrate --enable-history" before deploying`) case !recording && recorded: return ErrUnsettingRecording } @@ -882,6 +882,32 @@ func (db *DeploymentState) Persist() error { return db.unlockedSave() } +// PersistDeploymentHistoryMarker atomically replaces the local direct state with the tombstone a +// DMS-backed deployment uses. It does not mutate the open in-memory state, so a failed remote push +// can still be retried from the direct state during this process. +func (db *DeploymentState) PersistDeploymentHistoryMarker() error { + db.mu.Lock() + defer db.mu.Unlock() + + if db.Path == "" { + return errors.New("internal error: DeploymentState must be opened first") + } + + header := db.Data.Header + header.Features = maps.Clone(header.Features) + if header.Features == nil { + header.Features = make(map[string]struct{}, 1) + } + header.Features[FeatureDeploymentHistory] = struct{}{} + header.Serial = 0 + header.CLIVersion = build.GetInfo().Version + + return db.unlockedWrite(Database{ + Header: header, + State: map[string]ResourceEntry{}, + }) +} + // IsOpen reports whether the state has been opened (for read or write). It lets // callers probe the state without risking the panic in AssertOpenedForReadOrWrite, // e.g. code paths shared with the terraform engine where the state DB is never opened. @@ -970,7 +996,12 @@ func (db *DeploymentState) ExportState(ctx context.Context) resourcestate.Export // the WAL. A torn write would therefore leave a state file that Open rejects // next to an intact WAL it never reads. func (db *DeploymentState) unlockedSave() error { - data, err := json.MarshalIndent(db.dataForFile(), "", " ") + return db.unlockedWrite(db.dataForFile()) +} + +// unlockedWrite persists data at db.Path. Callers must hold db.mu. +func (db *DeploymentState) unlockedWrite(state Database) error { + data, err := json.MarshalIndent(state, "", " ") if err != nil { return err } diff --git a/bundle/direct/dstate/state_test.go b/bundle/direct/dstate/state_test.go index c8d78d150d6..7d939a18e54 100644 --- a/bundle/direct/dstate/state_test.go +++ b/bundle/direct/dstate/state_test.go @@ -248,6 +248,32 @@ func TestGetOrInitLineageReadableBeforeWriteAndPersisted(t *testing.T) { mustFinalize(t, &reopened) } +func TestPersistDeploymentHistoryMarker(t *testing.T) { + path := filepath.Join(t.TempDir(), "state.json") + state := NewDatabase("test-lineage", 7) + state.State["resources.jobs.my_job"] = ResourceEntry{ + ID: "123", + State: json.RawMessage(`{"name":"my job"}`), + } + + var db DeploymentState + db.OpenWithData(path, state) + require.NoError(t, db.PersistDeploymentHistoryMarker()) + + data, err := os.ReadFile(path) + require.NoError(t, err) + var persisted Database + require.NoError(t, json.Unmarshal(data, &persisted)) + assert.Equal(t, "test-lineage", persisted.Lineage) + assert.Zero(t, persisted.Serial) + assert.Contains(t, persisted.Features, FeatureDeploymentHistory) + assert.Empty(t, persisted.State) + + // The command can still retry from its open direct state when the remote tombstone upload fails. + assert.Equal(t, 7, db.Data.Serial) + assert.Contains(t, db.Data.State, "resources.jobs.my_job") +} + // TestOpenFailureLeavesStateClosed pins that a failed Open leaves the receiver // closed. Open assigns db.Path before the steps that can fail, so an unreadable // state file used to leave Path set: the next Open on the same value panicked diff --git a/bundle/phases/dms.go b/bundle/phases/dms.go index 9ceea29509d..dec3b7172d7 100644 --- a/bundle/phases/dms.go +++ b/bundle/phases/dms.go @@ -70,27 +70,37 @@ func actionToSDK(a deployplan.ActionType) (bundledeployments.OperationActionType func createOrUpdateDeployment(ctx context.Context, b *bundle.Bundle, current *bundledeployments.Deployment) { db := &b.DeploymentBundle dmsClient := db.StateDB.DmsClient() + deploymentID, err := ensureDeployment(ctx, b, db.StateDB.DeploymentID, current, dmsClient) + if err != nil { + logdiag.LogError(ctx, err) + return + } + db.StateDB.DeploymentID = deploymentID + + // A first deploy had no deployment to read at startup, so its id enters the history here. + bundle.ApplyFuncContext(ctx, b, func(_ context.Context, b *bundle.Bundle) { + b.Config.Bundle.Deployment.DeploymentID = deploymentID + }) +} + +// ensureDeployment creates the DMS deployment when absent and otherwise synchronizes its metadata. +// It returns the server-owned deployment ID used by the version and operation APIs. +func ensureDeployment(ctx context.Context, b *bundle.Bundle, deploymentID string, current *bundledeployments.Deployment, dmsClient *dms.Client) (string, error) { metadata := deploymentMetadata(b) - deploymentID := db.StateDB.DeploymentID if deploymentID == "" { id, err := dmsClient.CreateDeployment(ctx, b.Config.Workspace.StatePath, metadata) if err != nil { - logdiag.LogError(ctx, fmt.Errorf("failed to create deployment: %w", err)) - return + return "", fmt.Errorf("failed to create deployment: %w", err) } - deploymentID = id - db.StateDB.DeploymentID = deploymentID - } else if mask := metadata.StaleFields(current); mask != "" { + return id, nil + } + + if mask := metadata.StaleFields(current); mask != "" { if err := dmsClient.UpdateDeployment(ctx, deploymentID, metadata, mask); err != nil { - logdiag.LogError(ctx, fmt.Errorf("failed to update deployment: %w", err)) - return + return "", fmt.Errorf("failed to update deployment: %w", err) } } - - // A first deploy had no deployment to read at startup, so its id enters the history here. - bundle.ApplyFuncContext(ctx, b, func(_ context.Context, b *bundle.Bundle) { - b.Config.Bundle.Deployment.DeploymentID = deploymentID - }) + return deploymentID, nil } // startVersion claims the version the run settled on and opens the buffer that records @@ -116,20 +126,12 @@ func startVersion(ctx context.Context, b *bundle.Bundle, versionType dms.Version // The server rejects this unless the version number exceeds last_version_id and // previous_version_id matches it, which is what makes claiming the number up front // safe: a deploy that took it in the meantime is reported, not overwritten. - var gitInfo *bundledeployments.GitInfo - if git := b.Config.Bundle.Git; git.Branch != "" || git.Commit != "" || git.OriginURL != "" { - gitInfo = &bundledeployments.GitInfo{ - Branch: git.Branch, - Commit: git.Commit, - OriginUrl: git.OriginURL, - } - } version, err := dmsClient.CreateVersion(ctx, deploymentID, versionID, dms.CreateVersionRequest{ CliVersion: build.GetInfo().Version, VersionType: versionType, PreviousVersionId: previousVersionID, Operations: staged, - GitInfo: gitInfo, + GitInfo: versionGitInfo(b), }) if err != nil { return fmt.Errorf("failed to create deployment version: %w", err) @@ -140,6 +142,19 @@ func startVersion(ctx context.Context, b *bundle.Bundle, versionType dms.Version return nil } +// versionGitInfo returns source-control provenance when the bundle resolved any of it. +func versionGitInfo(b *bundle.Bundle) *bundledeployments.GitInfo { + git := b.Config.Bundle.Git + if git.Branch == "" && git.Commit == "" && git.OriginURL == "" { + return nil + } + return &bundledeployments.GitInfo{ + Branch: git.Branch, + Commit: git.Commit, + OriginUrl: git.OriginURL, + } +} + // logDeploymentVersion logs the deployment version URL. Workspace ID is omitted // so the page stays clickable in a terminal and redirects correctly without it. func logDeploymentVersion(ctx context.Context, b *bundle.Bundle) { diff --git a/bundle/phases/migrate_to_dms.go b/bundle/phases/migrate_to_dms.go new file mode 100644 index 00000000000..7e36bc064da --- /dev/null +++ b/bundle/phases/migrate_to_dms.go @@ -0,0 +1,176 @@ +package phases + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "maps" + "slices" + + "github.com/databricks/cli/bundle" + "github.com/databricks/cli/bundle/deploy/lock" + "github.com/databricks/cli/bundle/direct/dstate" + "github.com/databricks/cli/bundle/statemgmt" + "github.com/databricks/cli/internal/build" + "github.com/databricks/cli/libs/cmdio" + "github.com/databricks/cli/libs/dms" + "github.com/databricks/cli/libs/log" + "github.com/databricks/cli/libs/logdiag" + "github.com/databricks/databricks-sdk-go/service/bundledeployments" +) + +type migrationResource struct { + key string + id string + recorded json.RawMessage +} + +// MigrateToDMS copies an existing direct deployment into DMS and changes resources.json to the +// deployment-history marker only after every resource and the rollback backup are durable. +func MigrateToDMS(ctx context.Context, b *bundle.Bundle) { + log.Info(ctx, "Phase: migrate to deployment history") + + bundle.ApplyContext(ctx, b, lock.Acquire(lock.GoalMigrate)) + if logdiag.HasError(ctx) { + return + } + defer bundle.ApplyContext(ctx, b, lock.Release(lock.GoalMigrate)) + + dmsClient, err := dms.NewClient(b.WorkspaceClient(ctx)) + if err != nil { + logdiag.LogError(ctx, err) + return + } + + deploymentID, current, err := dms.LookupDeployment(ctx, b.WorkspaceClient(ctx), b.Config.Workspace.StatePath) + if err != nil { + logdiag.LogError(ctx, err) + return + } + deploymentID, err = ensureDeployment(ctx, b, deploymentID, current, dmsClient) + if err != nil { + logdiag.LogError(ctx, err) + return + } + + recorded, err := dmsClient.ListResources(ctx, deploymentID) + if err != nil { + logdiag.LogError(ctx, err) + return + } + resources, staged, err := migrationResources(b.DeploymentBundle.StateDB.Data.State, recorded) + if err != nil { + logdiag.LogError(ctx, err) + return + } + + if len(staged) > 0 { + if err := recordMigrationVersion(ctx, b, dmsClient, deploymentID, current, resources, staged); err != nil { + logdiag.LogError(ctx, err) + return + } + } + + if err := statemgmt.BackupRemoteResourcesState(ctx, b); err != nil { + logdiag.LogError(ctx, err) + return + } + if err := b.DeploymentBundle.StateDB.PersistDeploymentHistoryMarker(); err != nil { + logdiag.LogError(ctx, fmt.Errorf("persisting deployment history marker: %w", err)) + return + } + statemgmt.PushResourcesState(ctx, b) + if logdiag.HasError(ctx) { + // Keep the local cache on direct state when the remote commit did not complete. Its higher + // serial wins over a possibly-written tombstone on the next retry. + if err := b.DeploymentBundle.StateDB.Persist(); err != nil { + logdiag.LogError(ctx, fmt.Errorf("restoring local direct state after failed migration commit: %w", err)) + } + return + } + + cmdio.LogString(ctx, fmt.Sprintf("Migrate: %d resources moved to deployment history", len(b.DeploymentBundle.StateDB.Data.State))) +} + +// migrationResources returns direct resources not already recorded with the same ID and state. +// A retry therefore resumes a partial migration instead of writing successful resources again. +func migrationResources(direct map[string]dstate.ResourceEntry, existing []dms.Resource) ([]migrationResource, []dms.StagedOperation, error) { + recordedByKey := make(map[string]dms.Resource, len(existing)) + for _, resource := range existing { + recordedByKey[resource.Key] = resource + } + + resources := make([]migrationResource, 0, len(direct)) + staged := make([]dms.StagedOperation, 0, len(direct)) + for _, key := range slices.Sorted(maps.Keys(direct)) { + entry := direct[key] + state, err := json.Marshal(dstate.RecordedState{State: entry.State, DependsOn: entry.DependsOn}) + if err != nil { + return nil, nil, fmt.Errorf("serializing state for %s: %w", key, err) + } + if previous, ok := recordedByKey[key]; ok && previous.ID == entry.ID && previous.State == string(state) { + continue + } + + resources = append(resources, migrationResource{key: key, id: entry.ID, recorded: state}) + staged = append(staged, dms.StagedOperation{ + ResourceKey: key, + ActionType: dms.OperationActionTypeMigrate, + }) + } + return resources, staged, nil +} + +// recordMigrationVersion writes one migration version and closes it as failed when any operation +// upload fails, leaving resources.json authoritative for a later retry. +func recordMigrationVersion(ctx context.Context, b *bundle.Bundle, dmsClient *dms.Client, deploymentID string, current *bundledeployments.Deployment, resources []migrationResource, staged []dms.StagedOperation) error { + lastVersionID := "" + if current != nil { + lastVersionID = current.LastVersionId + } + versionID, err := migrationVersionID(b.DeploymentBundle.StateDB.GetSerial(), lastVersionID) + if err != nil { + return err + } + + _, err = dmsClient.CreateVersion(ctx, deploymentID, versionID, dms.CreateVersionRequest{ + CliVersion: build.GetInfo().Version, + VersionType: dms.VersionTypeDeploy, + PreviousVersionId: lastVersionID, + Operations: staged, + GitInfo: versionGitInfo(b), + }) + if err != nil { + return fmt.Errorf("creating migration version: %w", err) + } + + buffer := dms.StartOperationBuffer(ctx, dmsClient, deploymentID, versionID) + for _, resource := range resources { + buffer.RecordOperation(ctx, resource.key, false, resource.id, resource.recorded) + } + recordErr := buffer.Drain() + + reason := bundledeployments.VersionCompleteVersionCompleteSuccess + if recordErr != nil { + reason = bundledeployments.VersionCompleteVersionCompleteFailure + } + completeErr := dmsClient.CompleteVersion(ctx, deploymentID, versionID, reason) + if completeErr != nil { + completeErr = fmt.Errorf("completing migration version: %w", completeErr) + } + if recordErr != nil { + recordErr = fmt.Errorf("recording migration resources: %w", recordErr) + } + return errors.Join(recordErr, completeErr) +} + +// migrationVersionID advances past both state stores so a retry never reuses a serial that either +// resources.json or DMS has already observed. +func migrationVersionID(directSerial int, lastDMSVersion string) (int, error) { + nextDMSVersion, err := dms.NextVersion(lastDMSVersion) + if err != nil { + return 0, err + } + return max(directSerial+1, nextDMSVersion), nil +} diff --git a/bundle/phases/migrate_to_dms_test.go b/bundle/phases/migrate_to_dms_test.go new file mode 100644 index 00000000000..ddfce2abca0 --- /dev/null +++ b/bundle/phases/migrate_to_dms_test.go @@ -0,0 +1,109 @@ +package phases + +import ( + "encoding/json" + "testing" + + "github.com/databricks/cli/bundle/deployplan" + "github.com/databricks/cli/bundle/direct/dstate" + "github.com/databricks/cli/libs/dms" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestMigrationResources(t *testing.T) { + direct := map[string]dstate.ResourceEntry{ + "resources.pipelines.second": { + ID: "pipeline-id", + State: json.RawMessage(`{"name":"second"}`), + }, + "resources.jobs.first": { + ID: "job-id", + State: json.RawMessage(`{"name":"first"}`), + DependsOn: []deployplan.DependsOnEntry{{Node: "resources.pipelines.second", Label: "${resources.pipelines.second.id}"}}, + }, + } + jobState, err := json.Marshal(dstate.RecordedState{ + State: direct["resources.jobs.first"].State, + DependsOn: direct["resources.jobs.first"].DependsOn, + }) + require.NoError(t, err) + + tests := []struct { + name string + existing []dms.Resource + wantKeys []string + }{ + { + name: "new migration includes every resource in stable order", + wantKeys: []string{"resources.jobs.first", "resources.pipelines.second"}, + }, + { + name: "retry skips a resource already recorded with identical state", + existing: []dms.Resource{{ + Key: "resources.jobs.first", + ID: "job-id", + State: string(jobState), + }}, + wantKeys: []string{"resources.pipelines.second"}, + }, + { + name: "retry rewrites a stale recorded resource", + existing: []dms.Resource{{ + Key: "resources.jobs.first", + ID: "old-job-id", + State: string(jobState), + }}, + wantKeys: []string{"resources.jobs.first", "resources.pipelines.second"}, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + resources, staged, err := migrationResources(direct, tt.existing) + require.NoError(t, err) + assert.Equal(t, tt.wantKeys, migrationResourceKeys(resources)) + assert.Equal(t, tt.wantKeys, stagedResourceKeys(staged)) + for _, operation := range staged { + assert.Equal(t, dms.OperationActionTypeMigrate, operation.ActionType) + } + }) + } +} + +func migrationResourceKeys(resources []migrationResource) []string { + keys := make([]string, len(resources)) + for i, resource := range resources { + keys[i] = resource.key + } + return keys +} + +func stagedResourceKeys(operations []dms.StagedOperation) []string { + keys := make([]string, len(operations)) + for i, operation := range operations { + keys[i] = operation.ResourceKey + } + return keys +} + +func TestMigrationVersionID(t *testing.T) { + tests := []struct { + name string + directSerial int + lastDMS string + wantVersionID int + }{ + {name: "direct state is newer", directSerial: 7, lastDMS: "3", wantVersionID: 8}, + {name: "DMS is newer after a partial retry", directSerial: 3, lastDMS: "7", wantVersionID: 8}, + {name: "new DMS deployment follows direct state", directSerial: 1, wantVersionID: 2}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + got, err := migrationVersionID(tt.directSerial, tt.lastDMS) + require.NoError(t, err) + assert.Equal(t, tt.wantVersionID, got) + }) + } +} diff --git a/bundle/statemgmt/state_push.go b/bundle/statemgmt/state_push.go index c000e316e49..09ea3570c0a 100644 --- a/bundle/statemgmt/state_push.go +++ b/bundle/statemgmt/state_push.go @@ -3,6 +3,7 @@ package statemgmt import ( "context" "errors" + "fmt" "io/fs" "os" @@ -43,6 +44,29 @@ func PushResourcesState(ctx context.Context, b *bundle.Bundle) { } } +// BackupRemoteResourcesState writes the selected direct state to resources.json.backup without +// changing the active state. Migration treats any failure as fatal because this is its rollback +// point once DMS has a complete copy. +func BackupRemoteResourcesState(ctx context.Context, b *bundle.Bundle) error { + f, err := deploy.StateFiler(ctx, b) + if err != nil { + return err + } + + remotePath, localPath := b.StateFilenameDirect(ctx) + local, err := os.Open(localPath) + if err != nil { + return fmt.Errorf("opening resources state for backup: %w", err) + } + defer local.Close() + + backupPath := remotePath + ".backup" + if err := f.Write(ctx, backupPath, local, filer.CreateParentDirectories, filer.OverwriteIfExists); err != nil { + return fmt.Errorf("writing resources state backup to %s: %w", backupPath, err) + } + return nil +} + func BackupRemoteTerraformState(ctx context.Context, b *bundle.Bundle) { f, err := deploy.StateFiler(ctx, b) if err != nil { diff --git a/cmd/bundle/deployment/migrate.go b/cmd/bundle/deployment/migrate.go index be7b07c9a67..218f4898b73 100644 --- a/cmd/bundle/deployment/migrate.go +++ b/cmd/bundle/deployment/migrate.go @@ -1,6 +1,7 @@ package deployment import ( + "context" "encoding/json" "errors" "fmt" @@ -13,6 +14,8 @@ import ( "github.com/databricks/cli/bundle/direct/dresources" "github.com/databricks/cli/bundle/direct/dstate" "github.com/databricks/cli/bundle/migrate" + "github.com/databricks/cli/bundle/phases" + "github.com/databricks/cli/bundle/statemgmt" "github.com/databricks/cli/cmd/bundle/utils" "github.com/databricks/cli/cmd/root" "github.com/databricks/cli/libs/cmdio" @@ -52,9 +55,10 @@ func getCommonArgs(cmd *cobra.Command) string { } func newMigrateCommand() *cobra.Command { + var enableHistory bool cmd := &cobra.Command{ Use: "migrate", - Short: "Migrate from Terraform to Direct deployment engine", + Short: "Migrate bundle deployment state", Long: `This command converts your bundle from using Terraform for deployment to using the Direct deployment engine. It reads resource IDs from the existing Terraform state and creates a Direct deployment state file (resources.json) with the same @@ -62,6 +66,10 @@ lineage and incremented serial number. Note, the migration is performed locally only. To finalize it, run 'bundle deploy'. This will synchronize the state file to the workspace so that subsequent deploys of this bundle use direct deployment engine as well. + +With --enable-history, this command instead migrates an existing Direct deployment +to the deployment metadata service. The migration completes before any later deploy, +and resources.json is backed up before DMS becomes authoritative. `, Args: root.NoArgs, } @@ -69,6 +77,7 @@ to the workspace so that subsequent deploys of this bundle use direct deployment // --noplancheck kept for backward compatibility; the plan check was removed // because the command no longer invokes the Terraform engine. cmd.Flags().Bool("noplancheck", false, "No-op (kept for compatibility).") + cmd.Flags().BoolVar(&enableHistory, "enable-history", false, "Migrate an existing Direct deployment to deployment history.") cmd.RunE = func(cmd *cobra.Command, args []string) error { extraArgsStr := getCommonArgs(cmd) @@ -78,10 +87,11 @@ to the workspace so that subsequent deploys of this bundle use direct deployment cmd.SetContext(env.Set(cmd.Context(), engine.EnvVar, "")) opts := utils.ProcessOptions{ - AlwaysPull: true, + AlwaysPull: true, + SkipEnforcingDeploymentHistorySetting: enableHistory, // Same options as regular deploy, to ensure bundle config is in the same state FastValidate: true, - Build: true, + Build: !enableHistory, } b, stateDesc, err := utils.ProcessBundleRet(cmd, opts) @@ -89,6 +99,9 @@ to the workspace so that subsequent deploys of this bundle use direct deployment return err } ctx := cmd.Context() + if enableHistory { + return migrateToDeploymentHistory(ctx, b, stateDesc) + } if stateDesc.Lineage == "" { cmdio.LogString(ctx, `Error: This command migrates the existing Terraform state file (terraform.tfstate) to a direct deployment state file (resources.json). However, no existing local or remote state was found. @@ -203,3 +216,34 @@ To undo the migration, remove %s and rename %s to %s return cmd } + +func migrateToDeploymentHistory(ctx context.Context, b *bundle.Bundle, stateDesc *statemgmt.StateDesc) error { + if !utils.ConfiguresDeploymentHistory(ctx, b) { + return errors.New(`deployment history is not enabled for this target + +Set experimental.deployment_history: true, then rerun "databricks bundle deployment migrate --enable-history"`) + } + if stateDesc.SourcePath == "" { + return errors.New("no existing Direct deployment state was found; run bundle deploy to create a deployment with history enabled") + } + if !stateDesc.Engine.IsDirect() { + return errors.New(`migration to deployment history requires Direct deployment state + +First disable experimental.deployment_history and run "databricks bundle deploy" to migrate the Terraform state. Then re-enable deployment history and rerun this command`) + } + if stateDesc.IsDMS() { + cmdio.LogString(ctx, "Deployment already records deployment history.") + return nil + } + + _, localPath := b.StateFilenameDirect(ctx) + if err := b.DeploymentBundle.StateDB.Open(ctx, localPath, dstate.WithRecovery(true), dstate.WithWrite(false), dstate.WithDeploymentHistory(false), dstate.OpenDmsArgs{}); err != nil { + return err + } + + phases.MigrateToDMS(ctx, b) + if logdiag.HasError(ctx) { + return root.ErrAlreadyPrinted + } + return nil +} diff --git a/cmd/bundle/utils/process.go b/cmd/bundle/utils/process.go index aa7ce653a09..13293b2a5ec 100644 --- a/cmd/bundle/utils/process.go +++ b/cmd/bundle/utils/process.go @@ -4,7 +4,6 @@ import ( "context" "errors" "fmt" - "path" "path/filepath" "strconv" "time" @@ -33,8 +32,6 @@ import ( "github.com/databricks/cli/libs/logdiag" "github.com/databricks/cli/libs/sync" "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" @@ -334,6 +331,12 @@ func ProcessBundleRet(cmd *cobra.Command, opts ProcessOptions) (b *bundle.Bundle // was removed in v1.20.0, so a migration failure is fatal - there is no engine to fall // back to. Read-only commands that do not need state keep reading the Terraform state. if b.MigratingToDirect && needsState { + if ConfiguresDeploymentHistory(ctx, b) { + logdiag.LogError(ctx, errors.New(`cannot migrate Terraform state directly to deployment history + +First disable experimental.deployment_history and run "databricks bundle deploy" to migrate to Direct state. Then re-enable deployment history and run "databricks bundle deployment migrate --enable-history"`)) + return b, stateDesc, root.ErrAlreadyPrinted + } if err := migrateTerraformToDirect(ctx, b, stateDesc); err != nil { logdiag.LogError(ctx, err) return b, stateDesc, root.ErrAlreadyPrinted @@ -372,7 +375,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 = dms.LookupDeployment(ctx, b.WorkspaceClient(ctx), b.Config.Workspace.StatePath) if err != nil { logdiag.LogError(ctx, err) return b, stateDesc, root.ErrAlreadyPrinted @@ -588,36 +591,6 @@ func ResolveEngineSetting(ctx context.Context, b *bundle.Bundle) (engine.EngineS return engine.EngineSetting{Type: engine.Default, Source: engine.SourceDefault, IsDefault: true}, nil } -// Lookup and return the deployment object from ${workspace.state_path}/resources.deployment.json -// -// 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) -// that returns an empty id and version unless both halves are there, leaving the caller to call -// CreateDeployment to create or finalize it. -// -// 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 - } - if err != nil { - return "", nil, fmt.Errorf("looking up deployment at %s: %w", nodePath, err) - } - deploymentID := strconv.FormatInt(obj.ObjectId, 10) - deployment, err := w.BundleDeployments.GetDeployment(ctx, bundledeployments.GetDeploymentRequest{ - Name: dms.DeploymentName(deploymentID), - }) - if err != nil { - return "", nil, err - } - return deploymentID, deployment, nil -} - // parseLastVersionID parses the deployment's last recorded version, which the service reports // as a string. It returns 0 when the deployment does not exist yet or has no recorded version. func parseLastVersionID(dmsDeployment *bundledeployments.Deployment) (int, error) { @@ -643,7 +616,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 := dms.LookupDeployment(ctx, b.WorkspaceClient(ctx), b.Config.Workspace.StatePath) if err != nil { return err } @@ -662,7 +635,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 := ConfiguresDeploymentHistory(ctx, b) if stateDesc.SourcePath == "" { if configured { stateDesc.Features = map[string]struct{}{dstate.FeatureDeploymentHistory: {}} @@ -677,26 +650,28 @@ func enforceDeploymentHistorySetting(ctx context.Context, b *bundle.Bundle, stat if stateDesc.SourcePath == "" { return nil } - configured := configuresDeploymentHistory(ctx, b) + configured := ConfiguresDeploymentHistory(ctx, b) recorded := stateDesc.IsDMS() if configured == recorded { return nil } + if configured { + return errors.New(`deployment history is enabled for an existing Direct deployment that has not been migrated + +Run "databricks bundle deployment migrate --enable-history" before deploying`) + } if requireMatch { return fmt.Errorf(`deployment history setting (%t) does not match the existing state (%t) Update experimental.deployment_history to match the existing deployment, or run "databricks bundle destroy" to start over`, configured, recorded) - } - if configured { - return errors.New(`enabling experimental.deployment_history for an existing deployment is not supported - -Run "databricks bundle destroy" first, then deploy again with deployment history enabled`) } log.Warnf(ctx, "Deployment history setting (%t) does not match the existing state (%t). Using the existing state.", configured, recorded) return nil } -func configuresDeploymentHistory(ctx context.Context, b *bundle.Bundle) bool { +// ConfiguresDeploymentHistory reports whether bundle configuration or its environment override +// asks this target to record deployment history. +func ConfiguresDeploymentHistory(ctx context.Context, b *bundle.Bundle) bool { configured := b.Config.Experimental != nil && b.Config.Experimental.DeploymentHistory return bundleenv.RecordsDeploymentHistory(ctx, configured) } diff --git a/libs/dms/client.go b/libs/dms/client.go index 605b4bbd4db..9419b3ed966 100644 --- a/libs/dms/client.go +++ b/libs/dms/client.go @@ -17,6 +17,10 @@ import ( // match DeploymentWhsClient.DEPLOYMENT_NODE_NAME on the service side. const DeploymentNodeName = "resources.deployment.json" +// OperationActionTypeMigrate records the one-time transfer of an already-managed resource into +// deployment history. Keep the literal until the generated SDK includes the newly added API enum. +const OperationActionTypeMigrate bundledeployments.OperationActionType = "OPERATION_ACTION_TYPE_MIGRATE" + // statePrefix is what a bundle state key carries and a DMS resource key does not: state calls a // job "resources.jobs.foo", DMS calls it "jobs.foo". Every exported name here takes the state // form; the prefix comes off where a request is built, and back on where a resource is read. diff --git a/libs/dms/deployment.go b/libs/dms/deployment.go new file mode 100644 index 00000000000..ce25275b602 --- /dev/null +++ b/libs/dms/deployment.go @@ -0,0 +1,37 @@ +package dms + +import ( + "context" + "errors" + "fmt" + "path" + "strconv" + + "github.com/databricks/databricks-sdk-go" + "github.com/databricks/databricks-sdk-go/apierr" + "github.com/databricks/databricks-sdk-go/service/bundledeployments" +) + +// LookupDeployment returns the DMS deployment stored under statePath. A missing deployment returns +// empty values so callers can create it. A workspace node without a service record remains an error; +// CreateDeployment owns recovery for that partial-create case. +func LookupDeployment(ctx context.Context, w *databricks.WorkspaceClient, statePath string) (string, *bundledeployments.Deployment, 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, nil + } + if err != nil { + return "", nil, fmt.Errorf("looking up deployment at %s: %w", nodePath, err) + } + + deploymentID := strconv.FormatInt(obj.ObjectId, 10) + deployment, err := w.BundleDeployments.GetDeployment(ctx, bundledeployments.GetDeploymentRequest{ + Name: DeploymentName(deploymentID), + }) + if err != nil { + return "", nil, err + } + return deploymentID, deployment, nil +} From 8879f9512bf2d78fb04087a9edc02fb67317355c Mon Sep 17 00:00:00 2001 From: Shreyas Goenka Date: Tue, 6 Oct 2026 17:18:49 +0000 Subject: [PATCH 2/2] Add changelog for DMS migration command --- .nextchanges/bundles/direct-dms-migration.md | 1 + 1 file changed, 1 insertion(+) create mode 100644 .nextchanges/bundles/direct-dms-migration.md diff --git a/.nextchanges/bundles/direct-dms-migration.md b/.nextchanges/bundles/direct-dms-migration.md new file mode 100644 index 00000000000..21f3858dc94 --- /dev/null +++ b/.nextchanges/bundles/direct-dms-migration.md @@ -0,0 +1 @@ +* Add `bundle deployment migrate --enable-history` to migrate existing Direct deployments to deployment history. ([#6950](https://github.com/databricks/cli/pull/6950))