diff --git a/.nextchanges/bundles/remote-sync-cluster-policies.md b/.nextchanges/bundles/remote-sync-cluster-policies.md new file mode 100644 index 00000000000..df06c0c9783 --- /dev/null +++ b/.nextchanges/bundles/remote-sync-cluster-policies.md @@ -0,0 +1 @@ +* Fix `bundle config-remote-sync` copying cluster policy values into shared configuration and breaking deployment to another target. ([#7170](https://github.com/databricks/cli/pull/7170)) diff --git a/acceptance/bundle/config-remote-sync/cluster_policy_cross_target/databricks.yml.tmpl b/acceptance/bundle/config-remote-sync/cluster_policy_cross_target/databricks.yml.tmpl new file mode 100644 index 00000000000..0730403ce78 --- /dev/null +++ b/acceptance/bundle/config-remote-sync/cluster_policy_cross_target/databricks.yml.tmpl @@ -0,0 +1,52 @@ +bundle: + name: test-bundle-$UNIQUE_NAME + +workspace: + root_path: ~/.bundle/$UNIQUE_NAME/${bundle.target} + +variables: + environment: + default: dev + +resources: + cluster_policies: + policy: + name: test-policy-$UNIQUE_NAME-${bundle.target} + definition: + custom_tags.workspace: + type: fixed + value: ${var.environment} + spark_env_vars.ENVIRONMENT: + type: fixed + value: ${var.environment} + workload_type.clients.jobs: + type: fixed + value: true + workload_type.clients.notebooks: + type: fixed + value: false + jobs: + policy_sync: + name: test-job-$UNIQUE_NAME-${bundle.target} + description: Before UI edit + job_clusters: + - job_cluster_key: main + new_cluster: + policy_id: ${resources.cluster_policies.policy.id} + spark_version: $DEFAULT_SPARK_VERSION + node_type_id: $NODE_TYPE_ID + num_workers: 1 + custom_tags: + owner: bundle-author + tasks: + - task_key: main + job_cluster_key: main + spark_python_task: + python_file: ./hello.py + +targets: + dev: + default: true + prod: + variables: + environment: prod diff --git a/acceptance/bundle/config-remote-sync/cluster_policy_cross_target/hello.py b/acceptance/bundle/config-remote-sync/cluster_policy_cross_target/hello.py new file mode 100644 index 00000000000..11b15b1a458 --- /dev/null +++ b/acceptance/bundle/config-remote-sync/cluster_policy_cross_target/hello.py @@ -0,0 +1 @@ +print("hello") diff --git a/acceptance/bundle/config-remote-sync/cluster_policy_cross_target/out.test.toml b/acceptance/bundle/config-remote-sync/cluster_policy_cross_target/out.test.toml new file mode 100644 index 00000000000..a4f8dd5ef74 --- /dev/null +++ b/acceptance/bundle/config-remote-sync/cluster_policy_cross_target/out.test.toml @@ -0,0 +1,3 @@ +Cloud = true +GOOS.windows = false +EnvMatrix.DMS = ["", "true"] diff --git a/acceptance/bundle/config-remote-sync/cluster_policy_cross_target/output.txt b/acceptance/bundle/config-remote-sync/cluster_policy_cross_target/output.txt new file mode 100644 index 00000000000..3bff099f4f4 --- /dev/null +++ b/acceptance/bundle/config-remote-sync/cluster_policy_cross_target/output.txt @@ -0,0 +1,87 @@ + +=== Deploy with the dev policy +>>> [CLI] bundle deploy -t dev +Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/[UNIQUE_NAME]/dev/files... +Created cluster_policies.policy +Created jobs.policy_sync +Files: 3 uploaded, 0 deleted +Resources: 2 created, 0 changed, 0 deleted, 0 unchanged + +=== Sync the UI edit into shared YAML +>>> [CLI] bundle config-remote-sync -t dev --save +Detected changes in 1 resource(s): + +Resource: resources.jobs.policy_sync + description: replace + + + +>>> diff.py databricks.yml.backup databricks.yml +--- databricks.yml.backup ++++ databricks.yml +@@ -29,5 +29,5 @@ + policy_sync: + name: test-job-[UNIQUE_NAME]-${bundle.target} +- description: Before UI edit ++ description: Edited in the UI + job_clusters: + - job_cluster_key: main + +=== Deploy shared YAML with the prod policy +>>> [CLI] bundle deploy -t prod +Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/[UNIQUE_NAME]/prod/files... +Created cluster_policies.policy +Created jobs.policy_sync +Files: 4 uploaded, 0 deleted +Resources: 2 created, 0 changed, 0 deleted, 0 unchanged + +=== Add a user tag beside the policy tag in the UI +>>> [CLI] bundle config-remote-sync -t dev --save +Detected changes in 1 resource(s): + +Resource: resources.jobs.policy_sync + job_clusters[job_cluster_key='main'].new_cluster.custom_tags['UserTag']: add + + + +>>> diff.py databricks.yml.backup databricks.yml +--- databricks.yml.backup ++++ databricks.yml +@@ -39,4 +39,5 @@ + custom_tags: + owner: bundle-author ++ UserTag: edited-in-ui + tasks: + - task_key: main + +=== Deploy the user tag with the prod policy +>>> [CLI] bundle deploy -t prod +Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/[UNIQUE_NAME]/prod/files... +Updated jobs.policy_sync +Files: 2 uploaded, 0 deleted +Resources: 0 created, 1 changed, 0 deleted, 1 unchanged + +>>> [CLI] jobs get [POLICY_SYNC_ID_2] -o json +{ + "UserTag": "edited-in-ui", + "owner": "bundle-author", + "workspace": "prod" +} + +>>> [CLI] bundle destroy -t prod --auto-approve +The following resources will be deleted: + delete resources.cluster_policies.policy + delete resources.jobs.policy_sync + +All files and directories at the following location will be deleted: /Workspace/Users/[USERNAME]/.bundle/[UNIQUE_NAME]/prod + +Destroy: 2 deleted + +>>> [CLI] bundle destroy -t dev --auto-approve +The following resources will be deleted: + delete resources.cluster_policies.policy + delete resources.jobs.policy_sync + +All files and directories at the following location will be deleted: /Workspace/Users/[USERNAME]/.bundle/[UNIQUE_NAME]/dev + +Destroy: 2 deleted diff --git a/acceptance/bundle/config-remote-sync/cluster_policy_cross_target/script b/acceptance/bundle/config-remote-sync/cluster_policy_cross_target/script new file mode 100644 index 00000000000..fe0dd42d506 --- /dev/null +++ b/acceptance/bundle/config-remote-sync/cluster_policy_cross_target/script @@ -0,0 +1,37 @@ +envsubst < databricks.yml.tmpl > databricks.yml + +cleanup() { + trace $CLI bundle destroy -t prod --auto-approve + trace $CLI bundle destroy -t dev --auto-approve +} +trap cleanup EXIT + +title "Deploy with the dev policy" +trace $CLI bundle deploy -t dev +job_id=$(read_id.py -t dev policy_sync) + +edit_resource.py jobs "$job_id" <<'EOF' +r["description"] = "Edited in the UI" +EOF + +title "Sync the UI edit into shared YAML" +cp databricks.yml databricks.yml.backup +trace $CLI bundle config-remote-sync -t dev --save +trace diff.py databricks.yml.backup databricks.yml + +title "Deploy shared YAML with the prod policy" +trace $CLI bundle deploy -t prod + +title "Add a user tag beside the policy tag in the UI" +edit_resource.py jobs "$job_id" <<'EOF' +r["job_clusters"][0]["new_cluster"]["custom_tags"]["UserTag"] = "edited-in-ui" +EOF + +cp databricks.yml databricks.yml.backup +trace $CLI bundle config-remote-sync -t dev --save +trace diff.py databricks.yml.backup databricks.yml + +title "Deploy the user tag with the prod policy" +trace $CLI bundle deploy -t prod +prod_job_id=$(read_id.py -t prod policy_sync) +trace $CLI jobs get "$prod_job_id" -o json | jq '.settings.job_clusters[0].new_cluster.custom_tags' diff --git a/acceptance/bundle/config-remote-sync/cluster_policy_cross_target/test.toml b/acceptance/bundle/config-remote-sync/cluster_policy_cross_target/test.toml new file mode 100644 index 00000000000..a799274577a --- /dev/null +++ b/acceptance/bundle/config-remote-sync/cluster_policy_cross_target/test.toml @@ -0,0 +1,3 @@ +Cloud = true +RecordRequests = false +Ignore = [".databricks", "databricks.yml", "databricks.yml.backup"] diff --git a/acceptance/bundle/config-remote-sync/cluster_policy_remote_addition/output.txt b/acceptance/bundle/config-remote-sync/cluster_policy_remote_addition/output.txt index bca88a73a63..3c7e9f868cf 100644 --- a/acceptance/bundle/config-remote-sync/cluster_policy_remote_addition/output.txt +++ b/acceptance/bundle/config-remote-sync/cluster_policy_remote_addition/output.txt @@ -10,6 +10,7 @@ Resources: 2 created, 0 changed, 0 deleted, 0 unchanged Detected changes in 1 resource(s): Resource: resources.jobs.test_job + tasks[task_key='test'].new_cluster.custom_tags: add tasks[task_key='test'].new_cluster.spark_conf: add @@ -17,14 +18,45 @@ Resource: resources.jobs.test_job >>> diff.py databricks.yml.backup databricks.yml --- databricks.yml.backup +++ databricks.yml -@@ -24,4 +24,6 @@ +@@ -24,4 +24,8 @@ node_type_id: [NODE_TYPE_ID] num_workers: 1 ++ custom_tags: ++ UserTag: edited-in-ui + spark_conf: + spark.sql.shuffle.partitions: "7" spark_python_task: python_file: ./hello.py +=== Add a task with user and policy cluster fields in the UI +>>> [CLI] bundle config-remote-sync --save +Detected changes in 1 resource(s): + +Resource: resources.jobs.test_job + tasks[task_key='added_in_ui']: add + tasks[task_key='test'].new_cluster.custom_tags: replace + + + +>>> diff.py databricks.yml.backup databricks.yml +--- databricks.yml.backup ++++ databricks.yml +@@ -30,2 +30,14 @@ + spark_python_task: + python_file: ./hello.py ++ - new_cluster: ++ custom_tags: ++ UserTag: new-task ++ node_type_id: [NODE_TYPE_ID] ++ num_workers: 1 ++ policy_id: ${resources.cluster_policies.policy.id} ++ spark_conf: ++ spark.sql.shuffle.partitions: "7" ++ spark_version: [DEFAULT_SPARK_VERSION] ++ spark_python_task: ++ python_file: /Workspace/Users/[USERNAME]/.bundle/[UNIQUE_NAME]/files/hello.py ++ task_key: added_in_ui + >>> [CLI] bundle destroy --auto-approve The following resources will be deleted: delete resources.cluster_policies.policy diff --git a/acceptance/bundle/config-remote-sync/cluster_policy_remote_addition/script b/acceptance/bundle/config-remote-sync/cluster_policy_remote_addition/script index afe89d9a834..f0d5f620c2c 100644 --- a/acceptance/bundle/config-remote-sync/cluster_policy_remote_addition/script +++ b/acceptance/bundle/config-remote-sync/cluster_policy_remote_addition/script @@ -1,4 +1,5 @@ envsubst < databricks.yml.tmpl > databricks.yml +add_repl "$DEFAULT_SPARK_VERSION" DEFAULT_SPARK_VERSION cleanup() { trace $CLI bundle destroy --auto-approve @@ -10,6 +11,22 @@ job_id=$(read_id.py test_job) edit_resource.py jobs "$job_id" <<'EOF' r["tasks"][0]["new_cluster"]["spark_conf"] = {"spark.sql.shuffle.partitions": "7"} +r["tasks"][0]["new_cluster"]["custom_tags"]["UserTag"] = "edited-in-ui" +EOF + +cp databricks.yml databricks.yml.backup +trace $CLI bundle config-remote-sync --save +trace diff.py databricks.yml.backup databricks.yml +rm databricks.yml.backup + +title "Add a task with user and policy cluster fields in the UI" +edit_resource.py jobs "$job_id" <<'EOF' +from copy import deepcopy + +task = deepcopy(r["tasks"][0]) +task["task_key"] = "added_in_ui" +task["new_cluster"]["custom_tags"]["UserTag"] = "new-task" +r["tasks"].append(task) EOF cp databricks.yml databricks.yml.backup diff --git a/acceptance/bundle/config-remote-sync/cluster_policy_replaces_array/databricks.yml.tmpl b/acceptance/bundle/config-remote-sync/cluster_policy_replaces_array/databricks.yml.tmpl new file mode 100644 index 00000000000..70c90e5c246 --- /dev/null +++ b/acceptance/bundle/config-remote-sync/cluster_policy_replaces_array/databricks.yml.tmpl @@ -0,0 +1,24 @@ +bundle: + name: test-bundle-$UNIQUE_NAME + +workspace: + root_path: ~/.bundle/$UNIQUE_NAME + +resources: + jobs: + test_job: + name: test-job-$UNIQUE_NAME + tasks: + - task_key: main + new_cluster: + policy_id: $POLICY_ID + spark_version: $DEFAULT_SPARK_VERSION + node_type_id: $NODE_TYPE_ID + num_workers: 1 + init_scripts: + - workspace: + destination: /Workspace/init-a.sh + - workspace: + destination: /Workspace/init-b.sh + spark_python_task: + python_file: ./hello.py diff --git a/acceptance/bundle/config-remote-sync/cluster_policy_replaces_array/hello.py b/acceptance/bundle/config-remote-sync/cluster_policy_replaces_array/hello.py new file mode 100644 index 00000000000..11b15b1a458 --- /dev/null +++ b/acceptance/bundle/config-remote-sync/cluster_policy_replaces_array/hello.py @@ -0,0 +1 @@ +print("hello") diff --git a/acceptance/bundle/config-remote-sync/cluster_policy_replaces_array/out.test.toml b/acceptance/bundle/config-remote-sync/cluster_policy_replaces_array/out.test.toml new file mode 100644 index 00000000000..8c2cb31de53 --- /dev/null +++ b/acceptance/bundle/config-remote-sync/cluster_policy_replaces_array/out.test.toml @@ -0,0 +1,3 @@ +Cloud = false +GOOS.windows = false +EnvMatrix.DMS = ["", "true"] diff --git a/acceptance/bundle/config-remote-sync/cluster_policy_replaces_array/output.txt b/acceptance/bundle/config-remote-sync/cluster_policy_replaces_array/output.txt new file mode 100644 index 00000000000..40c2085e824 --- /dev/null +++ b/acceptance/bundle/config-remote-sync/cluster_policy_replaces_array/output.txt @@ -0,0 +1,49 @@ + +>>> [CLI] bundle deploy +Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/[UNIQUE_NAME]/files... +Created jobs.test_job +Files: 5 uploaded, 0 deleted +Resources: 1 created, 0 changed, 0 deleted, 0 unchanged + +=== Replace authored scripts with a script from an updated policy +>>> [CLI] cluster-policies edit --json { + "policy_id": "[POLICY_ID]", + "name": "test-policy-[UNIQUE_NAME]", + "definition": "{\"init_scripts.0.volumes.destination\":{\"type\":\"fixed\",\"value\":\"/Volumes/catalog/schema/volume/policy.sh\"}}" +} + +>>> [CLI] bundle config-remote-sync --save +Detected changes in 1 resource(s): + +Resource: resources.jobs.test_job + tasks[task_key='main'].new_cluster.init_scripts: remove + + + +>>> diff.py databricks.yml.backup databricks.yml +--- databricks.yml.backup ++++ databricks.yml +@@ -16,9 +16,4 @@ + node_type_id: [NODE_TYPE_ID] + num_workers: 1 +- init_scripts: +- - workspace: +- destination: /Workspace/init-a.sh +- - workspace: +- destination: /Workspace/init-b.sh + spark_python_task: + python_file: ./hello.py + +=== Authored scripts are removed without copying the policy script +>>> [CLI] bundle validate -o json +true + +>>> [CLI] bundle destroy --auto-approve +The following resources will be deleted: + delete resources.jobs.test_job + +All files and directories at the following location will be deleted: /Workspace/Users/[USERNAME]/.bundle/[UNIQUE_NAME] + +Destroy: 1 deleted + +>>> [CLI] cluster-policies delete [POLICY_ID] diff --git a/acceptance/bundle/config-remote-sync/cluster_policy_replaces_array/policy.json.tmpl b/acceptance/bundle/config-remote-sync/cluster_policy_replaces_array/policy.json.tmpl new file mode 100644 index 00000000000..f60659b84ab --- /dev/null +++ b/acceptance/bundle/config-remote-sync/cluster_policy_replaces_array/policy.json.tmpl @@ -0,0 +1,4 @@ +{ + "name": "test-policy-$UNIQUE_NAME", + "definition": "{}" +} diff --git a/acceptance/bundle/config-remote-sync/cluster_policy_replaces_array/script b/acceptance/bundle/config-remote-sync/cluster_policy_replaces_array/script new file mode 100644 index 00000000000..428e6ec433c --- /dev/null +++ b/acceptance/bundle/config-remote-sync/cluster_policy_replaces_array/script @@ -0,0 +1,34 @@ +export POLICY_ID="" +bundle_configured=false +cleanup() { + if $bundle_configured; then + errcode trace $CLI bundle destroy --auto-approve + fi + if [ -n "$POLICY_ID" ]; then + errcode trace $CLI cluster-policies delete "$POLICY_ID" + fi +} +trap cleanup EXIT + +POLICY_ID=$($CLI cluster-policies create --json "$(envsubst < policy.json.tmpl)" -o json | jq -er .policy_id) +add_repl "$POLICY_ID" POLICY_ID +envsubst < databricks.yml.tmpl > databricks.yml +bundle_configured=true + +trace $CLI bundle deploy +job_id=$(read_id.py test_job) + +title "Replace authored scripts with a script from an updated policy" +trace $CLI cluster-policies edit --json "$(envsubst < updated-policy.json.tmpl)" +edit_resource.py jobs "$job_id" <<'PY' +r["tasks"][0]["new_cluster"]["init_scripts"] = [ + {"volumes": {"destination": "/Volumes/catalog/schema/volume/policy.sh"}} +] +PY + +cp databricks.yml databricks.yml.backup +trace $CLI bundle config-remote-sync --save +trace diff.py databricks.yml.backup databricks.yml + +title "Authored scripts are removed without copying the policy script" +trace $CLI bundle validate -o json | jq -e '.resources.jobs.test_job.tasks[0].new_cluster | has("init_scripts") | not' diff --git a/acceptance/bundle/config-remote-sync/cluster_policy_replaces_array/test.toml b/acceptance/bundle/config-remote-sync/cluster_policy_replaces_array/test.toml new file mode 100644 index 00000000000..daa61d46152 --- /dev/null +++ b/acceptance/bundle/config-remote-sync/cluster_policy_replaces_array/test.toml @@ -0,0 +1,2 @@ +RecordRequests = false +Ignore = [".databricks", "databricks.yml", "databricks.yml.backup"] diff --git a/acceptance/bundle/config-remote-sync/cluster_policy_replaces_array/updated-policy.json.tmpl b/acceptance/bundle/config-remote-sync/cluster_policy_replaces_array/updated-policy.json.tmpl new file mode 100644 index 00000000000..3909d7c2dfa --- /dev/null +++ b/acceptance/bundle/config-remote-sync/cluster_policy_replaces_array/updated-policy.json.tmpl @@ -0,0 +1,5 @@ +{ + "policy_id": "$POLICY_ID", + "name": "test-policy-$UNIQUE_NAME", + "definition": "{\"init_scripts.0.volumes.destination\":{\"type\":\"fixed\",\"value\":\"/Volumes/catalog/schema/volume/policy.sh\"}}" +} diff --git a/acceptance/bundle/config-remote-sync/external_cluster_policy_cross_target/databricks.yml.tmpl b/acceptance/bundle/config-remote-sync/external_cluster_policy_cross_target/databricks.yml.tmpl new file mode 100644 index 00000000000..038d9fabebb --- /dev/null +++ b/acceptance/bundle/config-remote-sync/external_cluster_policy_cross_target/databricks.yml.tmpl @@ -0,0 +1,38 @@ +bundle: + name: test-bundle-$UNIQUE_NAME + +workspace: + root_path: ~/.bundle/$UNIQUE_NAME/${bundle.target} + +variables: + policy_id: + description: Policy managed outside this bundle + +resources: + jobs: + policy_sync: + name: test-job-$UNIQUE_NAME-${bundle.target} + description: Before UI edit + job_clusters: + - job_cluster_key: main + new_cluster: + policy_id: ${var.policy_id} + spark_version: $DEFAULT_SPARK_VERSION + node_type_id: $NODE_TYPE_ID + num_workers: 1 + custom_tags: + owner: bundle-author + tasks: + - task_key: main + job_cluster_key: main + spark_python_task: + python_file: ./hello.py + +targets: + dev: + default: true + variables: + policy_id: $DEV_POLICY_ID + prod: + variables: + policy_id: $PROD_POLICY_ID diff --git a/acceptance/bundle/config-remote-sync/external_cluster_policy_cross_target/hello.py b/acceptance/bundle/config-remote-sync/external_cluster_policy_cross_target/hello.py new file mode 100644 index 00000000000..11b15b1a458 --- /dev/null +++ b/acceptance/bundle/config-remote-sync/external_cluster_policy_cross_target/hello.py @@ -0,0 +1 @@ +print("hello") diff --git a/acceptance/bundle/config-remote-sync/external_cluster_policy_cross_target/out.test.toml b/acceptance/bundle/config-remote-sync/external_cluster_policy_cross_target/out.test.toml new file mode 100644 index 00000000000..a4f8dd5ef74 --- /dev/null +++ b/acceptance/bundle/config-remote-sync/external_cluster_policy_cross_target/out.test.toml @@ -0,0 +1,3 @@ +Cloud = true +GOOS.windows = false +EnvMatrix.DMS = ["", "true"] diff --git a/acceptance/bundle/config-remote-sync/external_cluster_policy_cross_target/output.txt b/acceptance/bundle/config-remote-sync/external_cluster_policy_cross_target/output.txt new file mode 100644 index 00000000000..fb9a571c96a --- /dev/null +++ b/acceptance/bundle/config-remote-sync/external_cluster_policy_cross_target/output.txt @@ -0,0 +1,133 @@ + +=== Create policies outside the bundle +=== Deploy with the dev policy +>>> [CLI] bundle deploy -t dev +Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/[UNIQUE_NAME]/dev/files... +Created jobs.policy_sync +Files: 4 uploaded, 0 deleted +Resources: 1 created, 0 changed, 0 deleted, 0 unchanged + +=== Sync the UI edit into shared YAML +>>> [CLI] bundle config-remote-sync -t dev --save +Detected changes in 1 resource(s): + +Resource: resources.jobs.policy_sync + description: replace + + + +>>> diff.py databricks.yml.backup databricks.yml +--- databricks.yml.backup ++++ databricks.yml +@@ -13,5 +13,5 @@ + policy_sync: + name: test-job-[UNIQUE_NAME]-${bundle.target} +- description: Before UI edit ++ description: Edited in the UI + job_clusters: + - job_cluster_key: main + +=== Deploy shared YAML with the prod policy +>>> [CLI] bundle deploy -t prod +Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/[UNIQUE_NAME]/prod/files... +Created jobs.policy_sync +Files: 5 uploaded, 0 deleted +Resources: 1 created, 0 changed, 0 deleted, 0 unchanged + +=== A deleted policy skips policy-dependent fields but still syncs authored edits +>>> [CLI] cluster-policies delete [DEV_POLICY_ID] + +>>> [CLI] bundle config-remote-sync -t dev --save -o json +{ + "status": "incomplete", + "changes": { + "resources.jobs.policy_sync": { + "description": { + "operation": "replace", + "value": "Another UI edit" + }, + "job_clusters[job_cluster_key='main'].new_cluster.custom_tags['owner']": { + "operation": "replace", + "value": "edited-owner" + } + } + }, + "skippedChanges": [ + { + "resourceKey": "resources.jobs.policy_sync", + "path": "job_clusters[job_cluster_key='main'].new_cluster.custom_tags['workspace']", + "reason": "cluster_policy_not_found", + "policyId": "[DEV_POLICY_ID]" + }, + { + "resourceKey": "resources.jobs.policy_sync", + "path": "job_clusters[job_cluster_key='main'].new_cluster.spark_env_vars", + "reason": "cluster_policy_not_found", + "policyId": "[DEV_POLICY_ID]" + } + ] +} + +>>> diff.py databricks.yml.backup databricks.yml +--- databricks.yml.backup ++++ databricks.yml +@@ -13,5 +13,5 @@ + policy_sync: + name: test-job-[UNIQUE_NAME]-${bundle.target} +- description: Edited in the UI ++ description: Another UI edit + job_clusters: + - job_cluster_key: main +@@ -22,5 +22,5 @@ + num_workers: 1 + custom_tags: +- owner: bundle-author ++ owner: edited-owner + tasks: + - task_key: main + +=== Only policy-dependent fields remain on the next sync +>>> [CLI] bundle config-remote-sync -t dev -o json +{ + "status": "incomplete", + "changes": {}, + "skippedChanges": [ + { + "resourceKey": "resources.jobs.policy_sync", + "path": "job_clusters[job_cluster_key='main'].new_cluster.custom_tags['workspace']", + "reason": "cluster_policy_not_found", + "policyId": "[DEV_POLICY_ID]" + }, + { + "resourceKey": "resources.jobs.policy_sync", + "path": "job_clusters[job_cluster_key='main'].new_cluster.spark_env_vars", + "reason": "cluster_policy_not_found", + "policyId": "[DEV_POLICY_ID]" + } + ] +} + +=== Text output reports the incomplete sync +>>> [CLI] bundle config-remote-sync -t dev +Sync is incomplete: some remote changes were skipped. +Skipped resources.jobs.policy_sync.job_clusters[job_cluster_key='main'].new_cluster.custom_tags['workspace']: cluster policy [DEV_POLICY_ID] was not found. +Skipped resources.jobs.policy_sync.job_clusters[job_cluster_key='main'].new_cluster.spark_env_vars: cluster policy [DEV_POLICY_ID] was not found. +No changes could be synced. + +>>> [CLI] bundle destroy -t prod --auto-approve +The following resources will be deleted: + delete resources.jobs.policy_sync + +All files and directories at the following location will be deleted: /Workspace/Users/[USERNAME]/.bundle/[UNIQUE_NAME]/prod + +Destroy: 1 deleted + +>>> [CLI] bundle destroy -t dev --auto-approve +The following resources will be deleted: + delete resources.jobs.policy_sync + +All files and directories at the following location will be deleted: /Workspace/Users/[USERNAME]/.bundle/[UNIQUE_NAME]/dev + +Destroy: 1 deleted + +>>> [CLI] cluster-policies delete [PROD_POLICY_ID] diff --git a/acceptance/bundle/config-remote-sync/external_cluster_policy_cross_target/policy.json.tmpl b/acceptance/bundle/config-remote-sync/external_cluster_policy_cross_target/policy.json.tmpl new file mode 100644 index 00000000000..34dd6b9e484 --- /dev/null +++ b/acceptance/bundle/config-remote-sync/external_cluster_policy_cross_target/policy.json.tmpl @@ -0,0 +1,4 @@ +{ + "name": "test-external-policy-$UNIQUE_NAME-$POLICY_ENVIRONMENT", + "definition": "{\"custom_tags.workspace\":{\"type\":\"fixed\",\"value\":\"$POLICY_ENVIRONMENT\"},\"spark_env_vars.ENVIRONMENT\":{\"type\":\"fixed\",\"value\":\"$POLICY_ENVIRONMENT\"}}" +} diff --git a/acceptance/bundle/config-remote-sync/external_cluster_policy_cross_target/script b/acceptance/bundle/config-remote-sync/external_cluster_policy_cross_target/script new file mode 100644 index 00000000000..14735b63b36 --- /dev/null +++ b/acceptance/bundle/config-remote-sync/external_cluster_policy_cross_target/script @@ -0,0 +1,60 @@ +export DEV_POLICY_ID="" PROD_POLICY_ID="" +bundle_configured=false +cleanup() { + if $bundle_configured; then + errcode trace $CLI bundle destroy -t prod --auto-approve + errcode trace $CLI bundle destroy -t dev --auto-approve + fi + if [ -n "$DEV_POLICY_ID" ]; then + errcode trace $CLI cluster-policies delete "$DEV_POLICY_ID" + fi + if [ -n "$PROD_POLICY_ID" ]; then + errcode trace $CLI cluster-policies delete "$PROD_POLICY_ID" + fi +} +trap cleanup EXIT + +title "Create policies outside the bundle" +export POLICY_ENVIRONMENT=dev +DEV_POLICY_ID=$($CLI cluster-policies create --json "$(envsubst < policy.json.tmpl)" -o json | jq -er .policy_id) +add_repl "$DEV_POLICY_ID" DEV_POLICY_ID +export POLICY_ENVIRONMENT=prod +PROD_POLICY_ID=$($CLI cluster-policies create --json "$(envsubst < policy.json.tmpl)" -o json | jq -er .policy_id) +add_repl "$PROD_POLICY_ID" PROD_POLICY_ID + +envsubst < databricks.yml.tmpl > databricks.yml +bundle_configured=true + +title "Deploy with the dev policy" +trace $CLI bundle deploy -t dev +job_id=$(read_id.py -t dev policy_sync) + +edit_resource.py jobs "$job_id" <<'EOF' +r["description"] = "Edited in the UI" +EOF + +title "Sync the UI edit into shared YAML" +cp databricks.yml databricks.yml.backup +trace $CLI bundle config-remote-sync -t dev --save +trace diff.py databricks.yml.backup databricks.yml + +title "Deploy shared YAML with the prod policy" +trace $CLI bundle deploy -t prod + +title "A deleted policy skips policy-dependent fields but still syncs authored edits" +edit_resource.py jobs "$job_id" <<'EOF' +r["description"] = "Another UI edit" +r["job_clusters"][0]["new_cluster"]["custom_tags"]["owner"] = "edited-owner" +EOF +trace $CLI cluster-policies delete "$DEV_POLICY_ID" +DEV_POLICY_ID="" + +cp databricks.yml databricks.yml.backup +trace $CLI bundle config-remote-sync -t dev --save -o json | jq '{status, changes, skippedChanges}' +trace diff.py databricks.yml.backup databricks.yml + +title "Only policy-dependent fields remain on the next sync" +trace $CLI bundle config-remote-sync -t dev -o json | jq '{status, changes, skippedChanges}' + +title "Text output reports the incomplete sync" +trace $CLI bundle config-remote-sync -t dev diff --git a/acceptance/bundle/config-remote-sync/external_cluster_policy_cross_target/test.toml b/acceptance/bundle/config-remote-sync/external_cluster_policy_cross_target/test.toml new file mode 100644 index 00000000000..a799274577a --- /dev/null +++ b/acceptance/bundle/config-remote-sync/external_cluster_policy_cross_target/test.toml @@ -0,0 +1,3 @@ +Cloud = true +RecordRequests = false +Ignore = [".databricks", "databricks.yml", "databricks.yml.backup"] diff --git a/acceptance/bundle/config-remote-sync/missing_cluster_policy_structural_changes/databricks.yml.tmpl b/acceptance/bundle/config-remote-sync/missing_cluster_policy_structural_changes/databricks.yml.tmpl new file mode 100644 index 00000000000..c0b88dcf2ed --- /dev/null +++ b/acceptance/bundle/config-remote-sync/missing_cluster_policy_structural_changes/databricks.yml.tmpl @@ -0,0 +1,77 @@ +bundle: + name: test-bundle-$UNIQUE_NAME + +workspace: + root_path: ~/.bundle/$UNIQUE_NAME + +resources: + jobs: + array_change: + name: test-array_change-$UNIQUE_NAME + description: Before UI edit + tasks: + - task_key: main + new_cluster: + policy_id: $POLICY_ID + spark_version: $DEFAULT_SPARK_VERSION + node_type_id: $NODE_TYPE_ID + num_workers: 1 + init_scripts: + - workspace: + destination: /Workspace/init-a.sh + - workspace: + destination: /Workspace/init-b.sh + spark_python_task: + python_file: ./hello.py + log_storage_change: + name: test-log_storage_change-$UNIQUE_NAME + description: Before UI edit + tasks: + - task_key: main + new_cluster: + policy_id: $POLICY_ID + spark_version: $DEFAULT_SPARK_VERSION + node_type_id: $NODE_TYPE_ID + num_workers: 1 + cluster_log_conf: + dbfs: + destination: dbfs:/logs + spark_python_task: + python_file: ./hello.py + new_task: + name: test-new_task-$UNIQUE_NAME + description: Before UI edit + tasks: + - task_key: main + new_cluster: + policy_id: $POLICY_ID + spark_version: $DEFAULT_SPARK_VERSION + node_type_id: $NODE_TYPE_ID + num_workers: 1 + spark_python_task: + python_file: ./hello.py + worker_count_change: + name: test-worker_count_change-$UNIQUE_NAME + description: Before UI edit + tasks: + - task_key: main + new_cluster: + policy_id: $POLICY_ID + spark_version: $DEFAULT_SPARK_VERSION + node_type_id: $NODE_TYPE_ID + autoscale: + min_workers: 1 + max_workers: 4 + spark_python_task: + python_file: ./hello.py + unaffected: + name: test-unaffected-$UNIQUE_NAME + description: Before UI edit + tasks: + - task_key: main + new_cluster: + spark_version: $DEFAULT_SPARK_VERSION + node_type_id: $NODE_TYPE_ID + num_workers: 1 + spark_python_task: + python_file: ./hello.py diff --git a/acceptance/bundle/config-remote-sync/missing_cluster_policy_structural_changes/hello.py b/acceptance/bundle/config-remote-sync/missing_cluster_policy_structural_changes/hello.py new file mode 100644 index 00000000000..11b15b1a458 --- /dev/null +++ b/acceptance/bundle/config-remote-sync/missing_cluster_policy_structural_changes/hello.py @@ -0,0 +1 @@ +print("hello") diff --git a/acceptance/bundle/config-remote-sync/missing_cluster_policy_structural_changes/out.test.toml b/acceptance/bundle/config-remote-sync/missing_cluster_policy_structural_changes/out.test.toml new file mode 100644 index 00000000000..8c2cb31de53 --- /dev/null +++ b/acceptance/bundle/config-remote-sync/missing_cluster_policy_structural_changes/out.test.toml @@ -0,0 +1,3 @@ +Cloud = false +GOOS.windows = false +EnvMatrix.DMS = ["", "true"] diff --git a/acceptance/bundle/config-remote-sync/missing_cluster_policy_structural_changes/output.txt b/acceptance/bundle/config-remote-sync/missing_cluster_policy_structural_changes/output.txt new file mode 100644 index 00000000000..f5a92a71374 --- /dev/null +++ b/acceptance/bundle/config-remote-sync/missing_cluster_policy_structural_changes/output.txt @@ -0,0 +1,150 @@ + +>>> [CLI] bundle deploy +Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/[UNIQUE_NAME]/files... +Created jobs.array_change +Created jobs.log_storage_change +Created jobs.new_task +Created jobs.unaffected +Created jobs.worker_count_change +Files: 5 uploaded, 0 deleted +Resources: 5 created, 0 changed, 0 deleted, 0 unchanged + +>>> [CLI] cluster-policies delete [POLICY_ID] + +=== Selecting the unaffected job does not read policies for unselected jobs +>>> [CLI] bundle config-remote-sync --select-ids jobs:[UNAFFECTED_ID] -o json +{ + "status": "success", + "changes": { + "resources.jobs.unaffected": { + "description": { + "operation": "replace", + "value": "Unrelated UI edit" + } + } + }, + "skippedChanges": null +} + +>>> print_requests.py //policies/clusters/get --get --oneline + +=== A selected job's policy permission error fails sync without saving +>>> [CLI] bundle config-remote-sync --select-ids jobs:[ARRAY_CHANGE_ID] --save -o json +{ + "status": "failed", + "error": "failed to extract changes: filtering policy values for resources.jobs.array_change.tasks[task_key='main'].new_cluster.init_scripts (no configuration was saved): reading cluster policy [POLICY_ID]: Fault injected by test.", + "changes": {}, + "skippedChanges": null +} + +>>> diff.py databricks.yml.backup databricks.yml + +>>> print_requests.py //policies/clusters/get --get --oneline +{"method": "GET", "path": "/api/2.0/policies/clusters/get", "q": {"policy_id": "[POLICY_ID]"}} + +=== An indivisible array replacement leaves the whole selected job unchanged +>>> [CLI] bundle config-remote-sync --select-ids jobs:[ARRAY_CHANGE_ID] --save -o json +{ + "status": "incomplete", + "changes": {}, + "skippedChanges": [ + { + "resourceKey": "resources.jobs.array_change", + "reason": "cluster_policy_not_found", + "policyId": "[POLICY_ID]" + } + ] +} + +>>> diff.py databricks.yml.backup databricks.yml + +=== A log-storage switch leaves the whole selected job unchanged +>>> [CLI] bundle config-remote-sync --select-ids jobs:[LOG_STORAGE_CHANGE_ID] --save -o json +{ + "status": "incomplete", + "changes": {}, + "skippedChanges": [ + { + "resourceKey": "resources.jobs.log_storage_change", + "reason": "cluster_policy_not_found", + "policyId": "[POLICY_ID]" + } + ] +} + +>>> diff.py databricks.yml.backup databricks.yml + +=== Switching autoscale to a fixed worker count leaves the whole selected job unchanged +>>> [CLI] bundle config-remote-sync --select-ids jobs:[WORKER_COUNT_CHANGE_ID] --save -o json +{ + "status": "incomplete", + "changes": {}, + "skippedChanges": [ + { + "resourceKey": "resources.jobs.worker_count_change", + "reason": "cluster_policy_not_found", + "policyId": "[POLICY_ID]" + } + ] +} + +>>> diff.py databricks.yml.backup databricks.yml + +=== Unrelated edits still sync when other jobs have structural changes +>>> [CLI] bundle config-remote-sync --save -o json +{ + "status": "incomplete", + "changes": { + "resources.jobs.unaffected": { + "description": { + "operation": "replace", + "value": "Unrelated UI edit" + } + } + }, + "skippedChanges": [ + { + "resourceKey": "resources.jobs.array_change", + "reason": "cluster_policy_not_found", + "policyId": "[POLICY_ID]" + }, + { + "resourceKey": "resources.jobs.log_storage_change", + "reason": "cluster_policy_not_found", + "policyId": "[POLICY_ID]" + }, + { + "resourceKey": "resources.jobs.new_task", + "reason": "cluster_policy_not_found", + "policyId": "[POLICY_ID]" + }, + { + "resourceKey": "resources.jobs.worker_count_change", + "reason": "cluster_policy_not_found", + "policyId": "[POLICY_ID]" + } + ] +} + +>>> diff.py databricks.yml.backup databricks.yml +--- databricks.yml.backup ++++ databricks.yml +@@ -67,5 +67,5 @@ + unaffected: + name: test-unaffected-[UNIQUE_NAME] +- description: Before UI edit ++ description: Unrelated UI edit + tasks: + - task_key: main + +>>> [CLI] bundle destroy --auto-approve +The following resources will be deleted: + delete resources.jobs.array_change + delete resources.jobs.log_storage_change + delete resources.jobs.new_task + delete resources.jobs.unaffected + delete resources.jobs.worker_count_change + +All files and directories at the following location will be deleted: /Workspace/Users/[USERNAME]/.bundle/[UNIQUE_NAME] + +Destroy: 5 deleted diff --git a/acceptance/bundle/config-remote-sync/missing_cluster_policy_structural_changes/policy.json.tmpl b/acceptance/bundle/config-remote-sync/missing_cluster_policy_structural_changes/policy.json.tmpl new file mode 100644 index 00000000000..f60659b84ab --- /dev/null +++ b/acceptance/bundle/config-remote-sync/missing_cluster_policy_structural_changes/policy.json.tmpl @@ -0,0 +1,4 @@ +{ + "name": "test-policy-$UNIQUE_NAME", + "definition": "{}" +} diff --git a/acceptance/bundle/config-remote-sync/missing_cluster_policy_structural_changes/script b/acceptance/bundle/config-remote-sync/missing_cluster_policy_structural_changes/script new file mode 100644 index 00000000000..d7ddf2d349e --- /dev/null +++ b/acceptance/bundle/config-remote-sync/missing_cluster_policy_structural_changes/script @@ -0,0 +1,94 @@ +export POLICY_ID="" +bundle_configured=false +cleanup() { + if $bundle_configured; then + errcode trace $CLI bundle destroy --auto-approve + fi + if [ -n "$POLICY_ID" ]; then + errcode trace $CLI cluster-policies delete "$POLICY_ID" + fi + print_requests.py //policies/clusters/get --get --oneline > LOG.cleanup-requests +} +trap cleanup EXIT + +POLICY_ID=$($CLI cluster-policies create --json "$(envsubst < policy.json.tmpl)" -o json | jq -er .policy_id) +add_repl "$POLICY_ID" POLICY_ID +envsubst < databricks.yml.tmpl > databricks.yml +bundle_configured=true + +trace $CLI bundle deploy +array_job_id=$(read_id.py array_change) +log_job_id=$(read_id.py log_storage_change) +new_task_job_id=$(read_id.py new_task) +worker_count_job_id=$(read_id.py worker_count_change) +unaffected_job_id=$(read_id.py unaffected) + +edit_resource.py jobs "$log_job_id" <<'PY' +r["description"] = "Log storage changed in the UI" +r["tasks"][0]["new_cluster"]["cluster_log_conf"] = { + "volumes": {"destination": "/Volumes/catalog/schema/volume/logs"} +} +PY + +edit_resource.py jobs "$worker_count_job_id" <<'PY' +r["description"] = "Worker count changed in the UI" +cluster = r["tasks"][0]["new_cluster"] +del cluster["autoscale"] +cluster["num_workers"] = 4 +PY + +# Only the following array/task edits should pick up the updated policy's injected values. +$CLI cluster-policies edit --json "$(envsubst < updated-policy.json.tmpl)" +edit_resource.py jobs "$array_job_id" <<'PY' +r["description"] = "Array changed in the UI" +r["tasks"][0]["new_cluster"]["init_scripts"] = [ + {"volumes": {"destination": "/Volumes/catalog/schema/volume/policy.sh"}} +] +PY +edit_resource.py jobs "$new_task_job_id" <<'PY' +from copy import deepcopy + +r["description"] = "Task added in the UI" +task = deepcopy(r["tasks"][0]) +task["task_key"] = "added_in_ui" +r["tasks"].append(task) +PY +edit_resource.py jobs "$unaffected_job_id" <<'PY' +r["description"] = "Unrelated UI edit" +PY + +trace $CLI cluster-policies delete "$POLICY_ID" +POLICY_ID="" + +# Isolate sync requests; the unselected job must leave this fault for the selected job. +print_requests.py //policies/clusters/get --get --oneline > LOG.setup-requests +fault.py 'GET /api/2.0/policies/clusters/get' 403 0 1 PERMISSION_DENIED + +title "Selecting the unaffected job does not read policies for unselected jobs" +trace $CLI bundle config-remote-sync --select-ids "jobs:$unaffected_job_id" -o json | jq '{status, changes, skippedChanges}' +trace print_requests.py //policies/clusters/get --get --oneline + +title "A selected job's policy permission error fails sync without saving" +cp databricks.yml databricks.yml.backup +trace $CLI bundle config-remote-sync --select-ids "jobs:$array_job_id" --save -o json | jq '{status, error, changes, skippedChanges}' +trace diff.py databricks.yml.backup databricks.yml +cmp databricks.yml.backup databricks.yml +trace print_requests.py //policies/clusters/get --get --oneline + +title "An indivisible array replacement leaves the whole selected job unchanged" +cp databricks.yml databricks.yml.backup +trace $CLI bundle config-remote-sync --select-ids "jobs:$array_job_id" --save -o json | jq '{status, changes, skippedChanges}' +trace diff.py databricks.yml.backup databricks.yml + +title "A log-storage switch leaves the whole selected job unchanged" +trace $CLI bundle config-remote-sync --select-ids "jobs:$log_job_id" --save -o json | jq '{status, changes, skippedChanges}' +trace diff.py databricks.yml.backup databricks.yml + +title "Switching autoscale to a fixed worker count leaves the whole selected job unchanged" +trace $CLI bundle config-remote-sync --select-ids "jobs:$worker_count_job_id" --save -o json | jq '{status, changes, skippedChanges}' +trace diff.py databricks.yml.backup databricks.yml +cmp databricks.yml.backup databricks.yml + +title "Unrelated edits still sync when other jobs have structural changes" +trace $CLI bundle config-remote-sync --save -o json | jq '{status, changes, skippedChanges}' +trace diff.py databricks.yml.backup databricks.yml diff --git a/acceptance/bundle/config-remote-sync/missing_cluster_policy_structural_changes/test.toml b/acceptance/bundle/config-remote-sync/missing_cluster_policy_structural_changes/test.toml new file mode 100644 index 00000000000..b2b24e3e578 --- /dev/null +++ b/acceptance/bundle/config-remote-sync/missing_cluster_policy_structural_changes/test.toml @@ -0,0 +1,2 @@ +RecordRequests = true +Ignore = [".databricks", "databricks.yml", "databricks.yml.backup"] diff --git a/acceptance/bundle/config-remote-sync/missing_cluster_policy_structural_changes/updated-policy.json.tmpl b/acceptance/bundle/config-remote-sync/missing_cluster_policy_structural_changes/updated-policy.json.tmpl new file mode 100644 index 00000000000..f4435f0b2f7 --- /dev/null +++ b/acceptance/bundle/config-remote-sync/missing_cluster_policy_structural_changes/updated-policy.json.tmpl @@ -0,0 +1,5 @@ +{ + "policy_id": "$POLICY_ID", + "name": "test-policy-$UNIQUE_NAME", + "definition": "{\"spark_env_vars.ENVIRONMENT\":{\"type\":\"fixed\",\"value\":\"from-policy\"},\"init_scripts.0.volumes.destination\":{\"type\":\"fixed\",\"value\":\"/Volumes/catalog/schema/volume/policy.sh\"}}" +} diff --git a/acceptance/bundle/config-remote-sync/policy_injected_cluster_fields/databricks.yml.tmpl b/acceptance/bundle/config-remote-sync/policy_injected_cluster_fields/databricks.yml.tmpl index 62dd2814a7e..9d798b71b8a 100644 --- a/acceptance/bundle/config-remote-sync/policy_injected_cluster_fields/databricks.yml.tmpl +++ b/acceptance/bundle/config-remote-sync/policy_injected_cluster_fields/databricks.yml.tmpl @@ -6,12 +6,29 @@ targets: mode: development resources: + cluster_policies: + policy: + name: test-policy-$UNIQUE_NAME + definition: + custom_tags.CostCenter: + type: fixed + value: dev-1234 + custom_tags.Team: + type: fixed + value: finops + cluster_log_conf.type: + type: fixed + value: DBFS + cluster_log_conf.path: + type: fixed + value: dbfs:/cluster-logs/dev jobs: job1: max_concurrent_runs: 1 job_clusters: - job_cluster_key: shared new_cluster: + policy_id: ${resources.cluster_policies.policy.id} spark_version: $DEFAULT_SPARK_VERSION node_type_id: $NODE_TYPE_ID num_workers: 1 @@ -24,6 +41,7 @@ resources: notebook_task: notebook_path: /Users/{{workspace_user_name}}/notebook new_cluster: + policy_id: ${resources.cluster_policies.policy.id} spark_version: $DEFAULT_SPARK_VERSION node_type_id: $NODE_TYPE_ID num_workers: 1 diff --git a/acceptance/bundle/config-remote-sync/policy_injected_cluster_fields/output.txt b/acceptance/bundle/config-remote-sync/policy_injected_cluster_fields/output.txt index 019b581933a..d9a6b56d93a 100644 --- a/acceptance/bundle/config-remote-sync/policy_injected_cluster_fields/output.txt +++ b/acceptance/bundle/config-remote-sync/policy_injected_cluster_fields/output.txt @@ -1,7 +1,13 @@ Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/test-bundle-[UNIQUE_NAME]/default/files... +Created cluster_policies.policy Created jobs.job1 Files: 2 uploaded, 0 deleted -Resources: 1 created, 0 changed, 0 deleted, 0 unchanged +Resources: 2 created, 0 changed, 0 deleted, 0 unchanged + +=== Policy values alone do not need to be synced +>>> [CLI] bundle config-remote-sync +No changes detected. + === Simulate a cluster policy injecting custom_tags and cluster_log_conf These fields exist only remotely (the user never set them in config). @@ -20,7 +26,7 @@ Resource: resources.jobs.job1 >>> diff.py databricks.yml.backup databricks.yml --- databricks.yml.backup +++ databricks.yml -@@ -9,5 +9,5 @@ +@@ -25,5 +25,5 @@ jobs: job1: - max_concurrent_runs: 1 @@ -30,8 +36,9 @@ Resource: resources.jobs.job1 >>> [CLI] bundle destroy --auto-approve The following resources will be deleted: + delete resources.cluster_policies.policy delete resources.jobs.job1 All files and directories at the following location will be deleted: /Workspace/Users/[USERNAME]/.bundle/test-bundle-[UNIQUE_NAME]/default -Destroy: 1 deleted +Destroy: 2 deleted diff --git a/acceptance/bundle/config-remote-sync/policy_injected_cluster_fields/script b/acceptance/bundle/config-remote-sync/policy_injected_cluster_fields/script index bcc9844b03f..762fcfba5c2 100644 --- a/acceptance/bundle/config-remote-sync/policy_injected_cluster_fields/script +++ b/acceptance/bundle/config-remote-sync/policy_injected_cluster_fields/script @@ -10,6 +10,9 @@ trap cleanup EXIT $CLI bundle deploy job1_id="$(read_id.py job1)" +title "Policy values alone do not need to be synced" +trace $CLI bundle config-remote-sync + title "Simulate a cluster policy injecting custom_tags and cluster_log_conf" echo echo "These fields exist only remotely (the user never set them in config)." diff --git a/bundle/configsync/cluster_policy.go b/bundle/configsync/cluster_policy.go new file mode 100644 index 00000000000..30a4f50999a --- /dev/null +++ b/bundle/configsync/cluster_policy.go @@ -0,0 +1,293 @@ +package configsync + +import ( + "context" + "fmt" + "iter" + "maps" + "reflect" + "slices" + "strconv" + "strings" + + "github.com/databricks/cli/bundle" + "github.com/databricks/cli/libs/dyn" + "github.com/databricks/cli/libs/dyn/jsonloader" + "github.com/databricks/cli/libs/structs/structaccess" + "github.com/databricks/cli/libs/structs/structdiff" + "github.com/databricks/cli/libs/structs/structpath" + "github.com/databricks/cli/libs/structs/structtag" +) + +type clusterPolicyDefinition struct { + value dyn.Value + err error +} + +type clusterPolicyDefinitions map[string]clusterPolicyDefinition + +type clusterPolicyLookupError struct { + policyID string + err error +} + +func (e *clusterPolicyLookupError) Error() string { + return fmt.Sprintf("reading cluster policy %s: %v", e.policyID, e.err) +} + +func (e *clusterPolicyLookupError) Unwrap() error { return e.err } + +// get caches effective definitions, including policy-family overrides, and lookup failures per sync. +func (definitions clusterPolicyDefinitions) get(ctx context.Context, b *bundle.Bundle, id string) (dyn.Value, error) { + if definition, ok := definitions[id]; ok { + return definition.value, definition.err + } + policy, err := b.WorkspaceClient(ctx).ClusterPolicies.GetByPolicyId(ctx, id) + if err != nil { + lookupErr := &clusterPolicyLookupError{policyID: id, err: err} + definitions[id] = clusterPolicyDefinition{err: lookupErr} + return dyn.InvalidValue, lookupErr + } + definition, err := jsonloader.LoadJSON([]byte(policy.Definition), "cluster policy "+id) + if err != nil { + return dyn.InvalidValue, err + } + definitions[id] = clusterPolicyDefinition{value: definition} + return definition, nil +} + +type clusterPolicyFilter struct { + definitions clusterPolicyDefinitions + config any + remote any + clusterRoots []*structpath.PatternNode + sliceKeys map[string]structdiff.KeyFunc +} + +// filter removes omitted policy values from a normalized plan change, including +// whole-object additions. A nil result means the value comes entirely from the policy. +func (f *clusterPolicyFilter) filter(ctx context.Context, b *bundle.Bundle, path *structpath.PathNode, value any) (any, error) { + filtered, keep, err := f.prune(ctx, b, path, value, true) + if err != nil || !keep { + return nil, err + } + return filtered, nil +} + +// prune reports keep=false when the whole value comes from the cluster policy. +func (f *clusterPolicyFilter) prune(ctx context.Context, b *bundle.Bundle, path *structpath.PathNode, value any, isChangeRoot bool) (any, bool, error) { + fieldPath, err := structdiff.KeyedPath(f.remote, path, f.sliceKeys) + if err != nil { + return nil, false, err + } + if clusterPath, ok := f.clusterPath(fieldPath); ok { + // A change may start inside an array, so protecting only sequence + // nodes would still allow an element to be emptied or removed. + for _, node := range fieldPath.SkipPrefix(clusterPath.Len()).AsSlice() { + if _, indexed := node.Index(); indexed { + return value, true, nil + } + } + // Keep required fields, but still filter their optional children. + if isChangeRoot || !isRequiredField(f.remote, fieldPath) { + matches, err := f.isPolicyOnlySubtree(ctx, b, clusterPath, fieldPath, value) + if err != nil { + return nil, false, err + } + if matches { + return nil, false, nil + } + } + // Removing elements would change indexed policy constraints. + if _, sequence := value.([]any); sequence { + return value, true, nil + } + } + + switch value := value.(type) { + case map[string]any: + result := make(map[string]any, len(value)) + for _, key := range slices.Sorted(maps.Keys(value)) { + child := value[key] + filtered, keep, err := f.prune(ctx, b, structpath.NewStringKey(path, key), child, false) + if err != nil { + return nil, false, err + } + if keep { + result[key] = filtered + } + } + return result, true, nil + case []any: + // Arrays inside clusters return above; elements here cannot be dropped. + result := make([]any, len(value)) + for i, child := range value { + filtered, _, err := f.prune(ctx, b, structpath.NewIndex(path, i), child, false) + if err != nil { + return nil, false, err + } + result[i] = filtered + } + return result, true, nil + default: + return value, true, nil + } +} + +// clusterPath finds the cluster containing a field; the cluster itself is excluded. +func (f *clusterPolicyFilter) clusterPath(path *structpath.PathNode) (*structpath.PathNode, bool) { + for _, root := range f.clusterRoots { + if path.HasPatternPrefix(root) && path.Len() > root.Len() { + return path.Prefix(root.Len()), true + } + } + return nil, false +} + +// canSkipField allows scalar and map changes strictly inside a cluster spec. +// The caller must also check for removals paired with a skipped addition. +func (f *clusterPolicyFilter) canSkipField(path *structpath.PathNode) bool { + if _, ok := f.clusterPath(path); !ok { + return false + } + remote, err := structaccess.Get(f.remote, path) + if err != nil || remote == nil { + return false + } + typ := reflect.TypeOf(remote) + for typ.Kind() == reflect.Pointer { + typ = typ.Elem() + } + switch typ.Kind() { + case reflect.Struct, reflect.Slice, reflect.Array: + return false + default: + return true + } +} + +// isPolicyOnlySubtree requires every leaf to be absent from authored config +// and equal to a fixed value or an opted-in default in the remote cluster's policy. +func (f *clusterPolicyFilter) isPolicyOnlySubtree(ctx context.Context, b *bundle.Bundle, clusterPath, path *structpath.PathNode, value any) (bool, error) { + hasLeaves := false + for leafPath := range normalizedLeaves(path, value) { + hasLeaves = true + if authored, _ := structaccess.Get(f.config, leafPath); authored != nil { + return false, nil + } + } + if !hasLeaves { + return false, nil + } + + remotePolicyID, _ := structaccess.Get(f.remote, structpath.NewStringKey(clusterPath, "policy_id")) + policyID, _ := remotePolicyID.(string) + if policyID == "" { + return false, nil + } + definition, err := f.definitions.get(ctx, b, policyID) + if err != nil { + return false, err + } + // Remote cluster specs omit this input-only flag; use the authored setting. + applyDefaults, _ := structaccess.Get(f.config, structpath.NewStringKey(clusterPath, "apply_policy_default_values")) + for leafPath, leaf := range normalizedLeaves(path, value) { + expected := policyValueForAPIAttribute(definition, policyAttribute(leafPath.SkipPrefix(clusterPath.Len())), applyDefaults == true) + if !expected.IsValid() || expected.AsAny() == nil || !reflect.DeepEqual(expected.AsAny(), leaf) { + return false, nil + } + } + return true, nil +} + +// normalizedLeaves preserves the normalized change's scalars, including nulls. +// Empty containers have no leaves and cannot alone establish a policy match. +func normalizedLeaves(path *structpath.PathNode, value any) iter.Seq2[*structpath.PathNode, any] { + return func(yield func(*structpath.PathNode, any) bool) { + switch value := value.(type) { + case map[string]any: + for _, key := range slices.Sorted(maps.Keys(value)) { + for leafPath, leaf := range normalizedLeaves(structpath.NewStringKey(path, key), value[key]) { + if !yield(leafPath, leaf) { + return + } + } + } + case []any: + for i, child := range value { + for leafPath, leaf := range normalizedLeaves(structpath.NewIndex(path, i), child) { + if !yield(leafPath, leaf) { + return + } + } + } + default: + yield(path, value) + } + } +} + +func isRequiredField(remote any, path *structpath.PathNode) bool { + key, ok := path.StringKey() + if !ok { + return false + } + parent, _ := structaccess.Get(remote, path.Parent()) + typ := reflect.TypeOf(parent) + if typ == nil { + return false + } + for typ.Kind() == reflect.Pointer { + typ = typ.Elem() + } + field, _, ok := structaccess.FindStructFieldByKeyType(typ, key) + return ok && !structtag.JSONTag(field.Tag.Get("json")).OmitEmpty() +} + +// policyAttribute preserves dotted map keys, e.g. spark_conf["a.b"] becomes spark_conf.a.b. +func policyAttribute(path *structpath.PathNode) string { + var parts []string + for _, node := range path.AsSlice() { + if key, ok := node.StringKey(); ok { + parts = append(parts, key) + } else if index, ok := node.Index(); ok { + parts = append(parts, strconv.Itoa(index)) + } + } + return strings.Join(parts, ".") +} + +// policyValueForAPIAttribute translates the policy language's log-delivery aliases. +// https://docs.databricks.com/aws/en/admin/clusters/policy-definition +func policyValueForAPIAttribute(definition dyn.Value, attribute string, applyDefaults bool) dyn.Value { + if attribute == "cluster_log_conf.s3.region" { + attribute = "cluster_log_conf.region" + } + if storage, ok := strings.CutPrefix(attribute, "cluster_log_conf."); ok { + if storage, ok := strings.CutSuffix(storage, ".destination"); ok { + logType := policyRuleValue(definition, "cluster_log_conf.type", applyDefaults) + if !logType.IsValid() || logType.AsAny() != strings.ToUpper(storage) { + return dyn.InvalidValue + } + return policyRuleValue(definition, "cluster_log_conf.path", applyDefaults) + } + } + return policyRuleValue(definition, attribute, applyDefaults) +} + +// policyRuleValue returns the value supplied by a rule, or InvalidValue if it supplies none. +func policyRuleValue(definition dyn.Value, attribute string, applyDefaults bool) dyn.Value { + rule, err := dyn.GetByPath(definition, dyn.NewPath(dyn.Key(attribute))) + if err != nil { + return dyn.InvalidValue + } + ruleType, _ := dyn.Get(rule, "type") + var value dyn.Value + switch { + case ruleType.AsAny() == "fixed": + value, _ = dyn.Get(rule, "value") + case applyDefaults && ruleType.AsAny() != "forbidden": + value, _ = dyn.Get(rule, "defaultValue") + } + return value +} diff --git a/bundle/configsync/cluster_policy_test.go b/bundle/configsync/cluster_policy_test.go new file mode 100644 index 00000000000..b4d0f617162 --- /dev/null +++ b/bundle/configsync/cluster_policy_test.go @@ -0,0 +1,632 @@ +package configsync_test + +import ( + "encoding/json" + "errors" + "strings" + "testing" + + "github.com/databricks/cli/bundle" + "github.com/databricks/cli/bundle/configsync" + "github.com/databricks/cli/bundle/deployplan" + "github.com/databricks/cli/bundle/direct/dresources" + "github.com/databricks/cli/libs/dyn" + "github.com/databricks/cli/libs/dyn/jsonloader" + "github.com/databricks/cli/libs/structs/structaccess" + "github.com/databricks/databricks-sdk-go/apierr" + "github.com/databricks/databricks-sdk-go/experimental/mocks" + "github.com/databricks/databricks-sdk-go/service/compute" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/mock" + "github.com/stretchr/testify/require" +) + +func TestExtractChangesFiltersPolicyAppliedValues(t *testing.T) { + tests := []struct { + name string + attribute string + definition string + config any + remote any + applyDefaults bool + noPolicy bool + wantLookup bool + want any + }{ + { + name: "fixed S3 log configuration dropped", + wantLookup: true, + attribute: "cluster_log_conf", + definition: `{"cluster_log_conf.type":{"type":"fixed","value":"S3"},"cluster_log_conf.path":{"type":"fixed","value":"s3://logs"},"cluster_log_conf.region":{"type":"fixed","value":"us-west-2"}}`, + remote: map[string]any{"s3": map[string]any{"destination": "s3://logs", "region": "us-west-2"}}, + }, + { + name: "partial log match keeps complete destination", + wantLookup: true, + attribute: "cluster_log_conf", + definition: `{"cluster_log_conf.type":{"type":"fixed","value":"S3"},"cluster_log_conf.path":{"type":"fixed","value":"s3://logs"},"cluster_log_conf.region":{"type":"unlimited","defaultValue":"us-west-2"}}`, + applyDefaults: true, + remote: map[string]any{"s3": map[string]any{"destination": "s3://logs", "region": "us-east-1"}}, + want: map[string]any{"s3": map[string]any{"destination": "s3://logs", "region": "us-east-1"}}, + }, + { + name: "edited log destination keeps required field and drops policy region", + wantLookup: true, + attribute: "cluster_log_conf", + definition: `{"cluster_log_conf.path":{"type":"unlimited"},"cluster_log_conf.region":{"type":"fixed","value":"us-west-2"}}`, + remote: map[string]any{"s3": map[string]any{"destination": "s3://user-logs", "region": "us-west-2"}}, + want: map[string]any{"s3": map[string]any{"destination": "s3://user-logs"}}, + }, + { + name: "required parent retains user field and drops optional policy sibling", + wantLookup: true, + attribute: "workload_type", + definition: `{"workload_type.clients.notebooks":{"type":"fixed","value":false}}`, + remote: map[string]any{"clients": map[string]any{"jobs": true, "notebooks": false}}, + want: map[string]any{"clients": map[string]any{"jobs": true}}, + }, + { + name: "partial struct match drops optional policy fields", + wantLookup: true, + attribute: "autoscale", + definition: `{"autoscale.min_workers":{"type":"fixed","value":1}}`, + remote: map[string]any{"min_workers": int64(1), "max_workers": int64(4)}, + want: map[string]any{"max_workers": int64(4)}, + }, + { + name: "fixed scalar dropped when absent from config", + wantLookup: true, + attribute: "autotermination_minutes", + definition: `{"autotermination_minutes":{"type":"fixed","value":60}}`, + remote: int64(60), + }, + { + name: "fixed value dropped without apply flag", + wantLookup: true, + attribute: "spark_env_vars", + definition: `{"spark_env_vars.ENV":{"type":"fixed","value":"dev"}}`, + remote: map[string]any{"ENV": "dev"}, + }, + { + name: "fixed tag dropped and UI tag kept", + wantLookup: true, + attribute: "custom_tags", + definition: `{"custom_tags.Environment":{"type":"fixed","value":"dev"}}`, + remote: map[string]any{"Environment": "dev", "Owner": "user"}, + want: map[string]any{"Owner": "user"}, + }, + { + // The policy changed after deploy; the explicitly authored field still syncs. + name: "authored field takes remote value", + wantLookup: false, + attribute: "spark_env_vars", + definition: `{"spark_env_vars.ENV":{"type":"fixed","value":"dev"}}`, + config: map[string]any{"ENV": "other"}, + remote: map[string]any{"ENV": "dev"}, + want: map[string]any{"ENV": "dev"}, + }, + { + name: "authored false is kept in whole object changes", + wantLookup: false, + attribute: "workload_type", + definition: `{"workload_type.clients.notebooks":{"type":"fixed","value":false}}`, + config: map[string]any{"clients": map[string]any{"notebooks": false, "jobs": true}}, + remote: map[string]any{"clients": map[string]any{"notebooks": false, "jobs": false}}, + want: map[string]any{"clients": map[string]any{"notebooks": false, "jobs": false}}, + }, + { + name: "authored zero is kept in whole object changes", + wantLookup: false, + attribute: "autoscale", + definition: `{"autoscale.min_workers":{"type":"fixed","value":0}}`, + config: map[string]any{"min_workers": int64(0), "max_workers": int64(4)}, + remote: map[string]any{"min_workers": int64(0), "max_workers": int64(8)}, + want: map[string]any{"min_workers": int64(0), "max_workers": int64(8)}, + }, + { + name: "default dropped with apply flag", + wantLookup: true, + attribute: "spark_env_vars", + definition: `{"spark_env_vars.ENV":{"type":"allowlist","values":["dev","prod"],"defaultValue":"dev"}}`, + remote: map[string]any{"ENV": "dev"}, + applyDefaults: true, + }, + { + name: "default kept without apply flag", + wantLookup: true, + attribute: "spark_env_vars", + definition: `{"spark_env_vars.ENV":{"type":"allowlist","values":["dev","prod"],"defaultValue":"dev"}}`, + remote: map[string]any{"ENV": "dev"}, + want: map[string]any{"ENV": "dev"}, + }, + { + name: "non-default value kept", + wantLookup: true, + attribute: "spark_env_vars", + definition: `{"spark_env_vars.ENV":{"type":"allowlist","values":["dev","prod"],"defaultValue":"dev"}}`, + remote: map[string]any{"ENV": "prod"}, + applyDefaults: true, + want: map[string]any{"ENV": "prod"}, + }, + { + name: "policy-only subtree with fixed false is dropped", + wantLookup: true, + attribute: "workload_type", + definition: `{"workload_type.clients.notebooks":{"type":"fixed","value":false}}`, + remote: map[string]any{"clients": map[string]any{"notebooks": false}}, + }, + { + name: "zero default dropped", + wantLookup: true, + attribute: "autoscale", + definition: `{"autoscale.min_workers":{"type":"range","minValue":0,"defaultValue":0}}`, + remote: map[string]any{"min_workers": int64(0)}, + applyDefaults: true, + }, + { + name: "dotted map key matched as one key", + wantLookup: true, + attribute: "spark_conf", + definition: `{"spark_conf.spark.sql.shuffle.partitions":{"type":"fixed","value":"7"}}`, + remote: map[string]any{"spark.sql.shuffle.partitions": "7", "spark.sql.adaptive.enabled": "true"}, + want: map[string]any{"spark.sql.adaptive.enabled": "true"}, + }, + { + name: "indexed rule drops whole array", + wantLookup: true, + attribute: "init_scripts", + definition: `{"init_scripts.0.volumes.destination":{"type":"fixed","value":"/Volumes/c/s/v/init.sh"}}`, + remote: []any{map[string]any{"volumes": map[string]any{"destination": "/Volumes/c/s/v/init.sh"}}}, + }, + { + name: "partial array match keeps whole array", + wantLookup: true, + attribute: "init_scripts", + definition: `{"init_scripts.0.volumes.destination":{"type":"fixed","value":"/Volumes/c/s/v/policy.sh"}}`, + remote: []any{ + map[string]any{"volumes": map[string]any{"destination": "/Volumes/c/s/v/policy.sh"}}, + map[string]any{"volumes": map[string]any{"destination": "/Volumes/c/s/v/user.sh"}}, + }, + want: []any{ + map[string]any{"volumes": map[string]any{"destination": "/Volumes/c/s/v/policy.sh"}}, + map[string]any{"volumes": map[string]any{"destination": "/Volumes/c/s/v/user.sh"}}, + }, + }, + { + name: "log destination matches policy type and path", + wantLookup: true, + attribute: "cluster_log_conf", + definition: `{"cluster_log_conf.type":{"type":"fixed","value":"DBFS"},"cluster_log_conf.path":{"type":"fixed","value":"dbfs:/logs"}}`, + remote: map[string]any{"dbfs": map[string]any{"destination": "dbfs:/logs"}}, + }, + { + name: "log destination edits remain managed", + wantLookup: true, + attribute: "cluster_log_conf", + definition: `{"cluster_log_conf.type":{"type":"fixed","value":"DBFS"},"cluster_log_conf.path":{"type":"unlimited","defaultValue":"dbfs:/logs"}}`, + applyDefaults: true, + remote: map[string]any{"dbfs": map[string]any{"destination": "dbfs:/user-logs"}}, + want: map[string]any{"dbfs": map[string]any{"destination": "dbfs:/user-logs"}}, + }, + { + name: "tags without a policy sync", + wantLookup: false, + attribute: "custom_tags", + noPolicy: true, + remote: map[string]any{"Owner": "user"}, + want: map[string]any{"Owner": "user"}, + }, + { + name: "log configuration without a policy syncs", + wantLookup: false, + attribute: "cluster_log_conf", + noPolicy: true, + remote: map[string]any{"dbfs": map[string]any{"destination": "dbfs:/user-logs"}}, + want: map[string]any{"dbfs": map[string]any{"destination": "dbfs:/user-logs"}}, + }, + { + name: "unconstrained values sync", + wantLookup: true, + attribute: "custom_tags", + definition: `{}`, + remote: map[string]any{"Owner": "user"}, + want: map[string]any{"Owner": "user"}, + }, + { + name: "volume log destination matches policy type and path", + wantLookup: true, + attribute: "cluster_log_conf", + definition: `{"cluster_log_conf.type":{"type":"fixed","value":"VOLUMES"},"cluster_log_conf.path":{"type":"fixed","value":"/Volumes/c/s/v/logs"}}`, + remote: map[string]any{"volumes": map[string]any{"destination": "/Volumes/c/s/v/logs"}}, + }, + { + name: "authored sibling preserves required destination without policy lookup", + wantLookup: false, + attribute: "cluster_log_conf", + config: map[string]any{"s3": map[string]any{"region": "us-west-2"}}, + remote: map[string]any{"s3": map[string]any{"destination": "s3://logs", "region": "us-east-1"}}, + want: map[string]any{"s3": map[string]any{"destination": "s3://logs", "region": "us-east-1"}}, + }, + } + + for _, location := range []string{"task", "shared", "for_each", "standalone"} { + for _, tt := range tests { + t.Run(location+"/"+tt.name, func(t *testing.T) { + config := map[string]any{"policy_id": "policy", "apply_policy_default_values": tt.applyDefaults} + if tt.config != nil { + config[tt.attribute] = tt.config + } + remote := map[string]any{"policy_id": "policy", tt.attribute: tt.remote} + if tt.noPolicy { + delete(config, "policy_id") + delete(remote, "policy_id") + } + key, prefix, configRoot := policyClusterLocation(location, config) + _, _, remoteRoot := policyClusterLocation(location, remote) + path := prefix + tt.attribute + remoteState := typedPolicyResource(t, location, remoteRoot) + remoteValue, err := structaccess.GetByString(remoteState, path) + require.NoError(t, err) + entry := &deployplan.PlanEntry{ + RemoteState: remoteState, + Changes: deployplan.Changes{path: { + Action: deployplan.Update, + New: tt.config, + Remote: remoteValue, + }}, + } + b := policyTestBundle(t, key, configRoot) + // Only policy retrieval is mocked; the actual comparison and filtering run. + client := mocks.NewMockWorkspaceClient(t) + b.SetWorkpaceClient(client.WorkspaceClient) + if tt.wantLookup { + client.GetMockClusterPoliciesAPI().EXPECT().GetByPolicyId(mock.Anything, "policy"). + Return(&compute.Policy{Definition: tt.definition}, nil).Once() + } + + changes, skipped, err := configsync.ExtractChanges(t.Context(), b, &deployplan.Plan{Plan: map[string]*deployplan.PlanEntry{key: entry}}) + require.NoError(t, err) + assert.Empty(t, skipped) + if tt.want == nil { + assert.Empty(t, changes) + } else { + require.Contains(t, changes, key) + require.Contains(t, changes[key], path) + assert.Equal(t, tt.want, changes[key][path].Value) + } + }) + } + } +} + +func TestExtractChangesFailsOnPolicyLookupError(t *testing.T) { + for _, tt := range []struct { + name string + err error + }{ + {"permission denied", apierr.ErrPermissionDenied}, + {"service unavailable", errors.New("policy service unavailable")}, + } { + t.Run(tt.name, func(t *testing.T) { + lookupError := tt.err + b := policyTestBundle(t, "resources.clusters.test", map[string]any{"policy_id": "policy"}) + client := mocks.NewMockWorkspaceClient(t) + b.SetWorkpaceClient(client.WorkspaceClient) + client.GetMockClusterPoliciesAPI().EXPECT().GetByPolicyId(mock.Anything, "policy").Return(nil, lookupError).Once() + entry := &deployplan.PlanEntry{ + RemoteState: &dresources.ClusterRemote{PolicyId: "policy"}, + Changes: deployplan.Changes{ + "spark_env_vars": {Action: deployplan.Update, Remote: map[string]any{"ENV": "dev"}}, + }, + } + changes, skipped, err := configsync.ExtractChanges(t.Context(), b, &deployplan.Plan{Plan: map[string]*deployplan.PlanEntry{"resources.clusters.test": entry}}) + require.ErrorIs(t, err, lookupError) + assert.Empty(t, skipped) + assert.ErrorContains(t, err, "no configuration was saved") + assert.Nil(t, changes) + }) + } +} + +func TestExtractChangesDoesNotHidePolicyErrorsAfterStructuralSkip(t *testing.T) { + key := "resources.jobs.test" + b := policyTestBundle(t, key, map[string]any{"job_clusters": []any{ + map[string]any{"job_cluster_key": "a", "new_cluster": map[string]any{"policy_id": "missing"}}, + map[string]any{"job_cluster_key": "b", "new_cluster": map[string]any{"policy_id": "denied"}}, + }}) + client := mocks.NewMockWorkspaceClient(t) + b.SetWorkpaceClient(client.WorkspaceClient) + client.GetMockClusterPoliciesAPI().EXPECT().GetByPolicyId(mock.Anything, "missing").Return(nil, apierr.ErrNotFound).Once() + client.GetMockClusterPoliciesAPI().EXPECT().GetByPolicyId(mock.Anything, "denied").Return(nil, apierr.ErrPermissionDenied).Once() + remote := &dresources.JobRemote{} + require.NoError(t, json.Unmarshal([]byte(`{"job_clusters":[ + {"job_cluster_key":"a","new_cluster":{"policy_id":"missing"}}, + {"job_cluster_key":"b","new_cluster":{"policy_id":"denied"}} + ]}`), remote)) + entry := &deployplan.PlanEntry{ + RemoteState: remote, + Changes: deployplan.Changes{ + "job_clusters[job_cluster_key='a']": {Action: deployplan.Update, Remote: map[string]any{"new_cluster": map[string]any{"spark_env_vars": map[string]any{"ENV": "dev"}}}}, + "job_clusters[job_cluster_key='b'].new_cluster.spark_env_vars": {Action: deployplan.Update, Remote: map[string]string{"ENV": "dev"}}, + }, + } + changes, skipped, err := configsync.ExtractChanges(t.Context(), b, &deployplan.Plan{Plan: map[string]*deployplan.PlanEntry{key: entry}}) + assert.ErrorIs(t, err, apierr.ErrPermissionDenied) + assert.Nil(t, changes) + assert.Nil(t, skipped) +} + +func TestExtractChangesLooksUpMissingPolicyOnce(t *testing.T) { + b := policyTestBundle(t, "resources.clusters.test", map[string]any{ + "policy_id": "policy", "cluster_name": "before", + }) + client := mocks.NewMockWorkspaceClient(t) + b.SetWorkpaceClient(client.WorkspaceClient) + client.GetMockClusterPoliciesAPI().EXPECT().GetByPolicyId(mock.Anything, "policy"). + Return(nil, apierr.ErrResourceDoesNotExist).Once() + entry := &deployplan.PlanEntry{ + RemoteState: &dresources.ClusterRemote{ + PolicyId: "policy", ClusterName: "after", + CustomTags: map[string]string{"environment": "dev"}, + SparkEnvVars: map[string]string{"ENVIRONMENT": "dev"}, + }, + Changes: deployplan.Changes{ + "cluster_name": {Action: deployplan.Update, Old: "before", New: "before", Remote: "after"}, + "custom_tags": {Action: deployplan.Skip, Reason: deployplan.ReasonRemoteAddition, Remote: map[string]string{"environment": "dev"}}, + "spark_env_vars": {Action: deployplan.Skip, Reason: deployplan.ReasonRemoteAddition, Remote: map[string]string{"ENVIRONMENT": "dev"}}, + }, + } + changes, skipped, err := configsync.ExtractChanges(t.Context(), b, &deployplan.Plan{Plan: map[string]*deployplan.PlanEntry{"resources.clusters.test": entry}}) + require.NoError(t, err) + assert.Equal(t, configsync.Changes{"resources.clusters.test": { + "cluster_name": {Operation: configsync.OperationReplace, Value: "after"}, + }}, changes) + assert.Equal(t, []configsync.SkippedChange{ + {ResourceKey: "resources.clusters.test", Path: "custom_tags", Reason: configsync.SkipReasonClusterPolicyNotFound, PolicyID: "policy"}, + {ResourceKey: "resources.clusters.test", Path: "spark_env_vars", Reason: configsync.SkipReasonClusterPolicyNotFound, PolicyID: "policy"}, + }, skipped) +} + +func TestExtractChangesMatchesClustersByKeyAndRemotePolicy(t *testing.T) { + config := map[string]any{"job_clusters": []any{ + map[string]any{"job_cluster_key": "explicit", "new_cluster": map[string]any{ + "policy_id": "old-policy", "spark_env_vars": map[string]any{"ENV": "authored"}, + }}, + map[string]any{"job_cluster_key": "omitted", "new_cluster": map[string]any{"policy_id": "old-policy"}}, + map[string]any{"job_cluster_key": "also_omitted", "new_cluster": map[string]any{"policy_id": "old-policy"}}, + }} + remote := map[string]any{"job_clusters": []any{ + map[string]any{"job_cluster_key": "also_omitted", "new_cluster": map[string]any{ + "policy_id": "new-policy", "spark_env_vars": map[string]any{"ENV": "dev"}, + }}, + map[string]any{"job_cluster_key": "omitted", "new_cluster": map[string]any{ + "policy_id": "new-policy", "spark_env_vars": map[string]any{"ENV": "dev"}, + }}, + map[string]any{"job_cluster_key": "explicit", "new_cluster": map[string]any{ + "policy_id": "new-policy", "spark_env_vars": map[string]any{"ENV": "dev"}, + }}, + }} + entry := &deployplan.PlanEntry{ + RemoteState: typedPolicyResource(t, "shared", remote), + Changes: deployplan.Changes{"job_clusters": { + Action: deployplan.Update, New: config["job_clusters"], Remote: remote["job_clusters"], + }}, + } + b := policyTestBundle(t, "resources.jobs.test", config) + client := mocks.NewMockWorkspaceClient(t) + b.SetWorkpaceClient(client.WorkspaceClient) + // Both omitted clusters use the remote policy, fetched once despite the different config ID. + client.GetMockClusterPoliciesAPI().EXPECT().GetByPolicyId(mock.Anything, "new-policy"). + Return(&compute.Policy{Definition: `{"spark_env_vars.ENV":{"type":"fixed","value":"dev"}}`}, nil).Once() + + changes, skipped, err := configsync.ExtractChanges(t.Context(), b, &deployplan.Plan{Plan: map[string]*deployplan.PlanEntry{"resources.jobs.test": entry}}) + require.NoError(t, err) + assert.Empty(t, skipped) + require.Contains(t, changes, "resources.jobs.test") + require.Contains(t, changes["resources.jobs.test"], "job_clusters") + assert.Equal(t, []any{ + map[string]any{"job_cluster_key": "also_omitted", "new_cluster": map[string]any{"policy_id": "new-policy"}}, + map[string]any{"job_cluster_key": "omitted", "new_cluster": map[string]any{"policy_id": "new-policy"}}, + map[string]any{"job_cluster_key": "explicit", "new_cluster": map[string]any{ + "policy_id": "new-policy", "spark_env_vars": map[string]any{"ENV": "dev"}, + }}, + }, changes["resources.jobs.test"]["job_clusters"].Value) +} + +func TestExtractChangesDropsPolicyValueAtDottedLeaf(t *testing.T) { + key, prefix, config := policyClusterLocation("task", map[string]any{ + "policy_id": "policy", + "spark_conf": map[string]any{"spark.sql.adaptive.enabled": "true"}, + }) + _, _, remote := policyClusterLocation("task", map[string]any{ + "policy_id": "policy", + "spark_conf": map[string]any{"spark.sql.adaptive.enabled": "true", "spark.sql.shuffle.partitions": "7"}, + }) + b := policyTestBundle(t, key, config) + client := mocks.NewMockWorkspaceClient(t) + b.SetWorkpaceClient(client.WorkspaceClient) + client.GetMockClusterPoliciesAPI().EXPECT().GetByPolicyId(mock.Anything, "policy"). + Return(&compute.Policy{Definition: `{"spark_conf.spark.sql.shuffle.partitions":{"type":"fixed","value":"7"}}`}, nil).Once() + entry := &deployplan.PlanEntry{ + Action: deployplan.Skip, + RemoteState: typedPolicyResource(t, "task", remote), + Changes: deployplan.Changes{prefix + "spark_conf['spark.sql.shuffle.partitions']": { + Action: deployplan.Skip, Reason: deployplan.ReasonRemoteAddition, Remote: "7", + }}, + } + changes, skipped, err := configsync.ExtractChanges(t.Context(), b, &deployplan.Plan{Plan: map[string]*deployplan.PlanEntry{key: entry}}) + require.NoError(t, err) + assert.Empty(t, skipped) + assert.Empty(t, changes) +} + +func TestExtractChangesKeepsCompleteLogConfigurationWhenStorageChanges(t *testing.T) { + key, prefix, config := policyClusterLocation("task", map[string]any{ + "policy_id": "policy", "apply_policy_default_values": true, + "cluster_log_conf": map[string]any{"dbfs": map[string]any{"destination": "dbfs:/old-logs"}}, + }) + newDestination := map[string]any{"destination": "s3://logs", "region": "us-east-1"} + _, _, remote := policyClusterLocation("task", map[string]any{ + "policy_id": "policy", "cluster_log_conf": map[string]any{"s3": newDestination}, + }) + remoteState := typedPolicyResource(t, "task", remote) + path := prefix + "cluster_log_conf.s3" + remoteValue, err := structaccess.GetByString(remoteState, path) + require.NoError(t, err) + entry := &deployplan.PlanEntry{ + RemoteState: remoteState, + Changes: deployplan.Changes{ + path: {Action: deployplan.Update, Remote: remoteValue}, + prefix + "cluster_log_conf.dbfs": {Action: deployplan.Update, New: map[string]any{"destination": "dbfs:/old-logs"}}, + }, + } + b := policyTestBundle(t, key, config) + client := mocks.NewMockWorkspaceClient(t) + b.SetWorkpaceClient(client.WorkspaceClient) + client.GetMockClusterPoliciesAPI().EXPECT().GetByPolicyId(mock.Anything, "policy"). + Return(&compute.Policy{Definition: `{"cluster_log_conf.type":{"type":"fixed","value":"S3"},"cluster_log_conf.path":{"type":"fixed","value":"s3://logs"},"cluster_log_conf.region":{"type":"unlimited","defaultValue":"us-west-2"}}`}, nil).Once() + + changes, skipped, err := configsync.ExtractChanges(t.Context(), b, &deployplan.Plan{Plan: map[string]*deployplan.PlanEntry{key: entry}}) + require.NoError(t, err) + assert.Empty(t, skipped) + require.Contains(t, changes, key) + require.Contains(t, changes[key], path) + assert.Equal(t, newDestination, changes[key][path].Value) + assert.Equal(t, configsync.OperationRemove, changes[key][prefix+"cluster_log_conf.dbfs"].Operation) +} + +func TestExtractChangesUsesEffectivePolicyFamilyDefinition(t *testing.T) { + b := policyTestBundle(t, "resources.clusters.test", map[string]any{"policy_id": "family-policy"}) + client := mocks.NewMockWorkspaceClient(t) + b.SetWorkpaceClient(client.WorkspaceClient) + // GET supplies the merged definition; the bundle does not own this policy. + client.GetMockClusterPoliciesAPI().EXPECT().GetByPolicyId(mock.Anything, "family-policy").Return(&compute.Policy{ + PolicyFamilyId: "job-cluster", + PolicyFamilyDefinitionOverrides: `{"custom_tags.Environment":{"type":"fixed","value":"dev"}}`, + Definition: `{"custom_tags.Inherited":{"type":"fixed","value":"family"},"custom_tags.Environment":{"type":"fixed","value":"dev"}}`, + }, nil).Once() + remoteTags := map[string]string{"Inherited": "family", "Environment": "dev", "Owner": "user"} + entry := &deployplan.PlanEntry{ + RemoteState: &dresources.ClusterRemote{PolicyId: "family-policy", CustomTags: remoteTags}, + Changes: deployplan.Changes{ + "custom_tags": {Action: deployplan.Skip, Reason: deployplan.ReasonRemoteAddition, Remote: remoteTags}, + }, + } + changes, skipped, err := configsync.ExtractChanges(t.Context(), b, &deployplan.Plan{Plan: map[string]*deployplan.PlanEntry{"resources.clusters.test": entry}}) + require.NoError(t, err) + assert.Empty(t, skipped) + require.Contains(t, changes, "resources.clusters.test") + require.Contains(t, changes["resources.clusters.test"], "custom_tags") + assert.Equal(t, map[string]any{"Owner": "user"}, changes["resources.clusters.test"]["custom_tags"].Value) +} + +func TestExtractChangesDropsPolicyAdditionBesideAuthoredField(t *testing.T) { + b := policyTestBundle(t, "resources.clusters.test", map[string]any{ + "policy_id": "policy", "autoscale": map[string]any{"max_workers": int64(4)}, + }) + client := mocks.NewMockWorkspaceClient(t) + b.SetWorkpaceClient(client.WorkspaceClient) + client.GetMockClusterPoliciesAPI().EXPECT().GetByPolicyId(mock.Anything, "policy"). + Return(&compute.Policy{Definition: `{"autoscale.min_workers":{"type":"fixed","value":1}}`}, nil).Once() + remote := &compute.AutoScale{MinWorkers: 1, MaxWorkers: 4} + entry := &deployplan.PlanEntry{ + RemoteState: &dresources.ClusterRemote{PolicyId: "policy", Autoscale: remote}, + Changes: deployplan.Changes{"autoscale": {Action: deployplan.Update, New: &compute.AutoScale{MaxWorkers: 4}, Remote: remote}}, + } + changes, skipped, err := configsync.ExtractChanges(t.Context(), b, &deployplan.Plan{Plan: map[string]*deployplan.PlanEntry{"resources.clusters.test": entry}}) + require.NoError(t, err) + assert.Empty(t, skipped) + require.Contains(t, changes, "resources.clusters.test") + require.Contains(t, changes["resources.clusters.test"], "autoscale") + assert.Equal(t, map[string]any{"max_workers": int64(4)}, changes["resources.clusters.test"]["autoscale"].Value) +} + +func TestExtractChangesKeepsPolicyValuesInsidePositionalArray(t *testing.T) { + b := policyTestBundle(t, "resources.clusters.test", map[string]any{ + "policy_id": "policy", "init_scripts": []any{map[string]any{"workspace": map[string]any{"destination": "/Workspace/init.sh"}}}, + }) + volumeScript := &compute.VolumesStorageInfo{Destination: "/Volumes/c/s/v/init.sh"} + entry := &deployplan.PlanEntry{ + RemoteState: &dresources.ClusterRemote{PolicyId: "policy", InitScripts: []compute.InitScriptInfo{{Volumes: volumeScript}}}, + Changes: deployplan.Changes{"init_scripts[0].volumes": {Action: deployplan.Update, Remote: volumeScript}}, + } + // Array positions must be preserved regardless of whether the value matches a policy. + client := mocks.NewMockWorkspaceClient(t) + b.SetWorkpaceClient(client.WorkspaceClient) + changes, skipped, err := configsync.ExtractChanges(t.Context(), b, &deployplan.Plan{Plan: map[string]*deployplan.PlanEntry{"resources.clusters.test": entry}}) + require.NoError(t, err) + assert.Empty(t, skipped) + require.Contains(t, changes, "resources.clusters.test") + require.Contains(t, changes["resources.clusters.test"], "init_scripts[0].volumes") + assert.Equal(t, map[string]any{"destination": "/Volumes/c/s/v/init.sh"}, changes["resources.clusters.test"]["init_scripts[0].volumes"].Value) +} + +func TestExtractChangesDropsPolicyOnlyObjectAtChangeRoot(t *testing.T) { + b := policyTestBundle(t, "resources.clusters.test", map[string]any{"policy_id": "policy"}) + client := mocks.NewMockWorkspaceClient(t) + b.SetWorkpaceClient(client.WorkspaceClient) + client.GetMockClusterPoliciesAPI().EXPECT().GetByPolicyId(mock.Anything, "policy"). + Return(&compute.Policy{Definition: `{"workload_type.clients.notebooks":{"type":"fixed","value":false}}`}, nil).Once() + remote := typedPolicyResource(t, "standalone", map[string]any{ + "policy_id": "policy", "workload_type": map[string]any{"clients": map[string]any{"notebooks": false}}, + }) + clients, err := structaccess.GetByString(remote, "workload_type.clients") + require.NoError(t, err) + entry := &deployplan.PlanEntry{ + RemoteState: remote, + Changes: deployplan.Changes{"workload_type.clients": {Action: deployplan.Update, Remote: clients}}, + } + changes, skipped, err := configsync.ExtractChanges(t.Context(), b, &deployplan.Plan{Plan: map[string]*deployplan.PlanEntry{"resources.clusters.test": entry}}) + require.NoError(t, err) + assert.Empty(t, skipped) + assert.Empty(t, changes) +} + +func policyClusterLocation(location string, cluster map[string]any) (resourceKey, pathPrefix string, root map[string]any) { + switch location { + case "task": + return "resources.jobs.test", "tasks[task_key='main'].new_cluster.", map[string]any{ + "tasks": []any{map[string]any{"task_key": "main", "new_cluster": cluster}}, + } + case "shared": + return "resources.jobs.test", "job_clusters[job_cluster_key='main'].new_cluster.", map[string]any{ + "job_clusters": []any{map[string]any{"job_cluster_key": "main", "new_cluster": cluster}}, + } + case "for_each": + return "resources.jobs.test", "tasks[task_key='each'].for_each_task.task.new_cluster.", map[string]any{ + "tasks": []any{map[string]any{"task_key": "each", "for_each_task": map[string]any{ + "task": map[string]any{"task_key": "main", "new_cluster": cluster}, + }}}, + } + default: + return "resources.clusters.test", "", cluster + } +} + +func typedPolicyResource(t *testing.T, location string, value map[string]any) any { + t.Helper() + var resource any = &dresources.JobRemote{} + if location == "standalone" { + resource = &dresources.ClusterRemote{} + } + data, err := json.Marshal(value) + require.NoError(t, err) + require.NoError(t, json.Unmarshal(data, resource)) + return resource +} + +func policyTestBundle(t *testing.T, resourceKey string, config map[string]any) *bundle.Bundle { + t.Helper() + parts := strings.Split(resourceKey, ".") + data, err := json.Marshal(map[string]any{"resources": map[string]any{ + parts[1]: map[string]any{parts[2]: config}, + }}) + require.NoError(t, err) + value, err := jsonloader.LoadJSON(data, "test config") + require.NoError(t, err) + b := &bundle.Bundle{} + require.NoError(t, b.Config.Mutate(func(_ dyn.Value) (dyn.Value, error) { + return value, nil + })) + return b +} diff --git a/bundle/configsync/defaults.go b/bundle/configsync/defaults.go index 0981b0625d6..f28de8098c5 100644 --- a/bundle/configsync/defaults.go +++ b/bundle/configsync/defaults.go @@ -11,18 +11,12 @@ type ( skipIfEmptyOrDefault struct { defaults map[string]any } - // skipBackendDefault skips a field, regardless of its remote value, when it is - // absent from config. Used for fields the backend or a cluster policy may fill - // in: their remote-only value must not be synced into config. Fields the user - // does manage (present in config) still sync normally. - skipBackendDefault struct{} ) var ( alwaysSkip = skipAlways{} zeroOrNil = skipIfZeroOrNil{} emptyEmailNotifications = skipIfEmptyOrDefault{defaults: map[string]any{"no_alert_for_skipped_runs": false}} - backendDefault = skipBackendDefault{} ) // serverSideDefaults contains all hardcoded server-side defaults. @@ -60,13 +54,6 @@ var serverSideDefaults = map[string]any{ "resources.jobs.*.tasks[*].new_cluster.data_security_mode": "SINGLE_USER", // TODO this field is computed on some workspaces in integration tests, check why and if we can skip it "resources.jobs.*.tasks[*].new_cluster.enable_elastic_disk": alwaysSkip, // deprecated field "resources.jobs.*.tasks[*].new_cluster.single_user_name": alwaysSkip, - // custom_tags and cluster_log_conf are commonly injected by cluster policies - // when the user omits them, so they exist only remotely. Syncing them back leaks - // one environment's policy values into (often shared) config and breaks deploys in - // other environments. TODO: move to backend_defaults in jobs.yml once - // configsync filtering is migrated to the direct engine lifecycle metadata. - "resources.jobs.*.tasks[*].new_cluster.custom_tags": backendDefault, - "resources.jobs.*.tasks[*].new_cluster.cluster_log_conf": backendDefault, // Cluster fields (job_clusters) "resources.jobs.*.job_clusters[*].new_cluster.aws_attributes": alwaysSkip, @@ -75,8 +62,6 @@ var serverSideDefaults = map[string]any{ "resources.jobs.*.job_clusters[*].new_cluster.data_security_mode": "SINGLE_USER", // TODO this field is computed on some workspaces in integration tests, check why and if we can skip it "resources.jobs.*.job_clusters[*].new_cluster.enable_elastic_disk": alwaysSkip, // deprecated field "resources.jobs.*.job_clusters[*].new_cluster.single_user_name": alwaysSkip, - "resources.jobs.*.job_clusters[*].new_cluster.custom_tags": backendDefault, // see tasks[*].new_cluster.custom_tags - "resources.jobs.*.job_clusters[*].new_cluster.cluster_log_conf": backendDefault, // see tasks[*].new_cluster.cluster_log_conf // Standalone cluster fields "resources.clusters.*.aws_attributes": alwaysSkip, @@ -86,8 +71,6 @@ var serverSideDefaults = map[string]any{ "resources.clusters.*.driver_node_type_id": alwaysSkip, "resources.clusters.*.enable_elastic_disk": alwaysSkip, "resources.clusters.*.single_user_name": alwaysSkip, - "resources.clusters.*.custom_tags": backendDefault, // see jobs.*.tasks[*].new_cluster.custom_tags - "resources.clusters.*.cluster_log_conf": backendDefault, // see jobs.*.tasks[*].new_cluster.cluster_log_conf // Experiment fields "resources.experiments.*.artifact_location": alwaysSkip, @@ -146,9 +129,6 @@ func shouldSkipField(path string, value any, hasConfigValue bool) bool { if hasConfigValue { return false } - if _, ok := expected.(skipBackendDefault); ok { - return true - } if _, ok := expected.(skipIfZeroOrNil); ok { return value == nil || value == int64(0) } diff --git a/bundle/configsync/diff.go b/bundle/configsync/diff.go index e42f0a1d83a..6734b4e4620 100644 --- a/bundle/configsync/diff.go +++ b/bundle/configsync/diff.go @@ -2,16 +2,23 @@ package configsync import ( "context" + "errors" "fmt" + "maps" + "slices" "strings" + "sync" "github.com/databricks/cli/bundle" "github.com/databricks/cli/bundle/config" "github.com/databricks/cli/bundle/deployplan" + "github.com/databricks/cli/bundle/direct/dresources" "github.com/databricks/cli/libs/dyn" "github.com/databricks/cli/libs/dyn/convert" "github.com/databricks/cli/libs/log" "github.com/databricks/cli/libs/structs/structdiff" + "github.com/databricks/cli/libs/structs/structpath" + "github.com/databricks/databricks-sdk-go/apierr" ) type OperationType string @@ -137,12 +144,14 @@ func isPermissionsOrGrantsSubResource(resourceKey string) bool { return strings.HasSuffix(nodeType, ".permissions") || strings.HasSuffix(nodeType, ".grants") } -// ExtractChanges extracts the map of remote-vs-config changes from a deploy -// plan. -func ExtractChanges(ctx context.Context, b *bundle.Bundle, plan *deployplan.Plan) (Changes, error) { +// ExtractChanges separates applicable remote changes from changes skipped because +// their cluster policy is missing. Other lookup failures abort extraction. +func ExtractChanges(ctx context.Context, b *bundle.Bundle, plan *deployplan.Plan) (Changes, []SkippedChange, error) { changes := make(Changes) + var skipped []SkippedChange + policyDefinitions := make(clusterPolicyDefinitions) - for resourceKey, entry := range plan.Plan { + for _, resourceKey := range slices.Sorted(maps.Keys(plan.Plan)) { // permissions and grants are emitted as their own plan keys // ("resources...permissions" / ".grants"; see splitResourcePath // in bundle/direct/bundle_plan.go). config-remote-sync cannot write them back @@ -152,40 +161,127 @@ func ExtractChanges(ctx context.Context, b *bundle.Bundle, plan *deployplan.Plan if isPermissionsOrGrantsSubResource(resourceKey) { continue } + resourceChanges, resourceSkipped, err := extractResourceChanges(ctx, b, resourceKey, plan.Plan[resourceKey], policyDefinitions) + if err != nil { + return nil, nil, err + } + if len(resourceChanges) > 0 { + changes[resourceKey] = resourceChanges + } + skipped = append(skipped, resourceSkipped...) + log.Debugf(ctx, "Resource %s has %d changes", resourceKey, len(resourceChanges)) + } + return changes, skipped, nil +} - resourceChanges := make(ResourceChanges) +func extractResourceChanges(ctx context.Context, b *bundle.Bundle, resourceKey string, entry *deployplan.PlanEntry, policyDefinitions clusterPolicyDefinitions) (ResourceChanges, []SkippedChange, error) { + resourceChanges := make(ResourceChanges) + var resourceSkipped []SkippedChange + hasStructuralSkip, hasRemoval := false, false + getPolicyFilter := sync.OnceValues(func() (*clusterPolicyFilter, error) { + return policyFilterForResource(b, resourceKey, entry, policyDefinitions) + }) - if entry.Changes != nil { - for path, changeDesc := range entry.Changes { - if changeDesc.Action == deployplan.Skip && changeDesc.Reason != deployplan.ReasonRemoteAddition { - continue - } + for _, path := range slices.Sorted(maps.Keys(entry.Changes)) { + changeDesc := entry.Changes[path] + if changeDesc.Action == deployplan.Skip && changeDesc.Reason != deployplan.ReasonRemoteAddition { + continue + } - fullPath := resourceKey + "." + path - change, err := convertChangeDesc(fullPath, changeDesc) - if err != nil { - return nil, fmt.Errorf("failed to compute config change for path %s: %w", path, err) + fullPath := resourceKey + "." + path + change, err := convertChangeDesc(fullPath, changeDesc) + if err != nil { + return nil, nil, fmt.Errorf("failed to compute config change for path %s: %w", path, err) + } + if change.Operation == OperationSkip { + continue + } + policyFilter, err := getPolicyFilter() + if err != nil { + return nil, nil, err + } + if change.Value != nil && policyFilter != nil { + fieldPath, err := structpath.ParsePath(path) + if err != nil { + return nil, nil, fmt.Errorf("parsing change path %s: %w", fullPath, err) + } + change.Value, err = policyFilter.filter(ctx, b, fieldPath, change.Value) + if err != nil { + lookupErr, ok := errors.AsType[*clusterPolicyLookupError](err) + if !ok || !errors.Is(err, apierr.ErrNotFound) { + return nil, nil, fmt.Errorf("filtering policy values for %s (no configuration was saved): %w", fullPath, err) } - if change.Operation == OperationSkip { - continue + skip := SkippedChange{ + ResourceKey: resourceKey, Path: path, + Reason: SkipReasonClusterPolicyNotFound, PolicyID: lookupErr.policyID, } - // The state snapshot holds real per-field values, so New != Old - // means the local config diverged from the last deploy and this - // change overwrites that pending edit. - if !structdiff.IsEqual(changeDesc.Old, changeDesc.New) { - change.LocalEdit = true + hasStructuralSkip = hasStructuralSkip || !policyFilter.canSkipField(fieldPath) + resourceSkipped = append(resourceSkipped, skip) + continue + } + if change.Value == nil { + if change.Operation != OperationReplace { + continue } - change.Value = stripNamePrefix(fullPath, change.Value, b.Config.Presets.NamePrefix) - resourceChanges[path] = change + // A policy-only replacement still removes the previous authored value. + change.Operation = OperationRemove } } - - if len(resourceChanges) != 0 { - changes[resourceKey] = resourceChanges + // The state snapshot holds real per-field values, so New != Old + // means the local config diverged from the last deploy and this + // change overwrites that pending edit. + if !structdiff.IsEqual(changeDesc.Old, changeDesc.New) { + change.LocalEdit = true } - - log.Debugf(ctx, "Resource %s has %d changes", resourceKey, len(resourceChanges)) + change.Value = stripNamePrefix(fullPath, change.Value, b.Config.Presets.NamePrefix) + resourceChanges[path] = change + // A removal may be paired with a skipped addition, e.g. autoscale -> num_workers. + hasRemoval = hasRemoval || change.Operation == OperationRemove } + if hasStructuralSkip || (len(resourceSkipped) > 0 && hasRemoval) { + return nil, []SkippedChange{{ + ResourceKey: resourceKey, Reason: SkipReasonClusterPolicyNotFound, PolicyID: resourceSkipped[0].PolicyID, + }}, nil + } + return resourceChanges, resourceSkipped, nil +} - return changes, nil +func policyFilterForResource(b *bundle.Bundle, resourceKey string, entry *deployplan.PlanEntry, definitions clusterPolicyDefinitions) (*clusterPolicyFilter, error) { + filter := &clusterPolicyFilter{remote: entry.RemoteState, definitions: definitions} + switch entry.RemoteState.(type) { + case *dresources.JobRemote: + var err error + filter.clusterRoots, err = jobClusterRoots() + if err != nil { + return nil, err + } + filter.sliceKeys = (&dresources.ResourceJob{}).KeyedSlices() + case *dresources.ClusterRemote: + filter.clusterRoots = []*structpath.PatternNode{nil} // nil matches the resource root + default: + // Only jobs and clusters expand policy values into their remote state. + return nil, nil + } + // Skip entries have no NewState; authored values must come from bundle config. + config, err := b.Config.GetResourceConfig(resourceKey) + if err != nil { + return nil, fmt.Errorf("reading config for %s: %w", resourceKey, err) + } + if config == nil { + return nil, nil + } + filter.config = config + return filter, nil } + +var jobClusterRoots = sync.OnceValues(func() ([]*structpath.PatternNode, error) { + var roots []*structpath.PatternNode + for _, pattern := range []string{"tasks[*].new_cluster", "tasks[*].for_each_task.task.new_cluster", "job_clusters[*].new_cluster"} { + root, err := structpath.ParsePattern(pattern) + if err != nil { + return nil, err + } + roots = append(roots, root) + } + return roots, nil +}) diff --git a/bundle/configsync/diff_test.go b/bundle/configsync/diff_test.go index b16885f6b71..3aed70ac215 100644 --- a/bundle/configsync/diff_test.go +++ b/bundle/configsync/diff_test.go @@ -282,7 +282,8 @@ func TestExtractChangesSkipsPermissionsAndGrants(t *testing.T) { }, }} - changes, err := ExtractChanges(ctx, &bundle.Bundle{}, plan) + changes, skipped, err := ExtractChanges(ctx, &bundle.Bundle{}, plan) + assert.Empty(t, skipped) require.NoError(t, err) _, hasPerms := changes["resources.jobs.my_job.permissions"] diff --git a/bundle/configsync/output.go b/bundle/configsync/output.go index dce2692b84c..3672949e083 100644 --- a/bundle/configsync/output.go +++ b/bundle/configsync/output.go @@ -25,27 +25,44 @@ type SyncStatus string const ( StatusSuccess SyncStatus = "success" + // StatusIncomplete means SkippedChanges lists policy-dependent changes that were not synced. + StatusIncomplete SyncStatus = "incomplete" // StatusSkipped: nothing to sync against (state missing, or no selector // matched a deployed resource). Not an error for the caller. StatusSkipped SyncStatus = "skipped" StatusFailed SyncStatus = "failed" ) +const SkipReasonClusterPolicyNotFound = "cluster_policy_not_found" + +// SkippedChange identifies a change that could not be classified safely. +// An empty Path means that the entire resource was left unchanged. +type SkippedChange struct { + ResourceKey string `json:"resourceKey"` + Path string `json:"path,omitempty"` + Reason string `json:"reason"` + PolicyID string `json:"policyId"` +} + // DiffOutput represents the complete output of the config-remote-sync command type DiffOutput struct { - Status SyncStatus `json:"status"` - Error string `json:"error,omitempty"` - ErrorCode protos.BundleConfigRemoteSyncErrorCategory `json:"errorCode,omitempty"` - Files []FileChange `json:"files"` - Changes Changes `json:"changes"` + Status SyncStatus `json:"status"` + Error string `json:"error,omitempty"` + ErrorCode protos.BundleConfigRemoteSyncErrorCategory `json:"errorCode,omitempty"` + Files []FileChange `json:"files"` + Changes Changes `json:"changes"` + SkippedChanges []SkippedChange `json:"skippedChanges,omitempty"` } // WriteResult renders the result and records the error payload on stats for // telemetry. In JSON mode the process never fails: the outcome is carried in // DiffOutput.Status so the caller branches on it, not the exit code. Text mode // returns the error so the CLI still fails loudly for humans. -func WriteResult(out io.Writer, jsonOutput bool, stats *Stats, files []FileChange, changes Changes, err error) error { +func WriteResult(out io.Writer, jsonOutput bool, stats *Stats, files []FileChange, changes Changes, skipped []SkippedChange, err error) error { status := StatusSuccess + if len(skipped) > 0 { + status = StatusIncomplete + } // Read before the fallback below so unclassified errors carry no code. errorCode := stats.ErrorCategory if err != nil { @@ -65,12 +82,26 @@ func WriteResult(out io.Writer, jsonOutput bool, stats *Stats, files []FileChang if err != nil { return err } + if len(skipped) > 0 { + _, _ = fmt.Fprintln(out, "Sync is incomplete: some remote changes were skipped.") + for _, skip := range skipped { + if skip.Path == "" { + _, _ = fmt.Fprintf(out, "Left %s unchanged: cluster policy %s was not found.\n", skip.ResourceKey, skip.PolicyID) + } else { + _, _ = fmt.Fprintf(out, "Skipped %s.%s: cluster policy %s was not found.\n", skip.ResourceKey, skip.Path, skip.PolicyID) + } + } + } + if len(files) == 0 && len(skipped) > 0 { + _, _ = fmt.Fprintln(out, "No changes could be synced.") + return nil + } _, _ = io.WriteString(out, FormatTextOutput(changes)) _, _ = out.Write([]byte{'\n'}) return nil } - output := DiffOutput{Status: status, Files: files, Changes: changes} + output := DiffOutput{Status: status, Files: files, Changes: changes, SkippedChanges: skipped} if output.Changes == nil { // Serialize as {} rather than null so a consumer can iterate it unconditionally. output.Changes = Changes{} diff --git a/bundle/configsync/select.go b/bundle/configsync/select.go index 269150ed8b5..2db0e379d2b 100644 --- a/bundle/configsync/select.go +++ b/bundle/configsync/select.go @@ -7,6 +7,7 @@ import ( "slices" "strings" + "github.com/databricks/cli/bundle/deployplan" "github.com/databricks/cli/bundle/direct/dstate" "github.com/databricks/cli/libs/log" ) @@ -169,11 +170,11 @@ func describeStateIDs(byTypeID map[string]string, resourceType string) string { return fmt.Sprintf("deployed %s ids in state: %s%s", resourceType, strings.Join(listed, ", "), suffix) } -// FilterChanges returns the subset of changes that belong to the resources in +// FilterPlanEntries returns the entries that belong to the resources in // selected, a list of plan keys ("resources..") as returned by -// ResolveResourceSelectors. Change keys are plan keys too. +// ResolveResourceSelectors. // -// Selection is at the resource level: a change node is kept when it is the +// Selection is at the resource level: an entry is kept when it is the // selected resource itself or any node beneath it, matched by prefix on the "." // path boundary. This groups a resource's permissions/grants sub-nodes // ("resources...permissions") with their parent, so selecting a @@ -185,12 +186,12 @@ func describeStateIDs(byTypeID map[string]string, resourceType string) string { // dependencies matter only when planning (the plan always covers the full // resource set so ${resources.*} references resolve), not when deciding which // resources' configuration may be rewritten. -func FilterChanges(changes Changes, selected []string) Changes { - filtered := make(Changes) - for key, resourceChanges := range changes { +func FilterPlanEntries(entries map[string]*deployplan.PlanEntry, selected []string) map[string]*deployplan.PlanEntry { + filtered := make(map[string]*deployplan.PlanEntry) + for key, entry := range entries { for _, sel := range selected { if key == sel || strings.HasPrefix(key, sel+".") { - filtered[key] = resourceChanges + filtered[key] = entry break } } diff --git a/bundle/configsync/select_test.go b/bundle/configsync/select_test.go index 8fb1c787050..8c2f29d7e63 100644 --- a/bundle/configsync/select_test.go +++ b/bundle/configsync/select_test.go @@ -6,35 +6,36 @@ import ( "slices" "testing" + "github.com/databricks/cli/bundle/deployplan" "github.com/stretchr/testify/assert" ) -func TestFilterChanges(t *testing.T) { - changes := Changes{ +func TestFilterPlanEntries(t *testing.T) { + entries := map[string]*deployplan.PlanEntry{ "resources.jobs.foo": { - "max_concurrent_runs": {Operation: OperationReplace, Value: 5}, + Action: deployplan.Update, }, "resources.jobs.foo.permissions": { - "[0].level": {Operation: OperationReplace, Value: "CAN_MANAGE"}, + Action: deployplan.Update, }, // Boundary: shares the "resources.jobs.foo" prefix but is a different // resource, so selecting "resources.jobs.foo" must not pull it in. "resources.jobs.foobar": { - "name": {Operation: OperationReplace, Value: "foobar"}, + Action: deployplan.Update, }, "resources.jobs.bar": { - "name": {Operation: OperationReplace, Value: "bar"}, + Action: deployplan.Update, }, "resources.schemas.baz": { - "comment": {Operation: OperationAdd, Value: "c"}, + Action: deployplan.Update, }, "resources.schemas.baz.grants": { - "[0].principal": {Operation: OperationAdd, Value: "users"}, + Action: deployplan.Update, }, // A resource whose name is literally "permissions" is the resource itself, // not a sub-node, and is kept only when selected by its own key. "resources.jobs.permissions": { - "name": {Operation: OperationReplace, Value: "p"}, + Action: deployplan.Update, }, } @@ -82,16 +83,16 @@ func TestFilterChanges(t *testing.T) { for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { - got := FilterChanges(changes, tt.selected) + got := FilterPlanEntries(entries, tt.selected) assert.ElementsMatch(t, tt.wantKeys, slices.Collect(maps.Keys(got))) for _, key := range tt.wantKeys { - assert.Equal(t, changes[key], got[key]) + assert.Same(t, entries[key], got[key]) } }) } // The input map is never mutated. - assert.Len(t, changes, 7) + assert.Len(t, entries, 7) } func TestDescribeStateIDs(t *testing.T) { diff --git a/cmd/bundle/config_remote_sync.go b/cmd/bundle/config_remote_sync.go index 8044a28bf9c..132159db4b5 100644 --- a/cmd/bundle/config_remote_sync.go +++ b/cmd/bundle/config_remote_sync.go @@ -71,8 +71,9 @@ Examples: // Populated by PostStateFunc on success; rendered after the run so an // error raised anywhere in ProcessBundleRet takes the same output path. var ( - files []configsync.FileChange - changes configsync.Changes + files []configsync.FileChange + changes configsync.Changes + skippedChanges []configsync.SkippedChange ) _, _, err := utils.ProcessBundleRet(cmd, utils.ProcessOptions{ @@ -95,22 +96,14 @@ Examples: return fmt.Errorf("failed to detect changes: %w", err) } - detected, err := configsync.ExtractChanges(ctx, b, plan) - if err != nil { - stats.ErrorCategory = protos.BundleConfigRemoteSyncErrorCategoryDetectChangesFailed - return fmt.Errorf("failed to extract changes: %w", err) - } - stats.CollectChangeStats(ctx, detected) - // Record the ids present in state and the ids requested: on failure // they are what classifies the miss. stats.CollectStateIDs(slices.Collect(maps.Keys(configsync.IndexDeployedResources(&deployBundle.StateDB)))) if len(selectIDs) > 0 { stats.CollectSelectedIDs(selectIDs) - // Filter after planning, never before: the plan must cover every - // resource so ${resources.*} references resolve; only the emitted - // changes are restricted to the selected resources. + // Plan all resources so references resolve, then restrict extraction + // to selected entries to avoid unrelated policy lookups. selected, err := configsync.ResolveResourceSelectors(ctx, &deployBundle.StateDB, selectIDs) if err != nil { return err @@ -127,9 +120,16 @@ Examples: stats.ErrorCategory = protos.BundleConfigRemoteSyncErrorCategoryResourceNotInConfig return fmt.Errorf("deployed resources missing from the bundle configuration (renamed or removed since the last deploy): %s", strings.Join(missing, ", ")) } - detected = configsync.FilterChanges(detected, selected) + plan.Plan = configsync.FilterPlanEntries(plan.Plan, selected) } + detected, policySkipped, err := configsync.ExtractChanges(ctx, b, plan) + if err != nil { + stats.ErrorCategory = protos.BundleConfigRemoteSyncErrorCategoryDetectChangesFailed + return fmt.Errorf("failed to extract changes: %w", err) + } + stats.CollectChangeStats(ctx, detected) + // Loaded once and shared: ResolveChanges uses it to skip changes whose // parent is a variable reference, RestoreVariableReferences to restore refs. preResolved := configsync.LoadPreResolvedConfig(ctx, b) @@ -139,7 +139,7 @@ Examples: stats.ErrorCategory = protos.BundleConfigRemoteSyncErrorCategoryResolveFailed return fmt.Errorf("failed to resolve field changes: %w", err) } - stats.SkippedChangesCount = int64(skipped) + stats.SkippedChangesCount = int64(skipped + len(policySkipped)) if err := configsync.RestoreVariableReferences(ctx, b, fieldChanges, preResolved, &stats.Restore); err != nil { log.Warnf(ctx, "variable restoration skipped: %v", err) @@ -162,11 +162,12 @@ Examples: files = applied changes = detected + skippedChanges = policySkipped return nil }, }) - return configsync.WriteResult(cmd.OutOrStdout(), root.OutputType(cmd) == flags.OutputJSON, &stats, files, changes, err) + return configsync.WriteResult(cmd.OutOrStdout(), root.OutputType(cmd) == flags.OutputJSON, &stats, files, changes, skippedChanges, err) } return cmd diff --git a/libs/structs/structdiff/keyed_path.go b/libs/structs/structdiff/keyed_path.go new file mode 100644 index 00000000000..ef103efac0d --- /dev/null +++ b/libs/structs/structdiff/keyed_path.go @@ -0,0 +1,49 @@ +package structdiff + +import ( + "fmt" + "reflect" + + "github.com/databricks/cli/libs/structs/structaccess" + "github.com/databricks/cli/libs/structs/structpath" +) + +// KeyedPath replaces slice indices with the selectors used by GetStructDiff. +// Indices without a matching key function and existing selectors are preserved. +func KeyedPath(root any, path *structpath.PathNode, sliceKeys map[string]KeyFunc) (*structpath.PathNode, error) { + diff := &diffContext{sliceKeys: sliceKeys} + var result *structpath.PathNode + for _, node := range path.AsSlice() { + index, isIndex := node.Index() + switch { + case isIndex: + keyFunc := diff.findKeyFunc(result) + if keyFunc == nil { + result = structpath.NewIndex(result, index) + continue + } + element, err := structaccess.Get(root, node) + if err != nil { + return nil, err + } + caller, err := newKeyFuncCaller(keyFunc) + if err != nil { + return nil, err + } + if element == nil || !reflect.TypeOf(element).AssignableTo(caller.argType) { + return nil, fmt.Errorf("KeyFunc expects %v at %s, got %T", caller.argType, node, element) + } + key, value := caller.call(element) + result = structpath.NewKeyValue(result, key, value) + default: + if key, value, ok := node.KeyValue(); ok { + result = structpath.NewKeyValue(result, key, value) + } else if key, ok := node.StringKey(); ok { + result = structpath.NewStringKey(result, key) + } else { + return nil, fmt.Errorf("unsupported path node: %s", node) + } + } + } + return result, nil +} diff --git a/libs/structs/structdiff/keyed_path_test.go b/libs/structs/structdiff/keyed_path_test.go new file mode 100644 index 00000000000..3832dd46095 --- /dev/null +++ b/libs/structs/structdiff/keyed_path_test.go @@ -0,0 +1,60 @@ +package structdiff_test + +import ( + "testing" + + "github.com/databricks/cli/libs/structs/structdiff" + "github.com/databricks/cli/libs/structs/structpath" + "github.com/databricks/databricks-sdk-go/service/jobs" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestKeyedPathUsesDiffSelectors(t *testing.T) { + root := jobs.JobSettings{Tasks: []jobs.Task{ + {TaskKey: "second", DependsOn: []jobs.TaskDependency{{TaskKey: "first"}}}, + {TaskKey: "first"}, + }} + keys := map[string]structdiff.KeyFunc{ + "tasks": func(task jobs.Task) (string, string) { return "task_key", task.TaskKey }, + "tasks[*].depends_on": func(dependency jobs.TaskDependency) (string, string) { return "task_key", dependency.TaskKey }, + } + for _, tt := range []struct{ path, want string }{ + {"tasks[0].depends_on[0].task_key", "tasks[task_key='second'].depends_on[task_key='first'].task_key"}, + {"tasks[1].task_key", "tasks[task_key='first'].task_key"}, + {"tasks[task_key='second'].depends_on[0].task_key", "tasks[task_key='second'].depends_on[task_key='first'].task_key"}, + {"tasks[task_key='first'].task_key", "tasks[task_key='first'].task_key"}, + {"tags['a.b']", "tags['a.b']"}, + {"", ""}, + } { + t.Run(tt.path, func(t *testing.T) { + result, err := structdiff.KeyedPath(&root, structpath.MustParsePath(tt.path), keys) + require.NoError(t, err) + assert.Equal(t, tt.want, result.String()) + }) + } + path := structpath.MustParsePath("tasks[0].task_key") + result, err := structdiff.KeyedPath(&root, path, nil) + require.NoError(t, err) + assert.Equal(t, path.String(), result.String()) +} + +func TestKeyedPathRejectsInvalidKeyFunctionsAndIndices(t *testing.T) { + root := jobs.JobSettings{Tasks: []jobs.Task{{TaskKey: "main"}}} + for _, tt := range []struct { + name, path string + keys map[string]structdiff.KeyFunc + }{ + {"invalid function", "tasks[0]", map[string]structdiff.KeyFunc{"tasks": 1}}, + {"wrong argument type", "tasks[0]", map[string]structdiff.KeyFunc{"tasks": func(int) (string, string) { return "", "" }}}, + {"missing element", "tasks[1]", map[string]structdiff.KeyFunc{ + "tasks": func(task jobs.Task) (string, string) { return "task_key", task.TaskKey }, + "tasks[*].depends_on": func(dependency jobs.TaskDependency) (string, string) { return "task_key", dependency.TaskKey }, + }}, + } { + t.Run(tt.name, func(t *testing.T) { + _, err := structdiff.KeyedPath(&root, structpath.MustParsePath(tt.path), tt.keys) + assert.Error(t, err) + }) + } +}