Move cluster policy matcher and loader to libs/clusterpolicy

bundle/configsync/cluster_policy.go

@@ -7,8 +7,6 @@ import (
 	"maps"
 	"reflect"
 	"slices"
-	"strconv"
-	"strings"
 
 	"github.com/databricks/cli/bundle"
 	"github.com/databricks/cli/bundle/deployplan"
@@ -29,9 +27,12 @@ const clusterPolicyGate = "policy_id"
 // policy cannot be read is left out entirely and reported as skipped: its changes may
 // pair with each other (e.g. autoscale -> num_workers), so a partial result is unsafe.
 func FilterClusterPolicyValues(ctx context.Context, b *bundle.Bundle, plan *deployplan.Plan, changes Changes) (Changes, []SkippedChange, error) {
-	definitions := make(clusterPolicyDefinitions)
+	var loader *clusterpolicy.Loader
 	lookupPolicy := func(id string) (clusterpolicy.Definition, error) {
-		return definitions.get(ctx, b, id)
+		if loader == nil {
+			loader = clusterpolicy.NewLoader(b.WorkspaceClient(ctx).ClusterPolicies)
+		}
+		return loader.Get(ctx, id)
 	}
 	filtered := make(Changes, len(changes))
 	var skipped []SkippedChange
@@ -45,8 +46,8 @@ func FilterClusterPolicyValues(ctx context.Context, b *bundle.Bundle, plan *depl
 			continue
 		}
 		resourceChanges, err := filter.filterChanges(changes[resourceKey])
-		if lookupErr, ok := errors.AsType[*clusterPolicyLookupError](err); ok {
-			skipped = append(skipped, lookupErr.skippedChange(resourceKey))
+		if lookupErr, ok := errors.AsType[*clusterpolicy.LookupError](err); ok {
+			skipped = append(skipped, policyLookupSkip(resourceKey, lookupErr))
 			continue
 		}
 		if err != nil {
@@ -59,64 +60,17 @@ func FilterClusterPolicyValues(ctx context.Context, b *bundle.Bundle, plan *depl
 	return filtered, skipped, nil
 }
 
-type clusterPolicyDefinition struct {
-	rules clusterpolicy.Definition
-	err   error
-}
-
-type clusterPolicyDefinitions map[string]clusterPolicyDefinition
-
-type clusterPolicyLookupError struct {
-	policyID string
-	err      error
-}
-
-func (e *clusterPolicyLookupError) Error() string {
-	return fmt.Sprintf("reading cluster policy %s: %v", e.policyID, e.err)
-}
-
-func (e *clusterPolicyLookupError) Unwrap() error { return e.err }
-
-func (e *clusterPolicyLookupError) skippedChange(resourceKey string) SkippedChange {
-	skip := SkippedChange{ResourceKey: resourceKey, Reason: SkipReasonClusterPolicyReadFailed, PolicyID: e.policyID, Error: e.err.Error()}
+func policyLookupSkip(resourceKey string, err *clusterpolicy.LookupError) SkippedChange {
+	skip := SkippedChange{ResourceKey: resourceKey, Reason: SkipReasonClusterPolicyReadFailed, PolicyID: err.PolicyID, Error: err.Err.Error()}
 	switch {
-	case errors.Is(e.err, apierr.ErrNotFound):
+	case errors.Is(err.Err, apierr.ErrNotFound):
 		skip.Reason, skip.Error = SkipReasonClusterPolicyNotFound, ""
-	case errors.Is(e.err, apierr.ErrPermissionDenied):
+	case errors.Is(err.Err, apierr.ErrPermissionDenied):
 		skip.Reason = SkipReasonClusterPolicyPermissionDenied
 	}
 	return skip
 }
 
-// get caches definitions and failures per sync. The API returns the effective
-// definition, including policy-family overrides.
-func (definitions clusterPolicyDefinitions) get(ctx context.Context, b *bundle.Bundle, id string) (clusterpolicy.Definition, error) {
-	if definition, ok := definitions[id]; ok {
-		return definition.rules, definition.err
-	}
-	var rules clusterpolicy.Definition
-	policy, err := b.WorkspaceClient(ctx).ClusterPolicies.GetByPolicyId(ctx, id)
-	if err == nil {
-		rules, err = clusterpolicy.ParseDefinition(policy.Definition)
-	}
-	if err != nil {
-		err = &clusterPolicyLookupError{policyID: id, err: err}
-	}
-	definitions[id] = clusterPolicyDefinition{rules: rules, err: err}
-	return rules, err
-}
-
-// elementSupplies reports whether the policy element could have set value. The backend
-// rejects values that differ from a fixed element, so a fixed field is frozen whatever its
-// remote value; other elements only set their opted-in default.
-func elementSupplies(element clusterpolicy.Element, value any, applyDefaults bool) bool {
-	if element.Type == "fixed" {
-		return true
-	}
-	expected := element.EffectiveValue(applyDefaults)
-	return expected != nil && clusterpolicy.ValueEquals(expected, value)
-}
-
 type clusterPolicyFilter struct {
 	lookupPolicy func(string) (clusterpolicy.Definition, error)
 	config       any
@@ -263,7 +217,7 @@ func (f *clusterPolicyFilter) isPolicyOnlySubtree(clusterPath, path *structpath.
 		return false, err
 	}
 	for _, leaf := range leaves {
-		if !policySupplies(definition, leaf.path.SkipPrefix(clusterPath.Len()), leaf.value, applyDefaults == true) {
+		if !definition.IsPolicyValue(leaf.path.SkipPrefix(clusterPath.Len()), leaf.value, applyDefaults == true) {
 			return false, nil
 		}
 	}
@@ -333,44 +287,6 @@ func isRequiredField(remote any, path *structpath.PathNode) bool {
 	return ok && !structtag.JSONTag(field.Tag.Get("json")).OmitEmpty()
 }
 
-// policyAttributes renders a path relative to the cluster in the policy language,
-// once with indices and once with "*" in their place. Dotted map keys stay one key,
-// e.g. spark_conf["a.b"] becomes spark_conf.a.b. Map keys are not wildcarded: a rule
-// such as spark_conf.* constrains values but is not treated as supplying them.
-func policyAttributes(path *structpath.PathNode) (indexed, wildcard string) {
-	var indexedParts, wildcardParts []string
-	for _, node := range path.AsSlice() {
-		if key, ok := node.StringKey(); ok {
-			indexedParts = append(indexedParts, key)
-			wildcardParts = append(wildcardParts, key)
-		} else if index, ok := node.Index(); ok {
-			indexedParts = append(indexedParts, strconv.Itoa(index))
-			wildcardParts = append(wildcardParts, "*")
-		}
-	}
-	return strings.Join(indexedParts, "."), strings.Join(wildcardParts, ".")
-}
-
-// policySupplies reports whether the policy could have set a leaf, translating the
-// policy language's log-delivery aliases. Indexed rules take precedence over wildcards.
-// https://docs.databricks.com/aws/en/admin/clusters/policy-definition
-func policySupplies(definition clusterpolicy.Definition, path *structpath.PathNode, value any, applyDefaults bool) bool {
-	indexed, wildcard := policyAttributes(path)
-	if indexed == "cluster_log_conf.s3.region" {
-		return elementSupplies(definition["cluster_log_conf.region"], value, applyDefaults)
-	}
-	if storage, ok := strings.CutPrefix(indexed, "cluster_log_conf."); ok && strings.HasSuffix(storage, ".destination") {
-		storage = strings.ToUpper(strings.TrimSuffix(storage, ".destination"))
-		logType := definition["cluster_log_conf.type"].EffectiveValue(applyDefaults)
-		return clusterpolicy.ValueEquals(logType, storage) && elementSupplies(definition["cluster_log_conf.path"], value, applyDefaults)
-	}
-	element, ok := definition[indexed]
-	if !ok {
-		element = definition[wildcard]
-	}
-	return elementSupplies(element, value, applyDefaults)
-}
-
 func policyFilterForResource(b *bundle.Bundle, resourceKey string, remote any, lookupPolicy func(string) (clusterpolicy.Definition, error)) (*clusterPolicyFilter, error) {
 	filter := &clusterPolicyFilter{remote: remote, lookupPolicy: lookupPolicy}
 	switch remote.(type) {

libs/clusterpolicy/definition.go

@@ -5,6 +5,8 @@ import (
 	"reflect"
 	"strconv"
 	"strings"
+
+	"github.com/databricks/cli/libs/structs/structpath"
 )
 
 // Element is the part of a cluster policy element that materializes a value. The backend
@@ -72,3 +74,54 @@ func ValueEquals(policyValue, value any) bool {
 	}
 	return reflect.DeepEqual(policyValue, value)
 }
+
+// IsPolicyValue reports whether the policy, not the user, determines the value of the
+// cluster attribute at path, a path relative to the cluster spec. A fixed element
+// determines the field whatever its current value, because the backend rejects any other
+// value; other elements only when apply_policy_default_values is set and value equals
+// their defaultValue. This is a filtering predicate, not validation: true for a fixed
+// element does not mean value equals the element's value. The policy language's
+// log-delivery aliases are translated, and indexed rules take precedence over wildcards.
+// https://docs.databricks.com/aws/en/admin/clusters/policy-definition
+func (d Definition) IsPolicyValue(path *structpath.PathNode, value any, applyDefaults bool) bool {
+	indexed, wildcard := attributes(path)
+	if indexed == "cluster_log_conf.s3.region" {
+		return d["cluster_log_conf.region"].isPolicyValue(value, applyDefaults)
+	}
+	if storage, ok := strings.CutPrefix(indexed, "cluster_log_conf."); ok && strings.HasSuffix(storage, ".destination") {
+		storage = strings.ToUpper(strings.TrimSuffix(storage, ".destination"))
+		logType := d["cluster_log_conf.type"].EffectiveValue(applyDefaults)
+		return ValueEquals(logType, storage) && d["cluster_log_conf.path"].isPolicyValue(value, applyDefaults)
+	}
+	element, ok := d[indexed]
+	if !ok {
+		element = d[wildcard]
+	}
+	return element.isPolicyValue(value, applyDefaults)
+}
+
+func (e Element) isPolicyValue(value any, applyDefaults bool) bool {
+	if e.Type == "fixed" {
+		return true
+	}
+	expected := e.EffectiveValue(applyDefaults)
+	return expected != nil && ValueEquals(expected, value)
+}
+
+// attributes renders a path relative to the cluster in the policy language,
+// once with indices and once with "*" in their place. Dotted map keys stay one key,
+// e.g. spark_conf["a.b"] becomes spark_conf.a.b. Map keys are not wildcarded: a rule
+// such as spark_conf.* constrains values but is not treated as supplying them.
+func attributes(path *structpath.PathNode) (indexed, wildcard string) {
+	var indexedParts, wildcardParts []string
+	for _, node := range path.AsSlice() {
+		if key, ok := node.StringKey(); ok {
+			indexedParts = append(indexedParts, key)
+			wildcardParts = append(wildcardParts, key)
+		} else if index, ok := node.Index(); ok {
+			indexedParts = append(indexedParts, strconv.Itoa(index))
+			wildcardParts = append(wildcardParts, "*")
+		}
+	}
+	return strings.Join(indexedParts, "."), strings.Join(wildcardParts, ".")
+}

libs/clusterpolicy/definition_test.go

@@ -4,6 +4,7 @@ import (
 	"encoding/json"
 	"testing"
 
+	"github.com/databricks/cli/libs/structs/structpath"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 )
@@ -46,3 +47,50 @@ func TestValueEquals(t *testing.T) {
 		})
 	}
 }
+
+func TestDefinitionIsPolicyValue(t *testing.T) {
+	definition, err := ParseDefinition(`{
+		"spark_version": {"type": "fixed", "value": "15.4.x-scala2.12"},
+		"autotermination_minutes": {"type": "range", "defaultValue": 60},
+		"custom_tags.CostCenter": {"type": "fixed", "value": "dev-1234"},
+		"spark_conf.spark.sql.shuffle.partitions": {"type": "unlimited", "defaultValue": 7},
+		"spark_conf.*": {"type": "fixed", "value": "any"},
+		"init_scripts.0.volumes.destination": {"type": "allowlist", "values": ["/Volumes/a"]},
+		"init_scripts.*.volumes.destination": {"type": "fixed", "value": "/Volumes/shared"},
+		"ssh_public_keys.*": {"type": "fixed", "value": "ssh-rsa key"},
+		"cluster_log_conf.type": {"type": "fixed", "value": "DBFS"},
+		"cluster_log_conf.path": {"type": "fixed", "value": "dbfs:/logs"},
+		"cluster_log_conf.region": {"type": "fixed", "value": "us-west-2"}
+	}`)
+	require.NoError(t, err)
+
+	for _, tt := range []struct {
+		name          string
+		path          string
+		value         any
+		applyDefaults bool
+		want          bool
+	}{
+		{"fixed matches its value", "spark_version", "15.4.x-scala2.12", false, true},
+		{"fixed matches a stale value", "spark_version", "14.3.x-scala2.12", false, true},
+		{"default requires the flag", "autotermination_minutes", int64(60), false, false},
+		{"default matches with the flag", "autotermination_minutes", int64(60), true, true},
+		{"default requires equality", "autotermination_minutes", int64(30), true, false},
+		{"unconstrained attribute", "node_type_id", "i3.xlarge", true, false},
+		{"dotted map key is one attribute", "spark_conf['spark.sql.shuffle.partitions']", "7", true, true},
+		{"map wildcard does not supply entries", "spark_conf['spark.other']", "any", true, false},
+		{"map key in policy language", "custom_tags.CostCenter", "dev-1234", false, true},
+		{"indexed rule masks the wildcard", "init_scripts[0].volumes.destination", "/Volumes/shared", false, false},
+		{"wildcard applies to other indices", "init_scripts[1].volumes.destination", "/Volumes/other", false, true},
+		{"wildcard array element", "ssh_public_keys[3]", "other", false, true},
+		{"log destination matches type and path", "cluster_log_conf.dbfs.destination", "dbfs:/logs", false, true},
+		{"log destination of another storage", "cluster_log_conf.s3.destination", "dbfs:/logs", false, false},
+		{"log region alias", "cluster_log_conf.s3.region", "eu-west-1", false, true},
+	} {
+		t.Run(tt.name, func(t *testing.T) {
+			path, err := structpath.ParsePath(tt.path)
+			require.NoError(t, err)
+			assert.Equal(t, tt.want, definition.IsPolicyValue(path, tt.value, tt.applyDefaults))
+		})
+	}
+}

libs/clusterpolicy/loader.go

@@ -0,0 +1,54 @@
+package clusterpolicy
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/databricks/databricks-sdk-go/service/compute"
+)
+
+// LookupError wraps a failure to read or parse the policy with the given ID.
+type LookupError struct {
+	PolicyID string
+	Err      error
+}
+
+func (e *LookupError) Error() string {
+	return fmt.Sprintf("reading cluster policy %s: %v", e.PolicyID, e.Err)
+}
+
+func (e *LookupError) Unwrap() error { return e.Err }
+
+// Loader reads policy definitions by ID and caches both definitions and failures for
+// its lifetime, so one loader serves a single operation against one workspace.
+type Loader struct {
+	api         compute.ClusterPoliciesInterface
+	definitions map[string]loadedDefinition
+}
+
+type loadedDefinition struct {
+	rules Definition
+	err   error
+}
+
+func NewLoader(api compute.ClusterPoliciesInterface) *Loader {
+	return &Loader{api: api, definitions: make(map[string]loadedDefinition)}
+}
+
+// Get returns the effective definition of a policy, including policy-family overrides.
+// Failures are returned as *LookupError.
+func (l *Loader) Get(ctx context.Context, policyID string) (Definition, error) {
+	if definition, ok := l.definitions[policyID]; ok {
+		return definition.rules, definition.err
+	}
+	var rules Definition
+	policy, err := l.api.GetByPolicyId(ctx, policyID)
+	if err == nil {
+		rules, err = ParseDefinition(policy.Definition)
+	}
+	if err != nil {
+		err = &LookupError{PolicyID: policyID, Err: err}
+	}
+	l.definitions[policyID] = loadedDefinition{rules: rules, err: err}
+	return rules, err
+}

libs/clusterpolicy/loader_test.go

@@ -0,0 +1,56 @@
+package clusterpolicy
+
+import (
+	"errors"
+	"testing"
+
+	"github.com/databricks/databricks-sdk-go/apierr"
+	"github.com/databricks/databricks-sdk-go/experimental/mocks"
+	"github.com/databricks/databricks-sdk-go/service/compute"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/mock"
+	"github.com/stretchr/testify/require"
+)
+
+func TestLoaderCachesDefinitions(t *testing.T) {
+	api := mocks.NewMockWorkspaceClient(t).GetMockClusterPoliciesAPI()
+	api.EXPECT().GetByPolicyId(mock.Anything, "policy").Return(&compute.Policy{
+		Definition: `{"spark_version": {"type": "fixed", "value": "15.4.x-scala2.12"}}`,
+	}, nil).Once()
+
+	loader := NewLoader(api)
+	for range 2 {
+		definition, err := loader.Get(t.Context(), "policy")
+		require.NoError(t, err)
+		assert.Equal(t, Definition{"spark_version": {Type: "fixed", Value: "15.4.x-scala2.12"}}, definition)
+	}
+}
+
+func TestLoaderCachesFailures(t *testing.T) {
+	for _, tt := range []struct {
+		name   string
+		policy *compute.Policy
+		err    error
+	}{
+		{"api error", nil, apierr.ErrResourceDoesNotExist},
+		{"invalid definition", &compute.Policy{Definition: `{"spark_version": `}, nil},
+	} {
+		t.Run(tt.name, func(t *testing.T) {
+			api := mocks.NewMockWorkspaceClient(t).GetMockClusterPoliciesAPI()
+			api.EXPECT().GetByPolicyId(mock.Anything, "policy").Return(tt.policy, tt.err).Once()
+
+			loader := NewLoader(api)
+			for range 2 {
+				definition, err := loader.Get(t.Context(), "policy")
+				assert.Nil(t, definition)
+				lookupErr, ok := errors.AsType[*LookupError](err)
+				require.True(t, ok, "expected *LookupError, got %T", err)
+				assert.Equal(t, "policy", lookupErr.PolicyID)
+				if tt.err != nil {
+					assert.ErrorIs(t, err, apierr.ErrNotFound)
+					assert.EqualError(t, err, "reading cluster policy policy: "+tt.err.Error())
+				}
+			}
+		})
+	}
+}