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..dfa52c96c28 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,64 @@ 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 +=== Synced config deploys and converges under the policy +>>> [CLI] bundle deploy +Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/[UNIQUE_NAME]/files... +Updated jobs.test_job +Files: 1 uploaded, 0 deleted +Resources: 0 created, 1 changed, 0 deleted, 1 unchanged + +>>> [CLI] bundle plan +Plan: 0 to add, 0 to change, 0 to delete, 2 unchanged + +=== 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 + + + +>>> 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 + +=== Synced config deploys and converges under the policy +>>> [CLI] bundle deploy +Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/[UNIQUE_NAME]/files... +Updated jobs.test_job +Files: 1 uploaded, 0 deleted +Resources: 0 created, 1 changed, 0 deleted, 1 unchanged + +>>> [CLI] bundle plan +Plan: 0 to add, 0 to change, 0 to delete, 2 unchanged + >>> [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..17d54ce9879 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,9 +11,33 @@ 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 "Synced config deploys and converges under the policy" +trace $CLI bundle deploy +trace $CLI bundle plan + +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 +trace $CLI bundle config-remote-sync --save +trace diff.py databricks.yml.backup databricks.yml +rm databricks.yml.backup + +title "Synced config deploys and converges under the policy" +trace $CLI bundle deploy +trace $CLI bundle plan 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..92a0fca8cb3 --- /dev/null +++ b/acceptance/bundle/config-remote-sync/cluster_policy_replaces_array/output.txt @@ -0,0 +1,43 @@ + +>>> [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 +No changes detected. + + +>>> diff.py databricks.yml.backup databricks.yml + +=== Authored scripts are kept without copying the policy script +>>> [CLI] bundle validate -o json +true + +=== Synced config deploys and converges under the policy +>>> [CLI] bundle deploy +Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/[UNIQUE_NAME]/files... +Updated jobs.test_job +Files: 1 uploaded, 0 deleted +Resources: 0 created, 1 changed, 0 deleted, 0 unchanged + +>>> [CLI] bundle plan +Plan: 0 to add, 0 to change, 0 to delete, 1 unchanged + +>>> [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..6976de41762 --- /dev/null +++ b/acceptance/bundle/config-remote-sync/cluster_policy_replaces_array/script @@ -0,0 +1,38 @@ +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 kept without copying the policy script" +trace $CLI bundle validate -o json | jq -e '.resources.jobs.test_job.tasks[0].new_cluster.init_scripts | map(keys) == [["workspace"], ["workspace"]]' + +title "Synced config deploys and converges under the policy" +trace $CLI bundle deploy +trace $CLI bundle plan 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..e9a5e0e1c32 --- /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": "partial", + "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": "partial", + "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..07b108c7b6f --- /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}' | contains.py '"status": "partial"' +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}' | contains.py '"status": "partial"' + +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..79f6d4ec21e --- /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": "partial", + "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": "partial", + "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": "partial", + "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": "partial", + "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..9e6e395dc2b 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 @@ -28,10 +34,20 @@ Resource: resources.jobs.job1 job_clusters: - job_cluster_key: shared +=== Synced config deploys and converges under the policy +>>> [CLI] bundle deploy +Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/test-bundle-[UNIQUE_NAME]/default/files... +Files: 1 uploaded, 0 deleted +Resources: 0 created, 0 changed, 0 deleted, 2 unchanged + +>>> [CLI] bundle plan +Plan: 0 to add, 0 to change, 0 to delete, 2 unchanged + >>> [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..e9f0558a9ef 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)." @@ -42,3 +45,7 @@ title "Configuration changes" echo trace diff.py databricks.yml.backup databricks.yml rm databricks.yml.backup + +title "Synced config deploys and converges under the policy" +trace $CLI bundle deploy +trace $CLI bundle plan diff --git a/bundle/configsync/cluster_policy.go b/bundle/configsync/cluster_policy.go new file mode 100644 index 00000000000..d8982ae9e0c --- /dev/null +++ b/bundle/configsync/cluster_policy.go @@ -0,0 +1,516 @@ +package configsync + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "maps" + "reflect" + "slices" + "strconv" + "strings" + + "github.com/databricks/cli/bundle" + "github.com/databricks/cli/bundle/deployplan" + "github.com/databricks/cli/bundle/direct/dresources" + "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" + "github.com/databricks/databricks-sdk-go/apierr" +) + +// FilterClusterPolicyValues removes policy-applied values from already extracted +// and selected changes. The plan and input changes are read-only. +func FilterClusterPolicyValues(ctx context.Context, b *bundle.Bundle, plan *deployplan.Plan, changes Changes) (Changes, []SkippedChange, error) { + definitions := make(clusterPolicyDefinitions) + lookupPolicy := func(id string) (clusterPolicyRules, error) { + return definitions.get(ctx, b, id) + } + filtered := make(Changes, len(changes)) + var skipped []SkippedChange + for _, resourceKey := range slices.Sorted(maps.Keys(changes)) { + filter, err := policyFilterForResource(b, resourceKey, plan.Plan[resourceKey].RemoteState, lookupPolicy) + if err != nil { + return nil, nil, err + } + if filter == nil { + filtered[resourceKey] = changes[resourceKey] + continue + } + resourceChanges, resourceSkipped, err := filterClusterPolicyChanges(resourceKey, changes[resourceKey], filter) + if err != nil { + return nil, nil, err + } + if len(resourceChanges) > 0 { + filtered[resourceKey] = resourceChanges + } + skipped = append(skipped, resourceSkipped...) + } + return filtered, skipped, nil +} + +// filterClusterPolicyChanges preserves atomic replacements and paired removals +// when a missing policy prevents classifying part of a resource's changes. +func filterClusterPolicyChanges(resourceKey string, changes ResourceChanges, filter *clusterPolicyFilter) (ResourceChanges, []SkippedChange, error) { + filtered := make(ResourceChanges, len(changes)) + var skipped []SkippedChange + structural, removal := false, false + for _, path := range slices.Sorted(maps.Keys(changes)) { + change := *changes[path] + if change.Value != nil { + fieldPath, err := structpath.ParsePath(path) + if err != nil { + return nil, nil, fmt.Errorf("parsing change path %s.%s: %w", resourceKey, path, err) + } + value, keep, err := filter.prune(fieldPath, change.Value, true) + 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.%s (no configuration was saved): %w", resourceKey, path, err) + } + structural = structural || !filter.canSkipField(fieldPath) + skipped = append(skipped, SkippedChange{ + ResourceKey: resourceKey, Path: path, + Reason: SkipReasonClusterPolicyNotFound, PolicyID: lookupErr.policyID, + }) + continue + } + switch { + case !keep: + // Policy values never sync, even when they replaced an authored value. + continue + case change.Operation == OperationReplace && !reflect.DeepEqual(value, change.Value): + // Dropping policy values can leave exactly the authored value. + unchanged, err := filter.matchesAuthored(fieldPath, value) + if err != nil { + return nil, nil, fmt.Errorf("filtering policy values for %s.%s (no configuration was saved): %w", resourceKey, path, err) + } + if unchanged { + continue + } + } + change.Value = value + } + filtered[path] = &change + // A removal may be paired with a skipped addition, e.g. autoscale -> num_workers. + removal = removal || change.Operation == OperationRemove + } + // A structural change is always skipped too, so one length check covers both cases. + if len(skipped) > 0 && (structural || removal) { + return nil, []SkippedChange{{ + ResourceKey: resourceKey, Reason: SkipReasonClusterPolicyNotFound, PolicyID: skipped[0].PolicyID, + }}, nil + } + return filtered, skipped, nil +} + +type clusterPolicyRule struct { + Type string `json:"type"` + Value any `json:"value"` + DefaultValue any `json:"defaultValue"` +} + +type clusterPolicyRules map[string]clusterPolicyRule + +type clusterPolicyDefinition struct { + rules clusterPolicyRules + 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 definitions and lookup failures per sync. The API returns the +// effective definition, including policy-family overrides. +func (definitions clusterPolicyDefinitions) get(ctx context.Context, b *bundle.Bundle, id string) (clusterPolicyRules, error) { + if definition, ok := definitions[id]; ok { + return definition.rules, 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 nil, lookupErr + } + decoder := json.NewDecoder(strings.NewReader(policy.Definition)) + decoder.UseNumber() + var rules clusterPolicyRules + if err := decoder.Decode(&rules); err != nil { + return nil, fmt.Errorf("parsing cluster policy %s: %w", id, err) + } + definitions[id] = clusterPolicyDefinition{rules: rules} + return rules, nil +} + +// supplied returns the value a rule sets, or nil if it sets none. +func (r clusterPolicyRule) supplied(applyDefaults bool) any { + switch { + case r.Type == "fixed": + return r.Value + case applyDefaults && r.Type != "forbidden": + return r.DefaultValue + } + return nil +} + +// supplies reports whether the rule could have set value. The backend rejects +// values that differ from a fixed rule, so a fixed field is frozen whatever its +// remote value; other rules only set their opted-in default. +func (r clusterPolicyRule) supplies(value any, applyDefaults bool) bool { + if r.Type == "fixed" { + return true + } + expected := r.supplied(applyDefaults) + return expected != nil && policyValueEquals(expected, value) +} + +// policyValueEquals compares a decoded policy value with a normalized remote value. +// Numbers compare by the remote type, and string fields such as spark_conf entries +// match the literal of a JSON number or boolean. +func policyValueEquals(policyValue, value any) bool { + switch policyValue := policyValue.(type) { + case json.Number: + switch value := value.(type) { + case string: + return value == policyValue.String() + case int64: + i, err := policyValue.Int64() + return err == nil && i == value + case float64: + f, err := policyValue.Float64() + return err == nil && f == value + } + return false + case bool: + if value, ok := value.(string); ok { + return value == strconv.FormatBool(policyValue) + } + } + return reflect.DeepEqual(policyValue, value) +} + +type clusterPolicyFilter struct { + lookupPolicy func(string) (clusterPolicyRules, error) + config any + remote any + clusterRoots []*structpath.PatternNode + sliceKeys map[string]structdiff.KeyFunc +} + +type pruneAction int + +const ( + pruneDescend pruneAction = iota // filter children individually + pruneKeep // keep the value as is + pruneDrop // the whole value comes from the policy +) + +// prune removes omitted policy values from a normalized plan change, including +// whole-object additions. It reports keep=false when the whole value comes from the policy. +func (f *clusterPolicyFilter) prune(path *structpath.PathNode, value any, isChangeRoot bool) (any, bool, error) { + action, err := f.classify(path, value, isChangeRoot) + if err != nil { + return nil, false, err + } + switch action { + case pruneDrop: + return nil, false, nil + case pruneKeep: + return value, true, nil + default: + return f.pruneChildren(path, value) + } +} + +func (f *clusterPolicyFilter) pruneChildren(path *structpath.PathNode, value any) (any, bool, error) { + switch value := value.(type) { + case map[string]any: + result := make(map[string]any, len(value)) + for _, key := range slices.Sorted(maps.Keys(value)) { + filtered, keep, err := f.prune(structpath.NewStringKey(path, key), value[key], false) + if err != nil { + return nil, false, err + } + if keep { + result[key] = filtered + } + } + return result, true, nil + case []any: + // Arrays inside clusters are never descended into; elements here cannot be dropped. + result := make([]any, len(value)) + for i, child := range value { + filtered, _, err := f.prune(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 + } +} + +// classify decides how prune handles a value. Values outside cluster specs are +// descended into because they may contain whole clusters. +func (f *clusterPolicyFilter) classify(path *structpath.PathNode, value any, isChangeRoot bool) (pruneAction, error) { + clusterPath, ok := f.clusterPath(path) + if !ok { + return pruneDescend, nil + } + remotePolicyID, _ := structaccess.Get(f.remote, structpath.NewStringKey(clusterPath, "policy_id")) + policyID, _ := remotePolicyID.(string) + if policyID == "" { + return pruneKeep, nil + } + fieldPath := path + if f.sliceKeys != nil { + // Whole-object changes use remote indices; authored tasks may have another order. + var err error + fieldPath, err = structdiff.KeyedPath(f.remote, path, f.sliceKeys) + if err != nil { + return 0, err + } + clusterPath = fieldPath.Prefix(clusterPath.Len()) + } + // 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 pruneKeep, nil + } + } + // Keep required fields, but still filter their optional children. + if isChangeRoot || !isRequiredField(f.remote, fieldPath) { + matches, err := f.isPolicyOnlySubtree(clusterPath, fieldPath, value, policyID) + if err != nil { + return 0, err + } + if matches { + return pruneDrop, nil + } + } + // Removing elements would change indexed policy constraints. + if _, sequence := value.([]any); sequence { + return pruneKeep, nil + } + return pruneDescend, 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 + } + switch derefType(reflect.TypeOf(remote)).Kind() { + case reflect.Struct, reflect.Slice, reflect.Array: + return false + default: + return true + } +} + +// isPolicyOnlySubtree requires every leaf to be absent from authored config and +// set by the remote cluster's policy: under a fixed rule, or equal to an opted-in default. +func (f *clusterPolicyFilter) isPolicyOnlySubtree(clusterPath, path *structpath.PathNode, value any, policyID string) (bool, error) { + leaves := appendLeaves(nil, path, value) + if len(leaves) == 0 { + return false, nil + } + for _, leaf := range leaves { + authored, err := f.authoredValue(leaf.path) + if err != nil { + return false, err + } + if authored != nil { + return false, nil + } + } + + rules, err := f.lookupPolicy(policyID) + if err != nil { + return false, err + } + // Remote cluster specs omit this input-only flag; use the authored setting. + applyDefaults, err := f.authoredValue(structpath.NewStringKey(clusterPath, "apply_policy_default_values")) + if err != nil { + return false, err + } + for _, leaf := range leaves { + if !policySupplies(rules, leaf.path.SkipPrefix(clusterPath.Len()), leaf.value, applyDefaults == true) { + return false, nil + } + } + return true, nil +} + +// authoredValue validates selectors before looking up fields so duplicate task +// identities cannot make an authored value appear absent. +func (f *clusterPolicyFilter) authoredValue(path *structpath.PathNode) (any, error) { + var value any + keyed, err := structdiff.KeyedPath(f.config, path, f.sliceKeys) + if err == nil { + value, err = structaccess.Get(f.config, keyed) + } + if _, missing := errors.AsType[*structaccess.NotFoundError](err); missing { + return nil, nil + } + if err != nil { + return nil, fmt.Errorf("reading authored field %s: %w", path, err) + } + return value, nil +} + +func (f *clusterPolicyFilter) matchesAuthored(path *structpath.PathNode, value any) (bool, error) { + authored, err := f.authoredValue(path) + if err != nil { + return false, err + } + normalized, err := normalizeValue(authored) + if err != nil { + return false, err + } + return reflect.DeepEqual(normalized, value), nil +} + +type policyLeaf struct { + path *structpath.PathNode + value any +} + +// appendLeaves collects the normalized change's scalars, including nulls. +// Empty containers have no leaves and cannot alone establish a policy match. +func appendLeaves(leaves []policyLeaf, path *structpath.PathNode, value any) []policyLeaf { + switch value := value.(type) { + case map[string]any: + for _, key := range slices.Sorted(maps.Keys(value)) { + leaves = appendLeaves(leaves, structpath.NewStringKey(path, key), value[key]) + } + case []any: + for i, child := range value { + leaves = appendLeaves(leaves, structpath.NewIndex(path, i), child) + } + default: + leaves = append(leaves, policyLeaf{path: path, value: value}) + } + return leaves +} + +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 + } + field, _, ok := structaccess.FindStructFieldByKeyType(derefType(typ), key) + return ok && !structtag.JSONTag(field.Tag.Get("json")).OmitEmpty() +} + +func derefType(typ reflect.Type) reflect.Type { + for typ.Kind() == reflect.Pointer { + typ = typ.Elem() + } + return typ +} + +// policyAttributes renders a path relative to the cluster in the policy language, +// once with indices and once with "*" in their place. Dotted map keys stay one key, +// e.g. spark_conf["a.b"] becomes spark_conf.a.b. +func policyAttributes(path *structpath.PathNode) (indexed, wildcard string) { + var indexedParts, wildcardParts []string + for _, node := range path.AsSlice() { + if key, ok := node.StringKey(); ok { + indexedParts = append(indexedParts, key) + wildcardParts = append(wildcardParts, key) + } else if index, ok := node.Index(); ok { + indexedParts = append(indexedParts, strconv.Itoa(index)) + wildcardParts = append(wildcardParts, "*") + } + } + return strings.Join(indexedParts, "."), strings.Join(wildcardParts, ".") +} + +// policySupplies reports whether the policy could have set a leaf, translating the +// policy language's log-delivery aliases. Indexed rules take precedence over wildcards. +// https://docs.databricks.com/aws/en/admin/clusters/policy-definition +func policySupplies(rules clusterPolicyRules, path *structpath.PathNode, value any, applyDefaults bool) bool { + indexed, wildcard := policyAttributes(path) + if indexed == "cluster_log_conf.s3.region" { + return rules["cluster_log_conf.region"].supplies(value, applyDefaults) + } + if storage, ok := strings.CutPrefix(indexed, "cluster_log_conf."); ok && strings.HasSuffix(storage, ".destination") { + storage = strings.ToUpper(strings.TrimSuffix(storage, ".destination")) + logType := rules["cluster_log_conf.type"].supplied(applyDefaults) + return policyValueEquals(logType, storage) && rules["cluster_log_conf.path"].supplies(value, applyDefaults) + } + rule, ok := rules[indexed] + if !ok { + rule = rules[wildcard] + } + return rule.supplies(value, applyDefaults) +} + +func policyFilterForResource(b *bundle.Bundle, resourceKey string, remote any, lookupPolicy func(string) (clusterPolicyRules, error)) (*clusterPolicyFilter, error) { + filter := &clusterPolicyFilter{remote: remote, lookupPolicy: lookupPolicy} + switch remote.(type) { + case *dresources.JobRemote: + filter.clusterRoots = policyClusterRoots("jobs") + filter.sliceKeys = (&dresources.ResourceJob{}).KeyedSlices() + case *dresources.ClusterRemote: + filter.clusterRoots = policyClusterRoots("clusters") + 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) + } + filter.config = config + return filter, nil +} + +// policyClusterRoots reuses the direct engine's policy-gated remote addition rules, +// which mark where the backend applies policy values. A nil root is the resource itself. +func policyClusterRoots(resourceType string) []*structpath.PatternNode { + var roots []*structpath.PatternNode + for _, rule := range dresources.GetResourceConfig(resourceType).IgnoreRemoteAdditions { + if rule.WhenSet.String() == "policy_id" { + roots = append(roots, rule.Field) + } + } + return roots +} diff --git a/bundle/configsync/cluster_policy_test.go b/bundle/configsync/cluster_policy_test.go new file mode 100644 index 00000000000..d367f6ee6f9 --- /dev/null +++ b/bundle/configsync/cluster_policy_test.go @@ -0,0 +1,772 @@ +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 TestFilterClusterPolicyValuesFiltersPolicyAppliedValues(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: "authored zero scalar is kept", + attribute: "num_workers", + config: int64(0), + remote: int64(2), + want: int64(2), + }, + { + name: "authored false scalar is kept", + attribute: "enable_local_disk_encryption", + config: false, + remote: true, + want: true, + }, + { + 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: "wildcard rule drops whole array", + wantLookup: true, + attribute: "init_scripts", + definition: `{"init_scripts.*.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: "policy-only replacement keeps authored array", + wantLookup: true, + attribute: "init_scripts", + definition: `{"init_scripts.0.volumes.destination":{"type":"fixed","value":"/Volumes/c/s/v/policy.sh"}}`, + config: []any{map[string]any{"workspace": map[string]any{"destination": "/Workspace/init.sh"}}}, + remote: []any{map[string]any{"volumes": map[string]any{"destination": "/Volumes/c/s/v/policy.sh"}}}, + }, + { + // The policy changed after deploy; the backend rejects any other value for a fixed field. + name: "stale fixed value dropped", + wantLookup: true, + attribute: "custom_tags", + definition: `{"custom_tags.Environment":{"type":"fixed","value":"prod"}}`, + remote: map[string]any{"Environment": "dev", "Owner": "user"}, + want: map[string]any{"Owner": "user"}, + }, + { + name: "numeric default matches string map value", + wantLookup: true, + attribute: "spark_conf", + definition: `{"spark_conf.spark.sql.shuffle.partitions":{"type":"unlimited","defaultValue":7}}`, + applyDefaults: true, + remote: map[string]any{"spark.sql.shuffle.partitions": "7"}, + }, + { + 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 := filterPolicyTestChanges(t, 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 TestFilterClusterPolicyValuesFailsOnPolicyLookupError(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 := filterPolicyTestChanges(t, 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 TestFilterClusterPolicyValuesDoesNotHidePolicyErrorsAfterStructuralSkip(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 := filterPolicyTestChanges(t, 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 TestFilterClusterPolicyValuesLooksUpMissingPolicyOnce(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 := filterPolicyTestChanges(t, 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 TestFilterClusterPolicyValuesMatchesClustersByKeyAndRemotePolicy(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 := filterPolicyTestChanges(t, 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 TestFilterClusterPolicyValuesDropsPolicyValueAtDottedLeaf(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 := filterPolicyTestChanges(t, b, &deployplan.Plan{Plan: map[string]*deployplan.PlanEntry{key: entry}}) + require.NoError(t, err) + assert.Empty(t, skipped) + assert.Empty(t, changes) +} + +func TestFilterClusterPolicyValuesKeepsCompleteLogConfigurationWhenStorageChanges(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 := filterPolicyTestChanges(t, 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 TestFilterClusterPolicyValuesUsesEffectivePolicyFamilyDefinition(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 := filterPolicyTestChanges(t, 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 TestFilterClusterPolicyValuesDropsPolicyAdditionBesideAuthoredField(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: 8} + 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 := filterPolicyTestChanges(t, 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(8)}, changes["resources.clusters.test"]["autoscale"].Value) +} + +func TestFilterClusterPolicyValuesDropsReplacementLeftWithAuthoredValue(t *testing.T) { + b := policyTestBundle(t, "resources.clusters.test", map[string]any{ + "policy_id": "policy", "custom_tags": map[string]any{"Owner": "user"}, + }) + client := mocks.NewMockWorkspaceClient(t) + b.SetWorkpaceClient(client.WorkspaceClient) + client.GetMockClusterPoliciesAPI().EXPECT().GetByPolicyId(mock.Anything, "policy"). + Return(&compute.Policy{Definition: `{"custom_tags.Environment":{"type":"fixed","value":"dev"}}`}, nil).Once() + remoteTags := map[string]string{"Environment": "dev", "Owner": "user"} + entry := &deployplan.PlanEntry{ + RemoteState: &dresources.ClusterRemote{PolicyId: "policy", CustomTags: remoteTags}, + Changes: deployplan.Changes{"custom_tags": {Action: deployplan.Update, New: map[string]string{"Owner": "user"}, Remote: remoteTags}}, + } + changes, skipped, err := filterPolicyTestChanges(t, 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 TestFilterClusterPolicyValuesKeepsPolicyValuesInsidePositionalArray(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 := filterPolicyTestChanges(t, 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 TestFilterClusterPolicyValuesDropsPolicyOnlyObjectAtChangeRoot(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 := filterPolicyTestChanges(t, 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 TestFilterClusterPolicyValuesRequiresUniqueTaskIdentityOnlyWithPolicy(t *testing.T) { + for _, tt := range []struct{ name, duplicateIn, policyID string }{ + {"policy with duplicate authored tasks", "config", "policy"}, + {"policy with duplicate remote tasks", "remote", "policy"}, + {"no policy with duplicate authored tasks", "config", ""}, + {"no policy with duplicate remote tasks", "remote", ""}, + } { + t.Run(tt.name, func(t *testing.T) { + task := map[string]any{ + "task_key": "main", "new_cluster": map[string]any{ + "policy_id": tt.policyID, "spark_env_vars": map[string]any{"ENV": "dev"}, + }, + } + configTasks, remoteTasks := []any{task}, []any{task} + if tt.duplicateIn == "config" { + configTasks = append(configTasks, task) + } else { + remoteTasks = append(remoteTasks, task) + } + key := "resources.jobs.test" + b := policyTestBundle(t, key, map[string]any{"tasks": configTasks}) + entry := &deployplan.PlanEntry{ + RemoteState: typedPolicyResource(t, "task", map[string]any{"tasks": remoteTasks}), + Changes: deployplan.Changes{"tasks": {Action: deployplan.Update, Remote: remoteTasks}}, + } + changes, skipped, err := filterPolicyTestChanges(t, b, &deployplan.Plan{Plan: map[string]*deployplan.PlanEntry{key: entry}}) + if tt.policyID == "" { + require.NoError(t, err) + require.Contains(t, changes[key], "tasks") + assert.Equal(t, remoteTasks, changes[key]["tasks"].Value) + assert.Empty(t, skipped) + return + } + require.ErrorContains(t, err, "ambiguous") + assert.Nil(t, changes) + assert.Nil(t, skipped) + }) + } +} + +func TestFilterClusterPolicyValuesRejectsInvalidAuthoredPath(t *testing.T) { + key := "resources.clusters.test" + b := policyTestBundle(t, key, map[string]any{"policy_id": "policy"}) + client := mocks.NewMockWorkspaceClient(t) + b.SetWorkpaceClient(client.WorkspaceClient) + entry := &deployplan.PlanEntry{ + RemoteState: &dresources.ClusterRemote{PolicyId: "policy"}, + Changes: deployplan.Changes{ + "unrecognized_field": {Action: deployplan.Update, Remote: "value"}, + }, + } + changes, skipped, err := filterPolicyTestChanges(t, b, &deployplan.Plan{Plan: map[string]*deployplan.PlanEntry{key: entry}}) + require.ErrorContains(t, err, "reading authored field unrecognized_field") + assert.Nil(t, changes) + assert.Nil(t, skipped) +} + +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 +} + +func filterPolicyTestChanges(t *testing.T, b *bundle.Bundle, plan *deployplan.Plan) (configsync.Changes, []configsync.SkippedChange, error) { + t.Helper() + changes, err := configsync.ExtractChanges(t.Context(), b, plan) + require.NoError(t, err) + inputs := []any{plan, changes} + for _, entry := range plan.Plan { + // RemoteState is excluded from the plan's JSON representation. + inputs = append(inputs, entry.RemoteState) + } + before, err := json.Marshal(inputs) + require.NoError(t, err) + filtered, skipped, filterErr := configsync.FilterClusterPolicyValues(t.Context(), b, plan, changes) + after, err := json.Marshal(inputs) + require.NoError(t, err) + assert.Equal(t, string(before), string(after), "policy filtering must not mutate the plan or input changes") + return filtered, skipped, filterErr +} 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/output.go b/bundle/configsync/output.go index dce2692b84c..5bad38f71bf 100644 --- a/bundle/configsync/output.go +++ b/bundle/configsync/output.go @@ -25,27 +25,45 @@ type SyncStatus string const ( StatusSuccess SyncStatus = "success" + // StatusPartial: some remote changes could not be synced and are listed in + // SkippedChanges. Files and Changes hold the edits that were synced. + StatusPartial SyncStatus = "partial" // 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 = StatusPartial + } // Read before the fallback below so unclassified errors carry no code. errorCode := stats.ErrorCategory if err != nil { @@ -65,12 +83,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/cmd/bundle/config_remote_sync.go b/cmd/bundle/config_remote_sync.go index 8044a28bf9c..acbfd066b74 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{ @@ -130,6 +131,12 @@ Examples: detected = configsync.FilterChanges(detected, selected) } + detected, policySkipped, err := configsync.FilterClusterPolicyValues(ctx, b, plan, detected) + if err != nil { + stats.ErrorCategory = protos.BundleConfigRemoteSyncErrorCategoryDetectChangesFailed + return fmt.Errorf("failed to extract changes: %w", err) + } + // 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 +146,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 +169,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..99f735941a5 --- /dev/null +++ b/libs/structs/structdiff/keyed_path.go @@ -0,0 +1,84 @@ +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. +// Keyed elements must have nonempty, unique identities. +// Indices and selectors without a matching key function 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() + key, value, isSelector := node.KeyValue() + switch { + case isIndex || isSelector: + keyFunc := diff.findKeyFunc(result) + if keyFunc == nil { + if isIndex { + result = structpath.NewIndex(result, index) + } else { + result = structpath.NewKeyValue(result, key, value) + } + 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) + if key == "" || value == "" { + return nil, fmt.Errorf("empty identity at %s", node) + } + if err := validateUniqueIdentity(root, node.Parent(), caller, key, value); err != nil { + return nil, err + } + result = structpath.NewKeyValue(result, key, value) + default: + if key, ok := node.StringKey(); ok { + result = structpath.NewStringKey(result, key) + } else { + return nil, fmt.Errorf("unsupported path node: %s", node) + } + } + } + return result, nil +} + +func validateUniqueIdentity(root any, path *structpath.PathNode, caller *keyFuncCaller, key, value string) error { + parent, err := structaccess.Get(root, path) + if err != nil { + return err + } + sequence := reflect.ValueOf(parent) + if sequence.Kind() != reflect.Slice && sequence.Kind() != reflect.Array { + return fmt.Errorf("expected sequence at %s, got %T", path, parent) + } + if err := validateKeyFuncElementType(sequence, caller.argType); err != nil { + return err + } + matches := 0 + for i := range sequence.Len() { + k, v := caller.call(sequence.Index(i).Interface()) + if k == key && v == value { + matches++ + } + } + if matches != 1 { + return fmt.Errorf("ambiguous identity at %s: %s=%q matches %d elements", path, key, value, matches) + } + return 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..7ecd7ee0ebd --- /dev/null +++ b/libs/structs/structdiff/keyed_path_test.go @@ -0,0 +1,80 @@ +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) + }) + } +} + +func TestKeyedPathRejectsEmptyOrAmbiguousIdentities(t *testing.T) { + keys := map[string]structdiff.KeyFunc{ + "tasks": func(task jobs.Task) (string, string) { return "task_key", task.TaskKey }, + } + for _, tt := range []struct { + name, path string + tasks []jobs.Task + want string + }{ + {"duplicate index", "tasks[1].new_cluster", []jobs.Task{{TaskKey: "main"}, {TaskKey: "main"}}, "ambiguous identity"}, + {"duplicate selector", "tasks[task_key='main'].new_cluster", []jobs.Task{{TaskKey: "main"}, {TaskKey: "main"}}, "ambiguous identity"}, + {"empty identity", "tasks[0].new_cluster", []jobs.Task{{}}, "empty identity"}, + } { + t.Run(tt.name, func(t *testing.T) { + _, err := structdiff.KeyedPath(&jobs.JobSettings{Tasks: tt.tasks}, structpath.MustParsePath(tt.path), keys) + require.ErrorContains(t, err, tt.want) + }) + } +}