Do not report cluster fields supplied by a cluster policy as drift

.nextchanges/bundles/cluster-policy-no-drift.md

@@ -0,0 +1 @@
+* On the direct engine, `bundle plan` no longer reports a permanent update on a cluster that uses a cluster policy: when the cluster spec sets `policy_id`, a field present in the remote but absent from the bundle config is not treated as drift. ([#6531](https://github.com/databricks/cli/pull/6531))

acceptance/bin/prune_plan.py

@@ -0,0 +1,41 @@
+#!/usr/bin/env python3
+"""Prune the cloud-dependent parts of `bundle plan -o json` so the plan can be recorded as a
+golden that is identical across clouds.
+
+Reads plan JSON on stdin, writes the pruned JSON on stdout. Per plan node it removes:
+
+  - remote_state: the raw backend read (per-run driver/executor IPs, instance ids,
+    spark_context_id, timestamps, default_tags, ...), none of which is reproducible.
+  - changes entries whose reason is in --ignore-reasons (default: managed, backend_default) --
+    values the backend chooses, e.g. {aws,azure,gcp}_attributes or node types, which differ by
+    cloud.
+  - changes entries whose field path contains any substring in --ignore-keys -- for fields a
+    backend sets to different values (so the same field lands under different reasons on
+    different clouds and cannot be matched by reason alone, e.g. enable_elastic_disk).
+
+What survives is cloud-independent: the action, and the changes driven by config or policy.
+"""
+
+import argparse
+import json
+import sys
+
+parser = argparse.ArgumentParser()
+parser.add_argument("--ignore-reasons", default="managed,backend_default")
+parser.add_argument("--ignore-keys", default="")
+args = parser.parse_args()
+
+ignore_reasons = {r for r in args.ignore_reasons.split(",") if r}
+ignore_keys = [k for k in args.ignore_keys.split(",") if k]
+
+plan = json.load(sys.stdin)
+
+for node in plan.get("plan", {}).values():
+    node.pop("remote_state", None)
+    changes = node.get("changes") or {}
+    for path, change in list(changes.items()):
+        if change.get("reason") in ignore_reasons or any(k in path for k in ignore_keys):
+            del changes[path]
+
+json.dump(plan, sys.stdout, indent=2, sort_keys=True)
+print()

acceptance/bundle/resources/cluster_policies/policy_no_drift_variants/databricks.yml.tmpl

@@ -0,0 +1,74 @@
+bundle:
+  name: test-bundle-$UNIQUE_NAME
+
+workspace:
+  root_path: ~/.bundle/$UNIQUE_NAME
+
+resources:
+  cluster_policies:
+    my_policy:
+      name: test-policy-$UNIQUE_NAME
+      definition:
+        custom_tags.CostCenter:
+          type: fixed
+          value: from-policy
+
+  # A cluster spec can appear in several places, each with its own ignore_remote_additions
+  # rule. This fixture puts a policy-attached cluster in each place -- a standalone cluster, a
+  # task's new_cluster, a for_each task's new_cluster, and a pipeline cluster -- so the rule is
+  # exercised at every location in one plan. (job_clusters is covered by the sibling
+  # jobs/cluster_policy tests.) The pipeline cluster is the deliberate no-rule case; see below.
+  clusters:
+    standalone:
+      cluster_name: test-cluster-$UNIQUE_NAME
+      spark_version: $DEFAULT_SPARK_VERSION
+      node_type_id: $NODE_TYPE_ID
+      # Use the shared instance pool: a cold cluster boot made this the slowest test in the
+      # cloud suite (6-8 minutes per env), and the policy tag under test is unaffected.
+      instance_pool_id: $TEST_INSTANCE_POOL_ID
+      num_workers: 1
+      policy_id: ${resources.cluster_policies.my_policy.id}
+
+  jobs:
+    task_cluster:
+      name: test-task-cluster-$UNIQUE_NAME
+      tasks:
+        - task_key: t
+          new_cluster:
+            policy_id: ${resources.cluster_policies.my_policy.id}
+            spark_version: $DEFAULT_SPARK_VERSION
+            node_type_id: $NODE_TYPE_ID
+            num_workers: 1
+          spark_python_task:
+            python_file: ./hello_world.py
+
+    for_each_cluster:
+      name: test-for-each-cluster-$UNIQUE_NAME
+      tasks:
+        - task_key: outer
+          for_each_task:
+            inputs: "[1,2]"
+            task:
+              task_key: inner
+              new_cluster:
+                policy_id: ${resources.cluster_policies.my_policy.id}
+                spark_version: $DEFAULT_SPARK_VERSION
+                node_type_id: $NODE_TYPE_ID
+                num_workers: 1
+              spark_python_task:
+                python_file: ./hello_world.py
+
+  # A pipeline cluster carries policy_id too, but the Pipelines API does not expand the
+  # policy into the stored spec: it reads back exactly as authored, so there is no addition
+  # to suppress and no ignore_remote_additions rule for pipelines. Kept here to record that.
+  pipelines:
+    pipe:
+      name: test-pipeline-$UNIQUE_NAME
+      clusters:
+        - label: default
+          policy_id: ${resources.cluster_policies.my_policy.id}
+          node_type_id: $NODE_TYPE_ID
+          num_workers: 1
+      libraries:
+        - file:
+            path: ./hello_world.py

acceptance/bundle/resources/cluster_policies/policy_no_drift_variants/hello_world.py

@@ -0,0 +1 @@
+print("hello")

acceptance/bundle/resources/cluster_policies/policy_no_drift_variants/output.txt

@@ -0,0 +1,182 @@
+
+>>> [CLI] bundle deploy
+Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/[UNIQUE_NAME]/files...
+Created cluster_policies.my_policy
+Created clusters.standalone
+Created jobs.for_each_cluster
+Created jobs.task_cluster
+Created pipelines.pipe
+Files: 6 uploaded, 0 deleted
+Resources: 5 created, 0 changed, 0 deleted, 0 unchanged
+
+=== Every cluster-spec location converges: nothing to change on a second plan
+
+>>> [CLI] bundle plan
+Plan: 0 to add, 0 to change, 0 to delete, 5 unchanged
+
+=== The plan, minus remote_state and backend-chosen changes (managed, backend_default)
+
+>>> [CLI] bundle plan -o json
+{
+  "cli_version": "[CLI_VERSION]",
+  "lineage": "[UUID]",
+  "plan": {
+    "resources.cluster_policies.my_policy": {
+      "action": "skip"
+    },
+    "resources.clusters.standalone": {
+      "action": "skip",
+      "changes": {
+        "custom_tags": {
+          "action": "skip",
+          "reason": "remote_addition",
+          "remote": {
+            "CostCenter": "from-policy"
+          }
+        },
+        "enable_elastic_disk": {
+          "action": "skip",
+          "reason": "empty",
+          "remote": false
+        },
+        "lifecycle": {
+          "action": "skip",
+          "reason": "remote_addition",
+          "remote": {
+            "started": true
+          }
+        }
+      },
+      "depends_on": [
+        {
+          "label": "${resources.cluster_policies.my_policy.id}",
+          "node": "resources.cluster_policies.my_policy"
+        }
+      ]
+    },
+    "resources.jobs.for_each_cluster": {
+      "action": "skip",
+      "changes": {
+        "email_notifications": {
+          "action": "skip",
+          "reason": "empty",
+          "remote": {}
+        },
+        "tasks[task_key='outer'].email_notifications": {
+          "action": "skip",
+          "reason": "empty",
+          "remote": {}
+        },
+        "tasks[task_key='outer'].for_each_task.task.new_cluster.custom_tags": {
+          "action": "skip",
+          "reason": "remote_addition",
+          "remote": {
+            "CostCenter": "from-policy"
+          }
+        },
+        "tasks[task_key='outer'].timeout_seconds": {
+          "action": "skip",
+          "reason": "empty",
+          "remote": 0
+        },
+        "timeout_seconds": {
+          "action": "skip",
+          "reason": "empty",
+          "remote": 0
+        },
+        "webhook_notifications": {
+          "action": "skip",
+          "reason": "empty",
+          "remote": {}
+        }
+      },
+      "depends_on": [
+        {
+          "label": "${resources.cluster_policies.my_policy.id}",
+          "node": "resources.cluster_policies.my_policy"
+        }
+      ]
+    },
+    "resources.jobs.task_cluster": {
+      "action": "skip",
+      "changes": {
+        "email_notifications": {
+          "action": "skip",
+          "reason": "empty",
+          "remote": {}
+        },
+        "tasks[task_key='t'].email_notifications": {
+          "action": "skip",
+          "reason": "empty",
+          "remote": {}
+        },
+        "tasks[task_key='t'].new_cluster.custom_tags": {
+          "action": "skip",
+          "reason": "remote_addition",
+          "remote": {
+            "CostCenter": "from-policy"
+          }
+        },
+        "tasks[task_key='t'].new_cluster.data_security_mode": {
+          "action": "skip",
+          "reason": "remote_addition",
+          "remote": "SINGLE_USER"
+        },
+        "tasks[task_key='t'].new_cluster.enable_elastic_disk": {
+          "action": "skip",
+          "reason": "empty",
+          "remote": false
+        },
+        "tasks[task_key='t'].timeout_seconds": {
+          "action": "skip",
+          "reason": "empty",
+          "remote": 0
+        },
+        "timeout_seconds": {
+          "action": "skip",
+          "reason": "empty",
+          "remote": 0
+        },
+        "webhook_notifications": {
+          "action": "skip",
+          "reason": "empty",
+          "remote": {}
+        }
+      },
+      "depends_on": [
+        {
+          "label": "${resources.cluster_policies.my_policy.id}",
+          "node": "resources.cluster_policies.my_policy"
+        }
+      ]
+    },
+    "resources.pipelines.pipe": {
+      "action": "skip",
+      "changes": {},
+      "depends_on": [
+        {
+          "label": "${resources.cluster_policies.my_policy.id}",
+          "node": "resources.cluster_policies.my_policy"
+        }
+      ]
+    }
+  },
+  "plan_version": [PLAN_VERSION],
+  "serial": 1
+}
+
+>>> [CLI] bundle destroy --auto-approve
+The following resources will be deleted:
+  delete resources.cluster_policies.my_policy
+  delete resources.clusters.standalone
+  delete resources.jobs.for_each_cluster
+  delete resources.jobs.task_cluster
+  delete resources.pipelines.pipe
+
+This action will result in the deletion of the following Lakeflow Spark Declarative Pipelines along with the
+Streaming Tables (STs) and Materialized Views (MVs) managed by them. Set 'cascade_on_destroy: false' on a pipeline to retain datasets on pipeline deletion:
+  delete resources.pipelines.pipe
+
+All files and directories at the following location will be deleted: /Workspace/Users/[USERNAME]/.bundle/[UNIQUE_NAME]
+
+Destroy: 5 deleted

acceptance/bundle/resources/cluster_policies/policy_no_drift_variants/script

@@ -0,0 +1,14 @@
+envsubst < databricks.yml.tmpl > databricks.yml
+
+cleanup() {
+    trace $CLI bundle destroy --auto-approve
+}
+trap cleanup EXIT
+
+trace $CLI bundle deploy
+
+title "Every cluster-spec location converges: nothing to change on a second plan\n"
+trace $CLI bundle plan
+
+title "The plan, minus remote_state and backend-chosen changes (managed, backend_default)\n"
+trace $CLI bundle plan -o json | prune_plan.py

acceptance/bundle/resources/cluster_policies/policy_no_drift_variants/test.toml

@@ -0,0 +1,18 @@
+# Local-only. This test creates a standalone cluster, which only the local testserver can do
+# reliably: on a real workspace it boots at deploy and repeatedly hit capacity limits, and the
+# full remote_state is a per-run cluster GET no testserver reproduces. The suppression behavior
+# is verified against a real workspace by the jobs/cluster_policy tests; here we cover every
+# cluster-spec location (standalone cluster, task new_cluster, for_each new_cluster, pipeline).
+Cloud = false
+
+# Deployment-history recording (DMS) adds version-dependent plan fields and .databricks
+# artifacts, and it is orthogonal to drift suppression (covered by the jobs/cluster_policy
+# tests). Pin to the off value so the test runs once on the "direct" CI cell; DMS = [] would
+# instead run on BOTH the direct and directdms cells (same trap as DATABRICKS_BUNDLE_ENGINE=[]).
+EnvMatrix.DMS = [""]
+
+RecordRequests = false
+
+Ignore = [
+    "databricks.yml",
+]

acceptance/bundle/resources/jobs/cluster_policy/fixed_value_conflict/databricks.yml.tmpl

@@ -0,0 +1,33 @@
+bundle:
+  name: test-bundle-$UNIQUE_NAME
+
+workspace:
+  root_path: ~/.bundle/$UNIQUE_NAME
+
+resources:
+  cluster_policies:
+    my_policy:
+      name: test-policy-$UNIQUE_NAME
+      definition:
+        custom_tags.CostCenter:
+          type: fixed
+          value: from-fixed
+
+  jobs:
+    # The config sets the fixed attribute to a value the policy forbids.
+    j:
+      name: test-job-$UNIQUE_NAME
+      job_clusters:
+        - job_cluster_key: small
+          new_cluster:
+            policy_id: ${resources.cluster_policies.my_policy.id}
+            spark_version: $DEFAULT_SPARK_VERSION
+            node_type_id: $NODE_TYPE_ID
+            num_workers: 1
+            custom_tags:
+              CostCenter: not-what-the-policy-says
+      tasks:
+        - task_key: t
+          job_cluster_key: small
+          spark_python_task:
+            python_file: ./hello_world.py

acceptance/bundle/resources/jobs/cluster_policy/fixed_value_conflict/hello_world.py

@@ -0,0 +1 @@
+print("hello")