diff --git a/acceptance/bundle/dms/depends-on/databricks.yml b/acceptance/bundle/dms/depends-on/databricks.yml new file mode 100644 index 00000000000..f97e3a38809 --- /dev/null +++ b/acceptance/bundle/dms/depends-on/databricks.yml @@ -0,0 +1,13 @@ +bundle: + name: dms-depends-on + +experimental: + record_deployment_history: true + +resources: + jobs: + parent: + name: parent + child: + name: child + description: depends on ${resources.jobs.parent.id} diff --git a/acceptance/bundle/dms/depends-on/out.test.toml b/acceptance/bundle/dms/depends-on/out.test.toml new file mode 100644 index 00000000000..e90b6d5d1ba --- /dev/null +++ b/acceptance/bundle/dms/depends-on/out.test.toml @@ -0,0 +1,3 @@ +Local = true +Cloud = false +EnvMatrix.DATABRICKS_BUNDLE_ENGINE = ["direct"] diff --git a/acceptance/bundle/dms/depends-on/output.txt b/acceptance/bundle/dms/depends-on/output.txt new file mode 100644 index 00000000000..2bb8d06b9dc --- /dev/null +++ b/acceptance/bundle/dms/depends-on/output.txt @@ -0,0 +1,26 @@ + +=== Deploy a job that references another: depends_on is recorded alongside the config, since the reference is resolved to a literal in the config itself +>>> [CLI] bundle deploy +Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/dms-depends-on/default/files... +Deploying resources... +Updating deployment state... +Deployment complete! + +>>> print_requests.py //versions/1/operations --sort --oneline +{"method": "POST", "path": "/api/2.0/bundle/deployments/[NUMID]/versions/1/operations", "q": {"resource_key": "jobs.child"}, "body": {"action_type": "OPERATION_ACTION_TYPE_CREATE", "resource_id": "[NUMID]", "resource_key": "jobs.child", "state": {"state": {"deployment": {"kind": "BUNDLE", "metadata_file_path": "/Workspace/Users/[USERNAME]/.bundle/dms-depends-on/default/state/metadata.json"}, "description": "depends on [NUMID]", "edit_mode": "UI_LOCKED", "format": "MULTI_TASK", "max_concurrent_runs": 1, "name": "child", "queue": {"enabled": true}}, "depends_on": [{"node": "resources.jobs.parent", "label": "${resources.jobs.parent.id}"}]}, "status": "OPERATION_STATUS_SUCCEEDED"}} +{"method": "POST", "path": "/api/2.0/bundle/deployments/[NUMID]/versions/1/operations", "q": {"resource_key": "jobs.parent"}, "body": {"action_type": "OPERATION_ACTION_TYPE_CREATE", "resource_id": "[NUMID]", "resource_key": "jobs.parent", "state": {"state": {"deployment": {"kind": "BUNDLE", "metadata_file_path": "/Workspace/Users/[USERNAME]/.bundle/dms-depends-on/default/state/metadata.json"}, "edit_mode": "UI_LOCKED", "format": "MULTI_TASK", "max_concurrent_runs": 1, "name": "parent", "queue": {"enabled": true}}}, "status": "OPERATION_STATUS_SUCCEEDED"}} + +=== Wipe the local state, then destroy: depends_on comes back from DMS, so the child is still deleted before the parent it references +>>> [CLI] bundle destroy --auto-approve +The following resources will be deleted: + delete resources.jobs.child + delete resources.jobs.parent + +All files and directories at the following location will be deleted: /Workspace/Users/[USERNAME]/.bundle/dms-depends-on/default + +Deleting files... +Destroy complete! + +>>> print_requests.py //jobs --oneline +{"method": "POST", "path": "/api/2.2/jobs/delete", "body": {"job_id": [NUMID]}} +{"method": "POST", "path": "/api/2.2/jobs/delete", "body": {"job_id": [NUMID]}} diff --git a/acceptance/bundle/dms/depends-on/script b/acceptance/bundle/dms/depends-on/script new file mode 100644 index 00000000000..905d022619c --- /dev/null +++ b/acceptance/bundle/dms/depends-on/script @@ -0,0 +1,8 @@ +title "Deploy a job that references another: depends_on is recorded alongside the config, since the reference is resolved to a literal in the config itself" +trace $CLI bundle deploy +trace print_requests.py //versions/1/operations --sort --oneline + +title "Wipe the local state, then destroy: depends_on comes back from DMS, so the child is still deleted before the parent it references" +rm -rf .databricks +trace $CLI bundle destroy --auto-approve +trace print_requests.py //jobs --oneline diff --git a/acceptance/bundle/dms/existing-state/databricks.yml b/acceptance/bundle/dms/existing-state/databricks.yml new file mode 100644 index 00000000000..cfd64979342 --- /dev/null +++ b/acceptance/bundle/dms/existing-state/databricks.yml @@ -0,0 +1,10 @@ +bundle: + name: dms-existing-state + +experimental: + record_deployment_history: false + +resources: + jobs: + one: + name: one diff --git a/acceptance/bundle/dms/existing-state/out.test.toml b/acceptance/bundle/dms/existing-state/out.test.toml new file mode 100644 index 00000000000..e90b6d5d1ba --- /dev/null +++ b/acceptance/bundle/dms/existing-state/out.test.toml @@ -0,0 +1,3 @@ +Local = true +Cloud = false +EnvMatrix.DATABRICKS_BUNDLE_ENGINE = ["direct"] diff --git a/acceptance/bundle/dms/existing-state/output.txt b/acceptance/bundle/dms/existing-state/output.txt new file mode 100644 index 00000000000..229b3bea03f --- /dev/null +++ b/acceptance/bundle/dms/existing-state/output.txt @@ -0,0 +1,52 @@ + +=== Deploy without recording: the bundle gets ordinary direct-engine state, unknown to DMS +>>> [CLI] bundle deploy +Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/dms-existing-state/default/files... +Deploying resources... +Updating deployment state... +Deployment complete! + +>>> print_requests.py //api/2.0/bundle --sort --oneline + +=== Turning recording on afterwards is an error: those resources were never recorded, so treating DMS as authoritative would deploy them a second time +>>> update_file.py databricks.yml record_deployment_history: false record_deployment_history: true + +>>> musterr [CLI] bundle deploy +Error: target "default" was deployed without experimental.record_deployment_history and cannot be migrated to it: [TEST_TMP_DIR]/.databricks/bundle/default/resources.json tracks resources this deployment recorded before the feature was enabled. Use a new target, destroy this one and deploy it again, or unset experimental.record_deployment_history + + +=== No deployment was created in DMS +>>> print_requests.py //api/2.0/bundle --sort --oneline + +=== Still an error after wiping the local cache: deploy pulls the state file back from the workspace, so the resources are still tracked +>>> musterr [CLI] bundle deploy +Error: target "default" was deployed without experimental.record_deployment_history and cannot be migrated to it: [TEST_TMP_DIR]/.databricks/bundle/default/resources.json tracks resources this deployment recorded before the feature was enabled. Use a new target, destroy this one and deploy it again, or unset experimental.record_deployment_history + + +>>> print_requests.py //api/2.0/bundle --sort --oneline + +=== Destroy clears the tracked resources, so recording can be enabled afterwards +>>> update_file.py databricks.yml record_deployment_history: true record_deployment_history: false + +>>> [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/default + +Deleting files... +Destroy complete! + +>>> update_file.py databricks.yml record_deployment_history: false record_deployment_history: true + +>>> [CLI] bundle deploy +Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/dms-existing-state/default/files... +Deploying resources... +Updating deployment state... +Deployment complete! + +>>> print_requests.py //api/2.0/bundle --sort --oneline +{"method": "POST", "path": "/api/2.0/bundle/deployments", "body": {"initial_parent_path": "/Workspace/Users/[USERNAME]/.bundle/dms-existing-state/default/state", "target_name": "default"}} +{"method": "POST", "path": "/api/2.0/bundle/deployments/[NUMID]/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/[NUMID]/versions/1/complete", "body": {"completion_reason": "VERSION_COMPLETE_SUCCESS"}} +{"method": "POST", "path": "/api/2.0/bundle/deployments/[NUMID]/versions/1/operations", "q": {"resource_key": "jobs.one"}, "body": {"action_type": "OPERATION_ACTION_TYPE_CREATE", "resource_id": "[NUMID]", "resource_key": "jobs.one", "state": {"state": {"deployment": {"kind": "BUNDLE", "metadata_file_path": "/Workspace/Users/[USERNAME]/.bundle/dms-existing-state/default/state/metadata.json"}, "edit_mode": "UI_LOCKED", "format": "MULTI_TASK", "max_concurrent_runs": 1, "name": "one", "queue": {"enabled": true}}}, "status": "OPERATION_STATUS_SUCCEEDED"}} diff --git a/acceptance/bundle/dms/existing-state/script b/acceptance/bundle/dms/existing-state/script new file mode 100644 index 00000000000..ae9be95f701 --- /dev/null +++ b/acceptance/bundle/dms/existing-state/script @@ -0,0 +1,22 @@ +title "Deploy without recording: the bundle gets ordinary direct-engine state, unknown to DMS" +trace $CLI bundle deploy +trace print_requests.py //api/2.0/bundle --sort --oneline + +title "Turning recording on afterwards is an error: those resources were never recorded, so treating DMS as authoritative would deploy them a second time" +trace update_file.py databricks.yml "record_deployment_history: false" "record_deployment_history: true" +trace musterr $CLI bundle deploy + +title "No deployment was created in DMS" +trace print_requests.py //api/2.0/bundle --sort --oneline + +title "Still an error after wiping the local cache: deploy pulls the state file back from the workspace, so the resources are still tracked" +rm -rf .databricks +trace musterr $CLI bundle deploy +trace print_requests.py //api/2.0/bundle --sort --oneline + +title "Destroy clears the tracked resources, so recording can be enabled afterwards" +trace update_file.py databricks.yml "record_deployment_history: true" "record_deployment_history: false" +trace $CLI bundle destroy --auto-approve +trace update_file.py databricks.yml "record_deployment_history: false" "record_deployment_history: true" +trace $CLI bundle deploy +trace print_requests.py //api/2.0/bundle --sort --oneline diff --git a/acceptance/bundle/dms/multiple-resources/databricks.yml b/acceptance/bundle/dms/multiple-resources/databricks.yml new file mode 100644 index 00000000000..30f2f433b81 --- /dev/null +++ b/acceptance/bundle/dms/multiple-resources/databricks.yml @@ -0,0 +1,18 @@ +bundle: + name: dms-multiple-resources + +experimental: + record_deployment_history: true + +resources: + jobs: + one: + name: one + two: + name: two + three: + name: three + four: + name: four + five: + name: five diff --git a/acceptance/bundle/dms/multiple-resources/out.test.toml b/acceptance/bundle/dms/multiple-resources/out.test.toml new file mode 100644 index 00000000000..e90b6d5d1ba --- /dev/null +++ b/acceptance/bundle/dms/multiple-resources/out.test.toml @@ -0,0 +1,3 @@ +Local = true +Cloud = false +EnvMatrix.DATABRICKS_BUNDLE_ENGINE = ["direct"] diff --git a/acceptance/bundle/dms/multiple-resources/output.txt b/acceptance/bundle/dms/multiple-resources/output.txt new file mode 100644 index 00000000000..75086ceb5f7 --- /dev/null +++ b/acceptance/bundle/dms/multiple-resources/output.txt @@ -0,0 +1,25 @@ + +=== Deploy several resources: operations are uploaded from background workers, so exactly one is recorded per resource no matter what order the uploads finish in. The serialized state is dropped from the output here; bundle/dms/record covers it. +>>> [CLI] bundle deploy +Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/dms-multiple-resources/default/files... +Deploying resources... +Updating deployment state... +Deployment complete! + +>>> print_requests.py //versions/1/operations --sort --del-body state --oneline +{"method": "POST", "path": "/api/2.0/bundle/deployments/[NUMID]/versions/1/operations", "q": {"resource_key": "jobs.five"}, "body": {"action_type": "OPERATION_ACTION_TYPE_CREATE", "resource_id": "[NUMID]", "resource_key": "jobs.five", "status": "OPERATION_STATUS_SUCCEEDED"}} +{"method": "POST", "path": "/api/2.0/bundle/deployments/[NUMID]/versions/1/operations", "q": {"resource_key": "jobs.four"}, "body": {"action_type": "OPERATION_ACTION_TYPE_CREATE", "resource_id": "[NUMID]", "resource_key": "jobs.four", "status": "OPERATION_STATUS_SUCCEEDED"}} +{"method": "POST", "path": "/api/2.0/bundle/deployments/[NUMID]/versions/1/operations", "q": {"resource_key": "jobs.one"}, "body": {"action_type": "OPERATION_ACTION_TYPE_CREATE", "resource_id": "[NUMID]", "resource_key": "jobs.one", "status": "OPERATION_STATUS_SUCCEEDED"}} +{"method": "POST", "path": "/api/2.0/bundle/deployments/[NUMID]/versions/1/operations", "q": {"resource_key": "jobs.three"}, "body": {"action_type": "OPERATION_ACTION_TYPE_CREATE", "resource_id": "[NUMID]", "resource_key": "jobs.three", "status": "OPERATION_STATUS_SUCCEEDED"}} +{"method": "POST", "path": "/api/2.0/bundle/deployments/[NUMID]/versions/1/operations", "q": {"resource_key": "jobs.two"}, "body": {"action_type": "OPERATION_ACTION_TYPE_CREATE", "resource_id": "[NUMID]", "resource_key": "jobs.two", "status": "OPERATION_STATUS_SUCCEEDED"}} + +=== Redeploy with no changes: nothing is applied, so no operations are recorded and only the version is opened and completed +>>> [CLI] bundle deploy +Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/dms-multiple-resources/default/files... +Deploying resources... +Updating deployment state... +Deployment complete! + +>>> print_requests.py //api/2.0/bundle --sort --oneline +{"method": "POST", "path": "/api/2.0/bundle/deployments/[NUMID]/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/[NUMID]/versions/2/complete", "body": {"completion_reason": "VERSION_COMPLETE_SUCCESS"}} diff --git a/acceptance/bundle/dms/multiple-resources/script b/acceptance/bundle/dms/multiple-resources/script new file mode 100644 index 00000000000..e9b9339607e --- /dev/null +++ b/acceptance/bundle/dms/multiple-resources/script @@ -0,0 +1,7 @@ +title "Deploy several resources: operations are uploaded from background workers, so exactly one is recorded per resource no matter what order the uploads finish in. The serialized state is dropped from the output here; bundle/dms/record covers it." +trace $CLI bundle deploy +trace print_requests.py //versions/1/operations --sort --del-body state --oneline + +title "Redeploy with no changes: nothing is applied, so no operations are recorded and only the version is opened and completed" +trace $CLI bundle deploy +trace print_requests.py //api/2.0/bundle --sort --oneline diff --git a/acceptance/bundle/dms/no-resources/databricks.yml b/acceptance/bundle/dms/no-resources/databricks.yml new file mode 100644 index 00000000000..78fad3a292e --- /dev/null +++ b/acceptance/bundle/dms/no-resources/databricks.yml @@ -0,0 +1,5 @@ +bundle: + name: dms-no-resources + +experimental: + record_deployment_history: true diff --git a/acceptance/bundle/dms/no-resources/out.test.toml b/acceptance/bundle/dms/no-resources/out.test.toml new file mode 100644 index 00000000000..e90b6d5d1ba --- /dev/null +++ b/acceptance/bundle/dms/no-resources/out.test.toml @@ -0,0 +1,3 @@ +Local = true +Cloud = false +EnvMatrix.DATABRICKS_BUNDLE_ENGINE = ["direct"] diff --git a/acceptance/bundle/dms/no-resources/output.txt b/acceptance/bundle/dms/no-resources/output.txt new file mode 100644 index 00000000000..b400fb9a67e --- /dev/null +++ b/acceptance/bundle/dms/no-resources/output.txt @@ -0,0 +1,76 @@ + +=== First deploy of a bundle with no resources: the deployment is created, and its workspace node identifies it even though no resource state was written +>>> [CLI] bundle deploy +Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/dms-no-resources/default/files... +Deploying resources... +Deployment complete! + +>>> print_requests.py //api/2.0/bundle --sort --get +{ + "method": "POST", + "path": "/api/2.0/bundle/deployments", + "body": { + "initial_parent_path": "/Workspace/Users/[USERNAME]/.bundle/dms-no-resources/default/state", + "target_name": "default" + } +} +{ + "method": "POST", + "path": "/api/2.0/bundle/deployments/[NUMID]/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/[NUMID]/versions/1/complete", + "body": { + "completion_reason": "VERSION_COMPLETE_SUCCESS" + } +} + +>>> [CLI] workspace get-status /Workspace/Users/[USERNAME]/.bundle/dms-no-resources/default/state/resources.deployment.json +{ + "object_type": "FILE", + "path": "/Workspace/Users/[USERNAME]/.bundle/dms-no-resources/default/state/resources.deployment.json" +} + +=== Redeploy: the deployment is resolved from that node, so no second deployment is created +>>> [CLI] bundle deploy +Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/dms-no-resources/default/files... +Deploying resources... +Deployment complete! + +>>> print_requests.py //api/2.0/bundle --sort --get +{ + "method": "GET", + "path": "/api/2.0/bundle/deployments/[NUMID]" +} +{ + "method": "GET", + "path": "/api/2.0/bundle/deployments/[NUMID]/resources" +} +{ + "method": "POST", + "path": "/api/2.0/bundle/deployments/[NUMID]/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/[NUMID]/versions/2/complete", + "body": { + "completion_reason": "VERSION_COMPLETE_SUCCESS" + } +} diff --git a/acceptance/bundle/dms/no-resources/script b/acceptance/bundle/dms/no-resources/script new file mode 100644 index 00000000000..847935af5d1 --- /dev/null +++ b/acceptance/bundle/dms/no-resources/script @@ -0,0 +1,8 @@ +title "First deploy of a bundle with no resources: the deployment is created, and its workspace node identifies it even though no resource state was written" +trace $CLI bundle deploy +trace print_requests.py //api/2.0/bundle --sort --get +trace $CLI workspace get-status "/Workspace/Users/${CURRENT_USER_NAME}/.bundle/dms-no-resources/default/state/resources.deployment.json" | jq '{object_type,path}' + +title "Redeploy: the deployment is resolved from that node, so no second deployment is created" +trace $CLI bundle deploy +trace print_requests.py //api/2.0/bundle --sort --get diff --git a/acceptance/bundle/dms/record/databricks.yml b/acceptance/bundle/dms/record/databricks.yml new file mode 100644 index 00000000000..b20e6274310 --- /dev/null +++ b/acceptance/bundle/dms/record/databricks.yml @@ -0,0 +1,10 @@ +bundle: + name: dms-record + +experimental: + record_deployment_history: true + +resources: + jobs: + foo: + name: foo diff --git a/acceptance/bundle/dms/record/out.test.toml b/acceptance/bundle/dms/record/out.test.toml new file mode 100644 index 00000000000..e90b6d5d1ba --- /dev/null +++ b/acceptance/bundle/dms/record/out.test.toml @@ -0,0 +1,3 @@ +Local = true +Cloud = false +EnvMatrix.DATABRICKS_BUNDLE_ENGINE = ["direct"] diff --git a/acceptance/bundle/dms/record/output.txt b/acceptance/bundle/dms/record/output.txt new file mode 100644 index 00000000000..3266e0d3af8 --- /dev/null +++ b/acceptance/bundle/dms/record/output.txt @@ -0,0 +1,158 @@ + +=== 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": { + "initial_parent_path": "/Workspace/Users/[USERNAME]/.bundle/dms-record/default/state", + "target_name": "default" + } +} +{ + "method": "POST", + "path": "/api/2.0/bundle/deployments/[NUMID]/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/[NUMID]/versions/1/complete", + "body": { + "completion_reason": "VERSION_COMPLETE_SUCCESS" + } +} +{ + "method": "POST", + "path": "/api/2.0/bundle/deployments/[NUMID]/versions/1/operations", + "q": { + "resource_key": "jobs.foo" + }, + "body": { + "action_type": "OPERATION_ACTION_TYPE_CREATE", + "resource_id": "[NUMID]", + "resource_key": "jobs.foo", + "state": { + "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 deployment ID is the ID of the workspace node the service registered under initial_parent_path; the CLI stores nothing locally +>>> [CLI] workspace get-status /Workspace/Users/[USERNAME]/.bundle/dms-record/default/state/resources.deployment.json +{ + "object_type": "FILE", + "path": "/Workspace/Users/[USERNAME]/.bundle/dms-record/default/state/resources.deployment.json" +} + +>>> jq has("deployment_id") .databricks/bundle/default/resources.json +false + +=== The state records the feature flag, at the version that makes a CLI without it refuse the state rather than deploy against resources it cannot see +>>> jq {state_version, features} .databricks/bundle/default/resources.json +{ + "state_version": 3, + "features": { + "deployment_history": {} + } +} + +=== Redeploy after deleting the local cache: the deployment ID is resolved from that node, 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/[NUMID]/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/[NUMID]/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/[NUMID]" +} +{ + "method": "POST", + "path": "/api/2.0/bundle/deployments/[NUMID]/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/[NUMID]/versions/3/complete", + "body": { + "completion_reason": "VERSION_COMPLETE_SUCCESS" + } +} +{ + "method": "POST", + "path": "/api/2.0/bundle/deployments/[NUMID]/versions/3/operations", + "q": { + "resource_key": "jobs.foo" + }, + "body": { + "action_type": "OPERATION_ACTION_TYPE_DELETE", + "resource_key": "jobs.foo", + "status": "OPERATION_STATUS_SUCCEEDED" + } +} diff --git a/acceptance/bundle/dms/record/script b/acceptance/bundle/dms/record/script new file mode 100644 index 00000000000..3298c8fb131 --- /dev/null +++ b/acceptance/bundle/dms/record/script @@ -0,0 +1,19 @@ +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 deployment ID is the ID of the workspace node the service registered under initial_parent_path; the CLI stores nothing locally" +trace $CLI workspace get-status "/Workspace/Users/${CURRENT_USER_NAME}/.bundle/dms-record/default/state/resources.deployment.json" | jq '{object_type,path}' +trace jq 'has("deployment_id")' .databricks/bundle/default/resources.json + +title "The state records the feature flag, at the version that makes a CLI without it refuse the state rather than deploy against resources it cannot see" +trace jq '{state_version, features}' .databricks/bundle/default/resources.json + +title "Redeploy after deleting the local cache: the deployment ID is resolved from that node, 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 diff --git a/acceptance/bundle/dms/redeploy-after-destroy/databricks.yml b/acceptance/bundle/dms/redeploy-after-destroy/databricks.yml new file mode 100644 index 00000000000..8f79a2c0381 --- /dev/null +++ b/acceptance/bundle/dms/redeploy-after-destroy/databricks.yml @@ -0,0 +1,10 @@ +bundle: + name: dms-redeploy-after-destroy + +experimental: + record_deployment_history: true + +resources: + jobs: + foo: + name: foo diff --git a/acceptance/bundle/dms/redeploy-after-destroy/out.test.toml b/acceptance/bundle/dms/redeploy-after-destroy/out.test.toml new file mode 100644 index 00000000000..e90b6d5d1ba --- /dev/null +++ b/acceptance/bundle/dms/redeploy-after-destroy/out.test.toml @@ -0,0 +1,3 @@ +Local = true +Cloud = false +EnvMatrix.DATABRICKS_BUNDLE_ENGINE = ["direct"] diff --git a/acceptance/bundle/dms/redeploy-after-destroy/output.txt b/acceptance/bundle/dms/redeploy-after-destroy/output.txt new file mode 100644 index 00000000000..ff8694b5424 --- /dev/null +++ b/acceptance/bundle/dms/redeploy-after-destroy/output.txt @@ -0,0 +1,89 @@ + +=== Deploy, then destroy: the deployment record and its workspace node are both deleted, so nothing points at the destroyed deployment +>>> [CLI] bundle deploy +Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/dms-redeploy-after-destroy/default/files... +Deploying resources... +Updating deployment state... +Deployment complete! + +>>> [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-redeploy-after-destroy/default + +Deleting files... +Destroy complete! + +>>> musterr [CLI] workspace get-status /Workspace/Users/[USERNAME]/.bundle/dms-redeploy-after-destroy/default/state/resources.deployment.json +Error: Path (/Workspace/Users/[USERNAME]/.bundle/dms-redeploy-after-destroy/default/state/resources.deployment.json) doesn't exist. + +=== Deploy again: with no node to resolve, a fresh deployment is created at version 1, with its own workspace node +>>> [CLI] bundle deploy +Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/dms-redeploy-after-destroy/default/files... +Deploying resources... +Updating deployment state... +Deployment complete! + +>>> [CLI] workspace get-status /Workspace/Users/[USERNAME]/.bundle/dms-redeploy-after-destroy/default/state/resources.deployment.json +{ + "object_type": "FILE", + "path": "/Workspace/Users/[USERNAME]/.bundle/dms-redeploy-after-destroy/default/state/resources.deployment.json" +} + +>>> print_requests.py //api/2.0/bundle --sort --get +{ + "method": "POST", + "path": "/api/2.0/bundle/deployments", + "body": { + "initial_parent_path": "/Workspace/Users/[USERNAME]/.bundle/dms-redeploy-after-destroy/default/state", + "target_name": "default" + } +} +{ + "method": "POST", + "path": "/api/2.0/bundle/deployments/[NUMID]/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/[NUMID]/versions/1/complete", + "body": { + "completion_reason": "VERSION_COMPLETE_SUCCESS" + } +} +{ + "method": "POST", + "path": "/api/2.0/bundle/deployments/[NUMID]/versions/1/operations", + "q": { + "resource_key": "jobs.foo" + }, + "body": { + "action_type": "OPERATION_ACTION_TYPE_CREATE", + "resource_id": "[NUMID]", + "resource_key": "jobs.foo", + "state": { + "state": { + "deployment": { + "kind": "BUNDLE", + "metadata_file_path": "/Workspace/Users/[USERNAME]/.bundle/dms-redeploy-after-destroy/default/state/metadata.json" + }, + "edit_mode": "UI_LOCKED", + "format": "MULTI_TASK", + "max_concurrent_runs": 1, + "name": "foo", + "queue": { + "enabled": true + } + } + }, + "status": "OPERATION_STATUS_SUCCEEDED" + } +} diff --git a/acceptance/bundle/dms/redeploy-after-destroy/script b/acceptance/bundle/dms/redeploy-after-destroy/script new file mode 100644 index 00000000000..04c639019c9 --- /dev/null +++ b/acceptance/bundle/dms/redeploy-after-destroy/script @@ -0,0 +1,11 @@ +title "Deploy, then destroy: the deployment record and its workspace node are both deleted, so nothing points at the destroyed deployment" +trace $CLI bundle deploy +trace $CLI bundle destroy --auto-approve +print_requests.py //api/2.0/bundle --sort --get > /dev/null + +trace musterr $CLI workspace get-status "/Workspace/Users/${CURRENT_USER_NAME}/.bundle/dms-redeploy-after-destroy/default/state/resources.deployment.json" + +title "Deploy again: with no node to resolve, a fresh deployment is created at version 1, with its own workspace node" +trace $CLI bundle deploy +trace $CLI workspace get-status "/Workspace/Users/${CURRENT_USER_NAME}/.bundle/dms-redeploy-after-destroy/default/state/resources.deployment.json" | jq '{object_type,path}' +trace print_requests.py //api/2.0/bundle --sort --get diff --git a/acceptance/bundle/dms/test.toml b/acceptance/bundle/dms/test.toml new file mode 100644 index 00000000000..1e36331a16f --- /dev/null +++ b/acceptance/bundle/dms/test.toml @@ -0,0 +1,12 @@ +Local = true +Cloud = false + +# Deployment Metadata Service (DMS) recording is only supported by the direct +# engine; it is a no-op on terraform. +EnvMatrix.DATABRICKS_BUNDLE_ENGINE = ["direct"] + +RecordRequests = true + +Ignore = [ + '.databricks', +] diff --git a/bundle/config/experimental.go b/bundle/config/experimental.go index 658f1cea819..c3f1465d880 100644 --- a/bundle/config/experimental.go +++ b/bundle/config/experimental.go @@ -53,6 +53,10 @@ type Experimental struct { // RecordDeploymentHistory opts the bundle into the deployment metadata // service (DMS), which records deployment history and tracks what changed // across deployments. + // + // Only supported for a bundle with no deployed resources yet: DMS becomes the + // source of truth for resource state, and resources tracked in an existing + // state file cannot be handed over to it yet. See dstate.DeploymentState.Open. RecordDeploymentHistory bool `json:"record_deployment_history,omitempty"` } diff --git a/bundle/configsync/diff.go b/bundle/configsync/diff.go index ea45903508b..ca5b2c9410b 100644 --- a/bundle/configsync/diff.go +++ b/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 diff --git a/bundle/configsync/variables.go b/bundle/configsync/variables.go index 055a47dc934..433b607a037 100644 --- a/bundle/configsync/variables.go +++ b/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 } diff --git a/bundle/direct/bind.go b/bundle/direct/bind.go index 9760ce95666..ec910b2734e 100644 --- a/bundle/direct/bind.go +++ b/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 } diff --git a/bundle/direct/bundle_apply.go b/bundle/direct/bundle_apply.go index c4178c4e601..f29aa18a186 100644 --- a/bundle/direct/bundle_apply.go +++ b/bundle/direct/bundle_apply.go @@ -34,6 +34,11 @@ func (b *DeploymentBundle) Apply(ctx context.Context, client *databricks.Workspa return } + // Operations are recorded with DMS from background workers so a resource's + // deploy is not held up by the CreateOperation round trip. The queue is + // drained below, once every apply worker has finished recording. + opQueue := newOperationQueue(ctx, b.OpRec) + g.Run(defaultParallelism, func(resourceKey string, failedDependency *string) bool { entry, err := plan.WriteLockEntry(resourceKey) if err != nil { @@ -88,6 +93,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 := opQueue.record(ctx, resourceKey, action, "", nil, nil); err != nil { + logdiag.LogError(ctx, fmt.Errorf("%s: %w", errorPrefix, err)) + return false + } return true } @@ -116,6 +126,15 @@ 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. depends_on is recorded alongside the config + // because it cannot be recomputed from it (see dstate.RecordedState). + if err := opQueue.record(ctx, resourceKey, action, b.StateDB.GetResourceID(resourceKey), sv.Value, d.DependsOn); 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. @@ -139,6 +158,13 @@ func (b *DeploymentBundle) Apply(ctx context.Context, client *databricks.Workspa return true }) + + // Wait for the queued operations before returning: the caller completes the + // DMS version right after, and a version must not be completed with uploads + // still in flight. + if err := opQueue.close(); err != nil { + logdiag.LogError(ctx, err) + } } func (b *DeploymentBundle) LookupReferencePostDeploy(ctx context.Context, path *structpath.PathNode) (any, error) { diff --git a/bundle/direct/dstate/dms.go b/bundle/direct/dstate/dms.go new file mode 100644 index 00000000000..ec51e2c4479 --- /dev/null +++ b/bundle/direct/dstate/dms.go @@ -0,0 +1,80 @@ +package dstate + +import ( + "context" + "encoding/json" + "fmt" + + "github.com/databricks/cli/bundle/deployplan" + "github.com/databricks/databricks-sdk-go/service/bundledeployments" +) + +// RecordedState is what the CLI serializes into the DMS Operation.State field. +// +// It is an envelope rather than the bare resource config, because depends_on has +// to survive the round trip: DMS has no field for dependency edges, and they +// cannot be recomputed from the config once it is recorded (references are +// resolved to literals before serialization). Nesting depends_on inside the +// config instead would collide with resource fields of the same name, e.g. +// jobs.Task.depends_on. +// +// The shape deliberately matches the local ResourceEntry so both sides of the +// state round trip look the same. +type RecordedState struct { + State json.RawMessage `json:"state"` + DependsOn []deployplan.DependsOnEntry `json:"depends_on,omitempty"` +} + +// readDMSState replaces the file-derived resource state with the state recorded +// in DMS. Recording is only enabled for net-new deployments, so once a +// deployment exists DMS owns its resource set outright - including when that set +// is empty, which is a successful deploy of nothing rather than missing data. +// The caller holds db.mu. +func (db *DeploymentState) readDMSState(ctx context.Context, src *DMSSource) error { + resources, err := fetchDeploymentResources(ctx, src.Client, src.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 +} + +// 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 recorded RecordedState + if res.State != nil { + if err := json.Unmarshal(*res.State, &recorded); err != nil { + return nil, fmt.Errorf("interpreting state recorded for %s: %w", key, err) + } + } + + out[key] = ResourceEntry{ + ID: res.ResourceId, + State: recorded.State, + DependsOn: recorded.DependsOn, + } + } + return out, nil +} diff --git a/bundle/direct/dstate/dms_test.go b/bundle/direct/dstate/dms_test.go new file mode 100644 index 00000000000..35fe7acbb0c --- /dev/null +++ b/bundle/direct/dstate/dms_test.go @@ -0,0 +1,68 @@ +package dstate + +import ( + "context" + "encoding/json" + "testing" + + "github.com/databricks/cli/bundle/deployplan" + "github.com/databricks/databricks-sdk-go/listing" + "github.com/databricks/databricks-sdk-go/service/bundledeployments" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// fakeResourceLister serves a fixed set of resources from ListResources. It +// embeds the SDK interface so it satisfies it while only overriding the one +// method the read path uses. +type fakeResourceLister struct { + bundledeployments.BundleDeploymentsInterface + resources []bundledeployments.Resource +} + +func (f *fakeResourceLister) ListResources(ctx context.Context, req bundledeployments.ListResourcesRequest) listing.Iterator[bundledeployments.Resource] { + return listing.NewIterator( + &req, + func(ctx context.Context, r bundledeployments.ListResourcesRequest) (*bundledeployments.ListResourcesResponse, error) { + return &bundledeployments.ListResourcesResponse{Resources: f.resources}, nil + }, + func(resp *bundledeployments.ListResourcesResponse) []bundledeployments.Resource { + return resp.Resources + }, + func(resp *bundledeployments.ListResourcesResponse) *bundledeployments.ListResourcesRequest { + return nil + }, + ) +} + +func TestFetchDeploymentResourcesUnwrapsEnvelope(t *testing.T) { + recorded := json.RawMessage(`{"state":{"name":"foo"},"depends_on":[{"node":"resources.pipelines.bar","label":"${resources.pipelines.bar.id}"}]}`) + f := &fakeResourceLister{resources: []bundledeployments.Resource{ + {ResourceKey: "jobs.foo", ResourceId: "123", State: &recorded}, + {ResourceKey: "pipelines.bar", ResourceId: "456"}, + }} + + got, err := fetchDeploymentResources(t.Context(), f, "dep-1") + require.NoError(t, err) + + // depends_on comes back from the envelope, so a bundle whose local state was + // wiped still has the edges needed for delete ordering. + assert.Equal(t, map[string]ResourceEntry{ + "resources.jobs.foo": { + ID: "123", + State: json.RawMessage(`{"name":"foo"}`), + DependsOn: []deployplan.DependsOnEntry{{Node: "resources.pipelines.bar", Label: "${resources.pipelines.bar.id}"}}, + }, + "resources.pipelines.bar": {ID: "456"}, + }, got) +} + +func TestFetchDeploymentResourcesRejectsMalformedState(t *testing.T) { + recorded := json.RawMessage(`not json`) + f := &fakeResourceLister{resources: []bundledeployments.Resource{ + {ResourceKey: "jobs.foo", ResourceId: "123", State: &recorded}, + }} + + _, err := fetchDeploymentResources(t.Context(), f, "dep-1") + assert.ErrorContains(t, err, "interpreting state recorded for resources.jobs.foo") +} diff --git a/bundle/direct/dstate/migrate.go b/bundle/direct/dstate/migrate.go index e4d21a7054a..288fdcf863f 100644 --- a/bundle/direct/dstate/migrate.go +++ b/bundle/direct/dstate/migrate.go @@ -12,25 +12,34 @@ import ( "github.com/databricks/databricks-sdk-go/service/iam" ) +// knownFeatures lists the state feature flags this CLI implements. A state that +// records anything outside this set is refused by migrateState. +var knownFeatures = map[string]bool{ + FeatureDeploymentHistory: true, +} + // migrateState runs all necessary migrations on the database. // It is called after loading state from disk. func migrateState(db *Database) error { - // featureStateVersion states carry a feature list this CLI does not yet write or - // understand (see the featureStateVersion doc comment). A featureStateVersion - // state with no features is equivalent to currentStateVersion, so accept it and - // return without running the migrations below, leaving the on-disk version at - // featureStateVersion rather than flipping it down. One that records any feature - // depends on capabilities this CLI lacks, so refuse it and tell the user to upgrade. + // featureStateVersion states carry a feature list (see the featureStateVersion + // doc comment). A featureStateVersion state with no features is equivalent to + // currentStateVersion, so accept it and return without running the migrations + // below, leaving the on-disk version at featureStateVersion rather than flipping + // it down. Same for a state whose features this CLI implements. One that records + // a feature this CLI does not know depends on capabilities it lacks, so refuse it + // and tell the user to upgrade. if db.StateVersion == featureStateVersion { - if len(db.Features) == 0 { - return nil - } - features := make([]string, 0, len(db.Features)) + unknown := make([]string, 0, len(db.Features)) for name := range db.Features { - features = append(features, name) + if !knownFeatures[name] { + unknown = append(unknown, name) + } + } + if len(unknown) == 0 { + return nil } - slices.Sort(features) - return fmt.Errorf("the deployment state requires features this CLI does not support: %s; upgrade to the latest CLI version and see %s for more information", strings.Join(features, ", "), featuresDocURL) + slices.Sort(unknown) + return fmt.Errorf("the deployment state requires features this CLI does not support: %s; upgrade to the latest CLI version and see %s for more information", strings.Join(unknown, ", "), featuresDocURL) } if db.StateVersion == currentStateVersion { diff --git a/bundle/direct/dstate/state.go b/bundle/direct/dstate/state.go index f6c8fc8ba3c..b367e97001b 100644 --- a/bundle/direct/dstate/state.go +++ b/bundle/direct/dstate/state.go @@ -19,6 +19,7 @@ import ( "github.com/databricks/cli/libs/dyn" "github.com/databricks/cli/libs/log" "github.com/databricks/cli/libs/structs/structwalk" + "github.com/databricks/databricks-sdk-go/service/bundledeployments" "github.com/google/uuid" ) @@ -30,22 +31,28 @@ const ( maxWalEntrySize = 10 * 1024 * 1024 walSuffix = ".wal" - // featureStateVersion is the schema version a future CLI will write once it - // records deployment state "feature flags" (see Header.Features). This CLI does - // not write it and records no features; it exists now only so this CLI reads - // such states correctly (see migrateState): - // - featureStateVersion with no features -> accept and leave the version as-is - // - featureStateVersion with any feature -> refuse, tell the user to upgrade + // featureStateVersion is the schema version written once a state records + // deployment state "feature flags" (see Header.Features). Reading such a state + // (see migrateState): + // - featureStateVersion with no features -> accept, leave the version as-is + // - featureStateVersion with known features -> accept + // - featureStateVersion with unknown features -> refuse, tell the user to upgrade // // A featureStateVersion state with no features is equivalent to // currentStateVersion, but we deliberately do not flip the on-disk version down // to currentStateVersion: a state written at featureStateVersion stays at - // featureStateVersion. This is forward-compat scaffolding so that a later release - // can start writing featureStateVersion + features without older CLIs (with this - // change) either mishandling a feature they lack or rejecting a featureless state - // outright. featureStateVersion is always 3. + // featureStateVersion. That way a release can start writing + // featureStateVersion + features without older CLIs either mishandling a feature + // they lack or rejecting a featureless state outright. featureStateVersion is + // always 3. featureStateVersion = 3 + // FeatureDeploymentHistory marks a state whose resources live in the deployment + // metadata service rather than in the state file. A CLI that does not know this + // feature refuses the state instead of deploying against a resource set it + // cannot see - the file's resources are not authoritative. + FeatureDeploymentHistory = "deployment_history" + // supportedStateVersion is the highest schema version this CLI can read. It is // normally equal to currentStateVersion — the version this CLI reads is the // version it writes — and exceeds it only during a two-phase version bump like @@ -81,13 +88,27 @@ type Header struct { Serial int `json:"serial"` // Features maps each feature flag this state depends on to a (currently empty) - // value. This CLI writes no features; it only reads the field to detect a state - // that depends on features it lacks and refuse it (see migrateState). It is a - // map so a future CLI can attach per-feature data without reshaping the state. - // Empty/omitted for states that use no features. + // value. A CLI that does not implement one of them refuses the state rather than + // deploying against it (see migrateState). It is a map so a future CLI can attach + // per-feature data without reshaping the state. Empty/omitted for states that use + // no features. Features map[string]struct{} `json:"features,omitempty"` } +// hasFeature reports whether the state records the given feature flag. +func (h *Header) hasFeature(name string) bool { + _, ok := h.Features[name] + return ok +} + +// setFeature records a feature flag in the state. +func (h *Header) setFeature(name string) { + if h.Features == nil { + h.Features = make(map[string]struct{}) + } + h.Features[name] = struct{}{} +} + type Database struct { Header @@ -218,7 +239,26 @@ type ( WithWrite bool ) -func (db *DeploymentState) Open(ctx context.Context, path string, withRecovery WithRecovery, withWrite WithWrite) error { +// DMSSource tells Open to read resource state from the deployment metadata +// service instead of the state file. Callers pass it only when the bundle set +// experimental.record_deployment_history; a nil *DMSSource keeps Open file-only. +type DMSSource struct { + Client bundledeployments.BundleDeploymentsInterface + + // DeploymentID is resolved from the deployment's workspace node (see + // dms.ResolveDeploymentID), and empty before the first recorded deploy. + DeploymentID string + + // TargetName names the bundle target in the error Open returns for a state that + // predates the opt-in, since the feature is enabled per target. + TargetName string +} + +// Open reads the deployment state from disk (and recovers the WAL when +// withRecovery is set). With a non-nil dmsSource, resources come from DMS rather +// than the file. Lineage and serial always come from the file, since that is +// what the write path increments. +func (db *DeploymentState) Open(ctx context.Context, path string, withRecovery WithRecovery, withWrite WithWrite, dmsSource *DMSSource) error { db.mu.Lock() defer db.mu.Unlock() @@ -266,6 +306,26 @@ func (db *DeploymentState) Open(ctx context.Context, path string, withRecovery W return fmt.Errorf("migrating state %s: %w", path, err) } + if dmsSource != nil { + // An existing state file has to be marked as DMS-owned before its resources + // can be read from DMS. Recording only starts on an empty state, so a state + // with resources and no feature flag predates the opt-in: DMS never saw those + // resources and is authoritative for the whole set once enabled (see + // readDMSState), so they would look absent and be created a second time. + // Migrating such a target is not supported yet. + if len(db.Data.State) > 0 && !db.Data.hasFeature(FeatureDeploymentHistory) { + return fmt.Errorf("target %q was deployed without experimental.record_deployment_history and cannot be migrated to it: %s tracks resources this deployment recorded before the feature was enabled. Use a new target, destroy this one and deploy it again, or unset experimental.record_deployment_history", dmsSource.TargetName, path) + } + // Mark the state as DMS-owned so a CLI without this feature refuses it rather + // than deploying against a resource set it cannot see. + db.Data.setFeature(FeatureDeploymentHistory) + if dmsSource.DeploymentID != "" { + if err := db.readDMSState(ctx, dmsSource); err != nil { + return err + } + } + } + if withWrite { if err := os.MkdirAll(filepath.Dir(walPath), 0o755); err != nil { return fmt.Errorf("failed to create state directory: %w", err) @@ -275,13 +335,7 @@ func (db *DeploymentState) Open(ctx context.Context, path string, withRecovery W return fmt.Errorf("failed to open WAL file %s: %w", walPath, err) } db.walFile = walFile - walHead := Header{ - Lineage: db.GetOrInitLineage(), - Serial: db.Data.Serial + 1, - StateVersion: currentStateVersion, - CLIVersion: build.GetInfo().Version, - } - return appendJSONLine(db.walFile, walHead) + return appendJSONLine(db.walFile, db.newWalHeader()) } return nil @@ -368,6 +422,16 @@ func (db *DeploymentState) mergeWalIntoState(ctx context.Context) (bool, error) return false, fmt.Errorf("WAL serial (%d) is ahead of expected (%d), state may be corrupted", header.Serial, expectedSerial) } newSerial = header.Serial + + // Carry the WAL's features (and the version that goes with them) into the + // state being written, so recovering a WAL written by a DMS-recording deploy + // still produces a state marked as DMS-owned. + for name := range header.Features { + db.Data.setFeature(name) + } + if len(db.Data.Features) > 0 { + db.Data.StateVersion = featureStateVersion + } } else { var entry WALEntry if err := json.Unmarshal(line, &entry); err != nil { @@ -471,13 +535,25 @@ func (db *DeploymentState) UpgradeToWrite() error { } db.walFile = walFile - walHead := Header{ + return appendJSONLine(db.walFile, db.newWalHeader()) +} + +// newWalHeader builds the header for a fresh WAL. Features carry over from the +// state being written, and a state that records any feature is written at +// featureStateVersion so a CLI that lacks the feature refuses it instead of +// deploying against it. The caller holds db.mu. +func (db *DeploymentState) newWalHeader() Header { + version := currentStateVersion + if len(db.Data.Features) > 0 { + version = featureStateVersion + } + return Header{ Lineage: db.GetOrInitLineage(), Serial: db.Data.Serial + 1, - StateVersion: currentStateVersion, + StateVersion: version, CLIVersion: build.GetInfo().Version, + Features: db.Data.Features, } - return appendJSONLine(db.walFile, walHead) } func (db *DeploymentState) AssertOpenedForReadOrWrite() { diff --git a/bundle/direct/dstate/state_test.go b/bundle/direct/dstate/state_test.go index 11589944472..35050575121 100644 --- a/bundle/direct/dstate/state_test.go +++ b/bundle/direct/dstate/state_test.go @@ -20,14 +20,14 @@ func TestOpenSaveFinalizeRoundTrip(t *testing.T) { path := filepath.Join(t.TempDir(), "state.json") var db DeploymentState - require.NoError(t, db.Open(t.Context(), path, WithRecovery(true), WithWrite(true))) + require.NoError(t, db.Open(t.Context(), path, WithRecovery(true), WithWrite(true), nil)) require.NoError(t, db.SaveState("jobs.my_job", "123", map[string]string{"key": "val"}, nil)) mustFinalize(t, &db) // Re-open and verify persisted data. var db2 DeploymentState - require.NoError(t, db2.Open(t.Context(), path, WithRecovery(false), WithWrite(false))) + require.NoError(t, db2.Open(t.Context(), path, WithRecovery(false), WithWrite(false), nil)) assert.Equal(t, 1, db2.Data.Serial) assert.Equal(t, "123", db2.GetResourceID("jobs.my_job")) mustFinalize(t, &db2) @@ -37,7 +37,7 @@ func TestFinalizeWithNoEntriesDoesNotWriteStateFile(t *testing.T) { path := filepath.Join(t.TempDir(), "state.json") var db DeploymentState - require.NoError(t, db.Open(t.Context(), path, WithRecovery(true), WithWrite(true))) + require.NoError(t, db.Open(t.Context(), path, WithRecovery(true), WithWrite(true), nil)) mustFinalize(t, &db) _, err := os.Stat(path) @@ -93,10 +93,10 @@ func TestPanicOnDoubleOpen(t *testing.T) { path := filepath.Join(t.TempDir(), "state.json") var db DeploymentState - require.NoError(t, db.Open(t.Context(), path, WithRecovery(true), WithWrite(true))) + require.NoError(t, db.Open(t.Context(), path, WithRecovery(true), WithWrite(true), nil)) assert.Panics(t, func() { - _ = db.Open(t.Context(), path, WithRecovery(true), WithWrite(true)) + _ = db.Open(t.Context(), path, WithRecovery(true), WithWrite(true), nil) }) mustFinalize(t, &db) } @@ -107,12 +107,12 @@ func TestHeaderOnlyWALRecoveryDoesNotAdvanceSerial(t *testing.T) { // Commit serial 1 with one resource. var db DeploymentState - require.NoError(t, db.Open(t.Context(), path, WithRecovery(true), WithWrite(true))) + require.NoError(t, db.Open(t.Context(), path, WithRecovery(true), WithWrite(true), nil)) require.NoError(t, db.SaveState("jobs.my_job", "123", map[string]string{}, nil)) mustFinalize(t, &db) var committed DeploymentState - require.NoError(t, committed.Open(t.Context(), path, WithRecovery(false), WithWrite(false))) + require.NoError(t, committed.Open(t.Context(), path, WithRecovery(false), WithWrite(false), nil)) lineage := committed.Data.Lineage require.Equal(t, 1, committed.Data.Serial) mustFinalize(t, &committed) @@ -128,7 +128,7 @@ func TestHeaderOnlyWALRecoveryDoesNotAdvanceSerial(t *testing.T) { require.NoError(t, os.WriteFile(walPath, append(headerLine, '\n'), 0o600)) var recovered DeploymentState - require.NoError(t, recovered.Open(t.Context(), path, WithRecovery(true), WithWrite(false))) + require.NoError(t, recovered.Open(t.Context(), path, WithRecovery(true), WithWrite(false), nil)) assert.Equal(t, 1, recovered.Data.Serial) assert.Equal(t, "123", recovered.GetResourceID("jobs.my_job")) assert.NoFileExists(t, walPath) @@ -154,7 +154,16 @@ func TestEmptyFeatureStateAcceptedWithoutFlippingVersion(t *testing.T) { require.NoError(t, migrateState(empty)) assert.Equal(t, featureStateVersion, empty.StateVersion, "v3 + no features keeps its on-disk version, not flipped to v2") - // v3 that records a feature is refused: this CLI does not understand features. + // v3 recording a feature this CLI implements is accepted. + known := &Database{Header: Header{ + StateVersion: featureStateVersion, + Features: map[string]struct{}{FeatureDeploymentHistory: {}}, + }} + require.NoError(t, migrateState(known)) + assert.Equal(t, featureStateVersion, known.StateVersion) + + // v3 recording a feature this CLI does not know is refused: its resources may + // live somewhere this CLI cannot see. withFeature := &Database{Header: Header{ StateVersion: featureStateVersion, Features: map[string]struct{}{"future_feature": {}}, @@ -165,23 +174,33 @@ func TestEmptyFeatureStateAcceptedWithoutFlippingVersion(t *testing.T) { assert.Contains(t, err.Error(), "future_feature") assert.Contains(t, err.Error(), "upgrade to the latest CLI version") assert.Contains(t, err.Error(), featuresDocURL) + + // Only the unknown feature is named, so the message tells the user what to do. + mixed := &Database{Header: Header{ + StateVersion: featureStateVersion, + Features: map[string]struct{}{FeatureDeploymentHistory: {}, "future_feature": {}}, + }} + err = migrateState(mixed) + require.Error(t, err) + assert.Contains(t, err.Error(), "future_feature") + assert.NotContains(t, err.Error(), FeatureDeploymentHistory) } func TestDeleteState(t *testing.T) { path := filepath.Join(t.TempDir(), "state.json") var db DeploymentState - require.NoError(t, db.Open(t.Context(), path, WithRecovery(true), WithWrite(true))) + require.NoError(t, db.Open(t.Context(), path, WithRecovery(true), WithWrite(true), nil)) require.NoError(t, db.SaveState("jobs.my_job", "123", map[string]string{}, nil)) mustFinalize(t, &db) var db2 DeploymentState - require.NoError(t, db2.Open(t.Context(), path, WithRecovery(true), WithWrite(true))) + require.NoError(t, db2.Open(t.Context(), path, WithRecovery(true), WithWrite(true), nil)) require.NoError(t, db2.DeleteState("jobs.my_job")) mustFinalize(t, &db2) var db3 DeploymentState - require.NoError(t, db3.Open(t.Context(), path, WithRecovery(false), WithWrite(false))) + require.NoError(t, db3.Open(t.Context(), path, WithRecovery(false), WithWrite(false), nil)) assert.Equal(t, 2, db3.Data.Serial) assert.Empty(t, db3.GetResourceID("jobs.my_job")) mustFinalize(t, &db3) @@ -193,7 +212,7 @@ func TestGetOrInitLineageReadableBeforeWriteAndPersisted(t *testing.T) { // Fresh state opened read-only, as the deploy does before planning: no // lineage yet. var db DeploymentState - require.NoError(t, db.Open(t.Context(), path, WithRecovery(true), WithWrite(false))) + require.NoError(t, db.Open(t.Context(), path, WithRecovery(true), WithWrite(false), nil)) require.Empty(t, db.Data.Lineage) // GetOrInitLineage initializes the lineage and makes it readable before any @@ -210,7 +229,7 @@ func TestGetOrInitLineageReadableBeforeWriteAndPersisted(t *testing.T) { // Re-open: the persisted lineage matches the one read before the write. var reopened DeploymentState - require.NoError(t, reopened.Open(t.Context(), path, WithRecovery(false), WithWrite(false))) + require.NoError(t, reopened.Open(t.Context(), path, WithRecovery(false), WithWrite(false), nil)) assert.Equal(t, lineage, reopened.Data.Lineage) mustFinalize(t, &reopened) } diff --git a/bundle/direct/opqueue.go b/bundle/direct/opqueue.go new file mode 100644 index 00000000000..1f38c2b7dd6 --- /dev/null +++ b/bundle/direct/opqueue.go @@ -0,0 +1,197 @@ +package direct + +import ( + "context" + "fmt" + "sync" + + "github.com/databricks/cli/bundle/deployplan" + "github.com/databricks/cli/libs/log" +) + +const ( + // operationQueueSize bounds how many recorded operations wait for upload. + // Apply deploys at most defaultParallelism resources at a time, so a queue + // this deep means an apply worker practically never blocks on a free slot. + operationQueueSize = 10 + + // operationUploadWorkers is how many uploads run at a time. It is below + // operationQueueSize so a burst of operations is absorbed by the queue rather + // than by one request per resource. + operationUploadWorkers = 4 +) + +// operationQueue uploads recorded operations from background workers, so an apply +// worker does not wait for the CreateOperation round trip. +// +// At most one upload is in flight per resource key, so the last operation +// recorded for a resource is also the last one the service sees. +// +// Uploads are not fire-and-forget: close returns the first failure and fails the +// deploy. A dropped operation would leave DMS with an incomplete resource set, +// and since DMS then becomes the source of truth (see dstate.overlayDMSState), +// the next deploy would recreate resources that already exist. +type operationQueue struct { + uploader operationUploader + + // queue carries resource keys, not the operations themselves: a worker looks + // the operation up in pending when it picks the key up, which is what lets + // record collapse repeated writes to the same resource. + queue chan string + wg sync.WaitGroup + + // mu guards the fields below. + mu sync.Mutex + + // pending is the latest operation recorded per resource key that no worker has + // picked up yet. + pending map[string]recordedOperation + + // owned holds the resource keys that are already queued or being uploaded. + // Such a key is never queued a second time; recording writes to pending + // instead, which the owning worker re-checks after each upload. This is what + // keeps uploads for one resource sequential. + owned map[string]bool + + err error + closed bool +} + +// newOperationQueue starts the upload workers. It returns nil when uploader is +// nil (recording disabled), and every method is a no-op on a nil queue so callers +// do not have to branch. +// +// ctx is used for the uploads, so it must stay valid until close returns. +func newOperationQueue(ctx context.Context, uploader operationUploader) *operationQueue { + if uploader == nil { + return nil + } + + q := &operationQueue{ + uploader: uploader, + queue: make(chan string, operationQueueSize), + pending: make(map[string]recordedOperation), + owned: make(map[string]bool), + } + + q.wg.Add(operationUploadWorkers) + for range operationUploadWorkers { + go q.work(ctx) + } + + return q +} + +// record serializes an operation and queues it for upload. It makes no API call, +// so upload failures surface from close; an error here only means the applied +// resource could not be turned into a payload. +// +// An operation for a resource that is still waiting replaces it rather than +// queueing again: DMS keeps one state per key, so one upload records both. The +// merge keeps a queued create's action (see mergeAction). Best effort — only +// operations no worker has picked up yet are collapsed. +func (q *operationQueue) record(ctx context.Context, resourceKey string, action deployplan.ActionType, resourceID string, state any, dependsOn []deployplan.DependsOnEntry) error { + if q == nil { + return nil + } + + op, err := newRecordedOperation(action, resourceID, state, dependsOn) + if err != nil { + return err + } + + q.mu.Lock() + queued, waiting := q.pending[resourceKey] + if waiting { + op.action = mergeAction(queued.action, op.action) + } + q.pending[resourceKey] = op + owned := q.owned[resourceKey] + q.owned[resourceKey] = true + q.mu.Unlock() + + if waiting { + log.Debugf(ctx, "Coalescing queued deployment operation for %s", resourceKey) + } + + // Someone already owns this key, so pending is enough: a worker will pick the + // operation up. Queueing again would upload the resource twice in parallel. + if owned { + return nil + } + + q.queue <- resourceKey + return nil +} + +// close drains the queue and returns the first upload error. All callers of +// record must have returned first: record on a closed queue panics. Calling close +// more than once is safe, so callers can defer it and still check the error at a +// specific point. +func (q *operationQueue) close() error { + if q == nil { + return nil + } + + q.mu.Lock() + closed := q.closed + q.closed = true + q.mu.Unlock() + + if !closed { + close(q.queue) + q.wg.Wait() + } + + q.mu.Lock() + defer q.mu.Unlock() + return q.err +} + +func (q *operationQueue) work(ctx context.Context) { + defer q.wg.Done() + + for resourceKey := range q.queue { + // Keep uploading this key until nothing new was recorded for it, instead of + // putting it back on the queue: a worker sending to the channel it consumes + // from can deadlock once the queue is full. + for { + op, ok := q.take(resourceKey) + if !ok { + break + } + + if err := q.uploader.upload(ctx, resourceKey, op); err != nil { + q.setErr(fmt.Errorf("recording operation for %s with the deployment metadata service: %w", resourceKey, err)) + } + } + } +} + +// take claims the operation waiting for resourceKey. It reports false and gives +// up ownership when nothing is waiting, which is what lets the next record +// queue the key again. +func (q *operationQueue) take(resourceKey string) (recordedOperation, bool) { + q.mu.Lock() + defer q.mu.Unlock() + + op, ok := q.pending[resourceKey] + if !ok { + delete(q.owned, resourceKey) + return recordedOperation{}, false + } + + delete(q.pending, resourceKey) + return op, true +} + +// setErr keeps the first upload error; later ones are dropped because one failure +// is enough to fail the deploy. +func (q *operationQueue) setErr(err error) { + q.mu.Lock() + defer q.mu.Unlock() + + if q.err == nil { + q.err = err + } +} diff --git a/bundle/direct/opqueue_test.go b/bundle/direct/opqueue_test.go new file mode 100644 index 00000000000..6bf0da8107b --- /dev/null +++ b/bundle/direct/opqueue_test.go @@ -0,0 +1,252 @@ +package direct + +import ( + "context" + "errors" + "strconv" + "strings" + "sync" + "testing" + + "github.com/databricks/cli/bundle/deployplan" + "github.com/databricks/databricks-sdk-go/service/bundledeployments" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// fakeUploader records the uploads it receives and optionally blocks until +// release is closed, so a test can hold operations in the queue and observe +// coalescing. +type fakeUploader struct { + block chan struct{} + started chan string + err error + + mu sync.Mutex + uploads []string + actions map[string]bundledeployments.OperationActionType +} + +func (f *fakeUploader) upload(ctx context.Context, resourceKey string, op recordedOperation) error { + if f.started != nil { + f.started <- resourceKey + } + if f.block != nil { + <-f.block + } + + f.mu.Lock() + defer f.mu.Unlock() + f.uploads = append(f.uploads, resourceKey+"="+string(op.state)) + if f.actions == nil { + f.actions = map[string]bundledeployments.OperationActionType{} + } + f.actions[resourceKey] = op.action + return f.err +} + +func (f *fakeUploader) recorded() []string { + f.mu.Lock() + defer f.mu.Unlock() + return append([]string(nil), f.uploads...) +} + +func (f *fakeUploader) actionFor(resourceKey string) bundledeployments.OperationActionType { + f.mu.Lock() + defer f.mu.Unlock() + return f.actions[resourceKey] +} + +func recordState(t *testing.T, q *operationQueue, resourceKey, name string) { + t.Helper() + require.NoError(t, q.record(t.Context(), resourceKey, deployplan.Update, "id-1", map[string]string{"name": name}, nil)) +} + +func TestOperationQueueUploadsEachOperation(t *testing.T) { + f := &fakeUploader{} + q := newOperationQueue(t.Context(), f) + + for i := range 20 { + recordState(t, q, "resources.jobs.job"+strconv.Itoa(i), "n") + } + require.NoError(t, q.close()) + + assert.Len(t, f.recorded(), 20) +} + +func TestOperationQueueCoalescesQueuedOperationsForSameResource(t *testing.T) { + // Hold the first upload so later operations for the same resource pile up in + // the queue and are collapsed into one. + f := &fakeUploader{block: make(chan struct{}), started: make(chan string, 1)} + q := newOperationQueue(t.Context(), f) + + recordState(t, q, "resources.jobs.foo", "v1") + // Wait until a worker owns the key, so the operations below are queued behind + // an in-flight upload rather than racing it. + assert.Equal(t, "resources.jobs.foo", <-f.started) + + recordState(t, q, "resources.jobs.foo", "v2") + recordState(t, q, "resources.jobs.foo", "v3") + + close(f.block) + require.NoError(t, q.close()) + + // Two uploads, not three: v2 was superseded by v3 while both were queued, and + // the last recorded state is the one the service ends up with. + assert.Equal(t, []string{ + `resources.jobs.foo={"state":{"name":"v1"}}`, + `resources.jobs.foo={"state":{"name":"v3"}}`, + }, f.recorded()) +} + +func TestOperationQueueCoalescingKeepsCreateAction(t *testing.T) { + // Hold the first upload so the create below stays queued and the update + // coalesces into it. + f := &fakeUploader{block: make(chan struct{}), started: make(chan string, 1)} + q := newOperationQueue(t.Context(), f) + + recordState(t, q, "resources.jobs.hold", "v1") + assert.Equal(t, "resources.jobs.hold", <-f.started) + + // Occupy the remaining workers so nothing drains the key under test. + for i := range operationUploadWorkers - 1 { + recordState(t, q, "resources.jobs.hold"+strconv.Itoa(i), "v1") + assert.Equal(t, "resources.jobs.hold"+strconv.Itoa(i), <-f.started) + } + + require.NoError(t, q.record(t.Context(), "resources.jobs.foo", deployplan.Create, "id-1", map[string]string{"name": "created"}, nil)) + require.NoError(t, q.record(t.Context(), "resources.jobs.foo", deployplan.Update, "id-1", map[string]string{"name": "updated"}, nil)) + + close(f.block) + require.NoError(t, q.close()) + + // The state is the later one, but the action stays CREATE: recording an update + // would tell DMS the resource already existed before this deploy. + assert.Contains(t, f.recorded(), `resources.jobs.foo={"state":{"name":"updated"}}`) + assert.Equal(t, + bundledeployments.OperationActionTypeOperationActionTypeCreate, + f.actionFor("resources.jobs.foo")) +} + +func TestOperationQueueReturnsUploadError(t *testing.T) { + uploadErr := errors.New("boom") + f := &fakeUploader{err: uploadErr} + q := newOperationQueue(t.Context(), f) + + recordState(t, q, "resources.jobs.foo", "v1") + + err := q.close() + require.Error(t, err) + assert.ErrorIs(t, err, uploadErr) + assert.Contains(t, err.Error(), "resources.jobs.foo") +} + +func TestOperationQueueRecordRejectsUnsupportedAction(t *testing.T) { + f := &fakeUploader{} + q := newOperationQueue(t.Context(), f) + + // Serialization failures surface at record time, on the resource that caused + // them, rather than from the drain at the end of apply. + err := q.record(t.Context(), "resources.jobs.foo", deployplan.Skip, "id-1", nil, nil) + require.Error(t, err) + + require.NoError(t, q.close()) + assert.Empty(t, f.recorded()) +} + +func TestOperationQueueRecordRejectsOversizedState(t *testing.T) { + f := &fakeUploader{} + q := newOperationQueue(t.Context(), f) + + big := map[string]string{"name": strings.Repeat("x", maxOperationStateSize)} + err := q.record(t.Context(), "resources.jobs.foo", deployplan.Create, "id-1", big, nil) + require.ErrorContains(t, err, "exceeds the 65536 byte limit") + + require.NoError(t, q.close()) + assert.Empty(t, f.recorded()) +} + +func TestOperationQueueCloseIsIdempotent(t *testing.T) { + f := &fakeUploader{err: errors.New("boom")} + q := newOperationQueue(t.Context(), f) + + recordState(t, q, "resources.jobs.foo", "v1") + + require.Error(t, q.close()) + // A second close reports the same error instead of panicking on the already + // closed channel, so callers can both defer close and check it explicitly. + require.Error(t, q.close()) +} + +// serialUploader fails if two uploads for the same resource key ever overlap. +type serialUploader struct { + mu sync.Mutex + live map[string]bool + last map[string]string + uneven bool +} + +func (s *serialUploader) upload(ctx context.Context, resourceKey string, op recordedOperation) error { + s.mu.Lock() + if s.live[resourceKey] { + s.uneven = true + } + s.live[resourceKey] = true + s.mu.Unlock() + + s.mu.Lock() + defer s.mu.Unlock() + s.live[resourceKey] = false + s.last[resourceKey] = string(op.state) + return nil +} + +func TestOperationQueueUploadsOneResourceAtATime(t *testing.T) { + // Concurrent apply workers repeatedly record overlapping resource keys, the + // case where a coalesced key can be handed to a second worker while the first + // is still uploading it. The service keeps one state per key, so overlapping + // uploads for a key could land out of order and leave a stale state behind. + const ( + workers = 10 + perWorker = 5 + distinctKeyMod = 12 + ) + + ctx := t.Context() + u := &serialUploader{live: map[string]bool{}, last: map[string]string{}} + q := newOperationQueue(ctx, u) + + // Collect record errors instead of asserting inside the goroutines: testify + // assertions may only run on the goroutine running the test function. + errs := make(chan error, workers*perWorker) + var wg sync.WaitGroup + for w := range workers { + wg.Go(func() { + for i := range perWorker { + key := "resources.jobs.job" + strconv.Itoa((w*perWorker+i)%distinctKeyMod) + errs <- q.record(ctx, key, deployplan.Update, "id-1", map[string]string{"name": strconv.Itoa(w)}, nil) + } + }) + } + wg.Wait() + close(errs) + for err := range errs { + require.NoError(t, err) + } + require.NoError(t, q.close()) + + assert.False(t, u.uneven, "two uploads overlapped for the same resource key") + // Every distinct key was recorded, and close drained all of them. + assert.Len(t, u.last, distinctKeyMod) + assert.Empty(t, q.pending) + assert.Empty(t, q.owned) +} + +func TestNilOperationQueueIsNoOp(t *testing.T) { + // Recording is disabled: newOperationQueue returns nil and every method is a + // no-op, so Apply does not have to branch. + q := newOperationQueue(t.Context(), nil) + require.Nil(t, q) + require.NoError(t, q.record(t.Context(), "resources.jobs.foo", deployplan.Create, "id-1", nil, nil)) + require.NoError(t, q.close()) +} diff --git a/bundle/direct/oprecorder.go b/bundle/direct/oprecorder.go new file mode 100644 index 00000000000..91bcb65a4f6 --- /dev/null +++ b/bundle/direct/oprecorder.go @@ -0,0 +1,157 @@ +package direct + +import ( + "context" + "encoding/json" + "fmt" + "strings" + + "github.com/databricks/cli/bundle/deployplan" + "github.com/databricks/cli/bundle/direct/dstate" + "github.com/databricks/cli/libs/dyn" + "github.com/databricks/cli/libs/structs/structwalk" + "github.com/databricks/databricks-sdk-go/service/bundledeployments" +) + +// maxOperationStateSize is the largest serialized state DMS accepts per +// operation. Uploading more is rejected server-side, so fail early with a +// message that names the resource. +const maxOperationStateSize = 64 * 1024 + +// recordedOperation is an applied resource operation, serialized and waiting to be +// uploaded to the deployment metadata service (DMS). +// +// The payload is built on the apply worker rather than in the uploader so the +// queue does not hold on to the live resource struct, and so a malformed state +// fails the resource that produced it instead of the drain at the end of apply. +type recordedOperation struct { + action bundledeployments.OperationActionType + resourceID string + + // state is the serialized local config after the operation. It is nil for a + // delete, where the resource no longer exists. + state json.RawMessage +} + +// newRecordedOperation serializes an applied operation for upload. state is the +// local config after the operation and must be nil for delete operations. It +// errors when the serialized state exceeds maxOperationStateSize. +func newRecordedOperation(action deployplan.ActionType, resourceID string, state any, dependsOn []deployplan.DependsOnEntry) (recordedOperation, error) { + actionType, err := deployActionToSDK(action) + if err != nil { + return recordedOperation{}, err + } + + op := recordedOperation{action: actionType, resourceID: resourceID} + + // Operation.State carries the serialized state, which DMS serves back as + // resource state. Unset for delete: the resource is gone. + // + // Redact secrets, like dstate.SaveState does for the local state file: + // otherwise we leak them to the service and the read path writes them back + // into a local state file in plaintext. + if state != nil { + config, err := structwalk.RedactSensitiveFields(state, dyn.SensitiveValueRedacted) + if err != nil { + return recordedOperation{}, fmt.Errorf("serializing state: %w", err) + } + raw, err := json.Marshal(dstate.RecordedState{State: config, DependsOn: dependsOn}) + if err != nil { + return recordedOperation{}, fmt.Errorf("serializing state: %w", err) + } + if len(raw) > maxOperationStateSize { + return recordedOperation{}, fmt.Errorf("serialized state is %d bytes, which exceeds the %d byte limit for recording deployment history", len(raw), maxOperationStateSize) + } + op.state = raw + } + + return op, nil +} + +// mergeAction returns the action to record when a later operation coalesces into +// one still queued for the same resource (see operationQueue.record). The state +// uploaded is the later one, but the action must not be downgraded: Create and +// Recreate tell DMS the resource ID is new, and a subsequent Update only refines +// the state of that same new resource. Recording the pair as an Update would +// claim the resource already existed. A Delete is the exception - the resource is +// gone, so nothing earlier is worth reporting. +func mergeAction(queued, next bundledeployments.OperationActionType) bundledeployments.OperationActionType { + if next == bundledeployments.OperationActionTypeOperationActionTypeDelete { + return next + } + if queued == bundledeployments.OperationActionTypeOperationActionTypeCreate || + queued == bundledeployments.OperationActionTypeOperationActionTypeRecreate { + return queued + } + return next +} + +// operationUploader records an applied resource operation with DMS. Uploads run +// on the operationQueue workers, off the apply path. +type operationUploader interface { + upload(ctx context.Context, resourceKey string, op recordedOperation) error +} + +// operationRecorder uploads operations via the DMS CreateOperation API. +type operationRecorder struct { + client bundledeployments.BundleDeploymentsInterface + // parent is the version the operations are recorded under, formatted as + // "deployments/{deployment_id}/versions/{version_id}". + parent string +} + +// NewOperationRecorder returns an operationUploader backed by the DMS +// CreateOperation API. deploymentID and version identify the deployment version +// assigned by DMS that the operations are recorded under. +func NewOperationRecorder(client bundledeployments.BundleDeploymentsInterface, deploymentID string, version int64) operationUploader { + return &operationRecorder{ + client: client, + parent: fmt.Sprintf("deployments/%s/versions/%d", deploymentID, version), + } +} + +func (r *operationRecorder) upload(ctx context.Context, resourceKey string, op recordedOperation) error { + // DMS resource keys are unprefixed (e.g. "jobs.foo"), while the CLI's state + // keys carry a leading "resources." (e.g. "resources.jobs.foo"). Strip it on + // the way out; the read path re-adds it (see dstate.fetchDeploymentResources). + dmsKey := strings.TrimPrefix(resourceKey, "resources.") + + operation := bundledeployments.Operation{ + ActionType: op.action, + ResourceId: op.resourceID, + ResourceKey: dmsKey, + Status: bundledeployments.OperationStatusOperationStatusSucceeded, + } + if op.state != nil { + operation.State = &op.state + } + + _, err := r.client.CreateOperation(ctx, bundledeployments.CreateOperationRequest{ + Parent: r.parent, + ResourceKey: dmsKey, + Operation: operation, + }) + return err +} + +// deployActionToSDK maps a deployplan action to its DMS operation action type. +// Only actions that mutate a resource are recordable; Skip and Undefined never +// reach a recorder and are rejected rather than silently coerced. +func deployActionToSDK(a deployplan.ActionType) (bundledeployments.OperationActionType, error) { + switch a { + case deployplan.Create: + return bundledeployments.OperationActionTypeOperationActionTypeCreate, nil + case deployplan.Update: + return bundledeployments.OperationActionTypeOperationActionTypeUpdate, nil + case deployplan.UpdateWithID: + return bundledeployments.OperationActionTypeOperationActionTypeUpdateWithId, nil + case deployplan.Recreate: + return bundledeployments.OperationActionTypeOperationActionTypeRecreate, nil + case deployplan.Resize: + return bundledeployments.OperationActionTypeOperationActionTypeResize, nil + case deployplan.Delete: + return bundledeployments.OperationActionTypeOperationActionTypeDelete, nil + default: + return "", fmt.Errorf("cannot record operation: unsupported action %q", a) + } +} diff --git a/bundle/direct/oprecorder_test.go b/bundle/direct/oprecorder_test.go new file mode 100644 index 00000000000..556f7f59f90 --- /dev/null +++ b/bundle/direct/oprecorder_test.go @@ -0,0 +1,155 @@ +package direct + +import ( + "context" + "sync" + "testing" + + "github.com/databricks/cli/bundle/deployplan" + "github.com/databricks/cli/libs/dyn" + "github.com/databricks/databricks-sdk-go/service/bundledeployments" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +type fakeOpClient struct { + bundledeployments.BundleDeploymentsInterface + + mu sync.Mutex + requests []bundledeployments.CreateOperationRequest +} + +func (f *fakeOpClient) CreateOperation(ctx context.Context, req bundledeployments.CreateOperationRequest) (*bundledeployments.Operation, error) { + f.mu.Lock() + defer f.mu.Unlock() + f.requests = append(f.requests, req) + return &bundledeployments.Operation{}, nil +} + +// uploadOne records a single operation through the given uploader, mirroring what +// an operationQueue worker does. +func uploadOne(t *testing.T, u operationUploader, resourceKey string, action deployplan.ActionType, resourceID string, state any) { + t.Helper() + op, err := newRecordedOperation(action, resourceID, state, nil) + require.NoError(t, err) + require.NoError(t, u.upload(t.Context(), resourceKey, op)) +} + +func TestOperationRecorderStripsResourcePrefix(t *testing.T) { + f := &fakeOpClient{} + r := NewOperationRecorder(f, "dep-1", 2) + + uploadOne(t, r, "resources.jobs.foo", deployplan.Create, "job-123", map[string]string{"name": "foo"}) + + require.Len(t, f.requests, 1) + req := f.requests[0] + // The wire key drops the CLI-internal "resources." prefix, both in the query + // param and the operation body. + assert.Equal(t, "jobs.foo", req.ResourceKey) + assert.Equal(t, "jobs.foo", req.Operation.ResourceKey) + assert.Equal(t, "deployments/dep-1/versions/2", req.Parent) + assert.Equal(t, bundledeployments.OperationActionTypeOperationActionTypeCreate, req.Operation.ActionType) + assert.Equal(t, "job-123", req.Operation.ResourceId) + require.NotNil(t, req.Operation.State) +} + +func TestOperationRecorderDeleteHasNoState(t *testing.T) { + f := &fakeOpClient{} + r := NewOperationRecorder(f, "dep-1", 3) + + uploadOne(t, r, "resources.jobs.foo", deployplan.Delete, "", nil) + + require.Len(t, f.requests, 1) + assert.Equal(t, bundledeployments.OperationActionTypeOperationActionTypeDelete, f.requests[0].Operation.ActionType) + // Delete operations carry no serialized state. + assert.Nil(t, f.requests[0].Operation.State) +} + +func TestNewRecordedOperationRedactsSensitiveFields(t *testing.T) { + state := struct { + Name string `json:"name"` + Token string `json:"token" bundle:"sensitive"` + }{Name: "foo", Token: "super-secret"} + + op, err := newRecordedOperation(deployplan.Create, "job-123", state, nil) + require.NoError(t, err) + + // Sensitive fields are redacted before leaving the CLI, matching what + // dstate.SaveState writes to the local state file. + assert.JSONEq(t, + `{"state":{"name":"foo","token":"`+dyn.SensitiveValueRedacted+`"}}`, + string(op.state)) +} + +func TestNewRecordedOperationRecordsDependsOn(t *testing.T) { + // depends_on rides in an envelope alongside the config: it cannot be + // recomputed from the config, whose references are already resolved. + dependsOn := []deployplan.DependsOnEntry{{Node: "resources.jobs.bar", Label: "${resources.jobs.bar.id}"}} + + op, err := newRecordedOperation(deployplan.Create, "job-123", map[string]string{"name": "foo"}, dependsOn) + require.NoError(t, err) + + assert.JSONEq(t, + `{"state":{"name":"foo"},"depends_on":[{"node":"resources.jobs.bar","label":"${resources.jobs.bar.id}"}]}`, + string(op.state)) +} + +func TestNewRecordedOperationRejectsUnsupportedAction(t *testing.T) { + _, err := newRecordedOperation(deployplan.Skip, "job-123", nil, nil) + assert.Error(t, err) +} + +func TestMergeAction(t *testing.T) { + const ( + create = bundledeployments.OperationActionTypeOperationActionTypeCreate + recreate = bundledeployments.OperationActionTypeOperationActionTypeRecreate + update = bundledeployments.OperationActionTypeOperationActionTypeUpdate + resize = bundledeployments.OperationActionTypeOperationActionTypeResize + del = bundledeployments.OperationActionTypeOperationActionTypeDelete + ) + + cases := []struct { + queued, next, want bundledeployments.OperationActionType + }{ + // A queued create is not downgraded: the resource is still new. + {create, update, create}, + {create, resize, create}, + {recreate, update, recreate}, + {create, create, create}, + // A delete wins: the resource is gone, so the earlier action is moot. + {create, del, del}, + {update, del, del}, + // Neither side is a create, so the later action stands. + {update, resize, resize}, + {resize, update, update}, + {del, create, create}, + } + for _, c := range cases { + assert.Equal(t, c.want, mergeAction(c.queued, c.next), "queued %s, next %s", c.queued, c.next) + } +} + +func TestDeployActionToSDK(t *testing.T) { + cases := []struct { + action deployplan.ActionType + want bundledeployments.OperationActionType + }{ + {deployplan.Create, bundledeployments.OperationActionTypeOperationActionTypeCreate}, + {deployplan.Update, bundledeployments.OperationActionTypeOperationActionTypeUpdate}, + {deployplan.UpdateWithID, bundledeployments.OperationActionTypeOperationActionTypeUpdateWithId}, + {deployplan.Recreate, bundledeployments.OperationActionTypeOperationActionTypeRecreate}, + {deployplan.Resize, bundledeployments.OperationActionTypeOperationActionTypeResize}, + {deployplan.Delete, bundledeployments.OperationActionTypeOperationActionTypeDelete}, + } + for _, c := range cases { + got, err := deployActionToSDK(c.action) + require.NoError(t, err) + assert.Equal(t, c.want, got) + } + + // Skip and Undefined never reach a recorder and are rejected. + _, err := deployActionToSDK(deployplan.Skip) + assert.Error(t, err) + _, err = deployActionToSDK(deployplan.Undefined) + assert.Error(t, err) +} diff --git a/bundle/direct/pkg.go b/bundle/direct/pkg.go index 48a9c5a2ff7..03864af5da2 100644 --- a/bundle/direct/pkg.go +++ b/bundle/direct/pkg.go @@ -44,6 +44,13 @@ type DeploymentBundle struct { Plan *deployplan.Plan RemoteStateCache sync.Map StateCache structvar.Cache + + // OpRec uploads each applied resource operation to the deployment metadata + // service (DMS). It is nil unless the bundle opts into recording deployment + // history, in which case the phases package sets it after CreateVersion. + // Apply queues the operations and drains them before returning, so the + // uploads do not block the resources being deployed. + OpRec operationUploader } // SetRemoteState updates the remote state with type validation and marks as fresh. diff --git a/bundle/internal/schema/annotations.yml b/bundle/internal/schema/annotations.yml index 10832fe04a6..100d33356fd 100644 --- a/bundle/internal/schema/annotations.yml +++ b/bundle/internal/schema/annotations.yml @@ -178,6 +178,8 @@ experimental: "record_deployment_history": "description": |- Whether to record deployment history using the deployment metadata service (DMS), which tracks what changed across deployments. + + Only supported for a bundle with no deployed resources yet. "scripts": "description": |- The commands to run. diff --git a/bundle/phases/deploy.go b/bundle/phases/deploy.go index f65e50a940e..3d70a218b15 100644 --- a/bundle/phases/deploy.go +++ b/bundle/phases/deploy.go @@ -17,6 +17,7 @@ import ( "github.com/databricks/cli/bundle/deploy/snapshot" "github.com/databricks/cli/bundle/deploy/terraform" "github.com/databricks/cli/bundle/deployplan" + "github.com/databricks/cli/bundle/direct" "github.com/databricks/cli/bundle/libraries" "github.com/databricks/cli/bundle/metrics" "github.com/databricks/cli/bundle/permissions" @@ -24,6 +25,7 @@ import ( "github.com/databricks/cli/bundle/statemgmt" "github.com/databricks/cli/libs/agent" "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/cli/libs/sync" @@ -161,7 +163,21 @@ func Deploy(ctx context.Context, b *bundle.Bundle, outputHandler sync.OutputHand } // lock is acquired here + // + // Set up DMS recording of this deployment as a version. The version is not + // created until the plan is approved (below), so a cancelled deploy records + // nothing; the deferred CompleteVersion is a no-op until CreateVersion runs. + // CompleteVersion is deferred before lock.Release so it runs while the lock + // is still held (defers run last-in-first-out). + recorder, err := newDeploymentRecorder(ctx, b, stateEngine, dms.VersionTypeDeploy) + if err != nil { + logdiag.LogError(ctx, err) + return + } defer func() { + if err := recorder.CompleteVersion(ctx, !logdiag.HasError(ctx)); err != nil { + logdiag.LogError(ctx, err) + } bundle.ApplyContext(ctx, b, lock.Release(lock.GoalDeploy)) }() @@ -255,6 +271,24 @@ func Deploy(ctx context.Context, b *bundle.Bundle, outputHandler sync.OutputHand return } if haveApproval { + // Record the DMS version now that the plan is approved and the state WAL + // has been opened. CreateVersion requests version_id == last_version_id + 1; + // the server returns ABORTED if a concurrent deploy advanced the deployment + // since the plan was computed, so a stale plan is not applied. + if err := recorder.CreateVersion(ctx); err != nil { + logdiag.LogError(ctx, err) + return + } + if recorder != nil { + // Record operations under the version just created so DMS holds the + // deployed resource state. On a first deploy the deployment ID was only + // assigned by CreateVersion above, so this must come after it. + b.DeploymentBundle.OpRec = direct.NewOperationRecorder( + b.WorkspaceClient(ctx).BundleDeployments, + recorder.DeploymentID(), + recorder.Version(), + ) + } deployCore(ctx, b, plan, stateEngine, requestedEngine) } else { cmdio.LogString(ctx, "Deployment cancelled!") diff --git a/bundle/phases/destroy.go b/bundle/phases/destroy.go index 2496c7033ad..2925e80bca8 100644 --- a/bundle/phases/destroy.go +++ b/bundle/phases/destroy.go @@ -13,8 +13,10 @@ import ( "github.com/databricks/cli/bundle/deploy/lock" "github.com/databricks/cli/bundle/deploy/terraform" "github.com/databricks/cli/bundle/deployplan" + "github.com/databricks/cli/bundle/direct" "github.com/databricks/cli/libs/cmdio" "github.com/databricks/cli/libs/diag" + "github.com/databricks/cli/libs/dms" "github.com/databricks/cli/libs/log" "github.com/databricks/cli/libs/logdiag" "github.com/databricks/databricks-sdk-go/apierr" @@ -131,7 +133,19 @@ func Destroy(ctx context.Context, b *bundle.Bundle, engine engine.EngineType) { return } + // Set up DMS recording of this destroy as a version. The version is not + // created until the destroy is approved (below), so a cancelled destroy + // records nothing; the deferred CompleteVersion is a no-op until then. It is + // deferred before lock.Release so it runs while the lock is still held. + recorder, err := newDeploymentRecorder(ctx, b, engine, dms.VersionTypeDestroy) + if err != nil { + logdiag.LogError(ctx, err) + return + } defer func() { + if err := recorder.CompleteVersion(ctx, !logdiag.HasError(ctx)); err != nil { + logdiag.LogError(ctx, err) + } bundle.ApplyContext(ctx, b, lock.Release(lock.GoalDestroy)) }() @@ -188,6 +202,19 @@ func Destroy(ctx context.Context, b *bundle.Bundle, engine engine.EngineType) { return } } + // Record the DMS version now that the destroy is approved and the state WAL + // has been opened, then record each delete operation under it. + if err := recorder.CreateVersion(ctx); err != nil { + logdiag.LogError(ctx, err) + return + } + if recorder != nil { + b.DeploymentBundle.OpRec = direct.NewOperationRecorder( + b.WorkspaceClient(ctx).BundleDeployments, + recorder.DeploymentID(), + recorder.Version(), + ) + } destroyCore(ctx, b, plan, engine) } else { cmdio.LogString(ctx, "Destroy cancelled!") diff --git a/bundle/phases/dms.go b/bundle/phases/dms.go new file mode 100644 index 00000000000..3d2f4f54009 --- /dev/null +++ b/bundle/phases/dms.go @@ -0,0 +1,43 @@ +package phases + +import ( + "context" + + "github.com/databricks/cli/bundle" + "github.com/databricks/cli/bundle/config/engine" + "github.com/databricks/cli/libs/dms" +) + +// newDeploymentRecorder returns a dms.Recorder for the current deployment, or +// nil when DMS recording does not apply. A nil recorder is a no-op, so callers +// do not need to branch on it. +// +// Recording is enabled only when experimental.record_deployment_history is set +// AND the engine is direct: DMS resource state is tracked per direct-engine +// deployment. Returning nil for terraform leaves those deployments untouched. +// +// The deployment ID is resolved from the workspace, not local state (see +// dms.ResolveDeploymentID). The lookup happens here, after the deployment lock is +// held, so it sees any deployment a concurrent deploy created. It is empty on the +// first recorded deploy, where the recorder creates the deployment instead. +func newDeploymentRecorder(ctx context.Context, b *bundle.Bundle, eng engine.EngineType, versionType dms.VersionType) (*dms.Recorder, error) { + if b.Config.Experimental == nil || !b.Config.Experimental.RecordDeploymentHistory { + return nil, nil + } + if !eng.IsDirect() { + return nil, nil + } + + statePath := b.Config.Workspace.StatePath + deploymentID, err := dms.ResolveDeploymentID(ctx, b.WorkspaceClient(ctx), statePath) + if err != nil { + return nil, err + } + return dms.NewRecorder( + b.WorkspaceClient(ctx).BundleDeployments, + deploymentID, + statePath, + b.Config.Bundle.Target, + versionType, + ), nil +} diff --git a/bundle/schema/jsonschema.json b/bundle/schema/jsonschema.json index 4c78bd7c384..c52e96d5dc9 100644 --- a/bundle/schema/jsonschema.json +++ b/bundle/schema/jsonschema.json @@ -3016,7 +3016,7 @@ "$ref": "#/$defs/bool" }, "record_deployment_history": { - "description": "Whether to record deployment history using the deployment metadata service (DMS), which tracks what changed across deployments.", + "description": "Whether to record deployment history using the deployment metadata service (DMS), which tracks what changed across deployments.\n\nOnly supported for a bundle with no deployed resources yet.", "$ref": "#/$defs/bool" }, "scripts": { diff --git a/cmd/bundle/generate/dashboard.go b/cmd/bundle/generate/dashboard.go index 086ec1d600a..2b286bcad3d 100644 --- a/cmd/bundle/generate/dashboard.go +++ b/cmd/bundle/generate/dashboard.go @@ -404,7 +404,7 @@ func (d *dashboard) runForResource(ctx context.Context, b *bundle.Bundle) { var state statemgmt.ExportedResourcesMap if stateDesc.Engine.IsDirect() { _, localPath := b.StateFilenameDirect(ctx) - if err := b.DeploymentBundle.StateDB.Open(ctx, localPath, dstate.WithRecovery(true), dstate.WithWrite(false)); err != nil { + if err := b.DeploymentBundle.StateDB.Open(ctx, localPath, dstate.WithRecovery(true), dstate.WithWrite(false), nil); err != nil { logdiag.LogError(ctx, err) return } diff --git a/cmd/bundle/generate/genie_space.go b/cmd/bundle/generate/genie_space.go index 6d938c5e03d..48ecc92a6cd 100644 --- a/cmd/bundle/generate/genie_space.go +++ b/cmd/bundle/generate/genie_space.go @@ -322,7 +322,7 @@ func (g *genieSpace) runForResource(ctx context.Context, b *bundle.Bundle) { var state statemgmt.ExportedResourcesMap if stateDesc.Engine.IsDirect() { _, localPath := b.StateFilenameDirect(ctx) - if err := b.DeploymentBundle.StateDB.Open(ctx, localPath, dstate.WithRecovery(true), dstate.WithWrite(false)); err != nil { + if err := b.DeploymentBundle.StateDB.Open(ctx, localPath, dstate.WithRecovery(true), dstate.WithWrite(false), nil); err != nil { logdiag.LogError(ctx, err) return } diff --git a/cmd/bundle/utils/process.go b/cmd/bundle/utils/process.go index e4f232605ce..b000da60264 100644 --- a/cmd/bundle/utils/process.go +++ b/cmd/bundle/utils/process.go @@ -20,6 +20,7 @@ import ( "github.com/databricks/cli/cmd/root" "github.com/databricks/cli/internal/build" "github.com/databricks/cli/libs/diag" + "github.com/databricks/cli/libs/dms" "github.com/databricks/cli/libs/dyn" "github.com/databricks/cli/libs/log" "github.com/databricks/cli/libs/logdiag" @@ -211,7 +212,27 @@ func ProcessBundleRet(cmd *cobra.Command, opts ProcessOptions) (b *bundle.Bundle needDirectState := stateDesc.Engine.IsDirect() && (opts.InitIDs || opts.ErrorOnEmptyState || opts.Deploy || opts.ReadPlanPath != "" || opts.PreDeployChecks || opts.PostStateFunc != nil) if needDirectState { _, localPath := b.StateFilenameDirect(ctx) - if err := b.DeploymentBundle.StateDB.Open(ctx, localPath, dstate.WithRecovery(true), dstate.WithWrite(false)); err != nil { + + // When the bundle records deployment history, the deployment metadata + // service owns resource state, so hand Open a DMS source to read it from + // there instead of the file. The local identity (lineage/serial) still + // comes from the file. Reads open the state write-disabled, so no lineage + // is minted here. + var dmsSource *dstate.DMSSource + if b.Config.Experimental != nil && b.Config.Experimental.RecordDeploymentHistory { + w := b.WorkspaceClient(ctx) + deploymentID, err := dms.ResolveDeploymentID(ctx, w, b.Config.Workspace.StatePath) + if err != nil { + logdiag.LogError(ctx, err) + return b, stateDesc, root.ErrAlreadyPrinted + } + dmsSource = &dstate.DMSSource{ + Client: w.BundleDeployments, + DeploymentID: deploymentID, + TargetName: b.Config.Bundle.Target, + } + } + if err := b.DeploymentBundle.StateDB.Open(ctx, localPath, dstate.WithRecovery(true), dstate.WithWrite(false), dmsSource); err != nil { logdiag.LogError(ctx, err) return b, stateDesc, root.ErrAlreadyPrinted } diff --git a/libs/dms/recorder.go b/libs/dms/recorder.go new file mode 100644 index 00000000000..83f180ba3be --- /dev/null +++ b/libs/dms/recorder.go @@ -0,0 +1,257 @@ +package dms + +import ( + "context" + "errors" + "fmt" + "net/http" + "strconv" + "strings" + "time" + + "github.com/databricks/cli/internal/build" + "github.com/databricks/cli/libs/log" + "github.com/databricks/databricks-sdk-go/apierr" + "github.com/databricks/databricks-sdk-go/service/bundledeployments" +) + +// The server expires a version's lease if it does not receive a heartbeat +// within a 2-minute TTL; we heartbeat well inside that window. +const defaultHeartbeatInterval = 30 * time.Second + +// VersionType identifies the kind of deployment a version records. +type VersionType = bundledeployments.VersionType + +const ( + VersionTypeDeploy VersionType = bundledeployments.VersionTypeVersionTypeDeploy + VersionTypeDestroy VersionType = bundledeployments.VersionTypeVersionTypeDestroy +) + +// Recorder records a single deploy/destroy as a version with DMS. +// +// The server assigns the deployment ID on the first deploy, i.e. when the ID +// resolved from the workspace is empty (see ResolveDeploymentID). Later deploys +// resolve the same ID and reuse the record; a destroy deletes the record and its +// node, so the next deploy starts over from empty. +type Recorder struct { + svc bundledeployments.BundleDeploymentsInterface + deploymentID string + statePath string + targetName string + versionType VersionType + + // populated by CreateVersion + versionNum int64 + stopHeartbeat context.CancelFunc +} + +// NewRecorder returns a Recorder for the given deployment. deploymentID is the +// ID resolved from the deployment's workspace node, or empty if this bundle has +// not yet recorded a deployment (the server assigns one during CreateVersion). +// statePath is the bundle's remote state directory, under which DMS registers +// the deployment node. +func NewRecorder(svc bundledeployments.BundleDeploymentsInterface, deploymentID, statePath, targetName string, versionType VersionType) *Recorder { + return &Recorder{ + svc: svc, + deploymentID: deploymentID, + statePath: statePath, + targetName: targetName, + versionType: versionType, + } +} + +// DeploymentID returns the DMS deployment ID this recorder is bound to. It is +// empty until CreateVersion has created the deployment record (on a first +// deploy) and non-empty afterwards, so callers can parent operations under it. +func (r *Recorder) DeploymentID() string { + if r == nil { + return "" + } + return r.deploymentID +} + +// Version returns the version number claimed by CreateVersion. It is zero until +// CreateVersion has run; callers use it to parent operations under the version. +func (r *Recorder) Version() int64 { + if r == nil { + return 0 + } + return r.versionNum +} + +// CreateVersion registers a new version with DMS, claiming it for the duration +// of the deployment. A nil Recorder is a no-op, so callers can leave it nil +// when recording is disabled. +func (r *Recorder) CreateVersion(ctx context.Context) error { + if r == nil { + return nil + } + + versionID, err := r.createDeploymentVersion(ctx) + if err != nil { + return err + } + + versionNum, err := strconv.ParseInt(versionID, 10, 64) + if err != nil { + return fmt.Errorf("failed to parse version ID %q: %w", versionID, err) + } + r.versionNum = versionNum + r.stopHeartbeat = startHeartbeat(ctx, r.svc, r.deploymentID, versionID) + return nil +} + +// CompleteVersion finalizes the version created by CreateVersion. A nil +// Recorder, or one whose CreateVersion never ran, is a no-op. +func (r *Recorder) CompleteVersion(ctx context.Context, success bool) error { + if r == nil || r.stopHeartbeat == nil { + return nil + } + + r.stopHeartbeat() + + versionIDStr := strconv.FormatInt(r.versionNum, 10) + versionName := fmt.Sprintf("deployments/%s/versions/%s", r.deploymentID, versionIDStr) + + reason := bundledeployments.VersionCompleteVersionCompleteSuccess + if !success { + reason = bundledeployments.VersionCompleteVersionCompleteFailure + } + + _, err := r.svc.CompleteVersion(ctx, bundledeployments.CompleteVersionRequest{ + Name: versionName, + CompletionReason: reason, + }) + if err != nil { + return err + } + log.Infof(ctx, "Completed deployment version: deployment=%s version=%s reason=%s", r.deploymentID, versionIDStr, reason) + + // For destroy operations, delete the deployment record after the version + // completes successfully. + if success && r.versionType == VersionTypeDestroy { + err = r.svc.DeleteDeployment(ctx, bundledeployments.DeleteDeploymentRequest{ + Name: "deployments/" + r.deploymentID, + }) + if err != nil { + return fmt.Errorf("failed to delete deployment: %w", err) + } + } + + return nil +} + +// createDeploymentVersion ensures the deployment record exists, then creates a +// new version under it. With no deployment ID it creates the deployment and lets +// the server assign the ID; otherwise it reads the existing deployment to +// compute the next version number. +func (r *Recorder) createDeploymentVersion(ctx context.Context) (versionID string, err error) { + if r.deploymentID != "" { + // Existing deployment: read it to compute the next version number. A 404 is + // not recovered from by creating a second deployment. The service trashes the + // workspace node when it deletes the record, so a node that resolved but has + // no record means the two are out of sync, and creating another deployment + // would collide on the same node path. + dep, getErr := r.svc.GetDeployment(ctx, bundledeployments.GetDeploymentRequest{ + Name: "deployments/" + r.deploymentID, + }) + if getErr != nil { + return "", fmt.Errorf("failed to get deployment: %w", getErr) + } + lastVersion, parseErr := strconv.ParseInt(dep.LastVersionId, 10, 64) + if parseErr != nil { + return "", fmt.Errorf("failed to parse last_version_id %q: %w", dep.LastVersionId, parseErr) + } + versionID = strconv.FormatInt(lastVersion+1, 10) + } else { + // First deploy: create the deployment so the server assigns an ID. + // + // initial_parent_path is required. The service creates the deployment node + // under it, and that node's ID is the deployment ID ResolveDeploymentID reads + // back later. The folder already exists by now: the deployment lock lives in + // the same directory. + dep, createErr := r.svc.CreateDeployment(ctx, bundledeployments.CreateDeploymentRequest{ + Deployment: bundledeployments.Deployment{ + InitialParentPath: r.statePath, + TargetName: r.targetName, + }, + }) + if createErr != nil { + return "", fmt.Errorf("failed to create deployment: %w", createErr) + } + id, idErr := deploymentIDFromName(dep.Name) + if idErr != nil { + return "", idErr + } + r.deploymentID = id + versionID = "1" + } + + // The server validates that versionID equals last_version_id + 1 and returns + // ABORTED otherwise (e.g. a concurrent deploy already created this version). + version, versionErr := r.svc.CreateVersion(ctx, bundledeployments.CreateVersionRequest{ + Parent: "deployments/" + r.deploymentID, + VersionId: versionID, + Version: bundledeployments.Version{ + CliVersion: build.GetInfo().Version, + VersionType: r.versionType, + TargetName: r.targetName, + }, + }) + if versionErr != nil { + return "", fmt.Errorf("failed to create deployment version: %w", versionErr) + } + + log.Infof(ctx, "Created deployment version: deployment=%s version=%s", r.deploymentID, version.VersionId) + return versionID, nil +} + +// deploymentIDFromName extracts the deployment ID from a DMS resource name of +// the form "deployments/{deployment_id}". +func deploymentIDFromName(name string) (string, error) { + id, ok := strings.CutPrefix(name, "deployments/") + if !ok || id == "" { + return "", fmt.Errorf("unexpected deployment name %q from deployment metadata service", name) + } + return id, nil +} + +// startHeartbeat starts a background goroutine that sends heartbeats to keep +// the deployment version's lease alive. Returns a cancel function to stop it. +func startHeartbeat(ctx context.Context, svc bundledeployments.BundleDeploymentsInterface, deploymentID, versionID string) context.CancelFunc { + ctx, cancel := context.WithCancel(ctx) + versionName := fmt.Sprintf("deployments/%s/versions/%s", deploymentID, versionID) + + go func() { + ticker := time.NewTicker(defaultHeartbeatInterval) + defer ticker.Stop() + + for { + select { + case <-ctx.Done(): + return + case <-ticker.C: + _, err := svc.Heartbeat(ctx, bundledeployments.HeartbeatRequest{Name: versionName}) + if err != nil { + // A 409 ABORTED is expected if the version was completed + // between the ticker firing and the heartbeat. + if isAbortedErr(err) { + log.Debugf(ctx, "Heartbeat stopped: version already completed") + return + } + log.Warnf(ctx, "Failed to send deployment heartbeat: %v", err) + } else { + log.Debugf(ctx, "Deployment heartbeat sent: deployment=%s version=%s", deploymentID, versionID) + } + } + } + }() + + return cancel +} + +// isAbortedErr reports whether err is an HTTP 409 ABORTED from the DMS API. +func isAbortedErr(err error) bool { + apiErr, ok := errors.AsType[*apierr.APIError](err) + return ok && apiErr.StatusCode == http.StatusConflict && apiErr.ErrorCode == "ABORTED" +} diff --git a/libs/dms/recorder_test.go b/libs/dms/recorder_test.go new file mode 100644 index 00000000000..6c2f334c946 --- /dev/null +++ b/libs/dms/recorder_test.go @@ -0,0 +1,196 @@ +package dms + +import ( + "context" + "errors" + "fmt" + "testing" + + "github.com/databricks/databricks-sdk-go/apierr" + "github.com/databricks/databricks-sdk-go/service/bundledeployments" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// testStatePath is the bundle state directory the recorder registers the +// deployment node under; several tests assert it round-trips to the service. +const testStatePath = "/Workspace/Users/me/.bundle/proj/dev/state" + +// fakeDMS records the calls the recorder makes and lets a test script the +// server-side responses. It embeds the SDK interface so it satisfies it while +// only overriding the methods the recorder uses. +type fakeDMS struct { + bundledeployments.BundleDeploymentsInterface + + // scripted behavior + getDeployment func(id string) (*bundledeployments.Deployment, error) + + // assigned deployment ID for CreateDeployment (server-generated flow) + assignedID string + + // captured requests + created []bundledeployments.CreateDeploymentRequest + versions []bundledeployments.CreateVersionRequest + completed []bundledeployments.CompleteVersionRequest + deleted []string +} + +func (f *fakeDMS) CreateDeployment(ctx context.Context, req bundledeployments.CreateDeploymentRequest) (*bundledeployments.Deployment, error) { + f.created = append(f.created, req) + // The server always assigns the ID; it is the ID of the workspace node it + // creates under initial_parent_path. + return &bundledeployments.Deployment{Name: "deployments/" + f.assignedID}, nil +} + +func (f *fakeDMS) GetDeployment(ctx context.Context, req bundledeployments.GetDeploymentRequest) (*bundledeployments.Deployment, error) { + id := req.Name[len("deployments/"):] + return f.getDeployment(id) +} + +func (f *fakeDMS) CreateVersion(ctx context.Context, req bundledeployments.CreateVersionRequest) (*bundledeployments.Version, error) { + f.versions = append(f.versions, req) + return &bundledeployments.Version{VersionId: req.VersionId}, nil +} + +func (f *fakeDMS) CompleteVersion(ctx context.Context, req bundledeployments.CompleteVersionRequest) (*bundledeployments.Version, error) { + f.completed = append(f.completed, req) + return &bundledeployments.Version{}, nil +} + +func (f *fakeDMS) DeleteDeployment(ctx context.Context, req bundledeployments.DeleteDeploymentRequest) error { + f.deleted = append(f.deleted, req.Name) + return nil +} + +func (f *fakeDMS) Heartbeat(ctx context.Context, req bundledeployments.HeartbeatRequest) (*bundledeployments.HeartbeatResponse, error) { + return &bundledeployments.HeartbeatResponse{}, nil +} + +func TestRecorderFirstDeployCreatesDeploymentWithServerAssignedID(t *testing.T) { + f := &fakeDMS{assignedID: "server-generated-id"} + // A first deploy resolves no deployment ID from the workspace. + r := NewRecorder(f, "", testStatePath, "dev", VersionTypeDeploy) + + require.NoError(t, r.CreateVersion(t.Context())) + + // The server assigned the ID, and the recorder exposes it for the rest of the + // deploy (it parents the operations recorded under this version). + require.Len(t, f.created, 1) + assert.Equal(t, "server-generated-id", r.DeploymentID()) + // initial_parent_path is required: the service creates the deployment node + // under it, and that node is what ResolveDeploymentID looks up later. + assert.Equal(t, testStatePath, f.created[0].Deployment.InitialParentPath) + + // The first version is 1, parented under the assigned deployment. + require.Len(t, f.versions, 1) + assert.Equal(t, "1", f.versions[0].VersionId) + assert.Equal(t, "deployments/server-generated-id", f.versions[0].Parent) + assert.Equal(t, int64(1), r.Version()) + + require.NoError(t, r.CompleteVersion(t.Context(), true)) + require.Len(t, f.completed, 1) + assert.Equal(t, bundledeployments.VersionCompleteVersionCompleteSuccess, f.completed[0].CompletionReason) + assert.Empty(t, f.deleted) +} + +func TestRecorderSubsequentDeployReusesDeploymentAndIncrementsVersion(t *testing.T) { + f := &fakeDMS{ + getDeployment: func(id string) (*bundledeployments.Deployment, error) { + return &bundledeployments.Deployment{Name: "deployments/" + id, LastVersionId: "4"}, nil + }, + } + // A subsequent deploy passes the stored deployment ID. + r := NewRecorder(f, "stored-id", testStatePath, "dev", VersionTypeDeploy) + + require.NoError(t, r.CreateVersion(t.Context())) + + // No new deployment is created; the version increments to last_version_id + 1. + assert.Empty(t, f.created) + require.Len(t, f.versions, 1) + assert.Equal(t, "5", f.versions[0].VersionId) + assert.Equal(t, "stored-id", r.DeploymentID()) +} + +func TestRecorderGetDeploymentErrorFailsDeploy(t *testing.T) { + cases := map[string]error{ + // A resolved ID whose record is missing means the record and the workspace + // node it was resolved from are out of sync. Creating a second deployment + // would collide on the same node path, so fail instead. + "not found": fmt.Errorf("deployment: %w", apierr.ErrNotFound), + "other": errors.New("boom"), + } + for name, getErr := range cases { + t.Run(name, func(t *testing.T) { + f := &fakeDMS{ + getDeployment: func(id string) (*bundledeployments.Deployment, error) { + return nil, getErr + }, + } + r := NewRecorder(f, "stored-id", testStatePath, "dev", VersionTypeDeploy) + + err := r.CreateVersion(t.Context()) + assert.ErrorContains(t, err, "failed to get deployment") + assert.Empty(t, f.created) + }) + } +} + +func TestRecorderDestroyDeletesDeploymentOnSuccess(t *testing.T) { + f := &fakeDMS{ + getDeployment: func(id string) (*bundledeployments.Deployment, error) { + return &bundledeployments.Deployment{Name: "deployments/" + id, LastVersionId: "2"}, nil + }, + } + r := NewRecorder(f, "stored-id", testStatePath, "dev", VersionTypeDestroy) + + require.NoError(t, r.CreateVersion(t.Context())) + assert.Equal(t, bundledeployments.VersionTypeVersionTypeDestroy, f.versions[0].Version.VersionType) + + require.NoError(t, r.CompleteVersion(t.Context(), true)) + // A successful destroy deletes the deployment record. + require.Equal(t, []string{"deployments/stored-id"}, f.deleted) +} + +func TestRecorderFailedDestroyKeepsDeployment(t *testing.T) { + f := &fakeDMS{ + getDeployment: func(id string) (*bundledeployments.Deployment, error) { + return &bundledeployments.Deployment{Name: "deployments/" + id, LastVersionId: "2"}, nil + }, + } + r := NewRecorder(f, "stored-id", testStatePath, "dev", VersionTypeDestroy) + + require.NoError(t, r.CreateVersion(t.Context())) + require.NoError(t, r.CompleteVersion(t.Context(), false)) + + assert.Equal(t, bundledeployments.VersionCompleteVersionCompleteFailure, f.completed[0].CompletionReason) + // A failed destroy leaves the deployment in place. + assert.Empty(t, f.deleted) +} + +func TestNilRecorderIsNoOp(t *testing.T) { + var r *Recorder + assert.NoError(t, r.CreateVersion(t.Context())) + assert.NoError(t, r.CompleteVersion(t.Context(), true)) + assert.Empty(t, r.DeploymentID()) + assert.Zero(t, r.Version()) +} + +func TestRecorderCompleteVersionNoOpWithoutCreateVersion(t *testing.T) { + f := &fakeDMS{} + r := NewRecorder(f, "stored-id", testStatePath, "dev", VersionTypeDeploy) + // CompleteVersion before CreateVersion is a no-op (nothing was claimed). + require.NoError(t, r.CompleteVersion(t.Context(), true)) + assert.Empty(t, f.completed) +} + +func TestDeploymentIDFromName(t *testing.T) { + id, err := deploymentIDFromName("deployments/abc-123") + require.NoError(t, err) + assert.Equal(t, "abc-123", id) + + _, err = deploymentIDFromName("abc-123") + assert.Error(t, err) + + _, err = deploymentIDFromName("deployments/") + assert.Error(t, err) +} diff --git a/libs/dms/resolve.go b/libs/dms/resolve.go new file mode 100644 index 00000000000..9bbf5a84b56 --- /dev/null +++ b/libs/dms/resolve.go @@ -0,0 +1,40 @@ +package dms + +import ( + "context" + "errors" + "fmt" + "path" + "strconv" + + "github.com/databricks/databricks-sdk-go" + "github.com/databricks/databricks-sdk-go/apierr" +) + +// DeploymentNodeName is the workspace node DMS creates per deployment. Must +// match DeploymentWhsClient.DEPLOYMENT_NODE_NAME on the service side. +const DeploymentNodeName = "resources.deployment.json" + +// ResolveDeploymentID returns the DMS deployment ID for the bundle whose state +// lives under statePath, or empty if it has never recorded a deployment. +// +// The CLI stores the ID nowhere: DMS registers the deployment as a workspace +// node and that node's ID *is* the deployment ID, so a get-status is the lookup. +// This keeps the workspace the single source of truth — a destroyed deployment +// reports absent instead of leaving a dangling ID in the local state file. +func ResolveDeploymentID(ctx context.Context, w *databricks.WorkspaceClient, statePath string) (string, error) { + nodePath := path.Join(statePath, DeploymentNodeName) + + obj, err := w.Workspace.GetStatusByPath(ctx, nodePath) + if err != nil { + if errors.Is(err, apierr.ErrNotFound) || errors.Is(err, apierr.ErrResourceDoesNotExist) { + return "", nil + } + return "", fmt.Errorf("looking up deployment at %s: %w", nodePath, err) + } + + if obj.ObjectId == 0 { + return "", fmt.Errorf("deployment at %s has no object ID", nodePath) + } + return strconv.FormatInt(obj.ObjectId, 10), nil +} diff --git a/libs/dms/resolve_test.go b/libs/dms/resolve_test.go new file mode 100644 index 00000000000..6838f4dd15c --- /dev/null +++ b/libs/dms/resolve_test.go @@ -0,0 +1,69 @@ +package dms + +import ( + "net/http" + "net/http/httptest" + "testing" + + "github.com/databricks/databricks-sdk-go" + "github.com/databricks/databricks-sdk-go/config" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// newTestClient returns a workspace client pointed at a server that serves a +// single get-status response. +func newTestClient(t *testing.T, statusCode int, body string) *databricks.WorkspaceClient { + t.Helper() + + var gotPath string + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + gotPath = r.URL.Query().Get("path") + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(statusCode) + _, _ = w.Write([]byte(body)) + })) + t.Cleanup(srv.Close) + t.Cleanup(func() { + assert.Equal(t, nodePath, gotPath) + }) + + w, err := databricks.NewWorkspaceClient(&databricks.Config{ + Host: srv.URL, + Token: "token", + Credentials: config.PatCredentials{}, + }) + require.NoError(t, err) + return w +} + +const nodePath = "/Workspace/state/" + DeploymentNodeName + +func TestResolveDeploymentIDReturnsNodeID(t *testing.T) { + w := newTestClient(t, http.StatusOK, `{"object_type":"FILE","object_id":123456789,"path":"/Workspace/state/`+DeploymentNodeName+`"}`) + + // The workspace node ID is the deployment ID, so no local state is consulted. + id, err := ResolveDeploymentID(t.Context(), w, "/Workspace/state") + require.NoError(t, err) + assert.Equal(t, "123456789", id) +} + +func TestResolveDeploymentIDAbsentWhenNodeMissing(t *testing.T) { + w := newTestClient(t, http.StatusNotFound, `{"error_code":"RESOURCE_DOES_NOT_EXIST","message":"Path (/Workspace/state/`+DeploymentNodeName+`) doesn't exist."}`) + + // A bundle that never recorded a deployment, or whose deployment was + // destroyed (the service trashes the node), has no ID rather than an error. + id, err := ResolveDeploymentID(t.Context(), w, "/Workspace/state") + require.NoError(t, err) + assert.Empty(t, id) +} + +func TestResolveDeploymentIDPropagatesOtherErrors(t *testing.T) { + w := newTestClient(t, http.StatusForbidden, `{"error_code":"PERMISSION_DENIED","message":"nope"}`) + + // Anything other than a missing node is fatal: silently treating it as absent + // would create a second deployment for a bundle that already has one. + _, err := ResolveDeploymentID(t.Context(), w, "/Workspace/state") + require.Error(t, err) + assert.ErrorContains(t, err, "looking up deployment at /Workspace/state/"+DeploymentNodeName) +} diff --git a/libs/testserver/bundle.go b/libs/testserver/bundle.go new file mode 100644 index 00000000000..6dffbe34a75 --- /dev/null +++ b/libs/testserver/bundle.go @@ -0,0 +1,280 @@ +package testserver + +import ( + "bytes" + "encoding/json" + "path" + "slices" + "strconv" + + "github.com/databricks/databricks-sdk-go/service/bundledeployments" + "github.com/databricks/databricks-sdk-go/service/workspace" +) + +// Handlers for the Deployment Metadata Service (DMS) API under /api/2.0/bundle. +// State is kept in FakeWorkspace.dmsDeployments, keyed by deployment ID. + +// dmsDeploymentNodeName is the name of the workspace node the service creates +// for every deployment. It must match DEPLOYMENT_NODE_NAME on the service side +// (DeploymentWhsClient); the literal is repeated here rather than shared with +// the CLI so a test would catch the CLI drifting from the service. +const dmsDeploymentNodeName = "resources.deployment.json" + +// dmsDeployment holds a deployment record together with the versions and +// resources recorded under it, so the read APIs (ListVersions/ListResources) +// can serve back what deploys wrote. +type dmsDeployment struct { + deployment bundledeployments.Deployment + versions map[string]*bundledeployments.Version + // resources is the latest resource state per resource key, updated as + // operations are recorded. + resources map[string]bundledeployments.Resource + // lastSuccessfulVersionID is the highest version that completed + // successfully. The server advances last_successful_version_id only on + // success (unlike last_version_id), and the read path treats a non-empty + // value as "DMS owns the state". Tracked separately because the SDK + // Deployment struct does not yet carry the field (still stage:DEVELOPMENT). + lastSuccessfulVersionID string + // nodePath is the workspace node whose object ID is this deployment's ID. + // Kept so DeleteDeployment can trash the node, the way the service does. + nodePath string +} + +func (s *FakeWorkspace) CreateDeployment(req Request) Response { + var dep bundledeployments.Deployment + if err := json.Unmarshal(req.Body, &dep); err != nil { + return Response{StatusCode: 400, Body: map[string]string{"message": err.Error()}} + } + if dep.InitialParentPath == "" { + return Response{ + StatusCode: 400, + Body: map[string]string{"error_code": "INVALID_PARAMETER_VALUE", "message": "initial_parent_path is required"}, + } + } + + defer s.LockUnlock()() + + // The service registers the deployment as a workspace node under + // initial_parent_path and uses that node's ID as the deployment ID, so a + // get-status on the node path is how clients look the deployment back up. + nodePath := path.Join(dep.InitialParentPath, dmsDeploymentNodeName) + if resp, ok := s.requireParentDirectory(nodePath); !ok { + return resp + } + objectID := nextID() + s.files[nodePath] = FileEntry{ + Info: workspace.ObjectInfo{ + ObjectType: "FILE", + Path: nodePath, + ObjectId: objectID, + }, + } + + deploymentID := strconv.FormatInt(objectID, 10) + dep.Name = "deployments/" + deploymentID + dep.Status = bundledeployments.DeploymentStatusDeploymentStatusActive + s.dmsDeployments[deploymentID] = &dmsDeployment{ + deployment: dep, + versions: map[string]*bundledeployments.Version{}, + resources: map[string]bundledeployments.Resource{}, + nodePath: nodePath, + } + return Response{Body: dep} +} + +func (s *FakeWorkspace) GetDeployment(deploymentID string) Response { + defer s.LockUnlock()() + + d, ok := s.dmsDeployments[deploymentID] + if !ok { + return dmsNotFound("deployment " + deploymentID) + } + + body, err := deploymentBody(d) + if err != nil { + return Response{StatusCode: 500, Body: map[string]string{"message": err.Error()}} + } + return Response{Body: body} +} + +// deploymentBody renders a deployment the way the real server does: the typed +// fields plus last_successful_version_id, which the generated SDK struct does +// not carry yet (still stage:DEVELOPMENT) but the read path reads off the raw +// JSON. +// +// The extra field cannot be added by embedding Deployment in a wrapper struct: +// Deployment has its own MarshalJSON, which is promoted to the wrapper and +// silently drops any sibling field. +func deploymentBody(d *dmsDeployment) (map[string]any, error) { + raw, err := json.Marshal(d.deployment) + if err != nil { + return nil, err + } + + var body map[string]any + dec := json.NewDecoder(bytes.NewReader(raw)) + dec.UseNumber() + if err := dec.Decode(&body); err != nil { + return nil, err + } + + if d.lastSuccessfulVersionID != "" { + body["last_successful_version_id"] = d.lastSuccessfulVersionID + } + return body, nil +} + +func (s *FakeWorkspace) DeleteDeployment(deploymentID string) Response { + defer s.LockUnlock()() + + // The service trashes the deployment's workspace node, so a later get-status + // on the node path reports the deployment as absent. + if d, ok := s.dmsDeployments[deploymentID]; ok { + delete(s.files, d.nodePath) + } + delete(s.dmsDeployments, deploymentID) + return Response{Body: map[string]any{}} +} + +func (s *FakeWorkspace) CreateVersion(req Request, deploymentID string) Response { + versionID := req.URL.Query().Get("version_id") + + var version bundledeployments.Version + if err := json.Unmarshal(req.Body, &version); err != nil { + return Response{StatusCode: 400, Body: map[string]string{"message": err.Error()}} + } + + defer s.LockUnlock()() + + d, ok := s.dmsDeployments[deploymentID] + if !ok { + return dmsNotFound("deployment " + deploymentID) + } + + // Mirror the server-side optimistic concurrency check: the new version must + // be exactly last_version_id + 1. + want := "1" + if d.deployment.LastVersionId != "" { + last, _ := strconv.ParseInt(d.deployment.LastVersionId, 10, 64) + want = strconv.FormatInt(last+1, 10) + } + if versionID != want { + return dmsAborted("expected version " + want + ", got " + versionID) + } + + d.deployment.LastVersionId = versionID + version.Name = "deployments/" + deploymentID + "/versions/" + versionID + version.VersionId = versionID + version.Status = bundledeployments.VersionStatusVersionStatusInProgress + d.versions[versionID] = &version + return Response{Body: version} +} + +func (s *FakeWorkspace) CompleteVersion(req Request, deploymentID, versionID string) Response { + var completeReq bundledeployments.CompleteVersionRequest + if err := json.Unmarshal(req.Body, &completeReq); err != nil { + return Response{StatusCode: 400, Body: map[string]string{"message": err.Error()}} + } + + defer s.LockUnlock()() + + d, ok := s.dmsDeployments[deploymentID] + if !ok { + return dmsNotFound("deployment " + deploymentID) + } + v, ok := d.versions[versionID] + if !ok { + return dmsNotFound("version " + versionID) + } + + v.Status = bundledeployments.VersionStatusVersionStatusCompleted + v.CompletionReason = completeReq.CompletionReason + if completeReq.CompletionReason == bundledeployments.VersionCompleteVersionCompleteSuccess { + d.lastSuccessfulVersionID = versionID + } + return Response{Body: *v} +} + +func (s *FakeWorkspace) Heartbeat() Response { + return Response{Body: bundledeployments.HeartbeatResponse{}} +} + +func (s *FakeWorkspace) CreateOperation(req Request, deploymentID, versionID string) Response { + resourceKey := req.URL.Query().Get("resource_key") + + var op bundledeployments.Operation + if err := json.Unmarshal(req.Body, &op); err != nil { + return Response{StatusCode: 400, Body: map[string]string{"message": err.Error()}} + } + + defer s.LockUnlock()() + + d, ok := s.dmsDeployments[deploymentID] + if !ok { + return dmsNotFound("deployment " + deploymentID) + } + + op.Name = "deployments/" + deploymentID + "/versions/" + versionID + "/operations/" + resourceKey + op.ResourceKey = resourceKey + + // Reflect the operation onto the deployment-level resource set the way the + // backend does: a delete removes the resource, anything else upserts it. + if op.ActionType == bundledeployments.OperationActionTypeOperationActionTypeDelete { + delete(d.resources, resourceKey) + } else { + d.resources[resourceKey] = bundledeployments.Resource{ + Name: "deployments/" + deploymentID + "/resources/" + resourceKey, + ResourceKey: resourceKey, + ResourceId: op.ResourceId, + ResourceType: op.ResourceType, + LastActionType: op.ActionType, + LastVersionId: versionID, + State: op.State, + } + } + return Response{Body: op} +} + +func (s *FakeWorkspace) ListResources(deploymentID string) Response { + defer s.LockUnlock()() + + d, ok := s.dmsDeployments[deploymentID] + if !ok { + return dmsNotFound("deployment " + deploymentID) + } + + // Sort by resource key so the response order is deterministic. + keys := make([]string, 0, len(d.resources)) + for key := range d.resources { + keys = append(keys, key) + } + slices.Sort(keys) + + resources := make([]bundledeployments.Resource, 0, len(keys)) + for _, key := range keys { + resources = append(resources, d.resources[key]) + } + return Response{Body: bundledeployments.ListResourcesResponse{Resources: resources}} +} + +// dmsNotFound returns the RESOURCE_DOES_NOT_EXIST error shape the DMS API uses, +// which the SDK maps to apierr.ErrNotFound. +func dmsNotFound(what string) Response { + return Response{ + StatusCode: 404, + Body: map[string]string{ + "error_code": "RESOURCE_DOES_NOT_EXIST", + "message": what + " does not exist", + }, + } +} + +// dmsAborted returns the 409 ABORTED error the server uses for the version +// optimistic-concurrency check. +func dmsAborted(message string) Response { + return Response{ + StatusCode: 409, + Headers: map[string][]string{"Content-Type": {"application/json"}}, + Body: map[string]string{"error_code": "ABORTED", "message": message}, + } +} diff --git a/libs/testserver/bundle_test.go b/libs/testserver/bundle_test.go new file mode 100644 index 00000000000..4d28624ba3e --- /dev/null +++ b/libs/testserver/bundle_test.go @@ -0,0 +1,51 @@ +package testserver + +import ( + "encoding/json" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// TestDeploymentBodyKeepsTypedFieldsAndLastSuccessfulVersionID guards against +// serializing the deployment through a struct that embeds +// bundledeployments.Deployment: Deployment has its own MarshalJSON, which is +// promoted to the embedding struct and silently drops last_successful_version_id. +// The CLI read path treats a missing value as "DMS does not own the state", so +// losing the field here makes the whole overlay path untestable. +func TestDeploymentBodyKeepsTypedFieldsAndLastSuccessfulVersionID(t *testing.T) { + d := &dmsDeployment{lastSuccessfulVersionID: "2"} + d.deployment.Name = "deployments/abc" + d.deployment.LastVersionId = "3" + d.deployment.TargetName = "default" + + body, err := deploymentBody(d) + require.NoError(t, err) + + assert.Equal(t, "deployments/abc", body["name"]) + assert.Equal(t, "3", body["last_version_id"]) + assert.Equal(t, "default", body["target_name"]) + assert.Equal(t, "2", body["last_successful_version_id"]) + + // The response must round-trip as JSON the same way, since that is what the + // client actually reads. + raw, err := json.Marshal(body) + require.NoError(t, err) + assert.JSONEq(t, + `{"name":"deployments/abc","last_version_id":"3","target_name":"default","last_successful_version_id":"2"}`, + string(raw)) +} + +// TestDeploymentBodyOmitsUnsetLastSuccessfulVersionID checks that a deployment +// with no successful version does not advertise one: the read path must keep +// using the local state file in that case. +func TestDeploymentBodyOmitsUnsetLastSuccessfulVersionID(t *testing.T) { + d := &dmsDeployment{} + d.deployment.Name = "deployments/abc" + + body, err := deploymentBody(d) + require.NoError(t, err) + + assert.NotContains(t, body, "last_successful_version_id") +} diff --git a/libs/testserver/fake_workspace.go b/libs/testserver/fake_workspace.go index 8d6e8ee0dd3..a3c4519ccc4 100644 --- a/libs/testserver/fake_workspace.go +++ b/libs/testserver/fake_workspace.go @@ -227,6 +227,10 @@ type FakeWorkspace struct { // clusterVenvs caches Python venvs per existing cluster ID, // matching cloud behavior where libraries are cached on running clusters. clusterVenvs map[string]*clusterEnv + + // dmsDeployments holds Deployment Metadata Service (DMS) records, keyed by + // deployment ID. Each record carries its versions and latest resource state. + dmsDeployments map[string]*dmsDeployment } func (s *FakeWorkspace) LockUnlock() func() { @@ -378,6 +382,7 @@ func NewFakeWorkspace(url, token string) *FakeWorkspace { postgresImplicitBranches: map[string]bool{}, postgresImplicitEndpoints: map[string]bool{}, clusterVenvs: map[string]*clusterEnv{}, + dmsDeployments: map[string]*dmsDeployment{}, Alerts: map[string]sql.AlertV2{}, Experiments: map[string]ml.GetExperimentResponse{}, ModelRegistryModels: map[string]ml.Model{}, diff --git a/libs/testserver/handlers.go b/libs/testserver/handlers.go index 1d534c47431..3cbfa16c7e1 100644 --- a/libs/testserver/handlers.go +++ b/libs/testserver/handlers.go @@ -266,6 +266,32 @@ func AddDefaultHandlers(server *Server) { return req.Workspace.JobsCreate(req) }) + // Deployment Metadata Service (DMS) endpoints. + server.Handle("POST", "/api/2.0/bundle/deployments", func(req Request) any { + return req.Workspace.CreateDeployment(req) + }) + server.Handle("GET", "/api/2.0/bundle/deployments/{deployment_id}", func(req Request) any { + return req.Workspace.GetDeployment(req.Vars["deployment_id"]) + }) + server.Handle("DELETE", "/api/2.0/bundle/deployments/{deployment_id}", func(req Request) any { + return req.Workspace.DeleteDeployment(req.Vars["deployment_id"]) + }) + server.Handle("POST", "/api/2.0/bundle/deployments/{deployment_id}/versions", func(req Request) any { + return req.Workspace.CreateVersion(req, req.Vars["deployment_id"]) + }) + server.Handle("POST", "/api/2.0/bundle/deployments/{deployment_id}/versions/{version_id}/complete", func(req Request) any { + return req.Workspace.CompleteVersion(req, req.Vars["deployment_id"], req.Vars["version_id"]) + }) + server.Handle("POST", "/api/2.0/bundle/deployments/{deployment_id}/versions/{version_id}/heartbeat", func(req Request) any { + return req.Workspace.Heartbeat() + }) + server.Handle("POST", "/api/2.0/bundle/deployments/{deployment_id}/versions/{version_id}/operations", func(req Request) any { + return req.Workspace.CreateOperation(req, req.Vars["deployment_id"], req.Vars["version_id"]) + }) + server.Handle("GET", "/api/2.0/bundle/deployments/{deployment_id}/resources", func(req Request) any { + return req.Workspace.ListResources(req.Vars["deployment_id"]) + }) + server.Handle("POST", "/api/2.2/jobs/delete", func(req Request) any { var request jobs.DeleteJob if err := json.Unmarshal(req.Body, &request); err != nil {