From cbef82725ca3682c8ecc927d19defbb9a217f75c Mon Sep 17 00:00:00 2001 From: Shreyas Goenka Date: Wed, 9 Sep 2026 16:02:05 +0000 Subject: [PATCH 1/6] Replace hand-written DMS client with the Go SDK The libs/dms rawClient hand-wrote CreateVersion, UpdateDeployment, and UpdateOperation because the SDK lacked the operations field, the method, and int64 sequence_id handling. All three are in SDK v0.177.0, so route them through the generated client, using ForceSendFields to keep DMS's masked-but-empty fields on the wire. The testserver fake now reads sequence_id as the number the SDK sends (still emitting a string, like the real service). Co-authored-by: Isaac --- .../bundle/dms/declined-deploy/output.txt | 22 +- acceptance/bundle/dms/depends-on/output.txt | 4 +- .../dms/deployment-metadata-change/output.txt | 10 +- acceptance/bundle/dms/empty-grants/output.txt | 28 +- .../bundle/dms/existing-state/output.txt | 4 +- .../bundle/dms/failed-delete/output.txt | 22 +- .../bundle/dms/failed-recreate/output.txt | 24 +- .../dms/failed-update-permissions/output.txt | 4 +- .../bundle/dms/multiple-resources/output.txt | 10 +- .../operation-upload-fails-once/output.txt | 14 +- acceptance/bundle/dms/provenance/output.txt | 10 +- .../bundle/dms/record-failure/output.txt | 10 +- acceptance/bundle/dms/stale-plan/output.txt | 4 +- .../bundle/dms/successful-recreate/output.txt | 36 +-- .../dms/version-never-created/output.txt | 4 +- libs/dms/client.go | 242 ++++++++---------- libs/dms/client_test.go | 49 ++-- libs/testserver/bundledeployments.go | 27 +- 18 files changed, 244 insertions(+), 280 deletions(-) diff --git a/acceptance/bundle/dms/declined-deploy/output.txt b/acceptance/bundle/dms/declined-deploy/output.txt index 7cea41997aa..eefd538e632 100644 --- a/acceptance/bundle/dms/declined-deploy/output.txt +++ b/acceptance/bundle/dms/declined-deploy/output.txt @@ -28,13 +28,13 @@ Resources: 1 created, 0 changed, 0 deleted, 0 unchanged }, "body": { "cli_version": "[CLI_VERSION]", - "version_type": "VERSION_TYPE_DEPLOY", "operations": [ { - "resource_key": "schemas.foo", - "action_type": "OPERATION_ACTION_TYPE_CREATE" + "action_type": "OPERATION_ACTION_TYPE_CREATE", + "resource_key": "schemas.foo" } - ] + ], + "version_type": "VERSION_TYPE_DEPLOY" } } { @@ -46,7 +46,7 @@ Resources: 1 created, 0 changed, 0 deleted, 0 unchanged "body": { "error_message": "", "resource_id": "main.dms_declined_deploy_schema_[UNIQUE_NAME]", - "sequence_id": "0", + "sequence_id": 0, "state": "{\"state\":{\"catalog_name\":\"main\",\"name\":\"dms_declined_deploy_schema_[UNIQUE_NAME]\"}}", "status": "OPERATION_STATUS_SUCCEEDED" } @@ -174,14 +174,14 @@ false }, "body": { "cli_version": "[CLI_VERSION]", - "version_type": "VERSION_TYPE_DESTROY", - "previous_version_id": "1", "operations": [ { - "resource_key": "schemas.foo", - "action_type": "OPERATION_ACTION_TYPE_DELETE" + "action_type": "OPERATION_ACTION_TYPE_DELETE", + "resource_key": "schemas.foo" } - ] + ], + "previous_version_id": "1", + "version_type": "VERSION_TYPE_DESTROY" } } { @@ -192,7 +192,7 @@ false }, "body": { "error_message": "", - "sequence_id": "0", + "sequence_id": 0, "status": "OPERATION_STATUS_SUCCEEDED" } } diff --git a/acceptance/bundle/dms/depends-on/output.txt b/acceptance/bundle/dms/depends-on/output.txt index a1c7270e43c..7f5d4327e92 100644 --- a/acceptance/bundle/dms/depends-on/output.txt +++ b/acceptance/bundle/dms/depends-on/output.txt @@ -33,7 +33,7 @@ Resources: 2 created, 0 changed, 0 deleted, 0 unchanged "body": { "error_message": "", "resource_id": "[NUMID]", - "sequence_id": "0", + "sequence_id": 0, "state": "{\"state\":{\"deployment\":{\"deployment_id\":\"[NUMID]\",\"kind\":\"BUNDLE\",\"metadata_file_path\":\"/Workspace/Users/[USERNAME]/.bundle/dms-depends-on-[UNIQUE_NAME]/default/state/metadata.json\",\"version_id\":\"1\"},\"edit_mode\":\"UI_LOCKED\",\"format\":\"MULTI_TASK\",\"max_concurrent_runs\":1,\"name\":\"parent\",\"queue\":{\"enabled\":true}}}", "status": "OPERATION_STATUS_SUCCEEDED" } @@ -47,7 +47,7 @@ Resources: 2 created, 0 changed, 0 deleted, 0 unchanged "body": { "error_message": "", "resource_id": "[NUMID]", - "sequence_id": "0", + "sequence_id": 0, "state": "{\"state\":{\"deployment\":{\"deployment_id\":\"[NUMID]\",\"kind\":\"BUNDLE\",\"metadata_file_path\":\"/Workspace/Users/[USERNAME]/.bundle/dms-depends-on-[UNIQUE_NAME]/default/state/metadata.json\",\"version_id\":\"1\"},\"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" } diff --git a/acceptance/bundle/dms/deployment-metadata-change/output.txt b/acceptance/bundle/dms/deployment-metadata-change/output.txt index 00bcb2d0c12..2bd9572e5d7 100644 --- a/acceptance/bundle/dms/deployment-metadata-change/output.txt +++ b/acceptance/bundle/dms/deployment-metadata-change/output.txt @@ -28,13 +28,13 @@ Resources: 1 created, 0 changed, 0 deleted, 0 unchanged }, "body": { "cli_version": "[CLI_VERSION]", - "version_type": "VERSION_TYPE_DEPLOY", "operations": [ { - "resource_key": "schemas.foo", - "action_type": "OPERATION_ACTION_TYPE_CREATE" + "action_type": "OPERATION_ACTION_TYPE_CREATE", + "resource_key": "schemas.foo" } - ] + ], + "version_type": "VERSION_TYPE_DEPLOY" } } { @@ -46,7 +46,7 @@ Resources: 1 created, 0 changed, 0 deleted, 0 unchanged "body": { "error_message": "", "resource_id": "main.dms_metadata_schema_[UNIQUE_NAME]", - "sequence_id": "0", + "sequence_id": 0, "state": "{\"state\":{\"catalog_name\":\"main\",\"name\":\"dms_metadata_schema_[UNIQUE_NAME]\"}}", "status": "OPERATION_STATUS_SUCCEEDED" } diff --git a/acceptance/bundle/dms/empty-grants/output.txt b/acceptance/bundle/dms/empty-grants/output.txt index 25ffb4376e9..6ea903f709a 100644 --- a/acceptance/bundle/dms/empty-grants/output.txt +++ b/acceptance/bundle/dms/empty-grants/output.txt @@ -29,17 +29,17 @@ Resources: 2 created, 0 changed, 0 deleted, 0 unchanged }, "body": { "cli_version": "[CLI_VERSION]", - "version_type": "VERSION_TYPE_DEPLOY", "operations": [ { - "resource_key": "schemas.foo", - "action_type": "OPERATION_ACTION_TYPE_CREATE" + "action_type": "OPERATION_ACTION_TYPE_CREATE", + "resource_key": "schemas.foo" }, { - "resource_key": "schemas.foo.grants", - "action_type": "OPERATION_ACTION_TYPE_CREATE" + "action_type": "OPERATION_ACTION_TYPE_CREATE", + "resource_key": "schemas.foo.grants" } - ] + ], + "version_type": "VERSION_TYPE_DEPLOY" } } { @@ -51,7 +51,7 @@ Resources: 2 created, 0 changed, 0 deleted, 0 unchanged "body": { "error_message": "", "resource_id": "main.dms_empty_grants_[UNIQUE_NAME]", - "sequence_id": "0", + "sequence_id": 0, "state": "{\"state\":{\"catalog_name\":\"main\",\"name\":\"dms_empty_grants_[UNIQUE_NAME]\"}}", "status": "OPERATION_STATUS_SUCCEEDED" } @@ -65,7 +65,7 @@ Resources: 2 created, 0 changed, 0 deleted, 0 unchanged "body": { "error_message": "", "resource_id": "schema/main.dms_empty_grants_[UNIQUE_NAME]", - "sequence_id": "0", + "sequence_id": 0, "state": "{\"state\":{\"securable_type\":\"schema\",\"full_name\":\"main.dms_empty_grants_[UNIQUE_NAME]\",\"__embed__\":[{\"principal\":\"deco-test-user@databricks.com\",\"privileges\":[\"USE_SCHEMA\"]}]},\"depends_on\":[{\"node\":\"resources.schemas.foo\",\"label\":\"${resources.schemas.foo.id}\"}]}", "status": "OPERATION_STATUS_SUCCEEDED" } @@ -96,14 +96,14 @@ Resources: 0 created, 1 changed, 0 deleted, 1 unchanged }, "body": { "cli_version": "[CLI_VERSION]", - "version_type": "VERSION_TYPE_DEPLOY", - "previous_version_id": "1", "operations": [ { - "resource_key": "schemas.foo.grants", - "action_type": "OPERATION_ACTION_TYPE_UPDATE" + "action_type": "OPERATION_ACTION_TYPE_UPDATE", + "resource_key": "schemas.foo.grants" } - ] + ], + "previous_version_id": "1", + "version_type": "VERSION_TYPE_DEPLOY" } } { @@ -114,7 +114,7 @@ Resources: 0 created, 1 changed, 0 deleted, 1 unchanged }, "body": { "error_message": "", - "sequence_id": "0", + "sequence_id": 0, "status": "OPERATION_STATUS_SUCCEEDED" } } diff --git a/acceptance/bundle/dms/existing-state/output.txt b/acceptance/bundle/dms/existing-state/output.txt index 8dc33d924fe..0551f63fe66 100644 --- a/acceptance/bundle/dms/existing-state/output.txt +++ b/acceptance/bundle/dms/existing-state/output.txt @@ -63,8 +63,8 @@ Resources: 1 created, 0 changed, 0 deleted, 0 unchanged >>> print_requests.py --dms //api/2.0/bundle --oneline {"method": "POST", "path": "/api/2.0/bundle/deployments", "body": {"display_name": "dms-existing-state-[UNIQUE_NAME]", "initial_parent_path": "/Workspace/Users/[USERNAME]/.bundle/dms-existing-state-[UNIQUE_NAME]/default/state", "target_name": "default", "workspace_info": {"file_path": "/Workspace/Users/[USERNAME]/.bundle/dms-existing-state-[UNIQUE_NAME]/default/files", "root_path": "/Workspace/Users/[USERNAME]/.bundle/dms-existing-state-[UNIQUE_NAME]/default"}}} -{"method": "POST", "path": "/api/2.0/bundle/deployments/[NUMID]/versions", "q": {"version_id": "1"}, "body": {"cli_version": "[CLI_VERSION]", "version_type": "VERSION_TYPE_DEPLOY", "operations": [{"resource_key": "jobs.one", "action_type": "OPERATION_ACTION_TYPE_CREATE"}]}} -{"method": "PATCH", "path": "/api/2.0/bundle/deployments/[NUMID]/versions/1/operations/jobs.one", "q": {"update_mask": "state,error_message,resource_id,status"}, "body": {"error_message": "", "resource_id": "[NUMID]", "sequence_id": "0", "state": "{\"state\":{\"deployment\":{\"deployment_id\":\"[NUMID]\",\"kind\":\"BUNDLE\",\"metadata_file_path\":\"/Workspace/Users/[USERNAME]/.bundle/dms-existing-state-[UNIQUE_NAME]/default/state/metadata.json\",\"version_id\":\"1\"},\"edit_mode\":\"UI_LOCKED\",\"format\":\"MULTI_TASK\",\"max_concurrent_runs\":1,\"name\":\"one\",\"queue\":{\"enabled\":true}}}", "status": "OPERATION_STATUS_SUCCEEDED"}} +{"method": "POST", "path": "/api/2.0/bundle/deployments/[NUMID]/versions", "q": {"version_id": "1"}, "body": {"cli_version": "[CLI_VERSION]", "operations": [{"action_type": "OPERATION_ACTION_TYPE_CREATE", "resource_key": "jobs.one"}], "version_type": "VERSION_TYPE_DEPLOY"}} +{"method": "PATCH", "path": "/api/2.0/bundle/deployments/[NUMID]/versions/1/operations/jobs.one", "q": {"update_mask": "state,error_message,resource_id,status"}, "body": {"error_message": "", "resource_id": "[NUMID]", "sequence_id": 0, "state": "{\"state\":{\"deployment\":{\"deployment_id\":\"[NUMID]\",\"kind\":\"BUNDLE\",\"metadata_file_path\":\"/Workspace/Users/[USERNAME]/.bundle/dms-existing-state-[UNIQUE_NAME]/default/state/metadata.json\",\"version_id\":\"1\"},\"edit_mode\":\"UI_LOCKED\",\"format\":\"MULTI_TASK\",\"max_concurrent_runs\":1,\"name\":\"one\",\"queue\":{\"enabled\":true}}}", "status": "OPERATION_STATUS_SUCCEEDED"}} {"method": "POST", "path": "/api/2.0/bundle/deployments/[NUMID]/versions/1/complete", "body": {"completion_reason": "VERSION_COMPLETE_SUCCESS"}} >>> find.py resources[.]json[.]wal --expect 0 diff --git a/acceptance/bundle/dms/failed-delete/output.txt b/acceptance/bundle/dms/failed-delete/output.txt index ff4c1987517..6d5a6756c37 100644 --- a/acceptance/bundle/dms/failed-delete/output.txt +++ b/acceptance/bundle/dms/failed-delete/output.txt @@ -28,13 +28,13 @@ Resources: 1 created, 0 changed, 0 deleted, 0 unchanged }, "body": { "cli_version": "[CLI_VERSION]", - "version_type": "VERSION_TYPE_DEPLOY", "operations": [ { - "resource_key": "jobs.stuck", - "action_type": "OPERATION_ACTION_TYPE_CREATE" + "action_type": "OPERATION_ACTION_TYPE_CREATE", + "resource_key": "jobs.stuck" } - ] + ], + "version_type": "VERSION_TYPE_DEPLOY" } } { @@ -46,7 +46,7 @@ Resources: 1 created, 0 changed, 0 deleted, 0 unchanged "body": { "error_message": "", "resource_id": "[NUMID]", - "sequence_id": "0", + "sequence_id": 0, "state": "{\"state\":{\"deployment\":{\"deployment_id\":\"[NUMID]\",\"kind\":\"BUNDLE\",\"metadata_file_path\":\"/Workspace/Users/[USERNAME]/.bundle/dms-failed-delete-[UNIQUE_NAME]/default/state/metadata.json\",\"version_id\":\"1\"},\"edit_mode\":\"UI_LOCKED\",\"format\":\"MULTI_TASK\",\"max_concurrent_runs\":1,\"name\":\"stuck\",\"queue\":{\"enabled\":true}}}", "status": "OPERATION_STATUS_SUCCEEDED" } @@ -122,14 +122,14 @@ API message: Fault injected by test. }, "body": { "cli_version": "[CLI_VERSION]", - "version_type": "VERSION_TYPE_DESTROY", - "previous_version_id": "1", "operations": [ { - "resource_key": "jobs.stuck", - "action_type": "OPERATION_ACTION_TYPE_DELETE" + "action_type": "OPERATION_ACTION_TYPE_DELETE", + "resource_key": "jobs.stuck" } - ] + ], + "previous_version_id": "1", + "version_type": "VERSION_TYPE_DESTROY" } } { @@ -141,7 +141,7 @@ API message: Fault injected by test. "body": { "error_message": "deleting id=[NUMID]: Fault injected by test. (400 INVALID_PARAMETER_VALUE)", "resource_id": "[NUMID]", - "sequence_id": "0", + "sequence_id": 0, "state": "{\"state\":{\"deployment\":{\"deployment_id\":\"[NUMID]\",\"kind\":\"BUNDLE\",\"metadata_file_path\":\"/Workspace/Users/[USERNAME]/.bundle/dms-failed-delete-[UNIQUE_NAME]/default/state/metadata.json\",\"version_id\":\"1\"},\"edit_mode\":\"UI_LOCKED\",\"format\":\"MULTI_TASK\",\"max_concurrent_runs\":1,\"name\":\"stuck\",\"queue\":{\"enabled\":true}}}", "status": "OPERATION_STATUS_FAILED" } diff --git a/acceptance/bundle/dms/failed-recreate/output.txt b/acceptance/bundle/dms/failed-recreate/output.txt index 59e3e5d9eab..df298291ce6 100644 --- a/acceptance/bundle/dms/failed-recreate/output.txt +++ b/acceptance/bundle/dms/failed-recreate/output.txt @@ -28,13 +28,13 @@ Resources: 1 created, 0 changed, 0 deleted, 0 unchanged }, "body": { "cli_version": "[CLI_VERSION]", - "version_type": "VERSION_TYPE_DEPLOY", "operations": [ { - "resource_key": "schemas.foo", - "action_type": "OPERATION_ACTION_TYPE_CREATE" + "action_type": "OPERATION_ACTION_TYPE_CREATE", + "resource_key": "schemas.foo" } - ] + ], + "version_type": "VERSION_TYPE_DEPLOY" } } { @@ -46,7 +46,7 @@ Resources: 1 created, 0 changed, 0 deleted, 0 unchanged "body": { "error_message": "", "resource_id": "main.dms_failed_recreate_schema_[UNIQUE_NAME]", - "sequence_id": "0", + "sequence_id": 0, "state": "{\"state\":{\"catalog_name\":\"main\",\"comment\":\"v1\",\"name\":\"dms_failed_recreate_schema_[UNIQUE_NAME]\"}}", "status": "OPERATION_STATUS_SUCCEEDED" } @@ -124,14 +124,14 @@ Files: 2 uploaded, 0 deleted }, "body": { "cli_version": "[CLI_VERSION]", - "version_type": "VERSION_TYPE_DEPLOY", - "previous_version_id": "1", "operations": [ { - "resource_key": "schemas.foo", - "action_type": "OPERATION_ACTION_TYPE_RECREATE" + "action_type": "OPERATION_ACTION_TYPE_RECREATE", + "resource_key": "schemas.foo" } - ] + ], + "previous_version_id": "1", + "version_type": "VERSION_TYPE_DEPLOY" } } { @@ -142,7 +142,7 @@ Files: 2 uploaded, 0 deleted }, "body": { "error_message": "", - "sequence_id": "0", + "sequence_id": 0, "status": "OPERATION_STATUS_PENDING" } } @@ -154,7 +154,7 @@ Files: 2 uploaded, 0 deleted }, "body": { "error_message": "Fault injected by test. (400 INVALID_PARAMETER_VALUE)", - "sequence_id": "1", + "sequence_id": 1, "status": "OPERATION_STATUS_FAILED" } } diff --git a/acceptance/bundle/dms/failed-update-permissions/output.txt b/acceptance/bundle/dms/failed-update-permissions/output.txt index 2cf353e1368..a45e4a21c1a 100644 --- a/acceptance/bundle/dms/failed-update-permissions/output.txt +++ b/acceptance/bundle/dms/failed-update-permissions/output.txt @@ -104,6 +104,6 @@ Files: 1 uploaded, 0 deleted } >>> print_requests.py --dms //api/2.0/bundle --oneline -{"method": "POST", "path": "/api/2.0/bundle/deployments/[NUMID]/versions", "q": {"version_id": "2"}, "body": {"cli_version": "[CLI_VERSION]", "version_type": "VERSION_TYPE_DEPLOY", "previous_version_id": "1", "operations": [{"resource_key": "jobs.foo.permissions", "action_type": "OPERATION_ACTION_TYPE_UPDATE"}]}} -{"method": "PATCH", "path": "/api/2.0/bundle/deployments/[NUMID]/versions/2/operations/jobs.foo.permissions", "q": {"update_mask": "state,error_message,resource_id,status"}, "body": {"error_message": "updating id=/jobs/[NUMID]: Fault injected by test. (403 PERMISSION_DENIED)", "resource_id": "/jobs/[NUMID]", "sequence_id": "0", "state": "{\"state\":{\"object_id\":\"/jobs/[NUMID]\",\"__embed__\":[{\"level\":\"CAN_VIEW\",\"user_name\":\"viewer@example.com\"},{\"level\":\"IS_OWNER\",\"user_name\":\"[USERNAME]\"}]},\"depends_on\":[{\"node\":\"resources.jobs.foo\",\"label\":\"${resources.jobs.foo.id}\"}]}", "status": "OPERATION_STATUS_FAILED"}} +{"method": "POST", "path": "/api/2.0/bundle/deployments/[NUMID]/versions", "q": {"version_id": "2"}, "body": {"cli_version": "[CLI_VERSION]", "operations": [{"action_type": "OPERATION_ACTION_TYPE_UPDATE", "resource_key": "jobs.foo.permissions"}], "previous_version_id": "1", "version_type": "VERSION_TYPE_DEPLOY"}} +{"method": "PATCH", "path": "/api/2.0/bundle/deployments/[NUMID]/versions/2/operations/jobs.foo.permissions", "q": {"update_mask": "state,error_message,resource_id,status"}, "body": {"error_message": "updating id=/jobs/[NUMID]: Fault injected by test. (403 PERMISSION_DENIED)", "resource_id": "/jobs/[NUMID]", "sequence_id": 0, "state": "{\"state\":{\"object_id\":\"/jobs/[NUMID]\",\"__embed__\":[{\"level\":\"CAN_VIEW\",\"user_name\":\"viewer@example.com\"},{\"level\":\"IS_OWNER\",\"user_name\":\"[USERNAME]\"}]},\"depends_on\":[{\"node\":\"resources.jobs.foo\",\"label\":\"${resources.jobs.foo.id}\"}]}", "status": "OPERATION_STATUS_FAILED"}} {"method": "POST", "path": "/api/2.0/bundle/deployments/[NUMID]/versions/2/complete", "body": {"completion_reason": "VERSION_COMPLETE_FAILURE"}} diff --git a/acceptance/bundle/dms/multiple-resources/output.txt b/acceptance/bundle/dms/multiple-resources/output.txt index 40e3648de3e..e7c4d058f28 100644 --- a/acceptance/bundle/dms/multiple-resources/output.txt +++ b/acceptance/bundle/dms/multiple-resources/output.txt @@ -11,11 +11,11 @@ Files: 2 uploaded, 0 deleted Resources: 5 created, 0 changed, 0 deleted, 0 unchanged >>> print_requests.py --dms //versions/1/operations --sort --del-body state --oneline -{"method": "PATCH", "path": "/api/2.0/bundle/deployments/[NUMID]/versions/1/operations/jobs.five", "q": {"update_mask": "state,error_message,resource_id,status"}, "body": {"error_message": "", "resource_id": "[NUMID]", "sequence_id": "0", "status": "OPERATION_STATUS_SUCCEEDED"}} -{"method": "PATCH", "path": "/api/2.0/bundle/deployments/[NUMID]/versions/1/operations/jobs.four", "q": {"update_mask": "state,error_message,resource_id,status"}, "body": {"error_message": "", "resource_id": "[NUMID]", "sequence_id": "0", "status": "OPERATION_STATUS_SUCCEEDED"}} -{"method": "PATCH", "path": "/api/2.0/bundle/deployments/[NUMID]/versions/1/operations/jobs.one", "q": {"update_mask": "state,error_message,resource_id,status"}, "body": {"error_message": "", "resource_id": "[NUMID]", "sequence_id": "0", "status": "OPERATION_STATUS_SUCCEEDED"}} -{"method": "PATCH", "path": "/api/2.0/bundle/deployments/[NUMID]/versions/1/operations/jobs.three", "q": {"update_mask": "state,error_message,resource_id,status"}, "body": {"error_message": "", "resource_id": "[NUMID]", "sequence_id": "0", "status": "OPERATION_STATUS_SUCCEEDED"}} -{"method": "PATCH", "path": "/api/2.0/bundle/deployments/[NUMID]/versions/1/operations/jobs.two", "q": {"update_mask": "state,error_message,resource_id,status"}, "body": {"error_message": "", "resource_id": "[NUMID]", "sequence_id": "0", "status": "OPERATION_STATUS_SUCCEEDED"}} +{"method": "PATCH", "path": "/api/2.0/bundle/deployments/[NUMID]/versions/1/operations/jobs.five", "q": {"update_mask": "state,error_message,resource_id,status"}, "body": {"error_message": "", "resource_id": "[NUMID]", "sequence_id": 0, "status": "OPERATION_STATUS_SUCCEEDED"}} +{"method": "PATCH", "path": "/api/2.0/bundle/deployments/[NUMID]/versions/1/operations/jobs.four", "q": {"update_mask": "state,error_message,resource_id,status"}, "body": {"error_message": "", "resource_id": "[NUMID]", "sequence_id": 0, "status": "OPERATION_STATUS_SUCCEEDED"}} +{"method": "PATCH", "path": "/api/2.0/bundle/deployments/[NUMID]/versions/1/operations/jobs.one", "q": {"update_mask": "state,error_message,resource_id,status"}, "body": {"error_message": "", "resource_id": "[NUMID]", "sequence_id": 0, "status": "OPERATION_STATUS_SUCCEEDED"}} +{"method": "PATCH", "path": "/api/2.0/bundle/deployments/[NUMID]/versions/1/operations/jobs.three", "q": {"update_mask": "state,error_message,resource_id,status"}, "body": {"error_message": "", "resource_id": "[NUMID]", "sequence_id": 0, "status": "OPERATION_STATUS_SUCCEEDED"}} +{"method": "PATCH", "path": "/api/2.0/bundle/deployments/[NUMID]/versions/1/operations/jobs.two", "q": {"update_mask": "state,error_message,resource_id,status"}, "body": {"error_message": "", "resource_id": "[NUMID]", "sequence_id": 0, "status": "OPERATION_STATUS_SUCCEEDED"}} === The plan records the deployment_history feature >>> [CLI] bundle plan -o json diff --git a/acceptance/bundle/dms/operation-upload-fails-once/output.txt b/acceptance/bundle/dms/operation-upload-fails-once/output.txt index adf0a58883c..daaecd892f4 100644 --- a/acceptance/bundle/dms/operation-upload-fails-once/output.txt +++ b/acceptance/bundle/dms/operation-upload-fails-once/output.txt @@ -94,14 +94,14 @@ Files: 2 uploaded, 0 deleted }, "body": { "cli_version": "[CLI_VERSION]", - "version_type": "VERSION_TYPE_DEPLOY", - "previous_version_id": "1", "operations": [ { - "resource_key": "schemas.foo", - "action_type": "OPERATION_ACTION_TYPE_RECREATE" + "action_type": "OPERATION_ACTION_TYPE_RECREATE", + "resource_key": "schemas.foo" } - ] + ], + "previous_version_id": "1", + "version_type": "VERSION_TYPE_DEPLOY" } } { @@ -112,7 +112,7 @@ Files: 2 uploaded, 0 deleted }, "body": { "error_message": "", - "sequence_id": "0", + "sequence_id": 0, "status": "OPERATION_STATUS_PENDING" } } @@ -125,7 +125,7 @@ Files: 2 uploaded, 0 deleted "body": { "error_message": "", "resource_id": "dms_other_[UNIQUE_NAME].dms_upload_fails_once_schema_[UNIQUE_NAME]", - "sequence_id": "0", + "sequence_id": 0, "state": "{\"state\":{\"catalog_name\":\"dms_other_[UNIQUE_NAME]\",\"comment\":\"v1\",\"name\":\"dms_upload_fails_once_schema_[UNIQUE_NAME]\"}}", "status": "OPERATION_STATUS_SUCCEEDED" } diff --git a/acceptance/bundle/dms/provenance/output.txt b/acceptance/bundle/dms/provenance/output.txt index 17118bc1317..a2f7de239b9 100644 --- a/acceptance/bundle/dms/provenance/output.txt +++ b/acceptance/bundle/dms/provenance/output.txt @@ -15,7 +15,6 @@ Resources: 1 created, 0 changed, 0 deleted, 0 unchanged }, "body": { "cli_version": "[CLI_VERSION]", - "version_type": "VERSION_TYPE_DEPLOY", "git_info": { "branch": "main", "commit": "[COMMIT]", @@ -23,10 +22,11 @@ Resources: 1 created, 0 changed, 0 deleted, 0 unchanged }, "operations": [ { - "resource_key": "jobs.foo", - "action_type": "OPERATION_ACTION_TYPE_CREATE" + "action_type": "OPERATION_ACTION_TYPE_CREATE", + "resource_key": "jobs.foo" } - ] + ], + "version_type": "VERSION_TYPE_DEPLOY" } } { @@ -38,7 +38,7 @@ Resources: 1 created, 0 changed, 0 deleted, 0 unchanged "body": { "error_message": "", "resource_id": "[NUMID]", - "sequence_id": "0", + "sequence_id": 0, "state": "{\"state\":{\"deployment\":{\"deployment_id\":\"[NUMID]\",\"kind\":\"BUNDLE\",\"metadata_file_path\":\"/Workspace/Users/[USERNAME]/.bundle/dms-provenance-[UNIQUE_NAME]/dev/state/metadata.json\",\"version_id\":\"1\"},\"edit_mode\":\"UI_LOCKED\",\"format\":\"MULTI_TASK\",\"max_concurrent_runs\":4,\"name\":\"[dev [USERNAME]] foo\",\"queue\":{\"enabled\":true},\"tags\":{\"dev\":\"[USERNAME]\"}}}", "status": "OPERATION_STATUS_SUCCEEDED" } diff --git a/acceptance/bundle/dms/record-failure/output.txt b/acceptance/bundle/dms/record-failure/output.txt index 758468fefab..bf61dfb7392 100644 --- a/acceptance/bundle/dms/record-failure/output.txt +++ b/acceptance/bundle/dms/record-failure/output.txt @@ -49,13 +49,13 @@ Plan: 1 to add, 0 to change, 0 to delete, 0 unchanged }, "body": { "cli_version": "[CLI_VERSION]", - "version_type": "VERSION_TYPE_DEPLOY", "operations": [ { - "resource_key": "jobs.doomed", - "action_type": "OPERATION_ACTION_TYPE_CREATE" + "action_type": "OPERATION_ACTION_TYPE_CREATE", + "resource_key": "jobs.doomed" } - ] + ], + "version_type": "VERSION_TYPE_DEPLOY" } } { @@ -66,7 +66,7 @@ Plan: 1 to add, 0 to change, 0 to delete, 0 unchanged }, "body": { "error_message": "Node type nonexistent-node-type is not supported. Supported node types: [NODE_TYPES] (400 INVALID_PARAMETER_VALUE)", - "sequence_id": "0", + "sequence_id": 0, "status": "OPERATION_STATUS_FAILED" } } diff --git a/acceptance/bundle/dms/stale-plan/output.txt b/acceptance/bundle/dms/stale-plan/output.txt index 8c812b8727c..b96fa7dddf7 100644 --- a/acceptance/bundle/dms/stale-plan/output.txt +++ b/acceptance/bundle/dms/stale-plan/output.txt @@ -53,8 +53,8 @@ Error: plan lineage "" does not match state lineage "[UUID]"; the state may have === Only the first deploy recorded a version; the rejected replay created none >>> print_requests.py --dms //api/2.0/bundle --oneline {"method": "POST", "path": "/api/2.0/bundle/deployments", "body": {"display_name": "dms-stale-plan-[UNIQUE_NAME]", "initial_parent_path": "/Workspace/Users/[USERNAME]/.bundle/dms-stale-plan-[UNIQUE_NAME]/default/state", "target_name": "default", "workspace_info": {"file_path": "/Workspace/Users/[USERNAME]/.bundle/dms-stale-plan-[UNIQUE_NAME]/default/files", "root_path": "/Workspace/Users/[USERNAME]/.bundle/dms-stale-plan-[UNIQUE_NAME]/default"}}} -{"method": "POST", "path": "/api/2.0/bundle/deployments/[NUMID]/versions", "q": {"version_id": "1"}, "body": {"cli_version": "[CLI_VERSION]", "version_type": "VERSION_TYPE_DEPLOY", "operations": [{"resource_key": "jobs.foo", "action_type": "OPERATION_ACTION_TYPE_CREATE"}]}} -{"method": "PATCH", "path": "/api/2.0/bundle/deployments/[NUMID]/versions/1/operations/jobs.foo", "q": {"update_mask": "state,error_message,resource_id,status"}, "body": {"error_message": "", "resource_id": "[NUMID]", "sequence_id": "0", "state": "{\"state\":{\"deployment\":{\"deployment_id\":\"[NUMID]\",\"kind\":\"BUNDLE\",\"metadata_file_path\":\"/Workspace/Users/[USERNAME]/.bundle/dms-stale-plan-[UNIQUE_NAME]/default/state/metadata.json\",\"version_id\":\"1\"},\"edit_mode\":\"UI_LOCKED\",\"format\":\"MULTI_TASK\",\"max_concurrent_runs\":1,\"name\":\"foo\",\"queue\":{\"enabled\":true}}}", "status": "OPERATION_STATUS_SUCCEEDED"}} +{"method": "POST", "path": "/api/2.0/bundle/deployments/[NUMID]/versions", "q": {"version_id": "1"}, "body": {"cli_version": "[CLI_VERSION]", "operations": [{"action_type": "OPERATION_ACTION_TYPE_CREATE", "resource_key": "jobs.foo"}], "version_type": "VERSION_TYPE_DEPLOY"}} +{"method": "PATCH", "path": "/api/2.0/bundle/deployments/[NUMID]/versions/1/operations/jobs.foo", "q": {"update_mask": "state,error_message,resource_id,status"}, "body": {"error_message": "", "resource_id": "[NUMID]", "sequence_id": 0, "state": "{\"state\":{\"deployment\":{\"deployment_id\":\"[NUMID]\",\"kind\":\"BUNDLE\",\"metadata_file_path\":\"/Workspace/Users/[USERNAME]/.bundle/dms-stale-plan-[UNIQUE_NAME]/default/state/metadata.json\",\"version_id\":\"1\"},\"edit_mode\":\"UI_LOCKED\",\"format\":\"MULTI_TASK\",\"max_concurrent_runs\":1,\"name\":\"foo\",\"queue\":{\"enabled\":true}}}", "status": "OPERATION_STATUS_SUCCEEDED"}} {"method": "POST", "path": "/api/2.0/bundle/deployments/[NUMID]/versions/1/complete", "body": {"completion_reason": "VERSION_COMPLETE_SUCCESS"}} === Save a plan against the recorded deployment, then move the deployment on diff --git a/acceptance/bundle/dms/successful-recreate/output.txt b/acceptance/bundle/dms/successful-recreate/output.txt index 0ca7c7bea3c..56eedb8531a 100644 --- a/acceptance/bundle/dms/successful-recreate/output.txt +++ b/acceptance/bundle/dms/successful-recreate/output.txt @@ -28,13 +28,13 @@ Resources: 1 created, 0 changed, 0 deleted, 0 unchanged }, "body": { "cli_version": "[CLI_VERSION]", - "version_type": "VERSION_TYPE_DEPLOY", "operations": [ { - "resource_key": "schemas.foo", - "action_type": "OPERATION_ACTION_TYPE_CREATE" + "action_type": "OPERATION_ACTION_TYPE_CREATE", + "resource_key": "schemas.foo" } - ] + ], + "version_type": "VERSION_TYPE_DEPLOY" } } { @@ -46,7 +46,7 @@ Resources: 1 created, 0 changed, 0 deleted, 0 unchanged "body": { "error_message": "", "resource_id": "main.dms_successful_recreate_schema_[UNIQUE_NAME]", - "sequence_id": "0", + "sequence_id": 0, "state": "{\"state\":{\"catalog_name\":\"main\",\"comment\":\"v1\",\"name\":\"dms_successful_recreate_schema_[UNIQUE_NAME]\"}}", "status": "OPERATION_STATUS_SUCCEEDED" } @@ -117,14 +117,14 @@ Resources: 1 created, 0 changed, 1 deleted, 0 unchanged }, "body": { "cli_version": "[CLI_VERSION]", - "version_type": "VERSION_TYPE_DEPLOY", - "previous_version_id": "1", "operations": [ { - "resource_key": "schemas.foo", - "action_type": "OPERATION_ACTION_TYPE_RECREATE" + "action_type": "OPERATION_ACTION_TYPE_RECREATE", + "resource_key": "schemas.foo" } - ] + ], + "previous_version_id": "1", + "version_type": "VERSION_TYPE_DEPLOY" } } { @@ -135,7 +135,7 @@ Resources: 1 created, 0 changed, 1 deleted, 0 unchanged }, "body": { "error_message": "", - "sequence_id": "0", + "sequence_id": 0, "status": "OPERATION_STATUS_PENDING" } } @@ -148,7 +148,7 @@ Resources: 1 created, 0 changed, 1 deleted, 0 unchanged "body": { "error_message": "", "resource_id": "dms_other_[UNIQUE_NAME].dms_successful_recreate_schema_[UNIQUE_NAME]", - "sequence_id": "1", + "sequence_id": 1, "state": "{\"state\":{\"catalog_name\":\"dms_other_[UNIQUE_NAME]\",\"comment\":\"v1\",\"name\":\"dms_successful_recreate_schema_[UNIQUE_NAME]\"}}", "status": "OPERATION_STATUS_SUCCEEDED" } @@ -219,14 +219,14 @@ Destroy: 1 deleted }, "body": { "cli_version": "[CLI_VERSION]", - "version_type": "VERSION_TYPE_DESTROY", - "previous_version_id": "2", "operations": [ { - "resource_key": "schemas.foo", - "action_type": "OPERATION_ACTION_TYPE_DELETE" + "action_type": "OPERATION_ACTION_TYPE_DELETE", + "resource_key": "schemas.foo" } - ] + ], + "previous_version_id": "2", + "version_type": "VERSION_TYPE_DESTROY" } } { @@ -237,7 +237,7 @@ Destroy: 1 deleted }, "body": { "error_message": "", - "sequence_id": "0", + "sequence_id": 0, "status": "OPERATION_STATUS_SUCCEEDED" } } diff --git a/acceptance/bundle/dms/version-never-created/output.txt b/acceptance/bundle/dms/version-never-created/output.txt index 40832a01d1e..ebc07afee93 100644 --- a/acceptance/bundle/dms/version-never-created/output.txt +++ b/acceptance/bundle/dms/version-never-created/output.txt @@ -32,7 +32,7 @@ Plan: 1 to add, 0 to change, 0 to delete, 0 unchanged >>> print_requests.py --dms //api/2.0/bundle --get --oneline {"method": "POST", "path": "/api/2.0/bundle/deployments", "body": {"display_name": "dms-version-never-created", "initial_parent_path": "/Workspace/Users/[USERNAME]/.bundle/dms-version-never-created/default/state", "target_name": "default", "workspace_info": {"file_path": "/Workspace/Users/[USERNAME]/.bundle/dms-version-never-created/default/files", "root_path": "/Workspace/Users/[USERNAME]/.bundle/dms-version-never-created/default"}}} -{"method": "POST", "path": "/api/2.0/bundle/deployments/[NUMID]/versions", "q": {"version_id": "1"}, "body": {"cli_version": "[CLI_VERSION]", "version_type": "VERSION_TYPE_DEPLOY", "operations": [{"resource_key": "jobs.foo", "action_type": "OPERATION_ACTION_TYPE_CREATE"}]}} +{"method": "POST", "path": "/api/2.0/bundle/deployments/[NUMID]/versions", "q": {"version_id": "1"}, "body": {"cli_version": "[CLI_VERSION]", "operations": [{"action_type": "OPERATION_ACTION_TYPE_CREATE", "resource_key": "jobs.foo"}], "version_type": "VERSION_TYPE_DEPLOY"}} {"method": "GET", "path": "/api/2.0/bundle/deployments/[NUMID]"} {"method": "GET", "path": "/api/2.0/bundle/deployments/[NUMID]/resources"} {"method": "GET", "path": "/api/2.0/bundle/deployments/[NUMID]"} @@ -57,4 +57,4 @@ Files: 0 uploaded, 0 deleted {"method": "GET", "path": "/api/2.0/bundle/deployments/[NUMID]/resources"} {"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": "1"}, "body": {"cli_version": "[CLI_VERSION]", "version_type": "VERSION_TYPE_DEPLOY", "operations": [{"resource_key": "jobs.foo", "action_type": "OPERATION_ACTION_TYPE_CREATE"}]}} +{"method": "POST", "path": "/api/2.0/bundle/deployments/[NUMID]/versions", "q": {"version_id": "1"}, "body": {"cli_version": "[CLI_VERSION]", "operations": [{"action_type": "OPERATION_ACTION_TYPE_CREATE", "resource_key": "jobs.foo"}], "version_type": "VERSION_TYPE_DEPLOY"}} diff --git a/libs/dms/client.go b/libs/dms/client.go index 605b4bbd4db..5bfde8f8378 100644 --- a/libs/dms/client.go +++ b/libs/dms/client.go @@ -3,13 +3,11 @@ package dms import ( "context" "fmt" - "net/http" "strconv" "strings" - "github.com/databricks/cli/libs/auth" "github.com/databricks/databricks-sdk-go" - "github.com/databricks/databricks-sdk-go/client" + "github.com/databricks/databricks-sdk-go/common/types/fieldmask" "github.com/databricks/databricks-sdk-go/service/bundledeployments" ) @@ -22,26 +20,16 @@ const DeploymentNodeName = "resources.deployment.json" // form; the prefix comes off where a request is built, and back on where a resource is read. const statePrefix = "resources." -// Client carries the calls the CLI makes to DMS, as methods below. Each one goes out through one -// of two halves: the generated client for the calls it can express, and hand-written requests -// for the two it cannot. -// -// TODO: Remove this and replace with the SDK. +// Client carries the calls the CLI makes to DMS, as methods below. Each one goes out through the +// generated SDK client. type Client struct { // Service is the generated client. Service bundledeployments.BundleDeploymentsInterface - - // raw sends what the generated client cannot; see requester. - raw requester } // NewClient returns a Client for the workspace w. func NewClient(w *databricks.WorkspaceClient) (*Client, error) { - api, err := client.New(w.Config) - if err != nil { - return nil, err - } - return &Client{Service: w.BundleDeployments, raw: &rawClient{client: api}}, nil + return &Client{Service: w.BundleDeployments}, nil } // DeploymentName and versionName are the two resource-name formats the service uses. Every @@ -70,7 +58,12 @@ func (c *Client) CreateDeployment(ctx context.Context, parentPath string, metada // UpdateDeployment writes the fields mask names onto the deployment. The service ignores every // other field, so the mask is what decides the write. func (c *Client) UpdateDeployment(ctx context.Context, deploymentID string, metadata Metadata, mask string) error { - return c.raw.UpdateDeployment(ctx, deploymentID, metadata.deployment(), mask) + _, err := c.Service.UpdateDeployment(ctx, bundledeployments.UpdateDeploymentRequest{ + Name: DeploymentName(deploymentID), + Deployment: newDeploymentUpdate(metadata, mask), + UpdateMask: fieldmask.FieldMask{Paths: strings.Split(mask, ",")}, + }) + return err } // DeleteDeployment removes the deployment record, which a completed destroy does. @@ -82,7 +75,25 @@ func (c *Client) DeleteDeployment(ctx context.Context, deploymentID string) erro // CreateVersion claims the version and stages the operations body carries. func (c *Client) CreateVersion(ctx context.Context, deploymentID string, version int, body CreateVersionRequest) (*bundledeployments.Version, error) { - return c.raw.CreateVersion(ctx, deploymentID, strconv.Itoa(version), body) + operations := make([]bundledeployments.StagedOperation, len(body.Operations)) + for i, op := range body.Operations { + operations[i] = bundledeployments.StagedOperation{ + ActionType: op.ActionType, + ResourceKey: strings.TrimPrefix(op.ResourceKey, statePrefix), + } + } + + return c.Service.CreateVersion(ctx, bundledeployments.CreateVersionRequest{ + Parent: DeploymentName(deploymentID), + VersionId: strconv.Itoa(version), + Version: bundledeployments.Version{ + CliVersion: body.CliVersion, + VersionType: body.VersionType, + PreviousVersionId: body.PreviousVersionId, + GitInfo: body.GitInfo, + Operations: operations, + }, + }) } // CompleteVersion closes the version out, which is what stops the service expiring its lease. @@ -97,7 +108,20 @@ func (c *Client) CompleteVersion(ctx context.Context, deploymentID string, versi // UpdateOperation fills in one operation the version staged, and returns the sequence id the // next update for that resource must send. func (c *Client) UpdateOperation(ctx context.Context, deploymentID string, version int, stateKey, sequenceID string, update OperationUpdate) (string, error) { - return c.raw.UpdateOperation(ctx, deploymentID, version, stateKey, sequenceID, update) + operation, err := newOperationUpdate(update, sequenceID) + if err != nil { + return "", err + } + + result, err := c.Service.UpdateOperation(ctx, bundledeployments.UpdateOperationRequest{ + Name: versionName(deploymentID, version) + "/operations/" + strings.TrimPrefix(stateKey, statePrefix), + Operation: operation, + UpdateMask: fieldmask.FieldMask{Paths: strings.Split(update.Fields.Mask(), ",")}, + }) + if err != nil { + return "", err + } + return strconv.FormatInt(result.SequenceId, 10), nil } // deploymentIDFromName extracts the deployment ID from a DMS resource name of @@ -110,149 +134,85 @@ func deploymentIDFromName(name string) (string, error) { return id, nil } -// requester sends the two requests the generated client cannot express, so a test can capture -// what the CLI puts on the wire. Both are TODO(DMS): drop them once the spec catches up. -type requester interface { - // CreateVersion is hand-written because the generated struct has no operations: the field - // is at DEVELOPMENT stage, which keeps it out of the SDK until it is promoted. - CreateVersion(ctx context.Context, deploymentID, versionID string, body CreateVersionRequest) (*bundledeployments.Version, error) - - // UpdateDeployment is hand-written because the generated client has no such call yet. - UpdateDeployment(ctx context.Context, deploymentID string, deployment bundledeployments.Deployment, mask string) error - - // UpdateOperation is hand-written because the SDK types sequence_id as an int64 while - // the service sends a JSON string, so it cannot read the response. sequenceID is the - // token the previous update for this resource returned, or 0 for the first, which is - // what staging leaves. - UpdateOperation(ctx context.Context, deploymentID string, version int, stateKey, sequenceID string, update OperationUpdate) (next string, err error) -} - -// CreateVersionRequest is the CreateVersion request body. +// CreateVersionRequest is the input to Client.CreateVersion. type CreateVersionRequest struct { - CliVersion string `json:"cli_version"` - VersionType VersionType `json:"version_type"` + CliVersion string + VersionType VersionType // PreviousVersionId is the deployment's most recent version, unset for a // deployment's first version. - PreviousVersionId string `json:"previous_version_id,omitempty"` + PreviousVersionId string // GitInfo records where this version's source came from. The rest of the provenance - // display name, target, mode, workspace paths - belongs to the deployment. - GitInfo *bundledeployments.GitInfo `json:"git_info,omitempty"` - // Operations is every resource this version will touch; see StagedOperation. It sits in this - // body with the version's own fields because the request binds body: "version", and is input - // only - the response never carries it back. - Operations []StagedOperation `json:"operations,omitempty"` + GitInfo *bundledeployments.GitInfo + // Operations is every resource this version will touch; see StagedOperation. + Operations []StagedOperation } // StagedOperation is one resource the version will record an operation for. The service // creates it in OPERATION_STATUS_PENDING at sequence id 0, and the CLI fills in the outcome // with UpdateOperation as the resource is applied. type StagedOperation struct { - // ResourceKey is the bundle state key; the request carries the form the service uses. - ResourceKey string `json:"resource_key"` - ActionType bundledeployments.OperationActionType `json:"action_type"` -} - -// operationResponse is the part of an operation response the CLI reads back. -type operationResponse struct { - // SequenceId is the concurrency token for the next update, typed as the service sends it. - SequenceId string `json:"sequence_id,omitempty"` -} - -// rawClient sends the three DMS requests the generated SDK client cannot express yet, while the -// endpoints are at DEVELOPMENT stage. TODO(DMS): drop each once the SDK catches up. -// - UpdateDeployment: the SDK exposes no method for it. -// - CreateVersion: the SDK's Version type has no operations field, so a version cannot stage -// its operations through it. -// - UpdateOperation: the SDK types sequence_id as int64, but the service sends and expects it -// as a JSON string. -// -// CreateDeployment, DeleteDeployment, CompleteVersion and ListResources go through the SDK (see Client). -type rawClient struct { - client *client.DatabricksClient -} - -func (r *rawClient) CreateVersion(ctx context.Context, deploymentID, versionID string, body CreateVersionRequest) (*bundledeployments.Version, error) { - staged := make([]StagedOperation, len(body.Operations)) - for i, op := range body.Operations { - op.ResourceKey = strings.TrimPrefix(op.ResourceKey, statePrefix) - staged[i] = op - } - body.Operations = staged - - var version bundledeployments.Version - path := "/api/2.0/bundle/" + DeploymentName(deploymentID) + "/versions" - err := r.client.Do(ctx, http.MethodPost, path, - auth.WorkspaceIDHeaders(r.client.Config), - map[string]any{"version_id": versionID}, - body, &version) - if err != nil { - return nil, err + // ResourceKey is the bundle state key; CreateVersion strips the prefix to the form the service uses. + ResourceKey string + ActionType bundledeployments.OperationActionType +} + +// newDeploymentUpdate builds the deployment carrying exactly the masked fields, empty ones +// included: the service requires every masked field to be present in the body and reads an empty +// value as a clear (a target that stops setting mode clears deployment_mode). ForceSendFields +// keeps those empty values on the wire, which omitempty would drop. +func newDeploymentUpdate(metadata Metadata, mask string) bundledeployments.Deployment { + full := metadata.deployment() + var dep bundledeployments.Deployment + for path := range strings.SplitSeq(mask, ",") { + switch path { + case "display_name": + dep.DisplayName = full.DisplayName + dep.ForceSendFields = append(dep.ForceSendFields, "DisplayName") + case "target_name": + dep.TargetName = full.TargetName + dep.ForceSendFields = append(dep.ForceSendFields, "TargetName") + case "deployment_mode": + dep.DeploymentMode = full.DeploymentMode + dep.ForceSendFields = append(dep.ForceSendFields, "DeploymentMode") + case "workspace_info": + dep.WorkspaceInfo = full.WorkspaceInfo + dep.ForceSendFields = append(dep.ForceSendFields, "WorkspaceInfo") + } } - return &version, nil + return dep } -// newDeploymentUpdate builds the request body for update, holding exactly the masked fields for -// the same reason newUpdateRequest does. A map, not the SDK struct, whose omitempty tags would -// drop a masked field that is empty - which is how deployment_mode is cleared when a target stops -// setting mode. -func newDeploymentUpdate(deployment bundledeployments.Deployment, mask string) map[string]any { - values := map[string]any{ - "display_name": deployment.DisplayName, - "target_name": deployment.TargetName, - "deployment_mode": deployment.DeploymentMode, - "workspace_info": deployment.WorkspaceInfo, +// newOperationUpdate builds the operation carrying exactly the fields update.Fields masks, plus +// the sequence_id precondition. The service requires a masked field to be present and reads an +// empty value as a write (error_message="" clears it), so masked fields that can be empty are +// forced onto the wire; sequence_id is always sent and a freshly staged operation sits at 0, +// which omitempty would drop. State is the exception: an absent value is how the service is told +// the resource is gone, so a nil state is left off (never forced) while still named in the mask. +func newOperationUpdate(update OperationUpdate, sequenceID string) (bundledeployments.Operation, error) { + sequence, err := strconv.ParseInt(sequenceID, 10, 64) + if err != nil { + return bundledeployments.Operation{}, fmt.Errorf("invalid sequence id %q: %w", sequenceID, err) } - body := map[string]any{} - for field := range strings.SplitSeq(mask, ",") { - body[field] = values[field] + operation := bundledeployments.Operation{ + SequenceId: sequence, + ForceSendFields: []string{"SequenceId"}, } - return body -} - -func (r *rawClient) UpdateDeployment(ctx context.Context, deploymentID string, deployment bundledeployments.Deployment, mask string) error { - path := "/api/2.0/bundle/" + DeploymentName(deploymentID) - return r.client.Do(ctx, http.MethodPatch, path, - auth.WorkspaceIDHeaders(r.client.Config), - map[string]any{"update_mask": mask}, - newDeploymentUpdate(deployment, mask), nil) -} - -// newUpdateRequest builds the request body for update. A field is in the body when the mask -// names it and absent otherwise, which is what the service requires: it rejects an update -// whose mask names a field the body leaves out, and an empty value is how a field is cleared - -// no state means the resource is gone, no error_message means an earlier failure is resolved. -// A map, not a struct, so presence cannot drift from the mask through an omitempty tag. -func newUpdateRequest(update OperationUpdate, sequenceID string) map[string]any { - body := map[string]any{"sequence_id": sequenceID} - // A masked state with no value is how the service is told the resource is gone, so a nil - // state is left out rather than sent empty. if update.Fields.Has(FieldState) && update.State != nil { - body["state"] = string(update.State) - } - if update.Fields.Has(FieldResourceID) { - body["resource_id"] = update.ResourceID + operation.State = string(update.State) } if update.Fields.Has(FieldErrorMessage) { - body["error_message"] = update.ErrorMessage + operation.ErrorMessage = update.ErrorMessage + operation.ForceSendFields = append(operation.ForceSendFields, "ErrorMessage") } - if update.Fields.Has(FieldStatus) { - body["status"] = update.Status + if update.Fields.Has(FieldResourceID) { + operation.ResourceId = update.ResourceID + operation.ForceSendFields = append(operation.ForceSendFields, "ResourceId") } - return body -} - -func (r *rawClient) UpdateOperation(ctx context.Context, deploymentID string, version int, stateKey, sequenceID string, update OperationUpdate) (string, error) { - body := newUpdateRequest(update, sequenceID) - - var result operationResponse - path := "/api/2.0/bundle/" + versionName(deploymentID, version) + "/operations/" + strings.TrimPrefix(stateKey, statePrefix) - err := r.client.Do(ctx, http.MethodPatch, path, - auth.WorkspaceIDHeaders(r.client.Config), - map[string]any{"update_mask": update.Fields.Mask()}, - body, &result) - if err != nil { - return "", err + if update.Fields.Has(FieldStatus) { + operation.Status = update.Status + operation.ForceSendFields = append(operation.ForceSendFields, "Status") } - return result.SequenceId, nil + return operation, nil } diff --git a/libs/dms/client_test.go b/libs/dms/client_test.go index 867655bca4e..0888829e50d 100644 --- a/libs/dms/client_test.go +++ b/libs/dms/client_test.go @@ -30,9 +30,9 @@ func TestDeploymentIDFromName(t *testing.T) { func TestUpdateRequestSendsExactlyTheMaskedFields(t *testing.T) { // The service rejects an update whose mask names a field the body leaves out, and treats a - // field it does carry as written - so the body has to hold every masked field and nothing - // else, empty values included. Both masks the CLI builds are asserted on the wire by - // acceptance/bundle/dms; this pins the rule they both rely on. + // field it does carry as written - so the body has to hold every masked field, empty values + // included, which ForceSendFields is what secures. Both bodies the CLI builds are asserted on + // the wire by acceptance/bundle/dms; this pins the rule they both rely on. update := OperationUpdate{ Fields: DescribesResource, State: json.RawMessage(`{"state":{"name":"foo"}}`), @@ -41,27 +41,40 @@ func TestUpdateRequestSendsExactlyTheMaskedFields(t *testing.T) { } // A successful write reports no error, and the empty value is what clears an earlier one. - assert.Equal(t, map[string]any{ - "state": `{"state":{"name":"foo"}}`, - "resource_id": "job-1", + operation, err := newOperationUpdate(update, "3") + require.NoError(t, err) + assert.JSONEq(t, `{ + "sequence_id": 3, + "state": "{\"state\":{\"name\":\"foo\"}}", + "resource_id": "job-1", "error_message": "", - "status": bundledeployments.OperationStatusOperationStatusSucceeded, - "sequence_id": "3", - }, newUpdateRequest(update, "3")) + "status": "OPERATION_STATUS_SUCCEEDED" + }`, marshalBody(t, operation)) // A failure keeps the recorded state, so state is absent rather than empty: naming it // would clear what the resource last recorded. - assert.Equal(t, map[string]any{ + operation, err = newOperationUpdate(NewFailureUpdate("job-1", nil, errors.New("boom")), "3") + require.NoError(t, err) + assert.JSONEq(t, `{ + "sequence_id": 3, "error_message": "boom", - "status": bundledeployments.OperationStatusOperationStatusFailed, - "sequence_id": "3", - }, newUpdateRequest(NewFailureUpdate("job-1", nil, errors.New("boom")), "3")) + "status": "OPERATION_STATUS_FAILED" + }`, marshalBody(t, operation)) // The deployment's own fields follow the same rule. Clearing deployment_mode - a target that // stops setting mode - sends it empty, which the SDK struct's omitempty would have dropped. - deployment := Metadata{DisplayName: "b", TargetName: "t"}.deployment() - assert.Equal(t, map[string]any{ - "target_name": "t", - "deployment_mode": bundledeployments.DeploymentMode(""), - }, newDeploymentUpdate(deployment, "target_name,deployment_mode")) + deployment := newDeploymentUpdate(Metadata{DisplayName: "b", TargetName: "t"}, "target_name,deployment_mode") + assert.JSONEq(t, `{ + "target_name": "t", + "deployment_mode": "" + }`, marshalBody(t, deployment)) +} + +// marshalBody renders the request body the SDK sends for v, so a test sees the wire the +// ForceSendFields produce. +func marshalBody(t *testing.T, v any) string { + t.Helper() + body, err := json.Marshal(v) + require.NoError(t, err) + return string(body) } diff --git a/libs/testserver/bundledeployments.go b/libs/testserver/bundledeployments.go index 28e009793a0..0d6f03054ed 100644 --- a/libs/testserver/bundledeployments.go +++ b/libs/testserver/bundledeployments.go @@ -387,8 +387,9 @@ func (s *FakeWorkspace) Heartbeat() Response { return Response{Body: bundledeployments.HeartbeatResponse{}} } -// operationBody renders an operation the way the service does: sequence_id as a -// JSON string, which the SDK struct cannot express (it types the field int64). +// operationBody renders an operation the way the service does: sequence_id as a JSON string. The +// SDK reads it back into its int64 field through its number-or-string normalizer, so the fake +// keeps emitting a string to mirror the real service's proto3 JSON. func operationBody(op *bundledeployments.Operation) (map[string]any, error) { raw, err := json.Marshal(op) if err != nil { @@ -409,26 +410,16 @@ func operationBody(op *bundledeployments.Operation) (map[string]any, error) { // UpdateOperation applies a later write for a resource already recorded in this // version. sequence_id is the concurrency precondition and increments on success. func (s *FakeWorkspace) UpdateOperation(req Request, deploymentID, versionID, resourceKey string) Response { - // sequence_id arrives as a string, which the SDK struct cannot hold (it types the - // field int64), so read the body twice: once for the typed fields with that key - // removed, and once for the precondition alone. + // raw records which fields the body carried, which the update_mask rules turn on: a masked + // path with no value is how a field is cleared. var raw map[string]json.RawMessage if err := json.Unmarshal(req.Body, &raw); err != nil { return Response{StatusCode: 400, Body: map[string]string{"message": err.Error()}} } - var precondition struct { - SequenceId string `json:"sequence_id"` - } - if err := json.Unmarshal(req.Body, &precondition); err != nil { - return Response{StatusCode: 400, Body: map[string]string{"message": err.Error()}} - } - delete(raw, "sequence_id") - typedBody, err := json.Marshal(raw) - if err != nil { - return Response{StatusCode: 500, Body: map[string]string{"message": err.Error()}} - } + // The SDK sends sequence_id as a JSON number and the real service as a string; Operation reads + // either into its int64 field, so the whole body decodes straight into it. var op bundledeployments.Operation - if err := json.Unmarshal(typedBody, &op); err != nil { + if err := json.Unmarshal(req.Body, &op); err != nil { return Response{StatusCode: 400, Body: map[string]string{"message": err.Error()}} } @@ -493,7 +484,7 @@ func (s *FakeWorkspace) UpdateOperation(req Request, deploymentID, versionID, re if !ok { return dmsNotFound("operation " + opName) } - if precondition.SequenceId != strconv.FormatInt(existing.SequenceId, 10) { + if op.SequenceId != existing.SequenceId { return dmsAborted("sequence_id is outdated; the operation is at " + strconv.FormatInt(existing.SequenceId, 10)) } From 10f66706b0bcbf1ab6acc4e65412565fb7387d73 Mon Sep 17 00:00:00 2001 From: Shreyas Goenka Date: Thu, 10 Sep 2026 09:05:52 +0000 Subject: [PATCH 2/6] Remove the dms.Client wrapper; call the SDK directly Delete the Client struct and client.go. DeploymentState and OperationBuffer now hold bundledeployments.BundleDeploymentsInterface, and each SDK call is made at its call site. libs/dms keeps only the non-trivial builders that would otherwise push wire details into caller packages: the ForceSendFields update-mask builders, ListResources, and the name/prefix helpers. Pure refactor - acceptance goldens are unchanged. Co-authored-by: Isaac --- bundle/direct/dstate/state.go | 28 ++--- bundle/phases/destroy.go | 5 +- bundle/phases/dms.go | 51 +++++--- bundle/phases/dms_test.go | 15 ++- libs/dms/client.go | 218 ---------------------------------- libs/dms/client_test.go | 80 ------------- libs/dms/metadata.go | 56 ++++++++- libs/dms/metadata_test.go | 18 +++ libs/dms/operation.go | 40 +++++++ libs/dms/operation_buffer.go | 23 +++- libs/dms/operation_test.go | 51 ++++++++ libs/dms/resources.go | 6 +- 12 files changed, 241 insertions(+), 350 deletions(-) delete mode 100644 libs/dms/client.go delete mode 100644 libs/dms/client_test.go diff --git a/bundle/direct/dstate/state.go b/bundle/direct/dstate/state.go index 9f347750659..6acb1d4f45e 100644 --- a/bundle/direct/dstate/state.go +++ b/bundle/direct/dstate/state.go @@ -98,9 +98,9 @@ type DeploymentState struct { // history, in which case InitializeOperationBuffer installs it once the version exists. operationBuffer *dms.OperationBuffer - // dmsClient talks to the deployment metadata service. Open builds it from the workspace + // dmsService talks to the deployment metadata service. Open sets it from the workspace // client when the deployment records history; nil otherwise. - dmsClient *dms.Client + dmsService bundledeployments.BundleDeploymentsInterface // versionCompleted makes CompleteVersion a no-op after the first call, so a deferred safety-net // completion after an explicit one does nothing. @@ -180,7 +180,7 @@ type WALEntry struct { // (after approval) - which is why it is not an Open option. It also records the id a first deploy // just created, which Open could not know, so CompleteVersion later has it. func (db *DeploymentState) InitializeOperationBuffer(ctx context.Context, deploymentID string, versionID int) { - buf := dms.StartOperationBuffer(ctx, db.dmsClient, deploymentID, versionID) + buf := dms.StartOperationBuffer(ctx, db.dmsService, deploymentID, versionID) db.mu.Lock() defer db.mu.Unlock() @@ -224,7 +224,7 @@ func (db *DeploymentState) CompleteVersion(ctx context.Context, success bool) (b return false, nil } db.versionCompleted = true - deploymentID, client, versionID := db.DeploymentID, db.dmsClient, db.VersionID + deploymentID, service, versionID := db.DeploymentID, db.dmsService, db.VersionID db.mu.Unlock() // A recording failure fails the version even when the caller counted the deploy a success: the @@ -238,7 +238,11 @@ func (db *DeploymentState) CompleteVersion(ctx context.Context, success bool) (b if !success { reason = bundledeployments.VersionCompleteVersionCompleteFailure } - if err := client.CompleteVersion(ctx, deploymentID, versionID, reason); err != nil { + _, err := service.CompleteVersion(ctx, bundledeployments.CompleteVersionRequest{ + Name: dms.VersionName(deploymentID, versionID), + CompletionReason: reason, + }) + if err != nil { return false, err } log.Infof(ctx, "Completed deployment version: deployment=%s version=%d reason=%s", deploymentID, versionID, reason) @@ -426,10 +430,10 @@ func (db *DeploymentState) IsDeploymentMetadataService() bool { return db.isDeploymentMetadataService() } -// DmsClient returns the deployment metadata service client Open built from the workspace client, +// DmsService returns the deployment metadata service client Open set from the workspace client, // or nil when the deployment does not record history. -func (db *DeploymentState) DmsClient() *dms.Client { - return db.dmsClient +func (db *DeploymentState) DmsService() bundledeployments.BundleDeploymentsInterface { + return db.dmsService } // GetOrInitLineage returns the deployment lineage, generating and storing a new @@ -584,15 +588,11 @@ Run "databricks bundle destroy" first, then deploy again with deployment history if recorded { // The service is the source of truth for a recorded deployment; the file is a tombstone // carrying only the marker, and applyDMSState loads the resources the service holds. - client, err := dms.NewClient(cmdctx.WorkspaceClient(ctx)) - if err != nil { - return err - } - db.dmsClient = client + db.dmsService = cmdctx.WorkspaceClient(ctx).BundleDeployments db.DeploymentID = dmsDeployment.DeploymentID if dmsDeployment.DeploymentID != "" { - resources, err := db.dmsClient.ListResources(ctx, dmsDeployment.DeploymentID) + resources, err := dms.ListResources(ctx, db.dmsService, dmsDeployment.DeploymentID) if err != nil { return err } diff --git a/bundle/phases/destroy.go b/bundle/phases/destroy.go index d8ed0f75c02..18b00b6235d 100644 --- a/bundle/phases/destroy.go +++ b/bundle/phases/destroy.go @@ -20,6 +20,7 @@ import ( "github.com/databricks/cli/libs/log" "github.com/databricks/cli/libs/logdiag" "github.com/databricks/databricks-sdk-go/apierr" + "github.com/databricks/databricks-sdk-go/service/bundledeployments" ) func assertRootPathExists(ctx context.Context, b *bundle.Bundle) (bool, error) { @@ -166,7 +167,7 @@ func destroyCore(ctx context.Context, b *bundle.Bundle, plan *deployplan.Plan) { // A completed destroy's resources are gone, so its deployment record is deleted too. if completed { deploymentID := b.DeploymentBundle.StateDB.DeploymentID - if err := b.DeploymentBundle.StateDB.DmsClient().DeleteDeployment(ctx, deploymentID); err != nil { + if err := b.DeploymentBundle.StateDB.DmsService().DeleteDeployment(ctx, bundledeployments.DeleteDeploymentRequest{Name: dms.DeploymentName(deploymentID)}); err != nil { logdiag.LogError(ctx, fmt.Errorf("failed to delete deployment: %w", err)) return } @@ -298,7 +299,7 @@ func Destroy(ctx context.Context, b *bundle.Bundle) { logdiag.LogError(ctx, err) } else if completed { deploymentID := b.DeploymentBundle.StateDB.DeploymentID - if err := b.DeploymentBundle.StateDB.DmsClient().DeleteDeployment(ctx, deploymentID); err != nil { + if err := b.DeploymentBundle.StateDB.DmsService().DeleteDeployment(ctx, bundledeployments.DeleteDeploymentRequest{Name: dms.DeploymentName(deploymentID)}); err != nil { logdiag.LogError(ctx, fmt.Errorf("failed to delete deployment: %w", err)) } } diff --git a/bundle/phases/dms.go b/bundle/phases/dms.go index 9ceea29509d..eda1abc7d7f 100644 --- a/bundle/phases/dms.go +++ b/bundle/phases/dms.go @@ -16,15 +16,16 @@ import ( "github.com/databricks/cli/libs/log" "github.com/databricks/cli/libs/logdiag" "github.com/databricks/cli/libs/workspaceurls" + "github.com/databricks/databricks-sdk-go/common/types/fieldmask" "github.com/databricks/databricks-sdk-go/service/bundledeployments" ) // stagedOperations lists the resources the plan will touch, for CreateVersion to stage an // operation each. Skipped and undefined actions are left out: nothing is applied for them, so // their operations would stay pending and the service would hold no state for them. -func stagedOperations(plan *deployplan.Plan) ([]dms.StagedOperation, error) { +func stagedOperations(plan *deployplan.Plan) ([]bundledeployments.StagedOperation, error) { actions := plan.GetActions() - staged := make([]dms.StagedOperation, 0, len(actions)) + staged := make([]bundledeployments.StagedOperation, 0, len(actions)) for _, action := range actions { if action.ActionType == deployplan.Skip || action.ActionType == deployplan.Undefined { continue @@ -33,8 +34,8 @@ func stagedOperations(plan *deployplan.Plan) ([]dms.StagedOperation, error) { if err != nil { return nil, fmt.Errorf("%s: %w", action.ResourceKey, err) } - staged = append(staged, dms.StagedOperation{ - ResourceKey: action.ResourceKey, + staged = append(staged, bundledeployments.StagedOperation{ + ResourceKey: strings.TrimPrefix(action.ResourceKey, dms.StatePrefix), ActionType: actionType, }) } @@ -69,19 +70,31 @@ func actionToSDK(a deployplan.ActionType) (bundledeployments.OperationActionType // no deployment behind; a first deploy's new id is then stamped into the plan (StampDeploymentID). func createOrUpdateDeployment(ctx context.Context, b *bundle.Bundle, current *bundledeployments.Deployment) { db := &b.DeploymentBundle - dmsClient := db.StateDB.DmsClient() + dmsService := db.StateDB.DmsService() metadata := deploymentMetadata(b) deploymentID := db.StateDB.DeploymentID if deploymentID == "" { - id, err := dmsClient.CreateDeployment(ctx, b.Config.Workspace.StatePath, metadata) + dep := metadata.Deployment() + dep.InitialParentPath = b.Config.Workspace.StatePath + created, err := dmsService.CreateDeployment(ctx, bundledeployments.CreateDeploymentRequest{Deployment: dep}) + if err != nil { + logdiag.LogError(ctx, fmt.Errorf("failed to create deployment: %w", err)) + return + } + // The server assigns the id as the workspace node it creates under the parent path. + deploymentID, err = dms.DeploymentIDFromName(created.Name) if err != nil { logdiag.LogError(ctx, fmt.Errorf("failed to create deployment: %w", err)) return } - deploymentID = id db.StateDB.DeploymentID = deploymentID } else if mask := metadata.StaleFields(current); mask != "" { - if err := dmsClient.UpdateDeployment(ctx, deploymentID, metadata, mask); err != nil { + _, err := dmsService.UpdateDeployment(ctx, bundledeployments.UpdateDeploymentRequest{ + Name: dms.DeploymentName(deploymentID), + Deployment: dms.DeploymentUpdate(metadata, mask), + UpdateMask: fieldmask.FieldMask{Paths: strings.Split(mask, ",")}, + }) + if err != nil { logdiag.LogError(ctx, fmt.Errorf("failed to update deployment: %w", err)) return } @@ -96,10 +109,10 @@ func createOrUpdateDeployment(ctx context.Context, b *bundle.Bundle, current *bu // startVersion claims the version the run settled on and opens the buffer that records // each state write under it. Called after approval, so a declined deploy never claims a number. // A no-op when the bundle does not record deployment history. -func startVersion(ctx context.Context, b *bundle.Bundle, versionType dms.VersionType, staged []dms.StagedOperation) error { +func startVersion(ctx context.Context, b *bundle.Bundle, versionType dms.VersionType, staged []bundledeployments.StagedOperation) error { db := &b.DeploymentBundle - dmsClient := db.StateDB.DmsClient() - if dmsClient == nil { + dmsService := db.StateDB.DmsService() + if dmsService == nil { return nil } deploymentID := db.StateDB.DeploymentID @@ -124,12 +137,16 @@ func startVersion(ctx context.Context, b *bundle.Bundle, versionType dms.Version OriginUrl: git.OriginURL, } } - version, err := dmsClient.CreateVersion(ctx, deploymentID, versionID, dms.CreateVersionRequest{ - CliVersion: build.GetInfo().Version, - VersionType: versionType, - PreviousVersionId: previousVersionID, - Operations: staged, - GitInfo: gitInfo, + version, err := dmsService.CreateVersion(ctx, bundledeployments.CreateVersionRequest{ + Parent: dms.DeploymentName(deploymentID), + VersionId: strconv.Itoa(versionID), + Version: bundledeployments.Version{ + CliVersion: build.GetInfo().Version, + VersionType: versionType, + PreviousVersionId: previousVersionID, + GitInfo: gitInfo, + Operations: staged, + }, }) if err != nil { return fmt.Errorf("failed to create deployment version: %w", err) diff --git a/bundle/phases/dms_test.go b/bundle/phases/dms_test.go index f6854999db9..6b045f863ae 100644 --- a/bundle/phases/dms_test.go +++ b/bundle/phases/dms_test.go @@ -4,7 +4,6 @@ import ( "testing" "github.com/databricks/cli/bundle/deployplan" - "github.com/databricks/cli/libs/dms" "github.com/databricks/databricks-sdk-go/service/bundledeployments" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" @@ -23,11 +22,11 @@ func TestStagedOperationsCoversEveryTouchedResource(t *testing.T) { staged, err := stagedOperations(plan) require.NoError(t, err) - assert.ElementsMatch(t, []dms.StagedOperation{ - {ResourceKey: "resources.jobs.foo", ActionType: bundledeployments.OperationActionTypeOperationActionTypeCreate}, - {ResourceKey: "resources.pipelines.bar", ActionType: bundledeployments.OperationActionTypeOperationActionTypeRecreate}, - {ResourceKey: "resources.schemas.baz", ActionType: bundledeployments.OperationActionTypeOperationActionTypeDelete}, - {ResourceKey: "resources.clusters.small", ActionType: bundledeployments.OperationActionTypeOperationActionTypeResize}, + assert.ElementsMatch(t, []bundledeployments.StagedOperation{ + {ResourceKey: "jobs.foo", ActionType: bundledeployments.OperationActionTypeOperationActionTypeCreate}, + {ResourceKey: "pipelines.bar", ActionType: bundledeployments.OperationActionTypeOperationActionTypeRecreate}, + {ResourceKey: "schemas.baz", ActionType: bundledeployments.OperationActionTypeOperationActionTypeDelete}, + {ResourceKey: "clusters.small", ActionType: bundledeployments.OperationActionTypeOperationActionTypeResize}, }, staged) } @@ -43,8 +42,8 @@ func TestStagedOperationsLeavesOutUntouchedResources(t *testing.T) { staged, err := stagedOperations(plan) require.NoError(t, err) - assert.Equal(t, []dms.StagedOperation{ - {ResourceKey: "resources.jobs.touched", ActionType: bundledeployments.OperationActionTypeOperationActionTypeUpdate}, + assert.Equal(t, []bundledeployments.StagedOperation{ + {ResourceKey: "jobs.touched", ActionType: bundledeployments.OperationActionTypeOperationActionTypeUpdate}, }, staged) } diff --git a/libs/dms/client.go b/libs/dms/client.go deleted file mode 100644 index 5bfde8f8378..00000000000 --- a/libs/dms/client.go +++ /dev/null @@ -1,218 +0,0 @@ -package dms - -import ( - "context" - "fmt" - "strconv" - "strings" - - "github.com/databricks/databricks-sdk-go" - "github.com/databricks/databricks-sdk-go/common/types/fieldmask" - "github.com/databricks/databricks-sdk-go/service/bundledeployments" -) - -// DeploymentNodeName is the workspace node DMS creates per deployment. Must -// match DeploymentWhsClient.DEPLOYMENT_NODE_NAME on the service side. -const DeploymentNodeName = "resources.deployment.json" - -// statePrefix is what a bundle state key carries and a DMS resource key does not: state calls a -// job "resources.jobs.foo", DMS calls it "jobs.foo". Every exported name here takes the state -// form; the prefix comes off where a request is built, and back on where a resource is read. -const statePrefix = "resources." - -// Client carries the calls the CLI makes to DMS, as methods below. Each one goes out through the -// generated SDK client. -type Client struct { - // Service is the generated client. - Service bundledeployments.BundleDeploymentsInterface -} - -// NewClient returns a Client for the workspace w. -func NewClient(w *databricks.WorkspaceClient) (*Client, error) { - return &Client{Service: w.BundleDeployments}, nil -} - -// DeploymentName and versionName are the two resource-name formats the service uses. Every -// call builds its name here, so a caller only ever passes ids. -func DeploymentName(deploymentID string) string { - return "deployments/" + deploymentID -} - -func versionName(deploymentID string, version int) string { - return fmt.Sprintf("deployments/%s/versions/%d", deploymentID, version) -} - -// CreateDeployment registers a deployment under parentPath and returns the id the server -// assigned it, which is the id of the workspace node it creates there. -func (c *Client) CreateDeployment(ctx context.Context, parentPath string, metadata Metadata) (string, error) { - dep := metadata.deployment() - dep.InitialParentPath = parentPath - - created, err := c.Service.CreateDeployment(ctx, bundledeployments.CreateDeploymentRequest{Deployment: dep}) - if err != nil { - return "", err - } - return deploymentIDFromName(created.Name) -} - -// UpdateDeployment writes the fields mask names onto the deployment. The service ignores every -// other field, so the mask is what decides the write. -func (c *Client) UpdateDeployment(ctx context.Context, deploymentID string, metadata Metadata, mask string) error { - _, err := c.Service.UpdateDeployment(ctx, bundledeployments.UpdateDeploymentRequest{ - Name: DeploymentName(deploymentID), - Deployment: newDeploymentUpdate(metadata, mask), - UpdateMask: fieldmask.FieldMask{Paths: strings.Split(mask, ",")}, - }) - return err -} - -// DeleteDeployment removes the deployment record, which a completed destroy does. -func (c *Client) DeleteDeployment(ctx context.Context, deploymentID string) error { - return c.Service.DeleteDeployment(ctx, bundledeployments.DeleteDeploymentRequest{ - Name: DeploymentName(deploymentID), - }) -} - -// CreateVersion claims the version and stages the operations body carries. -func (c *Client) CreateVersion(ctx context.Context, deploymentID string, version int, body CreateVersionRequest) (*bundledeployments.Version, error) { - operations := make([]bundledeployments.StagedOperation, len(body.Operations)) - for i, op := range body.Operations { - operations[i] = bundledeployments.StagedOperation{ - ActionType: op.ActionType, - ResourceKey: strings.TrimPrefix(op.ResourceKey, statePrefix), - } - } - - return c.Service.CreateVersion(ctx, bundledeployments.CreateVersionRequest{ - Parent: DeploymentName(deploymentID), - VersionId: strconv.Itoa(version), - Version: bundledeployments.Version{ - CliVersion: body.CliVersion, - VersionType: body.VersionType, - PreviousVersionId: body.PreviousVersionId, - GitInfo: body.GitInfo, - Operations: operations, - }, - }) -} - -// CompleteVersion closes the version out, which is what stops the service expiring its lease. -func (c *Client) CompleteVersion(ctx context.Context, deploymentID string, version int, reason bundledeployments.VersionComplete) error { - _, err := c.Service.CompleteVersion(ctx, bundledeployments.CompleteVersionRequest{ - Name: versionName(deploymentID, version), - CompletionReason: reason, - }) - return err -} - -// UpdateOperation fills in one operation the version staged, and returns the sequence id the -// next update for that resource must send. -func (c *Client) UpdateOperation(ctx context.Context, deploymentID string, version int, stateKey, sequenceID string, update OperationUpdate) (string, error) { - operation, err := newOperationUpdate(update, sequenceID) - if err != nil { - return "", err - } - - result, err := c.Service.UpdateOperation(ctx, bundledeployments.UpdateOperationRequest{ - Name: versionName(deploymentID, version) + "/operations/" + strings.TrimPrefix(stateKey, statePrefix), - Operation: operation, - UpdateMask: fieldmask.FieldMask{Paths: strings.Split(update.Fields.Mask(), ",")}, - }) - if err != nil { - return "", err - } - return strconv.FormatInt(result.SequenceId, 10), 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, DeploymentName("")) - if !ok || id == "" { - return "", fmt.Errorf("unexpected deployment name %q from the deployment history service", name) - } - return id, nil -} - -// CreateVersionRequest is the input to Client.CreateVersion. -type CreateVersionRequest struct { - CliVersion string - VersionType VersionType - // PreviousVersionId is the deployment's most recent version, unset for a - // deployment's first version. - PreviousVersionId string - // GitInfo records where this version's source came from. The rest of the provenance - - // display name, target, mode, workspace paths - belongs to the deployment. - GitInfo *bundledeployments.GitInfo - // Operations is every resource this version will touch; see StagedOperation. - Operations []StagedOperation -} - -// StagedOperation is one resource the version will record an operation for. The service -// creates it in OPERATION_STATUS_PENDING at sequence id 0, and the CLI fills in the outcome -// with UpdateOperation as the resource is applied. -type StagedOperation struct { - // ResourceKey is the bundle state key; CreateVersion strips the prefix to the form the service uses. - ResourceKey string - ActionType bundledeployments.OperationActionType -} - -// newDeploymentUpdate builds the deployment carrying exactly the masked fields, empty ones -// included: the service requires every masked field to be present in the body and reads an empty -// value as a clear (a target that stops setting mode clears deployment_mode). ForceSendFields -// keeps those empty values on the wire, which omitempty would drop. -func newDeploymentUpdate(metadata Metadata, mask string) bundledeployments.Deployment { - full := metadata.deployment() - var dep bundledeployments.Deployment - for path := range strings.SplitSeq(mask, ",") { - switch path { - case "display_name": - dep.DisplayName = full.DisplayName - dep.ForceSendFields = append(dep.ForceSendFields, "DisplayName") - case "target_name": - dep.TargetName = full.TargetName - dep.ForceSendFields = append(dep.ForceSendFields, "TargetName") - case "deployment_mode": - dep.DeploymentMode = full.DeploymentMode - dep.ForceSendFields = append(dep.ForceSendFields, "DeploymentMode") - case "workspace_info": - dep.WorkspaceInfo = full.WorkspaceInfo - dep.ForceSendFields = append(dep.ForceSendFields, "WorkspaceInfo") - } - } - return dep -} - -// newOperationUpdate builds the operation carrying exactly the fields update.Fields masks, plus -// the sequence_id precondition. The service requires a masked field to be present and reads an -// empty value as a write (error_message="" clears it), so masked fields that can be empty are -// forced onto the wire; sequence_id is always sent and a freshly staged operation sits at 0, -// which omitempty would drop. State is the exception: an absent value is how the service is told -// the resource is gone, so a nil state is left off (never forced) while still named in the mask. -func newOperationUpdate(update OperationUpdate, sequenceID string) (bundledeployments.Operation, error) { - sequence, err := strconv.ParseInt(sequenceID, 10, 64) - if err != nil { - return bundledeployments.Operation{}, fmt.Errorf("invalid sequence id %q: %w", sequenceID, err) - } - - operation := bundledeployments.Operation{ - SequenceId: sequence, - ForceSendFields: []string{"SequenceId"}, - } - if update.Fields.Has(FieldState) && update.State != nil { - operation.State = string(update.State) - } - if update.Fields.Has(FieldErrorMessage) { - operation.ErrorMessage = update.ErrorMessage - operation.ForceSendFields = append(operation.ForceSendFields, "ErrorMessage") - } - if update.Fields.Has(FieldResourceID) { - operation.ResourceId = update.ResourceID - operation.ForceSendFields = append(operation.ForceSendFields, "ResourceId") - } - if update.Fields.Has(FieldStatus) { - operation.Status = update.Status - operation.ForceSendFields = append(operation.ForceSendFields, "Status") - } - return operation, nil -} diff --git a/libs/dms/client_test.go b/libs/dms/client_test.go deleted file mode 100644 index 0888829e50d..00000000000 --- a/libs/dms/client_test.go +++ /dev/null @@ -1,80 +0,0 @@ -package dms - -import ( - "encoding/json" - "errors" - "testing" - - "github.com/databricks/databricks-sdk-go/service/bundledeployments" - "github.com/stretchr/testify/assert" - "github.com/stretchr/testify/require" -) - -func TestClientNamesEveryResourceTheSameWay(t *testing.T) { - // One format each, so a call only ever passes ids. - assert.Equal(t, "deployments/dep-1", DeploymentName("dep-1")) - assert.Equal(t, "deployments/dep-1/versions/2", versionName("dep-1", 2)) -} - -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) -} - -func TestUpdateRequestSendsExactlyTheMaskedFields(t *testing.T) { - // The service rejects an update whose mask names a field the body leaves out, and treats a - // field it does carry as written - so the body has to hold every masked field, empty values - // included, which ForceSendFields is what secures. Both bodies the CLI builds are asserted on - // the wire by acceptance/bundle/dms; this pins the rule they both rely on. - update := OperationUpdate{ - Fields: DescribesResource, - State: json.RawMessage(`{"state":{"name":"foo"}}`), - ResourceID: "job-1", - Status: bundledeployments.OperationStatusOperationStatusSucceeded, - } - - // A successful write reports no error, and the empty value is what clears an earlier one. - operation, err := newOperationUpdate(update, "3") - require.NoError(t, err) - assert.JSONEq(t, `{ - "sequence_id": 3, - "state": "{\"state\":{\"name\":\"foo\"}}", - "resource_id": "job-1", - "error_message": "", - "status": "OPERATION_STATUS_SUCCEEDED" - }`, marshalBody(t, operation)) - - // A failure keeps the recorded state, so state is absent rather than empty: naming it - // would clear what the resource last recorded. - operation, err = newOperationUpdate(NewFailureUpdate("job-1", nil, errors.New("boom")), "3") - require.NoError(t, err) - assert.JSONEq(t, `{ - "sequence_id": 3, - "error_message": "boom", - "status": "OPERATION_STATUS_FAILED" - }`, marshalBody(t, operation)) - - // The deployment's own fields follow the same rule. Clearing deployment_mode - a target that - // stops setting mode - sends it empty, which the SDK struct's omitempty would have dropped. - deployment := newDeploymentUpdate(Metadata{DisplayName: "b", TargetName: "t"}, "target_name,deployment_mode") - assert.JSONEq(t, `{ - "target_name": "t", - "deployment_mode": "" - }`, marshalBody(t, deployment)) -} - -// marshalBody renders the request body the SDK sends for v, so a test sees the wire the -// ForceSendFields produce. -func marshalBody(t *testing.T, v any) string { - t.Helper() - body, err := json.Marshal(v) - require.NoError(t, err) - return string(body) -} diff --git a/libs/dms/metadata.go b/libs/dms/metadata.go index 33b7d49f3c4..80170b7e568 100644 --- a/libs/dms/metadata.go +++ b/libs/dms/metadata.go @@ -34,8 +34,8 @@ type Metadata struct { // among them: the service derives the deployment's from the version that carried it. var deploymentFields = []string{"display_name", "target_name", "deployment_mode", "workspace_info"} -// deployment renders the metadata the deployment owns. -func (m Metadata) deployment() bundledeployments.Deployment { +// Deployment renders the metadata the deployment owns. +func (m Metadata) Deployment() bundledeployments.Deployment { return bundledeployments.Deployment{ DisplayName: m.DisplayName, TargetName: m.TargetName, @@ -51,7 +51,7 @@ func (m Metadata) StaleFields(current *bundledeployments.Deployment) string { return strings.Join(deploymentFields, ",") } - want := m.deployment() + want := m.Deployment() var stale []string if want.DisplayName != current.DisplayName { stale = append(stale, "display_name") @@ -80,3 +80,53 @@ func NextVersion(lastVersionID string) (int, error) { } return last + 1, nil } + +// DeploymentNodeName is the workspace node DMS creates per deployment. Must +// match DeploymentWhsClient.DEPLOYMENT_NODE_NAME on the service side. +const DeploymentNodeName = "resources.deployment.json" + +// DeploymentName and VersionName are the two resource-name formats the service uses, so a +// caller only ever passes ids. +func DeploymentName(deploymentID string) string { + return "deployments/" + deploymentID +} + +func VersionName(deploymentID string, version int) string { + return fmt.Sprintf("deployments/%s/versions/%d", deploymentID, version) +} + +// 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, DeploymentName("")) + if !ok || id == "" { + return "", fmt.Errorf("unexpected deployment name %q from the deployment history service", name) + } + return id, nil +} + +// DeploymentUpdate builds the deployment carrying exactly the masked fields, empty ones +// included: the service requires every masked field to be present in the body and reads an empty +// value as a clear (a target that stops setting mode clears deployment_mode). ForceSendFields +// keeps those empty values on the wire, which omitempty would drop. +func DeploymentUpdate(metadata Metadata, mask string) bundledeployments.Deployment { + full := metadata.Deployment() + var dep bundledeployments.Deployment + for path := range strings.SplitSeq(mask, ",") { + switch path { + case "display_name": + dep.DisplayName = full.DisplayName + dep.ForceSendFields = append(dep.ForceSendFields, "DisplayName") + case "target_name": + dep.TargetName = full.TargetName + dep.ForceSendFields = append(dep.ForceSendFields, "TargetName") + case "deployment_mode": + dep.DeploymentMode = full.DeploymentMode + dep.ForceSendFields = append(dep.ForceSendFields, "DeploymentMode") + case "workspace_info": + dep.WorkspaceInfo = full.WorkspaceInfo + dep.ForceSendFields = append(dep.ForceSendFields, "WorkspaceInfo") + } + } + return dep +} diff --git a/libs/dms/metadata_test.go b/libs/dms/metadata_test.go index cddf4205cff..5122533dafc 100644 --- a/libs/dms/metadata_test.go +++ b/libs/dms/metadata_test.go @@ -23,3 +23,21 @@ func TestNextVersion(t *testing.T) { _, err := NextVersion("not-a-number") assert.ErrorContains(t, err, "last_version_id") } + +func TestResourceNames(t *testing.T) { + // One format each, so a call only ever passes ids. + assert.Equal(t, "deployments/dep-1", DeploymentName("dep-1")) + assert.Equal(t, "deployments/dep-1/versions/2", VersionName("dep-1", 2)) +} + +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/operation.go b/libs/dms/operation.go index 57577f5dc21..0bf44520522 100644 --- a/libs/dms/operation.go +++ b/libs/dms/operation.go @@ -3,12 +3,18 @@ package dms import ( "encoding/json" "fmt" + "strconv" "unicode/utf8" "github.com/databricks/cli/libs/diag" "github.com/databricks/databricks-sdk-go/service/bundledeployments" ) +// statePrefix is what a bundle state key carries and a DMS resource key does not: state calls a +// job "resources.jobs.foo", DMS calls it "jobs.foo". Exported names take the state form; the +// prefix comes off where a request is built, and back on where a resource is read. +const StatePrefix = "resources." + // maxStateSize is the largest serialized state DMS accepts per operation. More than this // and the resource cannot be recorded at all, so the deploy fails rather than leaving the // service holding a resource with no state. @@ -140,3 +146,37 @@ func (u OperationUpdate) Merge(newer OperationUpdate) OperationUpdate { return merged } + +// newOperationUpdate builds the operation carrying exactly the fields update.Fields masks, plus +// the sequence_id precondition. The service requires a masked field to be present and reads an +// empty value as a write (error_message="" clears it), so masked fields that can be empty are +// forced onto the wire; sequence_id is always sent and a freshly staged operation sits at 0, +// which omitempty would drop. State is the exception: an absent value is how the service is told +// the resource is gone, so a nil state is left off (never forced) while still named in the mask. +func newOperationUpdate(update OperationUpdate, sequenceID string) (bundledeployments.Operation, error) { + sequence, err := strconv.ParseInt(sequenceID, 10, 64) + if err != nil { + return bundledeployments.Operation{}, fmt.Errorf("invalid sequence id %q: %w", sequenceID, err) + } + + operation := bundledeployments.Operation{ + SequenceId: sequence, + ForceSendFields: []string{"SequenceId"}, + } + if update.Fields.Has(FieldState) && update.State != nil { + operation.State = string(update.State) + } + if update.Fields.Has(FieldErrorMessage) { + operation.ErrorMessage = update.ErrorMessage + operation.ForceSendFields = append(operation.ForceSendFields, "ErrorMessage") + } + if update.Fields.Has(FieldResourceID) { + operation.ResourceId = update.ResourceID + operation.ForceSendFields = append(operation.ForceSendFields, "ResourceId") + } + if update.Fields.Has(FieldStatus) { + operation.Status = update.Status + operation.ForceSendFields = append(operation.ForceSendFields, "Status") + } + return operation, nil +} diff --git a/libs/dms/operation_buffer.go b/libs/dms/operation_buffer.go index c4fbb8e379e..e17f424f13a 100644 --- a/libs/dms/operation_buffer.go +++ b/libs/dms/operation_buffer.go @@ -4,7 +4,12 @@ import ( "context" "encoding/json" "fmt" + "strconv" + "strings" "sync" + + "github.com/databricks/databricks-sdk-go/common/types/fieldmask" + "github.com/databricks/databricks-sdk-go/service/bundledeployments" ) // bufferedOperations caps how far ahead of the service a deploy may get; DMS is what the next @@ -19,7 +24,7 @@ const stagedSequenceID = "0" // path: writes are queued and sent on one background goroutine. It exists only while a bundle // records deployment history; callers hold a nil buffer otherwise and must not call it. type OperationBuffer struct { - client *Client + service bundledeployments.BundleDeploymentsInterface deploymentID string versionNum int @@ -56,9 +61,9 @@ type OperationBuffer struct { // StartOperationBuffer opens the buffer for the version the caller just created. The version // must already exist: operations record under it, and nothing here creates it. -func StartOperationBuffer(ctx context.Context, client *Client, deploymentID string, versionNum int) *OperationBuffer { +func StartOperationBuffer(ctx context.Context, service bundledeployments.BundleDeploymentsInterface, deploymentID string, versionNum int) *OperationBuffer { b := &OperationBuffer{ - client: client, + service: service, deploymentID: deploymentID, versionNum: versionNum, queue: make(chan string, bufferedOperations), @@ -155,13 +160,21 @@ func (b *OperationBuffer) write(ctx context.Context, key string, update Operatio sequenceID = stagedSequenceID } - next, err := b.client.UpdateOperation(ctx, b.deploymentID, b.versionNum, key, sequenceID, update) + operation, err := newOperationUpdate(update, sequenceID) + if err != nil { + return err + } + result, err := b.service.UpdateOperation(ctx, bundledeployments.UpdateOperationRequest{ + Name: VersionName(b.deploymentID, b.versionNum) + "/operations/" + strings.TrimPrefix(key, StatePrefix), + Operation: operation, + UpdateMask: fieldmask.FieldMask{Paths: strings.Split(update.Fields.Mask(), ",")}, + }) if err != nil { return err } // The next write for this resource echoes the sequence id this one earned. - b.sequenceIDs[key] = next + b.sequenceIDs[key] = strconv.FormatInt(result.SequenceId, 10) return nil } diff --git a/libs/dms/operation_test.go b/libs/dms/operation_test.go index 2bae9a81144..ef5928a0ef9 100644 --- a/libs/dms/operation_test.go +++ b/libs/dms/operation_test.go @@ -81,3 +81,54 @@ func TestMergeLetsADeleteClearTheState(t *testing.T) { assert.Nil(t, merged.State) } + +func TestUpdateRequestSendsExactlyTheMaskedFields(t *testing.T) { + // The service rejects an update whose mask names a field the body leaves out, and treats a + // field it does carry as written - so the body has to hold every masked field, empty values + // included, which ForceSendFields is what secures. Both bodies the CLI builds are asserted on + // the wire by acceptance/bundle/dms; this pins the rule they both rely on. + update := OperationUpdate{ + Fields: DescribesResource, + State: json.RawMessage(`{"state":{"name":"foo"}}`), + ResourceID: "job-1", + Status: bundledeployments.OperationStatusOperationStatusSucceeded, + } + + // A successful write reports no error, and the empty value is what clears an earlier one. + operation, err := newOperationUpdate(update, "3") + require.NoError(t, err) + assert.JSONEq(t, `{ + "sequence_id": 3, + "state": "{\"state\":{\"name\":\"foo\"}}", + "resource_id": "job-1", + "error_message": "", + "status": "OPERATION_STATUS_SUCCEEDED" + }`, marshalBody(t, operation)) + + // A failure keeps the recorded state, so state is absent rather than empty: naming it + // would clear what the resource last recorded. + operation, err = newOperationUpdate(NewFailureUpdate("job-1", nil, errors.New("boom")), "3") + require.NoError(t, err) + assert.JSONEq(t, `{ + "sequence_id": 3, + "error_message": "boom", + "status": "OPERATION_STATUS_FAILED" + }`, marshalBody(t, operation)) + + // The deployment's own fields follow the same rule. Clearing deployment_mode - a target that + // stops setting mode - sends it empty, which the SDK struct's omitempty would have dropped. + deployment := DeploymentUpdate(Metadata{DisplayName: "b", TargetName: "t"}, "target_name,deployment_mode") + assert.JSONEq(t, `{ + "target_name": "t", + "deployment_mode": "" + }`, marshalBody(t, deployment)) +} + +// marshalBody renders the request body the SDK sends for v, so a test sees the wire the +// ForceSendFields produce. +func marshalBody(t *testing.T, v any) string { + t.Helper() + body, err := json.Marshal(v) + require.NoError(t, err) + return string(body) +} diff --git a/libs/dms/resources.go b/libs/dms/resources.go index 11321161c15..93024eda3c2 100644 --- a/libs/dms/resources.go +++ b/libs/dms/resources.go @@ -20,8 +20,8 @@ type Resource struct { } // ListResources returns every resource DMS holds for the deployment. -func (c *Client) ListResources(ctx context.Context, deploymentID string) ([]Resource, error) { - it := c.Service.ListResources(ctx, bundledeployments.ListResourcesRequest{ +func ListResources(ctx context.Context, service bundledeployments.BundleDeploymentsInterface, deploymentID string) ([]Resource, error) { + it := service.ListResources(ctx, bundledeployments.ListResourcesRequest{ Parent: DeploymentName(deploymentID), }) @@ -32,7 +32,7 @@ func (c *Client) ListResources(ctx context.Context, deploymentID string) ([]Reso return nil, fmt.Errorf("listing resources from the deployment history service: %w", err) } out = append(out, Resource{ - Key: statePrefix + res.ResourceKey, + Key: StatePrefix + res.ResourceKey, ID: res.ResourceId, State: res.State, }) From c3585667543b92788dc18d95739fd3740ddd0ef1 Mon Sep 17 00:00:00 2001 From: Shreyas Goenka Date: Wed, 16 Sep 2026 14:33:53 +0000 Subject: [PATCH 3/6] Simplify: drop the dmsService field, shorten DMS comments Use w.BundleDeployments directly at each call site instead of caching it on DeploymentState (field + DmsService accessor removed). Add OperationName for the operation resource name, shorten the DMS comments to 1-2 lines, and drop the newOperationUpdate unit test - acceptance/bundle/dms covers the ForceSendFields contract end to end. Co-authored-by: Isaac --- bundle/direct/dstate/state.go | 19 +++-------- bundle/phases/destroy.go | 4 +-- bundle/phases/dms.go | 8 ++--- libs/dms/metadata.go | 22 ++++++------ libs/dms/operation.go | 13 +++---- libs/dms/operation_buffer.go | 2 +- libs/dms/operation_test.go | 51 ---------------------------- libs/testserver/bundledeployments.go | 12 +++---- 8 files changed, 32 insertions(+), 99 deletions(-) diff --git a/bundle/direct/dstate/state.go b/bundle/direct/dstate/state.go index 6acb1d4f45e..0e473b01ea2 100644 --- a/bundle/direct/dstate/state.go +++ b/bundle/direct/dstate/state.go @@ -98,10 +98,6 @@ type DeploymentState struct { // history, in which case InitializeOperationBuffer installs it once the version exists. operationBuffer *dms.OperationBuffer - // dmsService talks to the deployment metadata service. Open sets it from the workspace - // client when the deployment records history; nil otherwise. - dmsService bundledeployments.BundleDeploymentsInterface - // versionCompleted makes CompleteVersion a no-op after the first call, so a deferred safety-net // completion after an explicit one does nothing. versionCompleted bool @@ -180,7 +176,7 @@ type WALEntry struct { // (after approval) - which is why it is not an Open option. It also records the id a first deploy // just created, which Open could not know, so CompleteVersion later has it. func (db *DeploymentState) InitializeOperationBuffer(ctx context.Context, deploymentID string, versionID int) { - buf := dms.StartOperationBuffer(ctx, db.dmsService, deploymentID, versionID) + buf := dms.StartOperationBuffer(ctx, cmdctx.WorkspaceClient(ctx).BundleDeployments, deploymentID, versionID) db.mu.Lock() defer db.mu.Unlock() @@ -224,7 +220,7 @@ func (db *DeploymentState) CompleteVersion(ctx context.Context, success bool) (b return false, nil } db.versionCompleted = true - deploymentID, service, versionID := db.DeploymentID, db.dmsService, db.VersionID + deploymentID, versionID := db.DeploymentID, db.VersionID db.mu.Unlock() // A recording failure fails the version even when the caller counted the deploy a success: the @@ -238,7 +234,7 @@ func (db *DeploymentState) CompleteVersion(ctx context.Context, success bool) (b if !success { reason = bundledeployments.VersionCompleteVersionCompleteFailure } - _, err := service.CompleteVersion(ctx, bundledeployments.CompleteVersionRequest{ + _, err := cmdctx.WorkspaceClient(ctx).BundleDeployments.CompleteVersion(ctx, bundledeployments.CompleteVersionRequest{ Name: dms.VersionName(deploymentID, versionID), CompletionReason: reason, }) @@ -430,12 +426,6 @@ func (db *DeploymentState) IsDeploymentMetadataService() bool { return db.isDeploymentMetadataService() } -// DmsService returns the deployment metadata service client Open set from the workspace client, -// or nil when the deployment does not record history. -func (db *DeploymentState) DmsService() bundledeployments.BundleDeploymentsInterface { - return db.dmsService -} - // GetOrInitLineage returns the deployment lineage, generating and storing a new // one if the state does not have one yet. It is the single place the lineage is // initialized, shared so the direct deployment engine (when it writes state, via @@ -588,11 +578,10 @@ Run "databricks bundle destroy" first, then deploy again with deployment history if recorded { // The service is the source of truth for a recorded deployment; the file is a tombstone // carrying only the marker, and applyDMSState loads the resources the service holds. - db.dmsService = cmdctx.WorkspaceClient(ctx).BundleDeployments db.DeploymentID = dmsDeployment.DeploymentID if dmsDeployment.DeploymentID != "" { - resources, err := dms.ListResources(ctx, db.dmsService, dmsDeployment.DeploymentID) + resources, err := dms.ListResources(ctx, cmdctx.WorkspaceClient(ctx).BundleDeployments, dmsDeployment.DeploymentID) if err != nil { return err } diff --git a/bundle/phases/destroy.go b/bundle/phases/destroy.go index 18b00b6235d..c141839a377 100644 --- a/bundle/phases/destroy.go +++ b/bundle/phases/destroy.go @@ -167,7 +167,7 @@ func destroyCore(ctx context.Context, b *bundle.Bundle, plan *deployplan.Plan) { // A completed destroy's resources are gone, so its deployment record is deleted too. if completed { deploymentID := b.DeploymentBundle.StateDB.DeploymentID - if err := b.DeploymentBundle.StateDB.DmsService().DeleteDeployment(ctx, bundledeployments.DeleteDeploymentRequest{Name: dms.DeploymentName(deploymentID)}); err != nil { + if err := b.WorkspaceClient(ctx).BundleDeployments.DeleteDeployment(ctx, bundledeployments.DeleteDeploymentRequest{Name: dms.DeploymentName(deploymentID)}); err != nil { logdiag.LogError(ctx, fmt.Errorf("failed to delete deployment: %w", err)) return } @@ -299,7 +299,7 @@ func Destroy(ctx context.Context, b *bundle.Bundle) { logdiag.LogError(ctx, err) } else if completed { deploymentID := b.DeploymentBundle.StateDB.DeploymentID - if err := b.DeploymentBundle.StateDB.DmsService().DeleteDeployment(ctx, bundledeployments.DeleteDeploymentRequest{Name: dms.DeploymentName(deploymentID)}); err != nil { + if err := b.WorkspaceClient(ctx).BundleDeployments.DeleteDeployment(ctx, bundledeployments.DeleteDeploymentRequest{Name: dms.DeploymentName(deploymentID)}); err != nil { logdiag.LogError(ctx, fmt.Errorf("failed to delete deployment: %w", err)) } } diff --git a/bundle/phases/dms.go b/bundle/phases/dms.go index eda1abc7d7f..9b275ec7033 100644 --- a/bundle/phases/dms.go +++ b/bundle/phases/dms.go @@ -70,7 +70,7 @@ func actionToSDK(a deployplan.ActionType) (bundledeployments.OperationActionType // no deployment behind; a first deploy's new id is then stamped into the plan (StampDeploymentID). func createOrUpdateDeployment(ctx context.Context, b *bundle.Bundle, current *bundledeployments.Deployment) { db := &b.DeploymentBundle - dmsService := db.StateDB.DmsService() + dmsService := b.WorkspaceClient(ctx).BundleDeployments metadata := deploymentMetadata(b) deploymentID := db.StateDB.DeploymentID if deploymentID == "" { @@ -81,7 +81,7 @@ func createOrUpdateDeployment(ctx context.Context, b *bundle.Bundle, current *bu logdiag.LogError(ctx, fmt.Errorf("failed to create deployment: %w", err)) return } - // The server assigns the id as the workspace node it creates under the parent path. + // The id is the workspace node the server makes under the parent path. deploymentID, err = dms.DeploymentIDFromName(created.Name) if err != nil { logdiag.LogError(ctx, fmt.Errorf("failed to create deployment: %w", err)) @@ -111,10 +111,10 @@ func createOrUpdateDeployment(ctx context.Context, b *bundle.Bundle, current *bu // A no-op when the bundle does not record deployment history. func startVersion(ctx context.Context, b *bundle.Bundle, versionType dms.VersionType, staged []bundledeployments.StagedOperation) error { db := &b.DeploymentBundle - dmsService := db.StateDB.DmsService() - if dmsService == nil { + if !db.StateDB.IsDeploymentMetadataService() { return nil } + dmsService := b.WorkspaceClient(ctx).BundleDeployments deploymentID := db.StateDB.DeploymentID // This run's version follows the one the state is anchored to, which is empty for the first. // InitializeOperationBuffer moves the anchor below, so callers afterwards read the created diff --git a/libs/dms/metadata.go b/libs/dms/metadata.go index 80170b7e568..f9e23a67db0 100644 --- a/libs/dms/metadata.go +++ b/libs/dms/metadata.go @@ -81,12 +81,12 @@ func NextVersion(lastVersionID string) (int, error) { return last + 1, nil } -// DeploymentNodeName is the workspace node DMS creates per deployment. Must -// match DeploymentWhsClient.DEPLOYMENT_NODE_NAME on the service side. +// DeploymentNodeName is the workspace node DMS makes for each deployment. +// Keep it equal to the service's DEPLOYMENT_NODE_NAME. const DeploymentNodeName = "resources.deployment.json" -// DeploymentName and VersionName are the two resource-name formats the service uses, so a -// caller only ever passes ids. +// DeploymentName, VersionName and OperationName build the resource names the +// service uses, so callers pass only ids. func DeploymentName(deploymentID string) string { return "deployments/" + deploymentID } @@ -95,8 +95,12 @@ func VersionName(deploymentID string, version int) string { return fmt.Sprintf("deployments/%s/versions/%d", deploymentID, version) } -// DeploymentIDFromName extracts the deployment ID from a DMS resource name of -// the form "deployments/{deployment_id}". +// OperationName drops the state prefix DMS keys don't use (see StatePrefix). +func OperationName(deploymentID string, version int, stateKey string) string { + return VersionName(deploymentID, version) + "/operations/" + strings.TrimPrefix(stateKey, StatePrefix) +} + +// DeploymentIDFromName pulls the id out of a "deployments/{id}" name. func DeploymentIDFromName(name string) (string, error) { id, ok := strings.CutPrefix(name, DeploymentName("")) if !ok || id == "" { @@ -105,10 +109,8 @@ func DeploymentIDFromName(name string) (string, error) { return id, nil } -// DeploymentUpdate builds the deployment carrying exactly the masked fields, empty ones -// included: the service requires every masked field to be present in the body and reads an empty -// value as a clear (a target that stops setting mode clears deployment_mode). ForceSendFields -// keeps those empty values on the wire, which omitempty would drop. +// DeploymentUpdate builds the deployment with only the masked fields. +// A masked field must be sent even when empty (empty = clear it), so ForceSendFields stops omitempty dropping it. func DeploymentUpdate(metadata Metadata, mask string) bundledeployments.Deployment { full := metadata.Deployment() var dep bundledeployments.Deployment diff --git a/libs/dms/operation.go b/libs/dms/operation.go index 0bf44520522..e19a993bc0e 100644 --- a/libs/dms/operation.go +++ b/libs/dms/operation.go @@ -10,9 +10,8 @@ import ( "github.com/databricks/databricks-sdk-go/service/bundledeployments" ) -// statePrefix is what a bundle state key carries and a DMS resource key does not: state calls a -// job "resources.jobs.foo", DMS calls it "jobs.foo". Exported names take the state form; the -// prefix comes off where a request is built, and back on where a resource is read. +// StatePrefix is the "resources." that bundle keys carry but DMS keys don't +// ("resources.jobs.foo" vs "jobs.foo"). Strip it when sending, add it back when reading. const StatePrefix = "resources." // maxStateSize is the largest serialized state DMS accepts per operation. More than this @@ -147,12 +146,8 @@ func (u OperationUpdate) Merge(newer OperationUpdate) OperationUpdate { return merged } -// newOperationUpdate builds the operation carrying exactly the fields update.Fields masks, plus -// the sequence_id precondition. The service requires a masked field to be present and reads an -// empty value as a write (error_message="" clears it), so masked fields that can be empty are -// forced onto the wire; sequence_id is always sent and a freshly staged operation sits at 0, -// which omitempty would drop. State is the exception: an absent value is how the service is told -// the resource is gone, so a nil state is left off (never forced) while still named in the mask. +// newOperationUpdate builds the operation from the masked fields, plus the sequence_id check. +// Force-send them (empty = clear, and 0 is a real sequence_id); never force state, since a missing state means the resource is gone. func newOperationUpdate(update OperationUpdate, sequenceID string) (bundledeployments.Operation, error) { sequence, err := strconv.ParseInt(sequenceID, 10, 64) if err != nil { diff --git a/libs/dms/operation_buffer.go b/libs/dms/operation_buffer.go index e17f424f13a..d04fe4aadd9 100644 --- a/libs/dms/operation_buffer.go +++ b/libs/dms/operation_buffer.go @@ -165,7 +165,7 @@ func (b *OperationBuffer) write(ctx context.Context, key string, update Operatio return err } result, err := b.service.UpdateOperation(ctx, bundledeployments.UpdateOperationRequest{ - Name: VersionName(b.deploymentID, b.versionNum) + "/operations/" + strings.TrimPrefix(key, StatePrefix), + Name: OperationName(b.deploymentID, b.versionNum, key), Operation: operation, UpdateMask: fieldmask.FieldMask{Paths: strings.Split(update.Fields.Mask(), ",")}, }) diff --git a/libs/dms/operation_test.go b/libs/dms/operation_test.go index ef5928a0ef9..2bae9a81144 100644 --- a/libs/dms/operation_test.go +++ b/libs/dms/operation_test.go @@ -81,54 +81,3 @@ func TestMergeLetsADeleteClearTheState(t *testing.T) { assert.Nil(t, merged.State) } - -func TestUpdateRequestSendsExactlyTheMaskedFields(t *testing.T) { - // The service rejects an update whose mask names a field the body leaves out, and treats a - // field it does carry as written - so the body has to hold every masked field, empty values - // included, which ForceSendFields is what secures. Both bodies the CLI builds are asserted on - // the wire by acceptance/bundle/dms; this pins the rule they both rely on. - update := OperationUpdate{ - Fields: DescribesResource, - State: json.RawMessage(`{"state":{"name":"foo"}}`), - ResourceID: "job-1", - Status: bundledeployments.OperationStatusOperationStatusSucceeded, - } - - // A successful write reports no error, and the empty value is what clears an earlier one. - operation, err := newOperationUpdate(update, "3") - require.NoError(t, err) - assert.JSONEq(t, `{ - "sequence_id": 3, - "state": "{\"state\":{\"name\":\"foo\"}}", - "resource_id": "job-1", - "error_message": "", - "status": "OPERATION_STATUS_SUCCEEDED" - }`, marshalBody(t, operation)) - - // A failure keeps the recorded state, so state is absent rather than empty: naming it - // would clear what the resource last recorded. - operation, err = newOperationUpdate(NewFailureUpdate("job-1", nil, errors.New("boom")), "3") - require.NoError(t, err) - assert.JSONEq(t, `{ - "sequence_id": 3, - "error_message": "boom", - "status": "OPERATION_STATUS_FAILED" - }`, marshalBody(t, operation)) - - // The deployment's own fields follow the same rule. Clearing deployment_mode - a target that - // stops setting mode - sends it empty, which the SDK struct's omitempty would have dropped. - deployment := DeploymentUpdate(Metadata{DisplayName: "b", TargetName: "t"}, "target_name,deployment_mode") - assert.JSONEq(t, `{ - "target_name": "t", - "deployment_mode": "" - }`, marshalBody(t, deployment)) -} - -// marshalBody renders the request body the SDK sends for v, so a test sees the wire the -// ForceSendFields produce. -func marshalBody(t *testing.T, v any) string { - t.Helper() - body, err := json.Marshal(v) - require.NoError(t, err) - return string(body) -} diff --git a/libs/testserver/bundledeployments.go b/libs/testserver/bundledeployments.go index 0d6f03054ed..9e2e864d63b 100644 --- a/libs/testserver/bundledeployments.go +++ b/libs/testserver/bundledeployments.go @@ -387,9 +387,8 @@ func (s *FakeWorkspace) Heartbeat() Response { return Response{Body: bundledeployments.HeartbeatResponse{}} } -// operationBody renders an operation the way the service does: sequence_id as a JSON string. The -// SDK reads it back into its int64 field through its number-or-string normalizer, so the fake -// keeps emitting a string to mirror the real service's proto3 JSON. +// operationBody renders sequence_id as a JSON string, like the real service. +// The SDK reads that string back into its int64 field, so this matches prod. func operationBody(op *bundledeployments.Operation) (map[string]any, error) { raw, err := json.Marshal(op) if err != nil { @@ -410,14 +409,13 @@ func operationBody(op *bundledeployments.Operation) (map[string]any, error) { // UpdateOperation applies a later write for a resource already recorded in this // version. sequence_id is the concurrency precondition and increments on success. func (s *FakeWorkspace) UpdateOperation(req Request, deploymentID, versionID, resourceKey string) Response { - // raw records which fields the body carried, which the update_mask rules turn on: a masked - // path with no value is how a field is cleared. + // raw tracks which fields the body sent. A masked field with no value means clear it. var raw map[string]json.RawMessage if err := json.Unmarshal(req.Body, &raw); err != nil { return Response{StatusCode: 400, Body: map[string]string{"message": err.Error()}} } - // The SDK sends sequence_id as a JSON number and the real service as a string; Operation reads - // either into its int64 field, so the whole body decodes straight into it. + // sequence_id comes as a number (SDK) or a string (real service). Operation reads + // either into int64, so the whole body decodes in one shot. var op bundledeployments.Operation if err := json.Unmarshal(req.Body, &op); err != nil { return Response{StatusCode: 400, Body: map[string]string{"message": err.Error()}} From 4b9230d989dcbb1bd834e53b5bbb78298e965aa0 Mon Sep 17 00:00:00 2001 From: Shreyas Goenka Date: Wed, 16 Sep 2026 15:00:22 +0000 Subject: [PATCH 4/6] Read the workspace client from context in the operation buffer StartOperationBuffer no longer takes a DMS service; the buffer reads w.BundleDeployments from the context it already runs on, so nothing caches it. Use the w := WorkspaceClient(ctx) local idiom at every DMS call site. Co-authored-by: Isaac --- bundle/direct/dstate/state.go | 8 +++++--- bundle/phases/destroy.go | 6 ++++-- bundle/phases/dms.go | 10 +++++----- libs/dms/operation_buffer.go | 8 ++++---- 4 files changed, 18 insertions(+), 14 deletions(-) diff --git a/bundle/direct/dstate/state.go b/bundle/direct/dstate/state.go index 0e473b01ea2..02b5d83fdd2 100644 --- a/bundle/direct/dstate/state.go +++ b/bundle/direct/dstate/state.go @@ -176,7 +176,7 @@ type WALEntry struct { // (after approval) - which is why it is not an Open option. It also records the id a first deploy // just created, which Open could not know, so CompleteVersion later has it. func (db *DeploymentState) InitializeOperationBuffer(ctx context.Context, deploymentID string, versionID int) { - buf := dms.StartOperationBuffer(ctx, cmdctx.WorkspaceClient(ctx).BundleDeployments, deploymentID, versionID) + buf := dms.StartOperationBuffer(ctx, deploymentID, versionID) db.mu.Lock() defer db.mu.Unlock() @@ -234,7 +234,8 @@ func (db *DeploymentState) CompleteVersion(ctx context.Context, success bool) (b if !success { reason = bundledeployments.VersionCompleteVersionCompleteFailure } - _, err := cmdctx.WorkspaceClient(ctx).BundleDeployments.CompleteVersion(ctx, bundledeployments.CompleteVersionRequest{ + w := cmdctx.WorkspaceClient(ctx) + _, err := w.BundleDeployments.CompleteVersion(ctx, bundledeployments.CompleteVersionRequest{ Name: dms.VersionName(deploymentID, versionID), CompletionReason: reason, }) @@ -581,7 +582,8 @@ Run "databricks bundle destroy" first, then deploy again with deployment history db.DeploymentID = dmsDeployment.DeploymentID if dmsDeployment.DeploymentID != "" { - resources, err := dms.ListResources(ctx, cmdctx.WorkspaceClient(ctx).BundleDeployments, dmsDeployment.DeploymentID) + w := cmdctx.WorkspaceClient(ctx) + resources, err := dms.ListResources(ctx, w.BundleDeployments, dmsDeployment.DeploymentID) if err != nil { return err } diff --git a/bundle/phases/destroy.go b/bundle/phases/destroy.go index c141839a377..8b5497d626a 100644 --- a/bundle/phases/destroy.go +++ b/bundle/phases/destroy.go @@ -167,7 +167,8 @@ func destroyCore(ctx context.Context, b *bundle.Bundle, plan *deployplan.Plan) { // A completed destroy's resources are gone, so its deployment record is deleted too. if completed { deploymentID := b.DeploymentBundle.StateDB.DeploymentID - if err := b.WorkspaceClient(ctx).BundleDeployments.DeleteDeployment(ctx, bundledeployments.DeleteDeploymentRequest{Name: dms.DeploymentName(deploymentID)}); err != nil { + w := b.WorkspaceClient(ctx) + if err := w.BundleDeployments.DeleteDeployment(ctx, bundledeployments.DeleteDeploymentRequest{Name: dms.DeploymentName(deploymentID)}); err != nil { logdiag.LogError(ctx, fmt.Errorf("failed to delete deployment: %w", err)) return } @@ -299,7 +300,8 @@ func Destroy(ctx context.Context, b *bundle.Bundle) { logdiag.LogError(ctx, err) } else if completed { deploymentID := b.DeploymentBundle.StateDB.DeploymentID - if err := b.WorkspaceClient(ctx).BundleDeployments.DeleteDeployment(ctx, bundledeployments.DeleteDeploymentRequest{Name: dms.DeploymentName(deploymentID)}); err != nil { + w := b.WorkspaceClient(ctx) + if err := w.BundleDeployments.DeleteDeployment(ctx, bundledeployments.DeleteDeploymentRequest{Name: dms.DeploymentName(deploymentID)}); err != nil { logdiag.LogError(ctx, fmt.Errorf("failed to delete deployment: %w", err)) } } diff --git a/bundle/phases/dms.go b/bundle/phases/dms.go index 9b275ec7033..534b749daaa 100644 --- a/bundle/phases/dms.go +++ b/bundle/phases/dms.go @@ -70,13 +70,13 @@ func actionToSDK(a deployplan.ActionType) (bundledeployments.OperationActionType // no deployment behind; a first deploy's new id is then stamped into the plan (StampDeploymentID). func createOrUpdateDeployment(ctx context.Context, b *bundle.Bundle, current *bundledeployments.Deployment) { db := &b.DeploymentBundle - dmsService := b.WorkspaceClient(ctx).BundleDeployments + w := b.WorkspaceClient(ctx) metadata := deploymentMetadata(b) deploymentID := db.StateDB.DeploymentID if deploymentID == "" { dep := metadata.Deployment() dep.InitialParentPath = b.Config.Workspace.StatePath - created, err := dmsService.CreateDeployment(ctx, bundledeployments.CreateDeploymentRequest{Deployment: dep}) + created, err := w.BundleDeployments.CreateDeployment(ctx, bundledeployments.CreateDeploymentRequest{Deployment: dep}) if err != nil { logdiag.LogError(ctx, fmt.Errorf("failed to create deployment: %w", err)) return @@ -89,7 +89,7 @@ func createOrUpdateDeployment(ctx context.Context, b *bundle.Bundle, current *bu } db.StateDB.DeploymentID = deploymentID } else if mask := metadata.StaleFields(current); mask != "" { - _, err := dmsService.UpdateDeployment(ctx, bundledeployments.UpdateDeploymentRequest{ + _, err := w.BundleDeployments.UpdateDeployment(ctx, bundledeployments.UpdateDeploymentRequest{ Name: dms.DeploymentName(deploymentID), Deployment: dms.DeploymentUpdate(metadata, mask), UpdateMask: fieldmask.FieldMask{Paths: strings.Split(mask, ",")}, @@ -114,7 +114,7 @@ func startVersion(ctx context.Context, b *bundle.Bundle, versionType dms.Version if !db.StateDB.IsDeploymentMetadataService() { return nil } - dmsService := b.WorkspaceClient(ctx).BundleDeployments + w := b.WorkspaceClient(ctx) deploymentID := db.StateDB.DeploymentID // This run's version follows the one the state is anchored to, which is empty for the first. // InitializeOperationBuffer moves the anchor below, so callers afterwards read the created @@ -137,7 +137,7 @@ func startVersion(ctx context.Context, b *bundle.Bundle, versionType dms.Version OriginUrl: git.OriginURL, } } - version, err := dmsService.CreateVersion(ctx, bundledeployments.CreateVersionRequest{ + version, err := w.BundleDeployments.CreateVersion(ctx, bundledeployments.CreateVersionRequest{ Parent: dms.DeploymentName(deploymentID), VersionId: strconv.Itoa(versionID), Version: bundledeployments.Version{ diff --git a/libs/dms/operation_buffer.go b/libs/dms/operation_buffer.go index d04fe4aadd9..ade63bc5a3a 100644 --- a/libs/dms/operation_buffer.go +++ b/libs/dms/operation_buffer.go @@ -8,6 +8,7 @@ import ( "strings" "sync" + "github.com/databricks/cli/libs/cmdctx" "github.com/databricks/databricks-sdk-go/common/types/fieldmask" "github.com/databricks/databricks-sdk-go/service/bundledeployments" ) @@ -24,7 +25,6 @@ const stagedSequenceID = "0" // path: writes are queued and sent on one background goroutine. It exists only while a bundle // records deployment history; callers hold a nil buffer otherwise and must not call it. type OperationBuffer struct { - service bundledeployments.BundleDeploymentsInterface deploymentID string versionNum int @@ -61,9 +61,8 @@ type OperationBuffer struct { // StartOperationBuffer opens the buffer for the version the caller just created. The version // must already exist: operations record under it, and nothing here creates it. -func StartOperationBuffer(ctx context.Context, service bundledeployments.BundleDeploymentsInterface, deploymentID string, versionNum int) *OperationBuffer { +func StartOperationBuffer(ctx context.Context, deploymentID string, versionNum int) *OperationBuffer { b := &OperationBuffer{ - service: service, deploymentID: deploymentID, versionNum: versionNum, queue: make(chan string, bufferedOperations), @@ -164,7 +163,8 @@ func (b *OperationBuffer) write(ctx context.Context, key string, update Operatio if err != nil { return err } - result, err := b.service.UpdateOperation(ctx, bundledeployments.UpdateOperationRequest{ + w := cmdctx.WorkspaceClient(ctx) + result, err := w.BundleDeployments.UpdateOperation(ctx, bundledeployments.UpdateOperationRequest{ Name: OperationName(b.deploymentID, b.versionNum, key), Operation: operation, UpdateMask: fieldmask.FieldMask{Paths: strings.Split(update.Fields.Mask(), ",")}, From 89c385987df11e809acbdf1964fa98de3de10bc3 Mon Sep 17 00:00:00 2001 From: Shreyas Goenka Date: Tue, 6 Oct 2026 14:21:10 +0000 Subject: [PATCH 5/6] Use ListResourcesAll directly and simplify the operation write Call w.BundleDeployments.ListResourcesAll in Open instead of the dms.ListResources wrapper. In the operation buffer, keep sequence ids as int64 (a missing key is the staged 0), have Fields.Mask return the SDK FieldMask, and make the request-body builder a method that cannot fail. Co-authored-by: Isaac --- bundle/direct/dstate/state.go | 15 +++++++++++++-- libs/dms/fields.go | 6 +++--- libs/dms/fields_test.go | 8 ++++---- libs/dms/operation.go | 30 ++++++++++++------------------ libs/dms/operation_buffer.go | 32 +++++++------------------------- libs/dms/resources.go | 28 ---------------------------- 6 files changed, 39 insertions(+), 80 deletions(-) diff --git a/bundle/direct/dstate/state.go b/bundle/direct/dstate/state.go index 02b5d83fdd2..a0be3d80629 100644 --- a/bundle/direct/dstate/state.go +++ b/bundle/direct/dstate/state.go @@ -583,9 +583,20 @@ Run "databricks bundle destroy" first, then deploy again with deployment history if dmsDeployment.DeploymentID != "" { w := cmdctx.WorkspaceClient(ctx) - resources, err := dms.ListResources(ctx, w.BundleDeployments, dmsDeployment.DeploymentID) + recorded, err := w.BundleDeployments.ListResourcesAll(ctx, bundledeployments.ListResourcesRequest{ + Parent: dms.DeploymentName(dmsDeployment.DeploymentID), + }) if err != nil { - return err + return fmt.Errorf("listing resources from the deployment history service: %w", err) + } + resources := make([]dms.Resource, len(recorded)) + for i, res := range recorded { + // DMS keys drop the "resources." prefix; add it back for local state. + resources[i] = dms.Resource{ + Key: dms.StatePrefix + res.ResourceKey, + ID: res.ResourceId, + State: res.State, + } } if err := db.applyDMSState(resources); err != nil { return err diff --git a/libs/dms/fields.go b/libs/dms/fields.go index e175e163aa5..aa244d3e24b 100644 --- a/libs/dms/fields.go +++ b/libs/dms/fields.go @@ -1,6 +1,6 @@ package dms -import "strings" +import "github.com/databricks/databricks-sdk-go/common/types/fieldmask" // Fields is the set of operation fields an update writes, sent as its update mask. The // service rejects any other path, so this is the whole vocabulary. @@ -51,12 +51,12 @@ func (f Fields) Has(other Fields) bool { } // Mask renders f as the update_mask the service expects, always in the same order. -func (f Fields) Mask() string { +func (f Fields) Mask() fieldmask.FieldMask { names := make([]string, 0, len(wireNames)) for _, w := range wireNames { if f.Has(w.field) { names = append(names, w.name) } } - return strings.Join(names, ",") + return fieldmask.FieldMask{Paths: names} } diff --git a/libs/dms/fields_test.go b/libs/dms/fields_test.go index 198a55efca8..daa6df3d919 100644 --- a/libs/dms/fields_test.go +++ b/libs/dms/fields_test.go @@ -8,10 +8,10 @@ import ( func TestFieldsMask(t *testing.T) { // The order is fixed, so the same set always sends the same mask. - assert.Equal(t, "state,error_message,resource_id,status", DescribesResource.Mask()) - assert.Equal(t, "error_message,status", KeepsState.Mask()) - assert.Equal(t, "state", FieldState.Mask()) - assert.Empty(t, Fields(0).Mask()) + assert.Equal(t, []string{"state", "error_message", "resource_id", "status"}, DescribesResource.Mask().Paths) + assert.Equal(t, []string{"error_message", "status"}, KeepsState.Mask().Paths) + assert.Equal(t, []string{"state"}, FieldState.Mask().Paths) + assert.Empty(t, Fields(0).Mask().Paths) } func TestFieldsHas(t *testing.T) { diff --git a/libs/dms/operation.go b/libs/dms/operation.go index e19a993bc0e..3bb2f0b00d8 100644 --- a/libs/dms/operation.go +++ b/libs/dms/operation.go @@ -3,7 +3,6 @@ package dms import ( "encoding/json" "fmt" - "strconv" "unicode/utf8" "github.com/databricks/cli/libs/diag" @@ -146,32 +145,27 @@ func (u OperationUpdate) Merge(newer OperationUpdate) OperationUpdate { return merged } -// newOperationUpdate builds the operation from the masked fields, plus the sequence_id check. +// operation builds the request body from the masked fields, plus the sequence_id check. // Force-send them (empty = clear, and 0 is a real sequence_id); never force state, since a missing state means the resource is gone. -func newOperationUpdate(update OperationUpdate, sequenceID string) (bundledeployments.Operation, error) { - sequence, err := strconv.ParseInt(sequenceID, 10, 64) - if err != nil { - return bundledeployments.Operation{}, fmt.Errorf("invalid sequence id %q: %w", sequenceID, err) - } - +func (u OperationUpdate) operation(sequenceID int64) bundledeployments.Operation { operation := bundledeployments.Operation{ - SequenceId: sequence, + SequenceId: sequenceID, ForceSendFields: []string{"SequenceId"}, } - if update.Fields.Has(FieldState) && update.State != nil { - operation.State = string(update.State) + if u.Fields.Has(FieldState) && u.State != nil { + operation.State = string(u.State) } - if update.Fields.Has(FieldErrorMessage) { - operation.ErrorMessage = update.ErrorMessage + if u.Fields.Has(FieldErrorMessage) { + operation.ErrorMessage = u.ErrorMessage operation.ForceSendFields = append(operation.ForceSendFields, "ErrorMessage") } - if update.Fields.Has(FieldResourceID) { - operation.ResourceId = update.ResourceID + if u.Fields.Has(FieldResourceID) { + operation.ResourceId = u.ResourceID operation.ForceSendFields = append(operation.ForceSendFields, "ResourceId") } - if update.Fields.Has(FieldStatus) { - operation.Status = update.Status + if u.Fields.Has(FieldStatus) { + operation.Status = u.Status operation.ForceSendFields = append(operation.ForceSendFields, "Status") } - return operation, nil + return operation } diff --git a/libs/dms/operation_buffer.go b/libs/dms/operation_buffer.go index ade63bc5a3a..efecb3a330a 100644 --- a/libs/dms/operation_buffer.go +++ b/libs/dms/operation_buffer.go @@ -4,12 +4,9 @@ import ( "context" "encoding/json" "fmt" - "strconv" - "strings" "sync" "github.com/databricks/cli/libs/cmdctx" - "github.com/databricks/databricks-sdk-go/common/types/fieldmask" "github.com/databricks/databricks-sdk-go/service/bundledeployments" ) @@ -17,10 +14,6 @@ import ( // plan reads. const bufferedOperations = 10 -// stagedSequenceID is what version creation leaves on a staged operation, so a resource's first -// update sends it as the precondition. -const stagedSequenceID = "0" - // OperationBuffer records each state write with DMS for one deployment version, off the apply // path: writes are queued and sent on one background goroutine. It exists only while a bundle // records deployment history; callers hold a nil buffer otherwise and must not call it. @@ -34,15 +27,14 @@ type OperationBuffer struct { done chan struct{} stopQueue func() - // sequenceIDs holds the token the last update for a resource returned. A resource absent - // from it has only what staging left, so its first update sends that. Unguarded: run is the - // only goroutine that writes, one update at a time. + // sequenceIDs holds the token the last update for a resource returned. A missing key reads as + // 0, which is what staging leaves. Unguarded: run is the only goroutine that writes, one update at a time. // // TODO: revisit and possibly deprecate. Sequence ids guard against an earlier update // overwriting a later one, which cannot happen here - updates for a resource are sent one at a // time from a single goroutine. UpdateOperation requires them today, so the client cannot // simply stop sending them. - sequenceIDs map[string]string + sequenceIDs map[string]int64 // mu guards the fields below. mu sync.Mutex @@ -69,7 +61,7 @@ func StartOperationBuffer(ctx context.Context, deploymentID string, versionNum i done: make(chan struct{}), pending: make(map[string]OperationUpdate), latestState: make(map[string]OperationUpdate), - sequenceIDs: make(map[string]string), + sequenceIDs: make(map[string]int64), } b.stopQueue = sync.OnceFunc(func() { close(b.queue) }) go b.run(ctx) @@ -154,28 +146,18 @@ func (b *OperationBuffer) run(ctx context.Context) { // write sends one update, at the sequence id the resource is at. func (b *OperationBuffer) write(ctx context.Context, key string, update OperationUpdate) error { - sequenceID, written := b.sequenceIDs[key] - if !written { - sequenceID = stagedSequenceID - } - - operation, err := newOperationUpdate(update, sequenceID) - if err != nil { - return err - } w := cmdctx.WorkspaceClient(ctx) result, err := w.BundleDeployments.UpdateOperation(ctx, bundledeployments.UpdateOperationRequest{ Name: OperationName(b.deploymentID, b.versionNum, key), - Operation: operation, - UpdateMask: fieldmask.FieldMask{Paths: strings.Split(update.Fields.Mask(), ",")}, + Operation: update.operation(b.sequenceIDs[key]), + UpdateMask: update.Fields.Mask(), }) if err != nil { return err } // The next write for this resource echoes the sequence id this one earned. - b.sequenceIDs[key] = strconv.FormatInt(result.SequenceId, 10) - + b.sequenceIDs[key] = result.SequenceId return nil } diff --git a/libs/dms/resources.go b/libs/dms/resources.go index 93024eda3c2..155903381e3 100644 --- a/libs/dms/resources.go +++ b/libs/dms/resources.go @@ -1,12 +1,5 @@ package dms -import ( - "context" - "fmt" - - "github.com/databricks/databricks-sdk-go/service/bundledeployments" -) - // Resource is what DMS holds for one resource of a deployment, as of the last operation // that recorded it. type Resource struct { @@ -18,24 +11,3 @@ type Resource struct { // stores, and empty when no operation recorded one. State string } - -// ListResources returns every resource DMS holds for the deployment. -func ListResources(ctx context.Context, service bundledeployments.BundleDeploymentsInterface, deploymentID string) ([]Resource, error) { - it := service.ListResources(ctx, bundledeployments.ListResourcesRequest{ - Parent: DeploymentName(deploymentID), - }) - - var out []Resource - for it.HasNext(ctx) { - res, err := it.Next(ctx) - if err != nil { - return nil, fmt.Errorf("listing resources from the deployment history service: %w", err) - } - out = append(out, Resource{ - Key: StatePrefix + res.ResourceKey, - ID: res.ResourceId, - State: res.State, - }) - } - return out, nil -} From d7c39029ac55ece15e076da17924aaf4ec6714a9 Mon Sep 17 00:00:00 2001 From: Shreyas Goenka Date: Tue, 6 Oct 2026 14:34:08 +0000 Subject: [PATCH 6/6] Update state-features golden for the SDK wire format Co-authored-by: Isaac --- acceptance/bundle/dms/state-features/output.txt | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/acceptance/bundle/dms/state-features/output.txt b/acceptance/bundle/dms/state-features/output.txt index c8fbbff65fd..00451468b55 100644 --- a/acceptance/bundle/dms/state-features/output.txt +++ b/acceptance/bundle/dms/state-features/output.txt @@ -44,7 +44,7 @@ All files and directories at the following location will be deleted: /Workspace/ Destroy: 1 deleted >>> print_requests.py --dms //api/2.0/bundle --oneline -{"method": "POST", "path": "/api/2.0/bundle/deployments/[NUMID]/versions", "q": {"version_id": "2"}, "body": {"cli_version": "[CLI_VERSION]", "version_type": "VERSION_TYPE_DESTROY", "previous_version_id": "1", "operations": [{"resource_key": "jobs.foo", "action_type": "OPERATION_ACTION_TYPE_DELETE"}]}} -{"method": "PATCH", "path": "/api/2.0/bundle/deployments/[NUMID]/versions/2/operations/jobs.foo", "q": {"update_mask": "state,error_message,status"}, "body": {"error_message": "", "sequence_id": "0", "status": "OPERATION_STATUS_SUCCEEDED"}} +{"method": "POST", "path": "/api/2.0/bundle/deployments/[NUMID]/versions", "q": {"version_id": "2"}, "body": {"cli_version": "[CLI_VERSION]", "operations": [{"action_type": "OPERATION_ACTION_TYPE_DELETE", "resource_key": "jobs.foo"}], "previous_version_id": "1", "version_type": "VERSION_TYPE_DESTROY"}} +{"method": "PATCH", "path": "/api/2.0/bundle/deployments/[NUMID]/versions/2/operations/jobs.foo", "q": {"update_mask": "state,error_message,status"}, "body": {"error_message": "", "sequence_id": 0, "status": "OPERATION_STATUS_SUCCEEDED"}} {"method": "POST", "path": "/api/2.0/bundle/deployments/[NUMID]/versions/2/complete", "body": {"completion_reason": "VERSION_COMPLETE_SUCCESS"}} {"method": "DELETE", "path": "/api/2.0/bundle/deployments/[NUMID]"}