From bb2b13f608cbcc949ee35a780d572407840c1cda Mon Sep 17 00:00:00 2001 From: ilyakuz-db <181845775+ilyakuz-db@users.noreply.github.com> Date: Fri, 23 Oct 2026 14:50:29 +0000 Subject: [PATCH 1/4] Avoid syncing cluster policy values into shared bundle config --- .../bundles/remote-sync-cluster-policies.md | 1 + .../databricks.yml.tmpl | 52 +++ .../cluster_policy_cross_target/hello.py | 1 + .../cluster_policy_cross_target/out.test.toml | 3 + .../cluster_policy_cross_target/output.txt | 54 +++ .../cluster_policy_cross_target/script | 23 + .../cluster_policy_cross_target/test.toml | 3 + .../cluster_policy_remote_addition/output.txt | 5 +- .../cluster_policy_remote_addition/script | 1 + .../databricks.yml.tmpl | 18 + .../policy_injected_cluster_fields/output.txt | 8 +- bundle/configsync/cluster_policy.go | 227 ++++++++++ bundle/configsync/cluster_policy_test.go | 403 ++++++++++++++++++ bundle/configsync/defaults.go | 20 - bundle/configsync/diff.go | 16 + 15 files changed, 811 insertions(+), 24 deletions(-) create mode 100644 .nextchanges/bundles/remote-sync-cluster-policies.md create mode 100644 acceptance/bundle/config-remote-sync/cluster_policy_cross_target/databricks.yml.tmpl create mode 100644 acceptance/bundle/config-remote-sync/cluster_policy_cross_target/hello.py create mode 100644 acceptance/bundle/config-remote-sync/cluster_policy_cross_target/out.test.toml create mode 100644 acceptance/bundle/config-remote-sync/cluster_policy_cross_target/output.txt create mode 100644 acceptance/bundle/config-remote-sync/cluster_policy_cross_target/script create mode 100644 acceptance/bundle/config-remote-sync/cluster_policy_cross_target/test.toml create mode 100644 bundle/configsync/cluster_policy.go create mode 100644 bundle/configsync/cluster_policy_test.go diff --git a/.nextchanges/bundles/remote-sync-cluster-policies.md b/.nextchanges/bundles/remote-sync-cluster-policies.md new file mode 100644 index 00000000000..5aea51dca65 --- /dev/null +++ b/.nextchanges/bundles/remote-sync-cluster-policies.md @@ -0,0 +1 @@ +* Fix `bundle config-remote-sync` copying cluster policy values into shared configuration and breaking deployment to another target. 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..ed673e67bba --- /dev/null +++ b/acceptance/bundle/config-remote-sync/cluster_policy_cross_target/output.txt @@ -0,0 +1,54 @@ + +=== 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 +>>> errcode [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 + +>>> [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..801e745c116 --- /dev/null +++ b/acceptance/bundle/config-remote-sync/cluster_policy_cross_target/script @@ -0,0 +1,23 @@ +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 errcode $CLI bundle deploy -t prod 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..682e6f12a28 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,9 +18,11 @@ 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: 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..640244ae158 100644 --- a/acceptance/bundle/config-remote-sync/cluster_policy_remote_addition/script +++ b/acceptance/bundle/config-remote-sync/cluster_policy_remote_addition/script @@ -10,6 +10,7 @@ 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 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..ad9560a34e1 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,8 @@ 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 === 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 +21,7 @@ Resource: resources.jobs.job1 >>> diff.py databricks.yml.backup databricks.yml --- databricks.yml.backup +++ databricks.yml -@@ -9,5 +9,5 @@ +@@ -25,5 +25,5 @@ jobs: job1: - max_concurrent_runs: 1 @@ -30,8 +31,9 @@ Resource: resources.jobs.job1 >>> [CLI] bundle destroy --auto-approve The following resources will be deleted: + delete resources.cluster_policies.policy delete resources.jobs.job1 All files and directories at the following location will be deleted: /Workspace/Users/[USERNAME]/.bundle/test-bundle-[UNIQUE_NAME]/default -Destroy: 1 deleted +Destroy: 2 deleted diff --git a/bundle/configsync/cluster_policy.go b/bundle/configsync/cluster_policy.go new file mode 100644 index 00000000000..cb677aee939 --- /dev/null +++ b/bundle/configsync/cluster_policy.go @@ -0,0 +1,227 @@ +package configsync + +import ( + "context" + "fmt" + "reflect" + "strconv" + "strings" + + "github.com/databricks/cli/bundle" + "github.com/databricks/cli/bundle/deployplan" + "github.com/databricks/cli/libs/dyn" + "github.com/databricks/cli/libs/dyn/jsonloader" + "github.com/databricks/cli/libs/log" + "github.com/databricks/cli/libs/structs/structaccess" + "github.com/databricks/cli/libs/structs/structpath" +) + +type clusterPolicyDefinitions map[string]dyn.Value + +// get reads the effective definition, including policy-family overrides, once per sync. +func (definitions clusterPolicyDefinitions) get(ctx context.Context, b *bundle.Bundle, id string) (dyn.Value, error) { + if definition, ok := definitions[id]; ok { + return definition, nil + } + policy, err := b.WorkspaceClient(ctx).ClusterPolicies.GetByPolicyId(ctx, id) + if err != nil { + return dyn.InvalidValue, fmt.Errorf("reading cluster policy %s: %w", id, err) + } + definition, err := jsonloader.LoadJSON([]byte(policy.Definition), "cluster policy "+id) + if err != nil { + return dyn.InvalidValue, err + } + definitions[id] = definition + return definition, nil +} + +type clusterPolicyFilter struct { + definitions clusterPolicyDefinitions + resourceKey string + config any + remote any + standalone bool +} + +func newClusterPolicyFilter(resourceKey string, entry *deployplan.PlanEntry, definitions clusterPolicyDefinitions) *clusterPolicyFilter { + if !strings.HasPrefix(resourceKey, "resources.jobs.") && !strings.HasPrefix(resourceKey, "resources.clusters.") { + return nil + } + if entry.RemoteState == nil { + return nil + } + return &clusterPolicyFilter{ + definitions: definitions, + resourceKey: resourceKey, + remote: entry.RemoteState, + standalone: strings.HasPrefix(resourceKey, "resources.clusters."), + } +} + +// filter removes only omitted values that match the attached policy. It also handles +// whole-object additions, where the plan does not expose individual policy-applied leaves. +// A nil result means the value should be omitted from YAML. +func (f *clusterPolicyFilter) filter(ctx context.Context, b *bundle.Bundle, path *structpath.PathNode, value any) (any, error) { + if value == nil { + return nil, nil + } + applied, err := f.isPolicyApplied(ctx, b, path, value) + if err != nil { + return nil, err + } + if applied { + return nil, nil + } + + switch value := value.(type) { + case map[string]any: + result := make(map[string]any, len(value)) + for key, child := range value { + filtered, err := f.filter(ctx, b, structpath.NewStringKey(path, key), child) + if err != nil { + return nil, err + } + if filtered != nil { + result[key] = filtered + } + } + if len(value) > 0 && len(result) == 0 { + return nil, nil + } + _, attribute, _ := clusterPolicyAttribute(path, f.standalone) + if (attribute == "cluster_log_conf" || strings.HasPrefix(attribute, "cluster_log_conf.")) && !reflect.DeepEqual(value, result) { + // Log destinations are required. Removing only policy-matched leaves + // can leave an invalid object, such as an S3 region without a destination. + log.Warnf(ctx, "Keeping policy-applied values in %s to preserve the complete log configuration; review this field before deploying with another policy", path) + return value, nil + } + return result, nil + case []any: + result := make([]any, 0, len(value)) + for i, child := range value { + childPath := structpath.NewIndex(path, i) + // Match cluster owners by identity even when the UI reorders tasks. + if object, ok := child.(map[string]any); ok { + field, _ := path.StringKey() + var key string + switch field { + case "tasks": + key = "task_key" + case "job_clusters": + key = "job_cluster_key" + } + if name, ok := object[key].(string); ok && key != "" { + childPath = structpath.NewKeyValue(path, key, name) + } + } + filtered, err := f.filter(ctx, b, childPath, child) + if err != nil { + return nil, err + } + if filtered != nil { + result = append(result, filtered) + } + } + if len(value) > 0 && len(result) == 0 { + return nil, nil + } + // Indexed policy rules depend on positions. A partial removal would + // move a user-authored element into a different policy constraint. + if len(result) != len(value) { + log.Warnf(ctx, "Keeping policy-applied values in %s to preserve array positions; review this field before deploying with another policy", path) + return value, nil + } + return result, nil + default: + return value, nil + } +} + +func (f *clusterPolicyFilter) isPolicyApplied(ctx context.Context, b *bundle.Bundle, path *structpath.PathNode, value any) (bool, error) { + clusterPath, attribute, ok := clusterPolicyAttribute(path, f.standalone) + if !ok { + return false, nil + } + remotePolicyID, _ := structaccess.Get(f.remote, structpath.NewStringKey(clusterPath, "policy_id")) + policyID, _ := remotePolicyID.(string) + if policyID == "" { + return false, nil + } + if f.config == nil { + // Skip-only plans omit NewState. Load bundle config only when a + // policy-backed cluster field needs comparison, not for unrelated edits. + config, err := b.Config.GetResourceConfig(f.resourceKey) + if err != nil { + return false, fmt.Errorf("reading config for %s: %w", f.resourceKey, err) + } + if config == nil { + // A deleted resource has no source configuration to filter against. + return false, nil + } + f.config = config + } + // Missing fields and newly added keyed elements have no authored value. + if configValue, _ := structaccess.Get(f.config, path); configValue != nil { + return false, nil + } + definition, err := f.definitions.get(ctx, b, policyID) + if err != nil { + return false, err + } + // Jobs GET omits this input-only flag; use the authored setting. + applyDefaults, _ := structaccess.Get(f.config, structpath.NewStringKey(clusterPath, "apply_policy_default_values")) + return isPolicyAppliedValue(definition, attribute, value, applyDefaults == true), nil +} + +// clusterPolicyAttribute returns the enclosing cluster path and policy attribute name, +// preserving map keys such as spark_conf["a.b"] as the attribute spark_conf.a.b. +func clusterPolicyAttribute(path *structpath.PathNode, standalone bool) (*structpath.PathNode, string, bool) { + var clusterPath *structpath.PathNode + var parts []string + insideCluster := standalone + for _, node := range path.AsSlice() { + if key, ok := node.StringKey(); ok { + if !insideCluster && key == "new_cluster" { + clusterPath = node + insideCluster = true + continue + } + if insideCluster { + parts = append(parts, key) + } + } else if index, ok := node.Index(); ok && insideCluster { + parts = append(parts, strconv.Itoa(index)) + } + } + return clusterPath, strings.Join(parts, "."), insideCluster && len(parts) > 0 +} + +func isPolicyAppliedValue(definition dyn.Value, attribute string, remoteValue any, applyDefaults bool) bool { + // Policies describe log delivery with type/path, while the API nests the + // destination under its storage type (e.g. cluster_log_conf.dbfs.destination). + // https://docs.databricks.com/aws/en/admin/clusters/policy-definition + if attribute == "cluster_log_conf.s3.region" { + attribute = "cluster_log_conf.region" + } + if storage, ok := strings.CutPrefix(attribute, "cluster_log_conf."); ok { + if storage, ok := strings.CutSuffix(storage, ".destination"); ok { + return isPolicyAppliedValue(definition, "cluster_log_conf.type", strings.ToUpper(storage), applyDefaults) && + isPolicyAppliedValue(definition, "cluster_log_conf.path", remoteValue, applyDefaults) + } + } + rule, err := dyn.GetByPath(definition, dyn.NewPath(dyn.Key(attribute))) + if err != nil { + return false + } + typeValue, _ := dyn.Get(rule, "type") + var policyValue dyn.Value + switch { + case typeValue.AsAny() == "fixed": + policyValue, _ = dyn.Get(rule, "value") + case applyDefaults && typeValue.AsAny() != "forbidden": + policyValue, _ = dyn.Get(rule, "defaultValue") + default: + return false + } + return policyValue.IsValid() && policyValue.AsAny() != nil && reflect.DeepEqual(policyValue.AsAny(), remoteValue) +} diff --git a/bundle/configsync/cluster_policy_test.go b/bundle/configsync/cluster_policy_test.go new file mode 100644 index 00000000000..c34ef9c96c9 --- /dev/null +++ b/bundle/configsync/cluster_policy_test.go @@ -0,0 +1,403 @@ +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/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 TestExtractChangesDropsPolicyAppliedValues(t *testing.T) { + tests := []struct { + name string + attribute string + definition string + config any + remote any + applyDefaults bool + noPolicy bool + want any + }{ + { + name: "fixed S3 log configuration dropped", + 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", + 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: "fixed scalar dropped when absent from config", + attribute: "autotermination_minutes", + definition: `{"autotermination_minutes":{"type":"fixed","value":60}}`, + remote: int64(60), + }, + { + name: "fixed value dropped without apply flag", + 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", + 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"}, + }, + { + name: "authored field takes remote value", + 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", + attribute: "workload_type", + definition: `{"workload_type.clients.notebooks":{"type":"fixed","value":false}}`, + config: map[string]any{"clients": map[string]any{"notebooks": false}}, + remote: map[string]any{"clients": map[string]any{"notebooks": false}}, + want: map[string]any{"clients": map[string]any{"notebooks": false}}, + }, + { + name: "authored zero is kept in whole object changes", + attribute: "autoscale", + definition: `{"autoscale.min_workers":{"type":"fixed","value":0}}`, + config: map[string]any{"min_workers": int64(0)}, + remote: map[string]any{"min_workers": int64(0)}, + want: map[string]any{"min_workers": int64(0)}, + }, + { + name: "default dropped with apply flag", + 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", + 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", + 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: "fixed false dropped and empty parents pruned", + 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", + 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", + 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", + attribute: "init_scripts", + definition: `{"init_scripts.0.volumes.destination":{"type":"fixed","value":"/Volumes/c/s/v/init.sh"}}`, + remote: []any{map[string]any{"volumes": map[string]any{"destination": "/Volumes/c/s/v/init.sh"}}}, + }, + { + name: "partial array match keeps whole array", + 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", + 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", + 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", + attribute: "custom_tags", + noPolicy: true, + remote: map[string]any{"Owner": "user"}, + want: map[string]any{"Owner": "user"}, + }, + { + name: "log configuration without a policy syncs", + 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", + attribute: "custom_tags", + definition: `{}`, + remote: map[string]any{"Owner": "user"}, + want: map[string]any{"Owner": "user"}, + }, + } + + 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.config == nil && !tt.noPolicy { + client.GetMockClusterPoliciesAPI().EXPECT().GetByPolicyId(mock.Anything, "policy"). + Return(&compute.Policy{Definition: tt.definition}, nil).Once() + } + + changes, err := configsync.ExtractChanges(t.Context(), b, &deployplan.Plan{Plan: map[string]*deployplan.PlanEntry{key: entry}}) + require.NoError(t, err) + if tt.want == nil { + assert.Empty(t, changes) + } else { + require.Contains(t, changes, key) + require.Contains(t, changes[key], path) + assert.Equal(t, tt.want, changes[key][path].Value) + } + }) + } + } +} + +func TestExtractChangesFailsOnPolicyLookupError(t *testing.T) { + b := policyTestBundle(t, "resources.clusters.test", map[string]any{"policy_id": "policy"}) + client := mocks.NewMockWorkspaceClient(t) + b.SetWorkpaceClient(client.WorkspaceClient) + lookupError := errors.New("policy access denied") + client.GetMockClusterPoliciesAPI().EXPECT().GetByPolicyId(mock.Anything, "policy").Return(nil, lookupError).Once() + entry := &deployplan.PlanEntry{ + RemoteState: &dresources.ClusterRemote{PolicyId: "policy"}, + Changes: deployplan.Changes{ + "spark_env_vars": {Action: deployplan.Update, Remote: map[string]any{"ENV": "dev"}}, + }, + } + changes, err := configsync.ExtractChanges(t.Context(), b, &deployplan.Plan{Plan: map[string]*deployplan.PlanEntry{"resources.clusters.test": entry}}) + assert.ErrorIs(t, err, lookupError) + assert.ErrorContains(t, err, "no configuration was saved") + assert.Nil(t, changes) +} + +func policyClusterLocation(location string, cluster map[string]any) (resourceKey, pathPrefix string, root map[string]any) { + switch location { + case "task": + return "resources.jobs.test", "tasks[task_key='main'].new_cluster.", map[string]any{ + "tasks": []any{map[string]any{"task_key": "main", "new_cluster": cluster}}, + } + case "shared": + return "resources.jobs.test", "job_clusters[job_cluster_key='main'].new_cluster.", map[string]any{ + "job_clusters": []any{map[string]any{"job_cluster_key": "main", "new_cluster": cluster}}, + } + case "for_each": + return "resources.jobs.test", "tasks[task_key='each'].for_each_task.task.new_cluster.", map[string]any{ + "tasks": []any{map[string]any{"task_key": "each", "for_each_task": map[string]any{ + "task": map[string]any{"task_key": "main", "new_cluster": cluster}, + }}}, + } + default: + return "resources.clusters.test", "", cluster + } +} + +func typedPolicyResource(t *testing.T, location string, value map[string]any) any { + t.Helper() + var resource any = &dresources.JobRemote{} + if location == "standalone" { + resource = &dresources.ClusterRemote{} + } + data, err := json.Marshal(value) + require.NoError(t, err) + require.NoError(t, json.Unmarshal(data, resource)) + return resource +} + +func TestExtractChangesMatchesClustersByKeyAndRemotePolicy(t *testing.T) { + config := map[string]any{"job_clusters": []any{ + map[string]any{"job_cluster_key": "explicit", "new_cluster": map[string]any{ + "policy_id": "old-policy", "spark_env_vars": map[string]any{"ENV": "authored"}, + }}, + map[string]any{"job_cluster_key": "omitted", "new_cluster": map[string]any{"policy_id": "old-policy"}}, + }} + remote := map[string]any{"job_clusters": []any{ + map[string]any{"job_cluster_key": "omitted", "new_cluster": map[string]any{ + "policy_id": "new-policy", "spark_env_vars": map[string]any{"ENV": "dev"}, + }}, + map[string]any{"job_cluster_key": "explicit", "new_cluster": map[string]any{ + "policy_id": "new-policy", "spark_env_vars": map[string]any{"ENV": "dev"}, + }}, + }} + entry := &deployplan.PlanEntry{ + RemoteState: typedPolicyResource(t, "shared", remote), + Changes: deployplan.Changes{"job_clusters": { + Action: deployplan.Update, New: config["job_clusters"], Remote: remote["job_clusters"], + }}, + } + b := policyTestBundle(t, "resources.jobs.test", config) + client := mocks.NewMockWorkspaceClient(t) + b.SetWorkpaceClient(client.WorkspaceClient) + // The omitted cluster uses the remote policy even though config has a different ID. + client.GetMockClusterPoliciesAPI().EXPECT().GetByPolicyId(mock.Anything, "new-policy"). + Return(&compute.Policy{Definition: `{"spark_env_vars.ENV":{"type":"fixed","value":"dev"}}`}, nil).Once() + + changes, err := configsync.ExtractChanges(t.Context(), b, &deployplan.Plan{Plan: map[string]*deployplan.PlanEntry{"resources.jobs.test": entry}}) + require.NoError(t, err) + require.Contains(t, changes, "resources.jobs.test") + require.Contains(t, changes["resources.jobs.test"], "job_clusters") + assert.Equal(t, []any{ + map[string]any{"job_cluster_key": "omitted", "new_cluster": map[string]any{"policy_id": "new-policy"}}, + map[string]any{"job_cluster_key": "explicit", "new_cluster": map[string]any{ + "policy_id": "new-policy", "spark_env_vars": map[string]any{"ENV": "dev"}, + }}, + }, changes["resources.jobs.test"]["job_clusters"].Value) +} + +func 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 TestExtractChangesDropsPolicyValueAtDottedLeaf(t *testing.T) { + key, prefix, config := policyClusterLocation("task", map[string]any{ + "policy_id": "policy", + "spark_conf": map[string]any{"spark.sql.adaptive.enabled": "true"}, + }) + _, _, remote := policyClusterLocation("task", map[string]any{ + "policy_id": "policy", + "spark_conf": map[string]any{"spark.sql.adaptive.enabled": "true", "spark.sql.shuffle.partitions": "7"}, + }) + b := policyTestBundle(t, key, config) + client := mocks.NewMockWorkspaceClient(t) + b.SetWorkpaceClient(client.WorkspaceClient) + client.GetMockClusterPoliciesAPI().EXPECT().GetByPolicyId(mock.Anything, "policy"). + Return(&compute.Policy{Definition: `{"spark_conf.spark.sql.shuffle.partitions":{"type":"fixed","value":"7"}}`}, nil).Once() + entry := &deployplan.PlanEntry{ + Action: deployplan.Skip, + RemoteState: typedPolicyResource(t, "task", remote), + Changes: deployplan.Changes{prefix + "spark_conf['spark.sql.shuffle.partitions']": { + Action: deployplan.Skip, Reason: deployplan.ReasonRemoteAddition, Remote: "7", + }}, + } + changes, err := configsync.ExtractChanges(t.Context(), b, &deployplan.Plan{Plan: map[string]*deployplan.PlanEntry{key: entry}}) + require.NoError(t, err) + assert.Empty(t, changes) +} + +func TestExtractChangesKeepsCompleteLogConfigurationWhenStorageChanges(t *testing.T) { + key, prefix, config := policyClusterLocation("task", map[string]any{ + "policy_id": "policy", "apply_policy_default_values": true, + "cluster_log_conf": map[string]any{"dbfs": map[string]any{"destination": "dbfs:/old-logs"}}, + }) + newDestination := map[string]any{"destination": "s3://logs", "region": "us-east-1"} + _, _, remote := policyClusterLocation("task", map[string]any{ + "policy_id": "policy", "cluster_log_conf": map[string]any{"s3": newDestination}, + }) + remoteState := typedPolicyResource(t, "task", remote) + path := prefix + "cluster_log_conf.s3" + remoteValue, err := structaccess.GetByString(remoteState, path) + require.NoError(t, err) + entry := &deployplan.PlanEntry{ + RemoteState: remoteState, + Changes: deployplan.Changes{ + path: {Action: deployplan.Update, Remote: remoteValue}, + prefix + "cluster_log_conf.dbfs": {Action: deployplan.Update, New: map[string]any{"destination": "dbfs:/old-logs"}}, + }, + } + b := policyTestBundle(t, key, config) + client := mocks.NewMockWorkspaceClient(t) + b.SetWorkpaceClient(client.WorkspaceClient) + client.GetMockClusterPoliciesAPI().EXPECT().GetByPolicyId(mock.Anything, "policy"). + Return(&compute.Policy{Definition: `{"cluster_log_conf.type":{"type":"fixed","value":"S3"},"cluster_log_conf.path":{"type":"fixed","value":"s3://logs"},"cluster_log_conf.region":{"type":"unlimited","defaultValue":"us-west-2"}}`}, nil).Once() + + changes, err := configsync.ExtractChanges(t.Context(), b, &deployplan.Plan{Plan: map[string]*deployplan.PlanEntry{key: entry}}) + require.NoError(t, err) + 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) +} diff --git a/bundle/configsync/defaults.go b/bundle/configsync/defaults.go index 0981b0625d6..f28de8098c5 100644 --- a/bundle/configsync/defaults.go +++ b/bundle/configsync/defaults.go @@ -11,18 +11,12 @@ type ( skipIfEmptyOrDefault struct { defaults map[string]any } - // skipBackendDefault skips a field, regardless of its remote value, when it is - // absent from config. Used for fields the backend or a cluster policy may fill - // in: their remote-only value must not be synced into config. Fields the user - // does manage (present in config) still sync normally. - skipBackendDefault struct{} ) var ( alwaysSkip = skipAlways{} zeroOrNil = skipIfZeroOrNil{} emptyEmailNotifications = skipIfEmptyOrDefault{defaults: map[string]any{"no_alert_for_skipped_runs": false}} - backendDefault = skipBackendDefault{} ) // serverSideDefaults contains all hardcoded server-side defaults. @@ -60,13 +54,6 @@ var serverSideDefaults = map[string]any{ "resources.jobs.*.tasks[*].new_cluster.data_security_mode": "SINGLE_USER", // TODO this field is computed on some workspaces in integration tests, check why and if we can skip it "resources.jobs.*.tasks[*].new_cluster.enable_elastic_disk": alwaysSkip, // deprecated field "resources.jobs.*.tasks[*].new_cluster.single_user_name": alwaysSkip, - // custom_tags and cluster_log_conf are commonly injected by cluster policies - // when the user omits them, so they exist only remotely. Syncing them back leaks - // one environment's policy values into (often shared) config and breaks deploys in - // other environments. TODO: move to backend_defaults in jobs.yml once - // configsync filtering is migrated to the direct engine lifecycle metadata. - "resources.jobs.*.tasks[*].new_cluster.custom_tags": backendDefault, - "resources.jobs.*.tasks[*].new_cluster.cluster_log_conf": backendDefault, // Cluster fields (job_clusters) "resources.jobs.*.job_clusters[*].new_cluster.aws_attributes": alwaysSkip, @@ -75,8 +62,6 @@ var serverSideDefaults = map[string]any{ "resources.jobs.*.job_clusters[*].new_cluster.data_security_mode": "SINGLE_USER", // TODO this field is computed on some workspaces in integration tests, check why and if we can skip it "resources.jobs.*.job_clusters[*].new_cluster.enable_elastic_disk": alwaysSkip, // deprecated field "resources.jobs.*.job_clusters[*].new_cluster.single_user_name": alwaysSkip, - "resources.jobs.*.job_clusters[*].new_cluster.custom_tags": backendDefault, // see tasks[*].new_cluster.custom_tags - "resources.jobs.*.job_clusters[*].new_cluster.cluster_log_conf": backendDefault, // see tasks[*].new_cluster.cluster_log_conf // Standalone cluster fields "resources.clusters.*.aws_attributes": alwaysSkip, @@ -86,8 +71,6 @@ var serverSideDefaults = map[string]any{ "resources.clusters.*.driver_node_type_id": alwaysSkip, "resources.clusters.*.enable_elastic_disk": alwaysSkip, "resources.clusters.*.single_user_name": alwaysSkip, - "resources.clusters.*.custom_tags": backendDefault, // see jobs.*.tasks[*].new_cluster.custom_tags - "resources.clusters.*.cluster_log_conf": backendDefault, // see jobs.*.tasks[*].new_cluster.cluster_log_conf // Experiment fields "resources.experiments.*.artifact_location": alwaysSkip, @@ -146,9 +129,6 @@ func shouldSkipField(path string, value any, hasConfigValue bool) bool { if hasConfigValue { return false } - if _, ok := expected.(skipBackendDefault); ok { - return true - } if _, ok := expected.(skipIfZeroOrNil); ok { return value == nil || value == int64(0) } diff --git a/bundle/configsync/diff.go b/bundle/configsync/diff.go index e42f0a1d83a..81e4c1eaf8a 100644 --- a/bundle/configsync/diff.go +++ b/bundle/configsync/diff.go @@ -12,6 +12,7 @@ import ( "github.com/databricks/cli/libs/dyn/convert" "github.com/databricks/cli/libs/log" "github.com/databricks/cli/libs/structs/structdiff" + "github.com/databricks/cli/libs/structs/structpath" ) type OperationType string @@ -141,6 +142,7 @@ func isPermissionsOrGrantsSubResource(resourceKey string) bool { // plan. func ExtractChanges(ctx context.Context, b *bundle.Bundle, plan *deployplan.Plan) (Changes, error) { changes := make(Changes) + policyDefinitions := make(clusterPolicyDefinitions) for resourceKey, entry := range plan.Plan { // permissions and grants are emitted as their own plan keys @@ -154,6 +156,7 @@ func ExtractChanges(ctx context.Context, b *bundle.Bundle, plan *deployplan.Plan } resourceChanges := make(ResourceChanges) + policyFilter := newClusterPolicyFilter(resourceKey, entry, policyDefinitions) if entry.Changes != nil { for path, changeDesc := range entry.Changes { @@ -169,6 +172,19 @@ func ExtractChanges(ctx context.Context, b *bundle.Bundle, plan *deployplan.Plan if change.Operation == OperationSkip { continue } + if change.Value != nil && policyFilter != nil { + fieldPath, err := structpath.ParsePath(path) + if err != nil { + return nil, fmt.Errorf("parsing change path %s: %w", fullPath, err) + } + change.Value, err = policyFilter.filter(ctx, b, fieldPath, change.Value) + if err != nil { + return nil, fmt.Errorf("filtering policy values for %s (no configuration was saved): %w", fullPath, err) + } + if change.Value == nil { + continue + } + } // The state snapshot holds real per-field values, so New != Old // means the local config diverged from the last deploy and this // change overwrites that pending edit. From d65985a8077396f0b4974f2fa75070273a42712d Mon Sep 17 00:00:00 2001 From: ilyakuz-db <181845775+ilyakuz-db@users.noreply.github.com> Date: Fri, 23 Oct 2026 14:52:18 +0000 Subject: [PATCH 2/4] Link the remote sync release note to its pull request --- .nextchanges/bundles/remote-sync-cluster-policies.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.nextchanges/bundles/remote-sync-cluster-policies.md b/.nextchanges/bundles/remote-sync-cluster-policies.md index 5aea51dca65..df06c0c9783 100644 --- a/.nextchanges/bundles/remote-sync-cluster-policies.md +++ b/.nextchanges/bundles/remote-sync-cluster-policies.md @@ -1 +1 @@ -* Fix `bundle config-remote-sync` copying cluster policy values into shared configuration and breaking deployment to another target. +* Fix `bundle config-remote-sync` copying cluster policy values into shared configuration and breaking deployment to another target. ([#7170](https://github.com/databricks/cli/pull/7170)) From 2f7808b4f731d2da8171a0e3e9d2a26b89e96deb Mon Sep 17 00:00:00 2001 From: ilyakuz-db <181845775+ilyakuz-db@users.noreply.github.com> Date: Fri, 23 Oct 2026 16:07:06 +0000 Subject: [PATCH 3/4] Refactor cluster policy sync filtering and expand coverage --- .../databricks.yml.tmpl | 38 +++ .../hello.py | 1 + .../out.test.toml | 3 + .../output.txt | 55 ++++ .../policy.json.tmpl | 4 + .../script | 35 +++ .../test.toml | 3 + bundle/configsync/cluster_policy.go | 278 +++++++++--------- bundle/configsync/cluster_policy_test.go | 104 +++++++ bundle/configsync/diff.go | 36 ++- libs/structs/structdiff/keyed_path.go | 48 +++ libs/structs/structdiff/keyed_path_test.go | 60 ++++ 12 files changed, 517 insertions(+), 148 deletions(-) create mode 100644 acceptance/bundle/config-remote-sync/external_cluster_policy_cross_target/databricks.yml.tmpl create mode 100644 acceptance/bundle/config-remote-sync/external_cluster_policy_cross_target/hello.py create mode 100644 acceptance/bundle/config-remote-sync/external_cluster_policy_cross_target/out.test.toml create mode 100644 acceptance/bundle/config-remote-sync/external_cluster_policy_cross_target/output.txt create mode 100644 acceptance/bundle/config-remote-sync/external_cluster_policy_cross_target/policy.json.tmpl create mode 100644 acceptance/bundle/config-remote-sync/external_cluster_policy_cross_target/script create mode 100644 acceptance/bundle/config-remote-sync/external_cluster_policy_cross_target/test.toml create mode 100644 libs/structs/structdiff/keyed_path.go create mode 100644 libs/structs/structdiff/keyed_path_test.go 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..68120b8905b --- /dev/null +++ b/acceptance/bundle/config-remote-sync/external_cluster_policy_cross_target/output.txt @@ -0,0 +1,55 @@ + +=== 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 +>>> errcode [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 + +>>> [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 [DEV_POLICY_ID] + +>>> [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..9629a24bcee --- /dev/null +++ b/acceptance/bundle/config-remote-sync/external_cluster_policy_cross_target/script @@ -0,0 +1,35 @@ +title "Create policies outside the bundle" +export POLICY_ENVIRONMENT=dev +export DEV_POLICY_ID +DEV_POLICY_ID=$($CLI cluster-policies create --json "$(envsubst < policy.json.tmpl)" -o json | jq -r .policy_id) +add_repl "$DEV_POLICY_ID" DEV_POLICY_ID +export POLICY_ENVIRONMENT=prod +export PROD_POLICY_ID +PROD_POLICY_ID=$($CLI cluster-policies create --json "$(envsubst < policy.json.tmpl)" -o json | jq -r .policy_id) +add_repl "$PROD_POLICY_ID" PROD_POLICY_ID + +envsubst < databricks.yml.tmpl > databricks.yml + +cleanup() { + trace $CLI bundle destroy -t prod --auto-approve + trace $CLI bundle destroy -t dev --auto-approve + trace $CLI cluster-policies delete "$DEV_POLICY_ID" + trace $CLI cluster-policies delete "$PROD_POLICY_ID" +} +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 errcode $CLI bundle deploy -t prod 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/bundle/configsync/cluster_policy.go b/bundle/configsync/cluster_policy.go index cb677aee939..91582062a6b 100644 --- a/bundle/configsync/cluster_policy.go +++ b/bundle/configsync/cluster_policy.go @@ -2,18 +2,20 @@ package configsync import ( "context" + "errors" "fmt" "reflect" "strconv" "strings" "github.com/databricks/cli/bundle" - "github.com/databricks/cli/bundle/deployplan" "github.com/databricks/cli/libs/dyn" + "github.com/databricks/cli/libs/dyn/convert" "github.com/databricks/cli/libs/dyn/jsonloader" - "github.com/databricks/cli/libs/log" "github.com/databricks/cli/libs/structs/structaccess" + "github.com/databricks/cli/libs/structs/structdiff" "github.com/databricks/cli/libs/structs/structpath" + "github.com/databricks/cli/libs/structs/structtag" ) type clusterPolicyDefinitions map[string]dyn.Value @@ -36,192 +38,174 @@ func (definitions clusterPolicyDefinitions) get(ctx context.Context, b *bundle.B } type clusterPolicyFilter struct { - definitions clusterPolicyDefinitions - resourceKey string - config any - remote any - standalone bool + definitions clusterPolicyDefinitions + config any + remote any + clusterRoots []*structpath.PatternNode + sliceKeys map[string]structdiff.KeyFunc } -func newClusterPolicyFilter(resourceKey string, entry *deployplan.PlanEntry, definitions clusterPolicyDefinitions) *clusterPolicyFilter { - if !strings.HasPrefix(resourceKey, "resources.jobs.") && !strings.HasPrefix(resourceKey, "resources.clusters.") { - return nil - } - if entry.RemoteState == nil { - return nil - } - return &clusterPolicyFilter{ - definitions: definitions, - resourceKey: resourceKey, - remote: entry.RemoteState, - standalone: strings.HasPrefix(resourceKey, "resources.clusters."), - } -} - -// filter removes only omitted values that match the attached policy. It also handles -// whole-object additions, where the plan does not expose individual policy-applied leaves. -// A nil result means the value should be omitted from YAML. -func (f *clusterPolicyFilter) filter(ctx context.Context, b *bundle.Bundle, path *structpath.PathNode, value any) (any, error) { - if value == nil { - return nil, nil - } - applied, err := f.isPolicyApplied(ctx, b, path, value) +// filter removes omitted policy values from a normalized plan change, including +// whole-object additions. A nil result means no value needs to be synced. +func (f *clusterPolicyFilter) filter(ctx context.Context, b *bundle.Bundle, path *structpath.PathNode, changeValue any) (any, error) { + normalized, err := convert.FromTyped(changeValue, dyn.NilValue) if err != nil { return nil, err } - if applied { - return nil, nil - } - - switch value := value.(type) { - case map[string]any: - result := make(map[string]any, len(value)) - for key, child := range value { - filtered, err := f.filter(ctx, b, structpath.NewStringKey(path, key), child) - if err != nil { - return nil, err - } - if filtered != nil { - result[key] = filtered - } - } - if len(value) > 0 && len(result) == 0 { - return nil, nil - } - _, attribute, _ := clusterPolicyAttribute(path, f.standalone) - if (attribute == "cluster_log_conf" || strings.HasPrefix(attribute, "cluster_log_conf.")) && !reflect.DeepEqual(value, result) { - // Log destinations are required. Removing only policy-matched leaves - // can leave an invalid object, such as an S3 region without a destination. - log.Warnf(ctx, "Keeping policy-applied values in %s to preserve the complete log configuration; review this field before deploying with another policy", path) - return value, nil + filtered, err := dyn.Walk(normalized, func(relative dyn.Path, value dyn.Value) (dyn.Value, error) { + fieldPath, err := structdiff.KeyedPath(f.remote, appendDynPath(path, relative), f.sliceKeys) + if err != nil { + return dyn.InvalidValue, err } - return result, nil - case []any: - result := make([]any, 0, len(value)) - for i, child := range value { - childPath := structpath.NewIndex(path, i) - // Match cluster owners by identity even when the UI reorders tasks. - if object, ok := child.(map[string]any); ok { - field, _ := path.StringKey() - var key string - switch field { - case "tasks": - key = "task_key" - case "job_clusters": - key = "job_cluster_key" - } - if name, ok := object[key].(string); ok && key != "" { - childPath = structpath.NewKeyValue(path, key, name) + for _, root := range f.clusterRoots { + if !fieldPath.HasPatternPrefix(root) || fieldPath.Len() == root.Len() { + continue + } + clusterPath := fieldPath.Prefix(root.Len()) + // A change may start inside an array, so protecting only sequence + // nodes would still allow an element to be emptied or removed. + for _, node := range fieldPath.SkipPrefix(root.Len()).AsSlice() { + if _, indexed := node.Index(); indexed { + return value, dyn.ErrSkip } } - filtered, err := f.filter(ctx, b, childPath, child) + // Keep required fields, but still filter their optional children. + if len(relative) > 0 && isRequiredField(f.remote, fieldPath) { + return value, nil + } + matches, err := f.matchesOmittedPolicySubtree(ctx, b, clusterPath, fieldPath, value) if err != nil { - return nil, err + return dyn.InvalidValue, err } - if filtered != nil { - result = append(result, filtered) + if matches { + return dyn.InvalidValue, dyn.ErrDrop } + // Removing elements would change indexed policy constraints. + if value.Kind() == dyn.KindSequence { + return value, dyn.ErrSkip + } + break } - if len(value) > 0 && len(result) == 0 { - return nil, nil - } - // Indexed policy rules depend on positions. A partial removal would - // move a user-authored element into a different policy constraint. - if len(result) != len(value) { - log.Warnf(ctx, "Keeping policy-applied values in %s to preserve array positions; review this field before deploying with another policy", path) - return value, nil - } - return result, nil - default: return value, nil + }) + if errors.Is(err, dyn.ErrDrop) { + return nil, nil + } + if err != nil { + return nil, err } + return filtered.AsAny(), nil } -func (f *clusterPolicyFilter) isPolicyApplied(ctx context.Context, b *bundle.Bundle, path *structpath.PathNode, value any) (bool, error) { - clusterPath, attribute, ok := clusterPolicyAttribute(path, f.standalone) - if !ok { - return false, nil - } - remotePolicyID, _ := structaccess.Get(f.remote, structpath.NewStringKey(clusterPath, "policy_id")) - policyID, _ := remotePolicyID.(string) - if policyID == "" { - return false, nil - } - if f.config == nil { - // Skip-only plans omit NewState. Load bundle config only when a - // policy-backed cluster field needs comparison, not for unrelated edits. - config, err := b.Config.GetResourceConfig(f.resourceKey) - if err != nil { - return false, fmt.Errorf("reading config for %s: %w", f.resourceKey, err) +// matchesOmittedPolicySubtree requires every leaf to be absent from authored config +// and equal to a fixed value or an opted-in default in the remote cluster's policy. +func (f *clusterPolicyFilter) matchesOmittedPolicySubtree(ctx context.Context, b *bundle.Bundle, clusterPath, path *structpath.PathNode, value dyn.Value) (bool, error) { + matches := true + leaves := 0 + _, err := dyn.Walk(value, func(relative dyn.Path, leaf dyn.Value) (dyn.Value, error) { + if !matches { + return leaf, dyn.ErrSkip } - if config == nil { - // A deleted resource has no source configuration to filter against. - return false, nil + if leaf.Kind() == dyn.KindMap || leaf.Kind() == dyn.KindSequence { + return leaf, nil } - f.config = config + leaves++ + fieldPath := appendDynPath(path, relative) + if authored, _ := structaccess.Get(f.config, fieldPath); authored != nil { + matches = false + return leaf, nil + } + remotePolicyID, _ := structaccess.Get(f.remote, structpath.NewStringKey(clusterPath, "policy_id")) + policyID, _ := remotePolicyID.(string) + if policyID == "" { + matches = false + return leaf, nil + } + definition, err := f.definitions.get(ctx, b, policyID) + if err != nil { + return dyn.InvalidValue, err + } + // Remote cluster specs omit this input-only flag; use the authored setting. + applyDefaults, _ := structaccess.Get(f.config, structpath.NewStringKey(clusterPath, "apply_policy_default_values")) + expected := policyValueForAPIAttribute(definition, policyAttribute(fieldPath.SkipPrefix(clusterPath.Len())), applyDefaults == true) + matches = expected.IsValid() && expected.AsAny() != nil && reflect.DeepEqual(expected.AsAny(), leaf.AsAny()) + return leaf, nil + }) + return matches && leaves > 0, err +} + +func isRequiredField(remote any, path *structpath.PathNode) bool { + key, ok := path.StringKey() + if !ok { + return false + } + parent, _ := structaccess.Get(remote, path.Parent()) + typ := reflect.TypeOf(parent) + if typ == nil { + return false } - // Missing fields and newly added keyed elements have no authored value. - if configValue, _ := structaccess.Get(f.config, path); configValue != nil { - return false, nil + for typ.Kind() == reflect.Pointer { + typ = typ.Elem() } - definition, err := f.definitions.get(ctx, b, policyID) - if err != nil { - return false, err + field, _, ok := structaccess.FindStructFieldByKeyType(typ, key) + return ok && !structtag.JSONTag(field.Tag.Get("json")).OmitEmpty() +} + +func appendDynPath(base *structpath.PathNode, suffix dyn.Path) *structpath.PathNode { + for _, component := range suffix { + if key := component.Key(); key != "" { + base = structpath.NewStringKey(base, key) + } else { + base = structpath.NewIndex(base, component.Index()) + } } - // Jobs GET omits this input-only flag; use the authored setting. - applyDefaults, _ := structaccess.Get(f.config, structpath.NewStringKey(clusterPath, "apply_policy_default_values")) - return isPolicyAppliedValue(definition, attribute, value, applyDefaults == true), nil + return base } -// clusterPolicyAttribute returns the enclosing cluster path and policy attribute name, -// preserving map keys such as spark_conf["a.b"] as the attribute spark_conf.a.b. -func clusterPolicyAttribute(path *structpath.PathNode, standalone bool) (*structpath.PathNode, string, bool) { - var clusterPath *structpath.PathNode +// policyAttribute preserves dotted map keys, e.g. spark_conf["a.b"] becomes spark_conf.a.b. +func policyAttribute(path *structpath.PathNode) string { var parts []string - insideCluster := standalone for _, node := range path.AsSlice() { if key, ok := node.StringKey(); ok { - if !insideCluster && key == "new_cluster" { - clusterPath = node - insideCluster = true - continue - } - if insideCluster { - parts = append(parts, key) - } - } else if index, ok := node.Index(); ok && insideCluster { + parts = append(parts, key) + } else if index, ok := node.Index(); ok { parts = append(parts, strconv.Itoa(index)) } } - return clusterPath, strings.Join(parts, "."), insideCluster && len(parts) > 0 + return strings.Join(parts, ".") } -func isPolicyAppliedValue(definition dyn.Value, attribute string, remoteValue any, applyDefaults bool) bool { - // Policies describe log delivery with type/path, while the API nests the - // destination under its storage type (e.g. cluster_log_conf.dbfs.destination). - // https://docs.databricks.com/aws/en/admin/clusters/policy-definition +// policyValueForAPIAttribute translates the policy language's log-delivery aliases. +// https://docs.databricks.com/aws/en/admin/clusters/policy-definition +func policyValueForAPIAttribute(definition dyn.Value, attribute string, applyDefaults bool) dyn.Value { if attribute == "cluster_log_conf.s3.region" { attribute = "cluster_log_conf.region" } if storage, ok := strings.CutPrefix(attribute, "cluster_log_conf."); ok { if storage, ok := strings.CutSuffix(storage, ".destination"); ok { - return isPolicyAppliedValue(definition, "cluster_log_conf.type", strings.ToUpper(storage), applyDefaults) && - isPolicyAppliedValue(definition, "cluster_log_conf.path", remoteValue, applyDefaults) + logType := policyRuleValue(definition, "cluster_log_conf.type", applyDefaults) + if !logType.IsValid() || logType.AsAny() != strings.ToUpper(storage) { + return dyn.InvalidValue + } + return policyRuleValue(definition, "cluster_log_conf.path", applyDefaults) } } + return policyRuleValue(definition, attribute, applyDefaults) +} + +// policyRuleValue returns the value supplied by a rule, or InvalidValue if it supplies none. +func policyRuleValue(definition dyn.Value, attribute string, applyDefaults bool) dyn.Value { rule, err := dyn.GetByPath(definition, dyn.NewPath(dyn.Key(attribute))) if err != nil { - return false + return dyn.InvalidValue } - typeValue, _ := dyn.Get(rule, "type") - var policyValue dyn.Value + ruleType, _ := dyn.Get(rule, "type") + var value dyn.Value switch { - case typeValue.AsAny() == "fixed": - policyValue, _ = dyn.Get(rule, "value") - case applyDefaults && typeValue.AsAny() != "forbidden": - policyValue, _ = dyn.Get(rule, "defaultValue") - default: - return false + case ruleType.AsAny() == "fixed": + value, _ = dyn.Get(rule, "value") + case applyDefaults && ruleType.AsAny() != "forbidden": + value, _ = dyn.Get(rule, "defaultValue") } - return policyValue.IsValid() && policyValue.AsAny() != nil && reflect.DeepEqual(policyValue.AsAny(), remoteValue) + return value } diff --git a/bundle/configsync/cluster_policy_test.go b/bundle/configsync/cluster_policy_test.go index c34ef9c96c9..dcc8392b159 100644 --- a/bundle/configsync/cluster_policy_test.go +++ b/bundle/configsync/cluster_policy_test.go @@ -45,6 +45,27 @@ func TestExtractChangesDropsPolicyAppliedValues(t *testing.T) { 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", + 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", + 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", + attribute: "autoscale", + definition: `{"autoscale.min_workers":{"type":"fixed","value":1}}`, + remote: map[string]any{"min_workers": int64(1), "max_workers": int64(4)}, + want: map[string]any{"max_workers": int64(4)}, + }, { name: "fixed scalar dropped when absent from config", attribute: "autotermination_minutes", @@ -401,3 +422,86 @@ func TestExtractChangesKeepsCompleteLogConfigurationWhenStorageChanges(t *testin assert.Equal(t, newDestination, changes[key][path].Value) assert.Equal(t, configsync.OperationRemove, changes[key][prefix+"cluster_log_conf.dbfs"].Operation) } + +func TestExtractChangesUsesEffectivePolicyFamilyDefinition(t *testing.T) { + b := policyTestBundle(t, "resources.clusters.test", map[string]any{"policy_id": "family-policy"}) + client := mocks.NewMockWorkspaceClient(t) + b.SetWorkpaceClient(client.WorkspaceClient) + // GET supplies the merged definition; the bundle does not own this policy. + client.GetMockClusterPoliciesAPI().EXPECT().GetByPolicyId(mock.Anything, "family-policy").Return(&compute.Policy{ + PolicyFamilyId: "job-cluster", + PolicyFamilyDefinitionOverrides: `{"custom_tags.Environment":{"type":"fixed","value":"dev"}}`, + Definition: `{"custom_tags.Inherited":{"type":"fixed","value":"family"},"custom_tags.Environment":{"type":"fixed","value":"dev"}}`, + }, nil).Once() + remoteTags := map[string]string{"Inherited": "family", "Environment": "dev", "Owner": "user"} + entry := &deployplan.PlanEntry{ + RemoteState: &dresources.ClusterRemote{PolicyId: "family-policy", CustomTags: remoteTags}, + Changes: deployplan.Changes{ + "custom_tags": {Action: deployplan.Skip, Reason: deployplan.ReasonRemoteAddition, Remote: remoteTags}, + }, + } + changes, err := configsync.ExtractChanges(t.Context(), b, &deployplan.Plan{Plan: map[string]*deployplan.PlanEntry{"resources.clusters.test": entry}}) + require.NoError(t, err) + require.Contains(t, changes, "resources.clusters.test") + require.Contains(t, changes["resources.clusters.test"], "custom_tags") + assert.Equal(t, map[string]any{"Owner": "user"}, changes["resources.clusters.test"]["custom_tags"].Value) +} + +func TestExtractChangesDropsPolicyAdditionBesideAuthoredField(t *testing.T) { + b := policyTestBundle(t, "resources.clusters.test", map[string]any{ + "policy_id": "policy", "autoscale": map[string]any{"max_workers": int64(4)}, + }) + client := mocks.NewMockWorkspaceClient(t) + b.SetWorkpaceClient(client.WorkspaceClient) + client.GetMockClusterPoliciesAPI().EXPECT().GetByPolicyId(mock.Anything, "policy"). + Return(&compute.Policy{Definition: `{"autoscale.min_workers":{"type":"fixed","value":1}}`}, nil).Once() + remote := &compute.AutoScale{MinWorkers: 1, MaxWorkers: 4} + entry := &deployplan.PlanEntry{ + RemoteState: &dresources.ClusterRemote{PolicyId: "policy", Autoscale: remote}, + Changes: deployplan.Changes{"autoscale": {Action: deployplan.Update, New: &compute.AutoScale{MaxWorkers: 4}, Remote: remote}}, + } + changes, err := configsync.ExtractChanges(t.Context(), b, &deployplan.Plan{Plan: map[string]*deployplan.PlanEntry{"resources.clusters.test": entry}}) + require.NoError(t, err) + require.Contains(t, changes, "resources.clusters.test") + require.Contains(t, changes["resources.clusters.test"], "autoscale") + assert.Equal(t, map[string]any{"max_workers": int64(4)}, changes["resources.clusters.test"]["autoscale"].Value) +} + +func TestExtractChangesKeepsPolicyValuesInsidePositionalArray(t *testing.T) { + b := policyTestBundle(t, "resources.clusters.test", map[string]any{ + "policy_id": "policy", "init_scripts": []any{map[string]any{"workspace": map[string]any{"destination": "/Workspace/init.sh"}}}, + }) + volumeScript := &compute.VolumesStorageInfo{Destination: "/Volumes/c/s/v/init.sh"} + entry := &deployplan.PlanEntry{ + RemoteState: &dresources.ClusterRemote{PolicyId: "policy", InitScripts: []compute.InitScriptInfo{{Volumes: volumeScript}}}, + Changes: deployplan.Changes{"init_scripts[0].volumes": {Action: deployplan.Update, Remote: volumeScript}}, + } + // Array positions must be preserved regardless of whether the value matches a policy. + client := mocks.NewMockWorkspaceClient(t) + b.SetWorkpaceClient(client.WorkspaceClient) + changes, err := configsync.ExtractChanges(t.Context(), b, &deployplan.Plan{Plan: map[string]*deployplan.PlanEntry{"resources.clusters.test": entry}}) + require.NoError(t, err) + require.Contains(t, changes, "resources.clusters.test") + require.Contains(t, changes["resources.clusters.test"], "init_scripts[0].volumes") + assert.Equal(t, map[string]any{"destination": "/Volumes/c/s/v/init.sh"}, changes["resources.clusters.test"]["init_scripts[0].volumes"].Value) +} + +func TestExtractChangesDropsPolicyOnlyObjectAtChangeRoot(t *testing.T) { + b := policyTestBundle(t, "resources.clusters.test", map[string]any{"policy_id": "policy"}) + client := mocks.NewMockWorkspaceClient(t) + b.SetWorkpaceClient(client.WorkspaceClient) + client.GetMockClusterPoliciesAPI().EXPECT().GetByPolicyId(mock.Anything, "policy"). + Return(&compute.Policy{Definition: `{"workload_type.clients.notebooks":{"type":"fixed","value":false}}`}, nil).Once() + remote := typedPolicyResource(t, "standalone", map[string]any{ + "policy_id": "policy", "workload_type": map[string]any{"clients": map[string]any{"notebooks": false}}, + }) + clients, err := structaccess.GetByString(remote, "workload_type.clients") + require.NoError(t, err) + entry := &deployplan.PlanEntry{ + RemoteState: remote, + Changes: deployplan.Changes{"workload_type.clients": {Action: deployplan.Update, Remote: clients}}, + } + changes, err := configsync.ExtractChanges(t.Context(), b, &deployplan.Plan{Plan: map[string]*deployplan.PlanEntry{"resources.clusters.test": entry}}) + require.NoError(t, err) + assert.Empty(t, changes) +} diff --git a/bundle/configsync/diff.go b/bundle/configsync/diff.go index 81e4c1eaf8a..01b654dd145 100644 --- a/bundle/configsync/diff.go +++ b/bundle/configsync/diff.go @@ -8,6 +8,7 @@ import ( "github.com/databricks/cli/bundle" "github.com/databricks/cli/bundle/config" "github.com/databricks/cli/bundle/deployplan" + "github.com/databricks/cli/bundle/direct/dresources" "github.com/databricks/cli/libs/dyn" "github.com/databricks/cli/libs/dyn/convert" "github.com/databricks/cli/libs/log" @@ -156,7 +157,10 @@ func ExtractChanges(ctx context.Context, b *bundle.Bundle, plan *deployplan.Plan } resourceChanges := make(ResourceChanges) - policyFilter := newClusterPolicyFilter(resourceKey, entry, policyDefinitions) + policyFilter, err := policyFilterForResource(b, resourceKey, entry, policyDefinitions) + if err != nil { + return nil, err + } if entry.Changes != nil { for path, changeDesc := range entry.Changes { @@ -205,3 +209,33 @@ func ExtractChanges(ctx context.Context, b *bundle.Bundle, plan *deployplan.Plan return changes, nil } + +func policyFilterForResource(b *bundle.Bundle, resourceKey string, entry *deployplan.PlanEntry, definitions clusterPolicyDefinitions) (*clusterPolicyFilter, error) { + filter := &clusterPolicyFilter{remote: entry.RemoteState, definitions: definitions} + switch entry.RemoteState.(type) { + case *dresources.JobRemote: + for _, pattern := range []string{"tasks[*].new_cluster", "tasks[*].for_each_task.task.new_cluster", "job_clusters[*].new_cluster"} { + root, err := structpath.ParsePattern(pattern) + if err != nil { + return nil, err + } + filter.clusterRoots = append(filter.clusterRoots, root) + } + filter.sliceKeys = (&dresources.ResourceJob{}).KeyedSlices() + case *dresources.ClusterRemote: + filter.clusterRoots = []*structpath.PatternNode{nil} // nil matches the resource root + default: + // Only jobs and clusters expand policy values into their remote state. + return nil, nil + } + // Skip entries have no NewState; authored values must come from bundle config. + config, err := b.Config.GetResourceConfig(resourceKey) + if err != nil { + return nil, fmt.Errorf("reading config for %s: %w", resourceKey, err) + } + if config == nil { + return nil, nil + } + filter.config = config + return filter, nil +} diff --git a/libs/structs/structdiff/keyed_path.go b/libs/structs/structdiff/keyed_path.go new file mode 100644 index 00000000000..b6355893870 --- /dev/null +++ b/libs/structs/structdiff/keyed_path.go @@ -0,0 +1,48 @@ +package structdiff + +import ( + "fmt" + "reflect" + + "github.com/databricks/cli/libs/structs/structaccess" + "github.com/databricks/cli/libs/structs/structpath" +) + +// KeyedPath replaces slice indices with the selectors used by GetStructDiff. +// Indices without a matching key function and existing selectors are preserved. +func KeyedPath(root any, path *structpath.PathNode, sliceKeys map[string]KeyFunc) (*structpath.PathNode, error) { + ctx := &diffContext{sliceKeys: sliceKeys} + var result *structpath.PathNode + for _, node := range path.AsSlice() { + index, isIndex := node.Index() + switch { + case isIndex: + keyFunc := ctx.findKeyFunc(result) + if keyFunc == nil { + result = structpath.NewIndex(result, index) + continue + } + element, err := structaccess.Get(root, node) + if err != nil { + return nil, err + } + caller, err := newKeyFuncCaller(keyFunc) + if err != nil { + return nil, err + } + if element == nil || !reflect.TypeOf(element).AssignableTo(caller.argType) { + return nil, fmt.Errorf("KeyFunc expects %v at %s, got %T", caller.argType, node, element) + } + key, value := caller.call(element) + result = structpath.NewKeyValue(result, key, value) + default: + if key, value, ok := node.KeyValue(); ok { + result = structpath.NewKeyValue(result, key, value) + } else { + key, _ := node.StringKey() + result = structpath.NewStringKey(result, key) + } + } + } + return result, nil +} diff --git a/libs/structs/structdiff/keyed_path_test.go b/libs/structs/structdiff/keyed_path_test.go new file mode 100644 index 00000000000..3832dd46095 --- /dev/null +++ b/libs/structs/structdiff/keyed_path_test.go @@ -0,0 +1,60 @@ +package structdiff_test + +import ( + "testing" + + "github.com/databricks/cli/libs/structs/structdiff" + "github.com/databricks/cli/libs/structs/structpath" + "github.com/databricks/databricks-sdk-go/service/jobs" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestKeyedPathUsesDiffSelectors(t *testing.T) { + root := jobs.JobSettings{Tasks: []jobs.Task{ + {TaskKey: "second", DependsOn: []jobs.TaskDependency{{TaskKey: "first"}}}, + {TaskKey: "first"}, + }} + keys := map[string]structdiff.KeyFunc{ + "tasks": func(task jobs.Task) (string, string) { return "task_key", task.TaskKey }, + "tasks[*].depends_on": func(dependency jobs.TaskDependency) (string, string) { return "task_key", dependency.TaskKey }, + } + for _, tt := range []struct{ path, want string }{ + {"tasks[0].depends_on[0].task_key", "tasks[task_key='second'].depends_on[task_key='first'].task_key"}, + {"tasks[1].task_key", "tasks[task_key='first'].task_key"}, + {"tasks[task_key='second'].depends_on[0].task_key", "tasks[task_key='second'].depends_on[task_key='first'].task_key"}, + {"tasks[task_key='first'].task_key", "tasks[task_key='first'].task_key"}, + {"tags['a.b']", "tags['a.b']"}, + {"", ""}, + } { + t.Run(tt.path, func(t *testing.T) { + result, err := structdiff.KeyedPath(&root, structpath.MustParsePath(tt.path), keys) + require.NoError(t, err) + assert.Equal(t, tt.want, result.String()) + }) + } + path := structpath.MustParsePath("tasks[0].task_key") + result, err := structdiff.KeyedPath(&root, path, nil) + require.NoError(t, err) + assert.Equal(t, path.String(), result.String()) +} + +func TestKeyedPathRejectsInvalidKeyFunctionsAndIndices(t *testing.T) { + root := jobs.JobSettings{Tasks: []jobs.Task{{TaskKey: "main"}}} + for _, tt := range []struct { + name, path string + keys map[string]structdiff.KeyFunc + }{ + {"invalid function", "tasks[0]", map[string]structdiff.KeyFunc{"tasks": 1}}, + {"wrong argument type", "tasks[0]", map[string]structdiff.KeyFunc{"tasks": func(int) (string, string) { return "", "" }}}, + {"missing element", "tasks[1]", map[string]structdiff.KeyFunc{ + "tasks": func(task jobs.Task) (string, string) { return "task_key", task.TaskKey }, + "tasks[*].depends_on": func(dependency jobs.TaskDependency) (string, string) { return "task_key", dependency.TaskKey }, + }}, + } { + t.Run(tt.name, func(t *testing.T) { + _, err := structdiff.KeyedPath(&root, structpath.MustParsePath(tt.path), tt.keys) + assert.Error(t, err) + }) + } +} From a14d1cdf101c232531f24d1ee60efcf81e9a849d Mon Sep 17 00:00:00 2001 From: ilyakuz-db <181845775+ilyakuz-db@users.noreply.github.com> Date: Fri, 23 Oct 2026 16:42:21 +0000 Subject: [PATCH 4/4] Replace deprecated traversal in cluster policy sync --- bundle/configsync/cluster_policy.go | 173 ++++++++++++++++------------ 1 file changed, 102 insertions(+), 71 deletions(-) diff --git a/bundle/configsync/cluster_policy.go b/bundle/configsync/cluster_policy.go index 91582062a6b..aaf9e125c8e 100644 --- a/bundle/configsync/cluster_policy.go +++ b/bundle/configsync/cluster_policy.go @@ -2,15 +2,16 @@ package configsync import ( "context" - "errors" "fmt" + "iter" + "maps" "reflect" + "slices" "strconv" "strings" "github.com/databricks/cli/bundle" "github.com/databricks/cli/libs/dyn" - "github.com/databricks/cli/libs/dyn/convert" "github.com/databricks/cli/libs/dyn/jsonloader" "github.com/databricks/cli/libs/structs/structaccess" "github.com/databricks/cli/libs/structs/structdiff" @@ -47,91 +48,132 @@ type clusterPolicyFilter struct { // filter removes omitted policy values from a normalized plan change, including // whole-object additions. A nil result means no value needs to be synced. -func (f *clusterPolicyFilter) filter(ctx context.Context, b *bundle.Bundle, path *structpath.PathNode, changeValue any) (any, error) { - normalized, err := convert.FromTyped(changeValue, dyn.NilValue) - if err != nil { +func (f *clusterPolicyFilter) filter(ctx context.Context, b *bundle.Bundle, path *structpath.PathNode, value any) (any, error) { + filtered, keep, err := f.prune(ctx, b, path, value, true) + if err != nil || !keep { return nil, err } - filtered, err := dyn.Walk(normalized, func(relative dyn.Path, value dyn.Value) (dyn.Value, error) { - fieldPath, err := structdiff.KeyedPath(f.remote, appendDynPath(path, relative), f.sliceKeys) - if err != nil { - return dyn.InvalidValue, err + return filtered, nil +} + +// prune reports keep=false when the whole value comes from the cluster policy. +func (f *clusterPolicyFilter) prune(ctx context.Context, b *bundle.Bundle, path *structpath.PathNode, value any, isChangeRoot bool) (any, bool, error) { + fieldPath, err := structdiff.KeyedPath(f.remote, path, f.sliceKeys) + if err != nil { + return nil, false, err + } + for _, root := range f.clusterRoots { + if !fieldPath.HasPatternPrefix(root) || fieldPath.Len() == root.Len() { + continue } - for _, root := range f.clusterRoots { - if !fieldPath.HasPatternPrefix(root) || fieldPath.Len() == root.Len() { - continue + clusterPath := fieldPath.Prefix(root.Len()) + // A change may start inside an array, so protecting only sequence + // nodes would still allow an element to be emptied or removed. + for _, node := range fieldPath.SkipPrefix(root.Len()).AsSlice() { + if _, indexed := node.Index(); indexed { + return value, true, nil } - clusterPath := fieldPath.Prefix(root.Len()) - // A change may start inside an array, so protecting only sequence - // nodes would still allow an element to be emptied or removed. - for _, node := range fieldPath.SkipPrefix(root.Len()).AsSlice() { - if _, indexed := node.Index(); indexed { - return value, dyn.ErrSkip - } + } + // Keep required fields, but still filter their optional children. + if isChangeRoot || !isRequiredField(f.remote, fieldPath) { + matches, err := f.isPolicyOnlySubtree(ctx, b, clusterPath, fieldPath, value) + if err != nil { + return nil, false, err } - // Keep required fields, but still filter their optional children. - if len(relative) > 0 && isRequiredField(f.remote, fieldPath) { - return value, nil + if matches { + return nil, false, nil } - matches, err := f.matchesOmittedPolicySubtree(ctx, b, clusterPath, fieldPath, value) + } + // Removing elements would change indexed policy constraints. + if _, sequence := value.([]any); sequence { + return value, true, nil + } + break + } + + switch value := value.(type) { + case map[string]any: + result := make(map[string]any, len(value)) + for _, key := range slices.Sorted(maps.Keys(value)) { + child := value[key] + filtered, keep, err := f.prune(ctx, b, structpath.NewStringKey(path, key), child, false) if err != nil { - return dyn.InvalidValue, err + return nil, false, err } - if matches { - return dyn.InvalidValue, dyn.ErrDrop + if keep { + result[key] = filtered } - // Removing elements would change indexed policy constraints. - if value.Kind() == dyn.KindSequence { - return value, dyn.ErrSkip + } + return result, true, nil + case []any: + // Arrays inside clusters return above; elements here cannot be dropped. + result := make([]any, len(value)) + for i, child := range value { + filtered, _, err := f.prune(ctx, b, structpath.NewIndex(path, i), child, false) + if err != nil { + return nil, false, err } - break + result[i] = filtered } - return value, nil - }) - if errors.Is(err, dyn.ErrDrop) { - return nil, nil - } - if err != nil { - return nil, err + return result, true, nil + default: + return value, true, nil } - return filtered.AsAny(), nil } -// matchesOmittedPolicySubtree requires every leaf to be absent from authored config +// isPolicyOnlySubtree requires every leaf to be absent from authored config // and equal to a fixed value or an opted-in default in the remote cluster's policy. -func (f *clusterPolicyFilter) matchesOmittedPolicySubtree(ctx context.Context, b *bundle.Bundle, clusterPath, path *structpath.PathNode, value dyn.Value) (bool, error) { - matches := true - leaves := 0 - _, err := dyn.Walk(value, func(relative dyn.Path, leaf dyn.Value) (dyn.Value, error) { - if !matches { - return leaf, dyn.ErrSkip - } - if leaf.Kind() == dyn.KindMap || leaf.Kind() == dyn.KindSequence { - return leaf, nil - } - leaves++ - fieldPath := appendDynPath(path, relative) +func (f *clusterPolicyFilter) isPolicyOnlySubtree(ctx context.Context, b *bundle.Bundle, clusterPath, path *structpath.PathNode, value any) (bool, error) { + hasLeaves := false + for fieldPath, leaf := range normalizedLeaves(path, value) { + hasLeaves = true if authored, _ := structaccess.Get(f.config, fieldPath); authored != nil { - matches = false - return leaf, nil + return false, nil } remotePolicyID, _ := structaccess.Get(f.remote, structpath.NewStringKey(clusterPath, "policy_id")) policyID, _ := remotePolicyID.(string) if policyID == "" { - matches = false - return leaf, nil + return false, nil } definition, err := f.definitions.get(ctx, b, policyID) if err != nil { - return dyn.InvalidValue, err + return false, err } // Remote cluster specs omit this input-only flag; use the authored setting. applyDefaults, _ := structaccess.Get(f.config, structpath.NewStringKey(clusterPath, "apply_policy_default_values")) expected := policyValueForAPIAttribute(definition, policyAttribute(fieldPath.SkipPrefix(clusterPath.Len())), applyDefaults == true) - matches = expected.IsValid() && expected.AsAny() != nil && reflect.DeepEqual(expected.AsAny(), leaf.AsAny()) - return leaf, nil - }) - return matches && leaves > 0, err + if !expected.IsValid() || expected.AsAny() == nil || !reflect.DeepEqual(expected.AsAny(), leaf) { + return false, nil + } + } + return hasLeaves, nil +} + +// normalizedLeaves preserves the normalized change's scalars, including nulls. +// Empty containers have no leaves and cannot alone establish a policy match. +func normalizedLeaves(path *structpath.PathNode, value any) iter.Seq2[*structpath.PathNode, any] { + return func(yield func(*structpath.PathNode, any) bool) { + switch value := value.(type) { + case map[string]any: + for _, key := range slices.Sorted(maps.Keys(value)) { + for leafPath, leaf := range normalizedLeaves(structpath.NewStringKey(path, key), value[key]) { + if !yield(leafPath, leaf) { + return + } + } + } + case []any: + for i, child := range value { + for leafPath, leaf := range normalizedLeaves(structpath.NewIndex(path, i), child) { + if !yield(leafPath, leaf) { + return + } + } + } + default: + yield(path, value) + } + } } func isRequiredField(remote any, path *structpath.PathNode) bool { @@ -151,17 +193,6 @@ func isRequiredField(remote any, path *structpath.PathNode) bool { return ok && !structtag.JSONTag(field.Tag.Get("json")).OmitEmpty() } -func appendDynPath(base *structpath.PathNode, suffix dyn.Path) *structpath.PathNode { - for _, component := range suffix { - if key := component.Key(); key != "" { - base = structpath.NewStringKey(base, key) - } else { - base = structpath.NewIndex(base, component.Index()) - } - } - return base -} - // policyAttribute preserves dotted map keys, e.g. spark_conf["a.b"] becomes spark_conf.a.b. func policyAttribute(path *structpath.PathNode) string { var parts []string