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..a30a2649e51 --- /dev/null +++ b/acceptance/bundle/config-remote-sync/cluster_policy_replaces_array/output.txt @@ -0,0 +1,33 @@ + +>>> [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 + +>>> [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..4c4a1b9393c --- /dev/null +++ b/acceptance/bundle/config-remote-sync/cluster_policy_replaces_array/script @@ -0,0 +1,34 @@ +export POLICY_ID="" +bundle_configured=false +cleanup() { + if $bundle_configured; then + errcode trace $CLI bundle destroy --auto-approve + fi + if [ -n "$POLICY_ID" ]; then + errcode trace $CLI cluster-policies delete "$POLICY_ID" + fi +} +trap cleanup EXIT + +POLICY_ID=$($CLI cluster-policies create --json "$(envsubst < policy.json.tmpl)" -o json | jq -er .policy_id) +add_repl "$POLICY_ID" POLICY_ID +envsubst < databricks.yml.tmpl > databricks.yml +bundle_configured=true + +trace $CLI bundle deploy +job_id=$(read_id.py test_job) + +title "Replace authored scripts with a script from an updated policy" +trace $CLI cluster-policies edit --json "$(envsubst < updated-policy.json.tmpl)" +edit_resource.py jobs "$job_id" <<'PY' +r["tasks"][0]["new_cluster"]["init_scripts"] = [ + {"volumes": {"destination": "/Volumes/catalog/schema/volume/policy.sh"}} +] +PY + +cp databricks.yml databricks.yml.backup +trace $CLI bundle config-remote-sync --save +trace diff.py databricks.yml.backup databricks.yml + +title "Authored scripts are 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"]]' 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..826a0721829 --- /dev/null +++ b/acceptance/bundle/config-remote-sync/external_cluster_policy_cross_target/output.txt @@ -0,0 +1,91 @@ + +=== 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 leaves the job unchanged, including its unrelated edits +>>> [CLI] cluster-policies delete [DEV_POLICY_ID] + +>>> [CLI] bundle config-remote-sync -t dev --save -o json +{ + "status": "partial", + "changes": {}, + "skippedChanges": [ + { + "resourceKey": "resources.jobs.policy_sync", + "reason": "cluster_policy_not_found", + "policyId": "[DEV_POLICY_ID]" + } + ] +} + +>>> diff.py databricks.yml.backup databricks.yml + +=== The job stays skipped on the next sync +>>> [CLI] bundle config-remote-sync -t dev -o json +{ + "status": "partial", + "changes": {}, + "skippedChanges": [ + { + "resourceKey": "resources.jobs.policy_sync", + "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. +Left resources.jobs.policy_sync unchanged: 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..44c2bb14c1f --- /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 leaves the job unchanged, including its unrelated 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 "The job stays skipped 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/databricks.yml.tmpl b/acceptance/bundle/config-remote-sync/missing_cluster_policy/databricks.yml.tmpl new file mode 100644 index 00000000000..c0b88dcf2ed --- /dev/null +++ b/acceptance/bundle/config-remote-sync/missing_cluster_policy/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/hello.py b/acceptance/bundle/config-remote-sync/missing_cluster_policy/hello.py new file mode 100644 index 00000000000..11b15b1a458 --- /dev/null +++ b/acceptance/bundle/config-remote-sync/missing_cluster_policy/hello.py @@ -0,0 +1 @@ +print("hello") diff --git a/acceptance/bundle/config-remote-sync/missing_cluster_policy/out.test.toml b/acceptance/bundle/config-remote-sync/missing_cluster_policy/out.test.toml new file mode 100644 index 00000000000..8c2cb31de53 --- /dev/null +++ b/acceptance/bundle/config-remote-sync/missing_cluster_policy/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/output.txt b/acceptance/bundle/config-remote-sync/missing_cluster_policy/output.txt new file mode 100644 index 00000000000..26b7afc429f --- /dev/null +++ b/acceptance/bundle/config-remote-sync/missing_cluster_policy/output.txt @@ -0,0 +1,156 @@ + +>>> [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 read error leaves the job unchanged without saving +>>> [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_permission_denied", + "policyId": "[POLICY_ID]", + "error": "Fault injected by test." + } + ] +} + +>>> 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 array replacement leaves the 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 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 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 jobs still sync when other jobs depend on the missing policy +>>> [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/policy.json.tmpl b/acceptance/bundle/config-remote-sync/missing_cluster_policy/policy.json.tmpl new file mode 100644 index 00000000000..f60659b84ab --- /dev/null +++ b/acceptance/bundle/config-remote-sync/missing_cluster_policy/policy.json.tmpl @@ -0,0 +1,4 @@ +{ + "name": "test-policy-$UNIQUE_NAME", + "definition": "{}" +} diff --git a/acceptance/bundle/config-remote-sync/missing_cluster_policy/script b/acceptance/bundle/config-remote-sync/missing_cluster_policy/script new file mode 100644 index 00000000000..236681bfdae --- /dev/null +++ b/acceptance/bundle/config-remote-sync/missing_cluster_policy/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 read error leaves the job unchanged 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, 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 array replacement leaves the 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 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 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 jobs still sync when other jobs depend on the missing policy" +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/test.toml b/acceptance/bundle/config-remote-sync/missing_cluster_policy/test.toml new file mode 100644 index 00000000000..b2b24e3e578 --- /dev/null +++ b/acceptance/bundle/config-remote-sync/missing_cluster_policy/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/updated-policy.json.tmpl b/acceptance/bundle/config-remote-sync/missing_cluster_policy/updated-policy.json.tmpl new file mode 100644 index 00000000000..f4435f0b2f7 --- /dev/null +++ b/acceptance/bundle/config-remote-sync/missing_cluster_policy/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..7364ff6b9de --- /dev/null +++ b/bundle/configsync/cluster_policy.go @@ -0,0 +1,404 @@ +package configsync + +import ( + "context" + "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/clusterpolicy" + "github.com/databricks/cli/libs/structs/structaccess" + "github.com/databricks/cli/libs/structs/structpath" + "github.com/databricks/cli/libs/structs/structtag" + "github.com/databricks/databricks-sdk-go/apierr" +) + +// clusterPolicyGate is the cluster spec field that attaches a policy. The direct engine's +// remote addition rules gated on it mark where the backend applies policy values. +const clusterPolicyGate = "policy_id" + +// FilterClusterPolicyValues removes policy-applied values from already extracted +// and selected changes. The plan and input changes are read-only. A resource whose +// policy cannot be read is left out entirely and reported as skipped: its changes may +// pair with each other (e.g. autoscale -> num_workers), so a partial result is unsafe. +func FilterClusterPolicyValues(ctx context.Context, b *bundle.Bundle, plan *deployplan.Plan, changes Changes) (Changes, []SkippedChange, error) { + definitions := make(clusterPolicyDefinitions) + lookupPolicy := func(id string) (clusterpolicy.Definition, 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, err := filter.filterChanges(changes[resourceKey]) + if lookupErr, ok := errors.AsType[*clusterPolicyLookupError](err); ok { + skipped = append(skipped, lookupErr.skippedChange(resourceKey)) + continue + } + if err != nil { + return nil, nil, fmt.Errorf("filtering policy values for %s: %w", resourceKey, err) + } + if len(resourceChanges) > 0 { + filtered[resourceKey] = resourceChanges + } + } + return filtered, skipped, nil +} + +type clusterPolicyDefinition struct { + rules clusterpolicy.Definition + 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 } + +func (e *clusterPolicyLookupError) skippedChange(resourceKey string) SkippedChange { + skip := SkippedChange{ResourceKey: resourceKey, Reason: SkipReasonClusterPolicyReadFailed, PolicyID: e.policyID, Error: e.err.Error()} + switch { + case errors.Is(e.err, apierr.ErrNotFound): + skip.Reason, skip.Error = SkipReasonClusterPolicyNotFound, "" + case errors.Is(e.err, apierr.ErrPermissionDenied): + skip.Reason = SkipReasonClusterPolicyPermissionDenied + } + return skip +} + +// get caches definitions and 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) (clusterpolicy.Definition, error) { + if definition, ok := definitions[id]; ok { + return definition.rules, definition.err + } + var rules clusterpolicy.Definition + policy, err := b.WorkspaceClient(ctx).ClusterPolicies.GetByPolicyId(ctx, id) + if err == nil { + rules, err = clusterpolicy.ParseDefinition(policy.Definition) + } + if err != nil { + err = &clusterPolicyLookupError{policyID: id, err: err} + } + definitions[id] = clusterPolicyDefinition{rules: rules, err: err} + return rules, err +} + +// elementSupplies reports whether the policy element could have set value. The backend +// rejects values that differ from a fixed element, so a fixed field is frozen whatever its +// remote value; other elements only set their opted-in default. +func elementSupplies(element clusterpolicy.Element, value any, applyDefaults bool) bool { + if element.Type == "fixed" { + return true + } + expected := element.EffectiveValue(applyDefaults) + return expected != nil && clusterpolicy.ValueEquals(expected, value) +} + +type clusterPolicyFilter struct { + lookupPolicy func(string) (clusterpolicy.Definition, error) + config any + remote any + clusterRoots []*structpath.PatternNode +} + +func (f *clusterPolicyFilter) filterChanges(changes ResourceChanges) (ResourceChanges, error) { + filtered := make(ResourceChanges, len(changes)) + 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, fmt.Errorf("parsing change path %s: %w", path, err) + } + value, keep, err := f.prune(fieldPath, change.Value, true) + if err != nil { + return nil, err + } + if !keep { + // Policy values never sync, even when they replaced an authored value. + continue + } + if change.Operation == OperationReplace && !reflect.DeepEqual(value, change.Value) { + // Dropping policy values can leave exactly the authored value. + unchanged, err := f.matchesAuthored(fieldPath, value) + if err != nil { + return nil, err + } + if unchanged { + continue + } + } + change.Value = value + } + filtered[path] = &change + } + return filtered, nil +} + +// prune removes policy values from a normalized change value, including whole-object +// additions. It reports keep=false when the whole value comes from the policy. Values +// outside cluster specs are descended into because they may contain whole clusters. +func (f *clusterPolicyFilter) prune(path *structpath.PathNode, value any, isChangeRoot bool) (any, bool, error) { + clusterPath, inCluster := f.clusterPath(path) + if !inCluster { + return f.pruneChildren(path, value) + } + remotePolicyID, _ := structaccess.Get(f.remote, structpath.NewStringKey(clusterPath, clusterPolicyGate)) + policyID, _ := remotePolicyID.(string) + if policyID == "" { + return value, true, nil + } + // 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 path.SkipPrefix(clusterPath.Len()).AsSlice() { + if _, indexed := node.Index(); indexed { + return value, true, nil + } + } + // Keep required fields, but still filter their optional children. + if isChangeRoot || !isRequiredField(f.remote, path) { + policyOnly, err := f.isPolicyOnlySubtree(clusterPath, path, value, policyID) + if err != nil { + return nil, false, err + } + if policyOnly { + return nil, false, nil + } + } + // Removing elements would change indexed policy constraints. + if _, sequence := value.([]any); sequence { + return value, true, nil + } + 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)) { + child, keep, err := f.prune(structpath.NewStringKey(path, key), value[key], false) + if err != nil { + return nil, false, err + } + if keep { + result[key] = child + } + } + return result, true, nil + case []any: + // Only reached outside cluster specs, e.g. a whole tasks array added remotely; + // arrays inside a cluster are kept as is above. Elements are never 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 + } +} + +// clusterPath finds the cluster spec containing a field; the spec 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 +} + +// isPolicyOnlySubtree requires every leaf to be absent from authored config and +// set by the remote cluster's policy: under a fixed element, 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 + } + } + + definition, 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(definition, leaf.path.SkipPrefix(clusterPath.Len()), leaf.value, applyDefaults == true) { + return false, nil + } + } + return true, nil +} + +// authoredValue returns the config value at path, or nil when the config does not declare +// it. Change paths select keyed slice elements by key, so they address the same element in +// config and remote regardless of order. +func (f *clusterPolicyFilter) authoredValue(path *structpath.PathNode) (any, error) { + value, err := structaccess.Get(f.config, path) + 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()) + parentValue := reflect.Indirect(reflect.ValueOf(parent)) + if !parentValue.IsValid() || parentValue.Kind() != reflect.Struct { + return false + } + field, _, ok := structaccess.FindStructFieldByKeyType(parentValue.Type(), key) + return ok && !structtag.JSONTag(field.Tag.Get("json")).OmitEmpty() +} + +// 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. Map keys are not wildcarded: a rule +// such as spark_conf.* constrains values but is not treated as supplying them. +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(definition clusterpolicy.Definition, path *structpath.PathNode, value any, applyDefaults bool) bool { + indexed, wildcard := policyAttributes(path) + if indexed == "cluster_log_conf.s3.region" { + return elementSupplies(definition["cluster_log_conf.region"], value, applyDefaults) + } + if storage, ok := strings.CutPrefix(indexed, "cluster_log_conf."); ok && strings.HasSuffix(storage, ".destination") { + storage = strings.ToUpper(strings.TrimSuffix(storage, ".destination")) + logType := definition["cluster_log_conf.type"].EffectiveValue(applyDefaults) + return clusterpolicy.ValueEquals(logType, storage) && elementSupplies(definition["cluster_log_conf.path"], value, applyDefaults) + } + element, ok := definition[indexed] + if !ok { + element = definition[wildcard] + } + return elementSupplies(element, value, applyDefaults) +} + +func policyFilterForResource(b *bundle.Bundle, resourceKey string, remote any, lookupPolicy func(string) (clusterpolicy.Definition, error)) (*clusterPolicyFilter, error) { + filter := &clusterPolicyFilter{remote: remote, lookupPolicy: lookupPolicy} + switch remote.(type) { + case *dresources.JobRemote: + filter.clusterRoots = policyClusterRoots("jobs") + 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() == clusterPolicyGate { + 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..ee864d0e9a0 --- /dev/null +++ b/bundle/configsync/cluster_policy_test.go @@ -0,0 +1,716 @@ +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 TestFilterClusterPolicyValuesSkipsResourceOnPolicyLookupError(t *testing.T) { + for _, tt := range []struct { + name string + err error + reason string + }{ + {"permission denied", apierr.ErrPermissionDenied, configsync.SkipReasonClusterPolicyPermissionDenied}, + {"service unavailable", errors.New("policy service unavailable"), configsync.SkipReasonClusterPolicyReadFailed}, + } { + t.Run(tt.name, func(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(nil, tt.err).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.NoError(t, err) + assert.Empty(t, changes) + assert.Equal(t, []configsync.SkippedChange{{ + ResourceKey: "resources.clusters.test", Reason: tt.reason, PolicyID: "policy", Error: tt.err.Error(), + }}, skipped) + }) + } +} + +func TestFilterClusterPolicyValuesSkipsResourceWhenPolicyIsMissing(t *testing.T) { + b := policyTestBundle(t, "resources.clusters.test", map[string]any{ + "policy_id": "policy", "cluster_name": "before", + }) + require.NoError(t, b.Config.Mutate(func(v dyn.Value) (dyn.Value, error) { + return dyn.SetByPath(v, dyn.MustPathFromString("resources.clusters.other"), dyn.V(map[string]dyn.Value{ + "cluster_name": dyn.V("other-before"), + })) + })) + client := mocks.NewMockWorkspaceClient(t) + b.SetWorkpaceClient(client.WorkspaceClient) + // Looked up once for the resource even though two additions depend on it. + client.GetMockClusterPoliciesAPI().EXPECT().GetByPolicyId(mock.Anything, "policy"). + Return(nil, apierr.ErrResourceDoesNotExist).Once() + plan := &deployplan.Plan{Plan: map[string]*deployplan.PlanEntry{ + "resources.clusters.test": { + 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"}}, + }, + }, + "resources.clusters.other": { + RemoteState: &dresources.ClusterRemote{ClusterName: "other-after"}, + Changes: deployplan.Changes{ + "cluster_name": {Action: deployplan.Update, Old: "other-before", New: "other-before", Remote: "other-after"}, + }, + }, + }} + changes, skipped, err := filterPolicyTestChanges(t, b, plan) + require.NoError(t, err) + // The unrelated cluster_name edit is held back too: it may pair with a skipped change. + assert.Equal(t, configsync.Changes{"resources.clusters.other": { + "cluster_name": {Operation: configsync.OperationReplace, Value: "other-after"}, + }}, changes) + assert.Equal(t, []configsync.SkippedChange{ + {ResourceKey: "resources.clusters.test", Reason: configsync.SkipReasonClusterPolicyNotFound, PolicyID: "policy"}, + }, skipped) +} + +func TestFilterClusterPolicyValuesUsesRemotePolicyPerCluster(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"}, + }}, + }} + env, authored := map[string]any{"ENV": "dev"}, map[string]any{"ENV": "authored"} + entry := &deployplan.PlanEntry{ + RemoteState: typedPolicyResource(t, "shared", remote), + Changes: deployplan.Changes{ + "job_clusters[job_cluster_key='explicit'].new_cluster.spark_env_vars": {Action: deployplan.Update, Old: authored, New: authored, Remote: env}, + "job_clusters[job_cluster_key='omitted'].new_cluster.spark_env_vars": {Action: deployplan.Skip, Reason: deployplan.ReasonRemoteAddition, Remote: env}, + "job_clusters[job_cluster_key='also_omitted'].new_cluster.spark_env_vars": {Action: deployplan.Skip, Reason: deployplan.ReasonRemoteAddition, Remote: env}, + }, + } + 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) + assert.Equal(t, configsync.Changes{"resources.jobs.test": { + "job_clusters[job_cluster_key='explicit'].new_cluster.spark_env_vars": {Operation: configsync.OperationReplace, Value: env}, + }}, changes) +} + +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 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..196a26db704 100644 --- a/bundle/configsync/output.go +++ b/bundle/configsync/output.go @@ -25,27 +25,60 @@ 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" + SkipReasonClusterPolicyPermissionDenied = "cluster_policy_permission_denied" + SkipReasonClusterPolicyReadFailed = "cluster_policy_read_failed" +) + +// SkippedChange identifies a resource left unchanged because its remote changes +// could not be classified safely. +type SkippedChange struct { + ResourceKey string `json:"resourceKey"` + Reason string `json:"reason"` + PolicyID string `json:"policyId"` + Error string `json:"error,omitempty"` +} + +func (s SkippedChange) describe() string { + switch s.Reason { + case SkipReasonClusterPolicyNotFound: + return fmt.Sprintf("Left %s unchanged: cluster policy %s was not found.", s.ResourceKey, s.PolicyID) + case SkipReasonClusterPolicyPermissionDenied: + return fmt.Sprintf("Left %s unchanged: no permission to read cluster policy %s.", s.ResourceKey, s.PolicyID) + default: + return fmt.Sprintf("Left %s unchanged: cluster policy %s could not be read: %s", s.ResourceKey, s.PolicyID, s.Error) + } +} + // 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 +98,22 @@ 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 { + _, _ = fmt.Fprintln(out, skip.describe()) + } + } + if len(files) == 0 && len(skipped) > 0 { + _, _ = fmt.Fprintln(out, "No changes could be synced.") + return nil + } _, _ = io.WriteString(out, FormatTextOutput(changes)) _, _ = out.Write([]byte{'\n'}) return nil } - output := DiffOutput{Status: status, Files: files, Changes: changes} + output := DiffOutput{Status: status, Files: files, Changes: changes, SkippedChanges: skipped} if output.Changes == nil { // Serialize as {} rather than null so a consumer can iterate it unconditionally. output.Changes = Changes{} diff --git a/bundle/configsync/output_test.go b/bundle/configsync/output_test.go index 1f43343c6e8..eef7ffbd2a0 100644 --- a/bundle/configsync/output_test.go +++ b/bundle/configsync/output_test.go @@ -3,6 +3,7 @@ package configsync import ( "os" "path/filepath" + "strings" "testing" "github.com/databricks/cli/bundle" @@ -86,3 +87,25 @@ func TestSaveFiles_EmptyList(t *testing.T) { err := SaveFiles(ctx, &bundle.Bundle{}, []FileChange{}) require.NoError(t, err) } + +func TestWriteResultTextReportsSkippedResources(t *testing.T) { + skipped := []SkippedChange{ + {ResourceKey: "resources.jobs.a", Reason: SkipReasonClusterPolicyNotFound, PolicyID: "p1"}, + {ResourceKey: "resources.jobs.b", Reason: SkipReasonClusterPolicyPermissionDenied, PolicyID: "p2", Error: "denied"}, + {ResourceKey: "resources.jobs.c", Reason: SkipReasonClusterPolicyReadFailed, PolicyID: "p3", Error: "boom"}, + } + var out strings.Builder + require.NoError(t, WriteResult(&out, false, &Stats{}, nil, nil, skipped, nil)) + assert.Equal(t, `Sync is incomplete: some remote changes were skipped. +Left resources.jobs.a unchanged: cluster policy p1 was not found. +Left resources.jobs.b unchanged: no permission to read cluster policy p2. +Left resources.jobs.c unchanged: cluster policy p3 could not be read: boom +No changes could be synced. +`, out.String()) + + out.Reset() + require.NoError(t, WriteResult(&out, true, &Stats{}, nil, nil, skipped[:1], nil)) + assert.Contains(t, out.String(), `"status": "partial"`) + assert.Contains(t, out.String(), `"reason": "cluster_policy_not_found"`) + assert.NotContains(t, out.String(), `"error"`) +} diff --git a/bundle/configsync/telemetry.go b/bundle/configsync/telemetry.go index 9fd0afccc83..86743a62efb 100644 --- a/bundle/configsync/telemetry.go +++ b/bundle/configsync/telemetry.go @@ -49,6 +49,11 @@ type Stats struct { // could not be attributed to a single source location. SkippedChangesCount int64 + // Resources left unchanged because a cluster policy could not be read, by cause. + ClusterPolicyNotFoundCount int64 + ClusterPolicyPermissionDeniedCount int64 + ClusterPolicyReadFailedCount int64 + Restore RestoreStats // Identity of the resource state the run actually read. Without these, a @@ -70,6 +75,21 @@ type Stats struct { ErrorCategory protos.BundleConfigRemoteSyncErrorCategory } +// CollectSkippedResources counts resources held back by cluster policy lookups, by +// reason. Only counts are kept; the resource keys and policy IDs stay in the output. +func (s *Stats) CollectSkippedResources(skipped []SkippedChange) { + for _, skip := range skipped { + switch skip.Reason { + case SkipReasonClusterPolicyNotFound: + s.ClusterPolicyNotFoundCount++ + case SkipReasonClusterPolicyPermissionDenied: + s.ClusterPolicyPermissionDeniedCount++ + default: + s.ClusterPolicyReadFailedCount++ + } + } +} + // CollectStateStats records which resource state the run read. Only the // state's identity is recorded (serial, system-generated lineage UUID, // local/remote origin, candidate count) — never a path or a target name. @@ -274,28 +294,31 @@ func (s *Stats) LogTelemetry(ctx context.Context) { telemetry.Log(ctx, protos.DatabricksCliLog{ BundleConfigRemoteSyncEvent: &protos.BundleConfigRemoteSyncEvent{ - Save: s.Save, - Engine: string(s.Engine), - ChangesTotal: s.ChangesTotal, - AddCount: s.AddCount, - ReplaceCount: s.ReplaceCount, - RemoveCount: s.RemoveCount, - RecreateForcingChanges: s.RecreateForcingChanges, - OverwrittenLocalEdits: s.OverwrittenLocalEdits, - ResourceChanges: resourceChanges, - FilesChangedCount: s.FilesChangedCount, - FilesWrittenCount: s.FilesWrittenCount, - SkippedChangesCount: s.SkippedChangesCount, - RefsRetargeted: s.Restore.Retargeted, - RefsFromSiblings: s.Restore.FromSiblings, - StateSerial: s.StateSerial, - StateLineage: s.StateLineage, - StateSource: s.StateSource, - StatesAvailableCount: s.StatesAvailableCount, - StateResourceIDs: s.StateResourceIDs, - SelectedResourceIDs: s.SelectedResourceIDs, - ErrorMessage: s.ErrorMessage, - ErrorCategory: s.ErrorCategory, + Save: s.Save, + Engine: string(s.Engine), + ChangesTotal: s.ChangesTotal, + AddCount: s.AddCount, + ReplaceCount: s.ReplaceCount, + RemoveCount: s.RemoveCount, + RecreateForcingChanges: s.RecreateForcingChanges, + OverwrittenLocalEdits: s.OverwrittenLocalEdits, + ResourceChanges: resourceChanges, + FilesChangedCount: s.FilesChangedCount, + FilesWrittenCount: s.FilesWrittenCount, + SkippedChangesCount: s.SkippedChangesCount, + ClusterPolicyNotFoundCount: s.ClusterPolicyNotFoundCount, + ClusterPolicyPermissionDeniedCount: s.ClusterPolicyPermissionDeniedCount, + ClusterPolicyReadFailedCount: s.ClusterPolicyReadFailedCount, + RefsRetargeted: s.Restore.Retargeted, + RefsFromSiblings: s.Restore.FromSiblings, + StateSerial: s.StateSerial, + StateLineage: s.StateLineage, + StateSource: s.StateSource, + StatesAvailableCount: s.StatesAvailableCount, + StateResourceIDs: s.StateResourceIDs, + SelectedResourceIDs: s.SelectedResourceIDs, + ErrorMessage: s.ErrorMessage, + ErrorCategory: s.ErrorCategory, }, }) } diff --git a/bundle/configsync/telemetry_test.go b/bundle/configsync/telemetry_test.go index d2a3da537f8..21ddd7afd8e 100644 --- a/bundle/configsync/telemetry_test.go +++ b/bundle/configsync/telemetry_test.go @@ -106,3 +106,18 @@ func TestRestoreStatsCountersNilSafe(t *testing.T) { s.incFromSiblings() }) } + +func TestCollectSkippedResources(t *testing.T) { + var stats Stats + stats.CollectSkippedResources([]SkippedChange{ + {ResourceKey: "resources.jobs.a", Reason: SkipReasonClusterPolicyNotFound, PolicyID: "p1"}, + {ResourceKey: "resources.jobs.b", Reason: SkipReasonClusterPolicyNotFound, PolicyID: "p1"}, + {ResourceKey: "resources.jobs.c", Reason: SkipReasonClusterPolicyPermissionDenied, PolicyID: "p2", Error: "denied"}, + {ResourceKey: "resources.clusters.d", Reason: SkipReasonClusterPolicyReadFailed, PolicyID: "p3", Error: "boom"}, + }) + assert.Equal(t, Stats{ + ClusterPolicyNotFoundCount: 2, + ClusterPolicyPermissionDeniedCount: 1, + ClusterPolicyReadFailedCount: 1, + }, stats) +} diff --git a/cmd/bundle/config_remote_sync.go b/cmd/bundle/config_remote_sync.go index 8044a28bf9c..67f74e926bf 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) @@ -140,6 +147,7 @@ Examples: return fmt.Errorf("failed to resolve field changes: %w", err) } stats.SkippedChangesCount = int64(skipped) + stats.CollectSkippedResources(policySkipped) if err := configsync.RestoreVariableReferences(ctx, b, fieldChanges, preResolved, &stats.Restore); err != nil { log.Warnf(ctx, "variable restoration skipped: %v", err) @@ -162,11 +170,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/clusterpolicy/definition.go b/libs/clusterpolicy/definition.go new file mode 100644 index 00000000000..ab22ebc864c --- /dev/null +++ b/libs/clusterpolicy/definition.go @@ -0,0 +1,74 @@ +package clusterpolicy + +import ( + "encoding/json" + "reflect" + "strconv" + "strings" +) + +// Element is the part of a cluster policy element that materializes a value. The backend +// supports more element types, but only these fields supply one: "fixed" elements set Value, +// every limiting type ("allowlist", "blocklist", "regex", "range", "unlimited") may carry +// DefaultValue. "forbidden" only rejects. +// https://docs.databricks.com/aws/en/admin/clusters/policy-definition +type Element struct { + Type string `json:"type"` + Value any `json:"value"` + DefaultValue any `json:"defaultValue"` +} + +// Definition maps policy attribute paths (e.g. "spark_version", "custom_tags.CostCenter") +// to their elements. +type Definition map[string]Element + +// ParseDefinition decodes a policy definition. Numbers stay json.Number so integer and +// float literals compare exactly against typed cluster fields. +func ParseDefinition(definition string) (Definition, error) { + decoder := json.NewDecoder(strings.NewReader(definition)) + decoder.UseNumber() + var result Definition + if err := decoder.Decode(&result); err != nil { + return nil, err + } + return result, nil +} + +// EffectiveValue returns the value the element materializes into a spec that omits the +// attribute, or nil if it supplies none. "fixed" applies whether or not the request sets +// apply_policy_default_values, "defaultValue" only when it does. Pinned against a real +// workspace by acceptance/bundle/resources/jobs/cluster_policy (fixed_addition, default_flag). +func (e Element) EffectiveValue(applyDefaults bool) any { + if e.Type == "fixed" { + return e.Value + } + if applyDefaults { + return e.DefaultValue + } + return nil +} + +// ValueEquals compares a decoded policy value with a cluster field value. Numbers compare +// by the field's type, and string fields such as spark_conf entries match the literal of a +// JSON number or boolean. +func ValueEquals(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) +} diff --git a/libs/clusterpolicy/definition_test.go b/libs/clusterpolicy/definition_test.go new file mode 100644 index 00000000000..385e711fdb8 --- /dev/null +++ b/libs/clusterpolicy/definition_test.go @@ -0,0 +1,48 @@ +package clusterpolicy + +import ( + "encoding/json" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestParseDefinitionKeepsNumbers(t *testing.T) { + definition, err := ParseDefinition(`{ + "autotermination_minutes": {"type": "fixed", "value": 60}, + "spark_conf.spark.sql.shuffle.partitions": {"type": "unlimited", "defaultValue": 7}, + "custom_tags.Team": {"type": "forbidden"} + }`) + require.NoError(t, err) + assert.Equal(t, json.Number("60"), definition["autotermination_minutes"].EffectiveValue(false)) + assert.Nil(t, definition["spark_conf.spark.sql.shuffle.partitions"].EffectiveValue(false)) + assert.Equal(t, json.Number("7"), definition["spark_conf.spark.sql.shuffle.partitions"].EffectiveValue(true)) + assert.Nil(t, definition["custom_tags.Team"].EffectiveValue(true)) + assert.Nil(t, definition["missing"].EffectiveValue(true)) + + _, err = ParseDefinition(`{"spark_version": `) + assert.Error(t, err) +} + +func TestValueEquals(t *testing.T) { + for _, tt := range []struct { + name string + policy, want any + equal bool + }{ + {"int field", json.Number("60"), int64(60), true}, + {"float field", json.Number("1.5"), 1.5, true}, + {"string field holds number literal", json.Number("7"), "7", true}, + {"string field holds bool literal", true, "true", true}, + {"different number", json.Number("60"), int64(61), false}, + {"number against bool", json.Number("1"), true, false}, + {"strings", "dev", "dev", true}, + {"bools", false, false, true}, + {"nil policy value", nil, "dev", false}, + } { + t.Run(tt.name, func(t *testing.T) { + assert.Equal(t, tt.equal, ValueEquals(tt.policy, tt.want)) + }) + } +} diff --git a/libs/telemetry/protos/bundle_config_remote_sync.go b/libs/telemetry/protos/bundle_config_remote_sync.go index 678198514bc..467acb09939 100644 --- a/libs/telemetry/protos/bundle_config_remote_sync.go +++ b/libs/telemetry/protos/bundle_config_remote_sync.go @@ -63,6 +63,13 @@ type BundleConfigRemoteSyncEvent struct { // during change detection, which never reaches ChangesTotal. SkippedChangesCount int64 `json:"skipped_changes_count,omitempty"` + // Number of selected resources left entirely unchanged because the cluster + // policy attached to one of their clusters could not be read, by cause. A + // nonzero value means the run finished with a "partial" status. + ClusterPolicyNotFoundCount int64 `json:"cluster_policy_not_found_count,omitempty"` + ClusterPolicyPermissionDeniedCount int64 `json:"cluster_policy_permission_denied_count,omitempty"` + ClusterPolicyReadFailedCount int64 `json:"cluster_policy_read_failed_count,omitempty"` + // Variable-reference restoration counts for the two mechanisms that can // write a current-target-scoped reference into a shared file (the source of // the cross-target "reference does not exist" failures). diff --git a/libs/testserver/cluster_policies.go b/libs/testserver/cluster_policies.go index ce29464a9e5..7351e357d11 100644 --- a/libs/testserver/cluster_policies.go +++ b/libs/testserver/cluster_policies.go @@ -4,10 +4,10 @@ import ( "encoding/json" "fmt" "maps" - "reflect" "slices" "strings" + "github.com/databricks/cli/libs/clusterpolicy" "github.com/databricks/databricks-sdk-go/service/compute" "github.com/databricks/databricks-sdk-go/service/jobs" ) @@ -128,51 +128,6 @@ func (s *FakeWorkspace) ClusterPoliciesDelete(req Request) any { return Response{} } -// policyElement is the part of a cluster policy element the fake applies. The backend -// supports more element types, but only these two fields materialize a value: -// "fixed" elements set Value, every limiting type ("allowlist", "blocklist", "regex", -// "range", "unlimited") may carry DefaultValue. "forbidden" only rejects and is ignored here. -type policyElement struct { - Type string `json:"type"` - Value any `json:"value"` - DefaultValue any `json:"defaultValue"` -} - -// effectiveValue returns the value this element materializes into a spec that omits the -// attribute, or nil if it materializes none. Mirrors the backend: "fixed" applies whether -// or not the request sets apply_policy_default_values, "defaultValue" only when it does. -// Pinned against a real workspace by -// acceptance/bundle/resources/jobs/cluster_policy (fixed_addition, default_flag). -func (e policyElement) effectiveValue(applyDefaults bool) any { - if e.Type == "fixed" { - return e.Value - } - if applyDefaults { - return e.DefaultValue - } - return nil -} - -// clusterPolicyValues returns the values the policy supplies, keyed by the policy's -// attribute path (e.g. "spark_version", "custom_tags.CostCenter"). -func (s *FakeWorkspace) clusterPolicyValues(policyID string, applyDefaults bool) map[string]any { - policy, ok := s.ClusterPolicies[policyID] - if !ok { - return nil - } - var elements map[string]policyElement - if err := json.Unmarshal([]byte(policy.Definition), &elements); err != nil { - return nil - } - values := make(map[string]any, len(elements)) - for path, element := range elements { - if value := element.effectiveValue(applyDefaults); value != nil { - values[path] = value - } - } - return values -} - // applyClusterPolicy fills in attributes the request omitted from the policy attached via // policyID. It returns the backend's validation message when a supplied value contradicts a // "fixed" element, or "" when the spec is acceptable; callers must hold the workspace lock and @@ -185,8 +140,12 @@ func (s *FakeWorkspace) applyClusterPolicy(spec any, policyID string, applyDefau if policyID == "" { return "" } - values := s.clusterPolicyValues(policyID, applyDefaults) - if len(values) == 0 { + policy, ok := s.ClusterPolicies[policyID] + if !ok { + return "" + } + definition, err := clusterpolicy.ParseDefinition(policy.Definition) + if err != nil || len(definition) == 0 { return "" } @@ -207,13 +166,17 @@ func (s *FakeWorkspace) applyClusterPolicy(spec any, policyID string, applyDefau // every key present, which makes fields the backend omits (e.g. the Jobs API dropping // apply_policy_default_values) serialize as explicit zeros. patch := map[string]any{} - for _, path := range slices.Sorted(maps.Keys(values)) { - value := values[path] + for _, path := range slices.Sorted(maps.Keys(definition)) { + element := definition[path] + value := element.EffectiveValue(applyDefaults) + if value == nil { + continue + } segments := strings.Split(path, ".") if existing, ok := lookup(doc, segments); ok { // A "fixed" element enforces its value as well as supplying it; anything else // only supplies a default, which the request is free to override. - if s.isFixed(policyID, path) && !reflect.DeepEqual(existing, value) { + if element.Type == "fixed" && !clusterpolicy.ValueEquals(value, existing) { // Message shape copied from the real backend for a nested attribute // ("custom_tags, CostCenter must be ..."); the wording for a top-level // attribute has not been observed, so it is only approximated here. @@ -234,15 +197,6 @@ func (s *FakeWorkspace) applyClusterPolicy(spec any, policyID string, applyDefau return "" } -// isFixed reports whether the policy's element at path is a "fixed" element. -func (s *FakeWorkspace) isFixed(policyID, path string) bool { - var elements map[string]policyElement - if err := json.Unmarshal([]byte(s.ClusterPolicies[policyID].Definition), &elements); err != nil { - return false - } - return elements[path].Type == "fixed" -} - // lookup returns the value at the given key path in doc. func lookup(doc map[string]any, path []string) (any, bool) { for _, key := range path[:len(path)-1] {