direct: Support wildcards in resources config; add managed job attributes (#4487)

## Changes
- Support wildcards in resources.yml
- Add job-related attributes to resources.yml. The list of patterns
comes from bundle/configsync/defaults.go, but I could not remove those
patterns there, we need to handle the case with new task added.

## Why
We need to handle these to ignore known drift.

## Tests
Existing tests.

---------

Co-authored-by: Claude Opus 4.6 <noreply@anthropic.com>

Author: denik

acceptance/bundle/resources/jobs/remote_matches_config/out.plan.direct.json

@@ -72,7 +72,7 @@
         },
         "tasks[task_key='main_task'].new_cluster.aws_attributes": {
           "action": "skip",
-          "reason": "server_side_default",
+          "reason": "managed",
           "remote": {
             "availability": "SPOT_WITH_FALLBACK",
             "zone_id": "us-east-1c"
@@ -85,7 +85,7 @@
         },
         "tasks[task_key='main_task'].new_cluster.enable_elastic_disk": {
           "action": "skip",
-          "reason": "server_side_default",
+          "reason": "managed",
           "remote": false
         },
         "tasks[task_key='main_task'].notebook_task.source": {

bundle/direct/bundle_plan.go

@@ -437,7 +437,7 @@ func addPerFieldActions(ctx context.Context, adapter *dresources.Adapter, change
 
 func findMatchingRule(path *structpath.PathNode, rules []dresources.FieldRule) (string, bool) {
 	for _, r := range rules {
-		if path.HasPrefix(r.Field) {
+		if path.HasPatternPrefix(r.Field) {
 			return r.Reason, true
 		}
 	}

bundle/direct/dresources/adapter.go

@@ -363,7 +363,7 @@ func (a *Adapter) GeneratedResourceConfig() *ResourceLifecycleConfig {
 
 func (a *Adapter) IsFieldInRecreateOnChanges(path *structpath.PathNode) bool {
 	for _, p := range a.resourceConfig.RecreateOnChanges {
-		if path.HasPrefix(p.Field) {
+		if path.HasPatternPrefix(p.Field) {
 			return true
 		}
 	}

bundle/direct/dresources/all_test.go

@@ -842,13 +842,13 @@ func TestGeneratedResourceConfig(t *testing.T) {
 
 func validateResourceConfig(t *testing.T, stateType reflect.Type, cfg *ResourceLifecycleConfig) {
 	for _, p := range cfg.RecreateOnChanges {
-		assert.NoError(t, structaccess.ValidatePath(stateType, p.Field), "RecreateOnChanges: %s", p.Field)
+		assert.NoError(t, structaccess.ValidatePattern(stateType, p.Field), "RecreateOnChanges: %s", p.Field)
 	}
 	for _, p := range cfg.UpdateIDOnChanges {
-		assert.NoError(t, structaccess.ValidatePath(stateType, p.Field), "UpdateIDOnChanges: %s", p.Field)
+		assert.NoError(t, structaccess.ValidatePattern(stateType, p.Field), "UpdateIDOnChanges: %s", p.Field)
 	}
 	for _, p := range cfg.IgnoreRemoteChanges {
-		assert.NoError(t, structaccess.ValidatePath(stateType, p.Field), "IgnoreRemoteChanges: %s", p.Field)
+		assert.NoError(t, structaccess.ValidatePattern(stateType, p.Field), "IgnoreRemoteChanges: %s", p.Field)
 	}
 }
 

bundle/direct/dresources/config.go

@@ -10,8 +10,8 @@ import (
 
 // FieldRule represents a field path with its reason for inclusion.
 type FieldRule struct {
-	Field  *structpath.PathNode `yaml:"field"`
-	Reason string               `yaml:"reason"`
+	Field  *structpath.PatternNode `yaml:"field"`
+	Reason string                  `yaml:"reason"`
 }
 
 // ResourceLifecycleConfig defines lifecycle behavior for a resource type.

bundle/direct/dresources/resources.yml

@@ -13,6 +13,29 @@
 
 resources:
 
+  jobs:
+    ignore_remote_changes:
+      - field: tasks[*].new_cluster.aws_attributes
+        # These are not "output_only", because they can be set by the user as well.
+        reason: managed
+      - field: tasks[*].new_cluster.azure_attributes
+        reason: managed
+      - field: tasks[*].new_cluster.gcp_attributes
+        reason: managed
+
+      - field: job_clusters[*].new_cluster.aws_attributes
+        reason: managed
+      - field: job_clusters[*].new_cluster.azure_attributes
+        reason: managed
+      - field: job_clusters[*].new_cluster.gcp_attributes
+        reason: managed
+
+      # can also be added to backend_defaults with value=false once we have that
+      - field: tasks[*].new_cluster.enable_elastic_disk
+        reason: managed
+      - field: job_clusters[*].new_cluster.enable_elastic_disk
+        reason: managed
+
   pipelines:
     recreate_on_changes:
       - field: storage

libs/structs/structpath/path.go

@@ -272,6 +272,10 @@ func (p *PatternNode) Len() int {
 	return (*PathNode)(p).Len()
 }
 
+func (p *PatternNode) Prefix(n int) *PatternNode {
+	return (*PatternNode)((*PathNode)(p).Prefix(n))
+}
+
 func (p *PatternNode) String() string {
 	return (*PathNode)(p).String()
 }
@@ -834,6 +838,56 @@ func nodesEqual(a, b *PathNode) bool {
 	return a.key == b.key && a.index == b.index && a.value == b.value
 }
 
+// HasPatternPrefix returns true if the path starts with the given pattern prefix.
+// Wildcards (.* and [*]) in the pattern match any single path component.
+// Both .* and [*] are treated interchangeably.
+func (p *PathNode) HasPatternPrefix(prefix *PatternNode) bool {
+	if prefix.IsRoot() {
+		return true
+	}
+
+	if p.IsRoot() {
+		return false
+	}
+
+	pLen := p.Len()
+	prefixLen := prefix.Len()
+
+	if prefixLen > pLen {
+		return false
+	}
+
+	pAtPrefixLen := p.Prefix(prefixLen)
+
+	currentP := pAtPrefixLen
+	currentPrefix := prefix
+
+	for currentPrefix != nil {
+		if !nodeMatchesPattern(currentP, currentPrefix) {
+			return false
+		}
+		currentP = currentP.prev
+		currentPrefix = (*PatternNode)((*PathNode)(currentPrefix).prev)
+	}
+
+	return true
+}
+
+// nodeMatchesPattern checks if a concrete path node matches a pattern node.
+// Wildcards (.* and [*]) match any node.
+func nodeMatchesPattern(concrete *PathNode, pattern *PatternNode) bool {
+	if concrete == nil && pattern == nil {
+		return true
+	}
+	if concrete == nil || pattern == nil {
+		return false
+	}
+	if pattern.DotStar() || pattern.BracketStar() {
+		return true
+	}
+	return nodesEqual(concrete, (*PathNode)(pattern))
+}
+
 // MarshalYAML implements yaml.Marshaler for PathNode.
 func (p *PathNode) MarshalYAML() (any, error) {
 	return p.String(), nil
@@ -856,3 +910,26 @@ func (p *PathNode) UnmarshalYAML(unmarshal func(any) error) error {
 	*p = *(*PathNode)(parsed)
 	return nil
 }
+
+// MarshalYAML implements yaml.Marshaler for PatternNode.
+func (p *PatternNode) MarshalYAML() (any, error) {
+	return p.String(), nil
+}
+
+// UnmarshalYAML implements yaml.Unmarshaler for PatternNode.
+// Wildcards (.* and [*]) are allowed.
+func (p *PatternNode) UnmarshalYAML(unmarshal func(any) error) error {
+	var s string
+	if err := unmarshal(&s); err != nil {
+		return err
+	}
+	parsed, err := parse(s, true)
+	if err != nil {
+		return err
+	}
+	if parsed == nil {
+		return nil
+	}
+	*p = *parsed
+	return nil
+}

libs/structs/structpath/path_test.go

@@ -883,6 +883,136 @@ func TestHasPrefix(t *testing.T) {
 	}
 }
 
+func TestHasPatternPrefix(t *testing.T) {
+	tests := []struct {
+		name     string
+		path     string
+		pattern  string
+		expected bool
+	}{
+		{
+			name:     "empty pattern",
+			path:     "a.b.c",
+			pattern:  "",
+			expected: true,
+		},
+		{
+			name:     "empty path",
+			path:     "",
+			pattern:  "a",
+			expected: false,
+		},
+		{
+			name:     "exact match no wildcards",
+			path:     "config",
+			pattern:  "config",
+			expected: true,
+		},
+		{
+			name:     "simple prefix no wildcards",
+			path:     "a.b",
+			pattern:  "a",
+			expected: true,
+		},
+
+		// .* wildcard
+		{
+			name:     "dot star matches field",
+			path:     "tasks.my_task.notebook_task",
+			pattern:  "tasks.*.notebook_task",
+			expected: true,
+		},
+		{
+			name:     "dot star matches any field",
+			path:     "tasks.other.notebook_task",
+			pattern:  "tasks.*.notebook_task",
+			expected: true,
+		},
+		{
+			name:     "dot star matches index",
+			path:     "items[0].name",
+			pattern:  "items.*.name",
+			expected: true,
+		},
+		{
+			name:     "dot star as prefix",
+			path:     "items[0].name.value",
+			pattern:  "items.*",
+			expected: true,
+		},
+
+		// [*] wildcard
+		{
+			name:     "bracket star matches index",
+			path:     "items[0].name",
+			pattern:  "items[*].name",
+			expected: true,
+		},
+		{
+			name:     "bracket star matches field",
+			path:     "tasks.my_task.notebook_task",
+			pattern:  "tasks[*].notebook_task",
+			expected: true,
+		},
+		{
+			name:     "bracket star matches key-value",
+			path:     "tasks[task_key='my_task'].notebook_task",
+			pattern:  "tasks[*].notebook_task",
+			expected: true,
+		},
+
+		// Multiple wildcards
+		{
+			name:     "multiple wildcards",
+			path:     "tasks[0].params[1].value",
+			pattern:  "tasks[*].params[*].value",
+			expected: true,
+		},
+
+		// Non-matches
+		{
+			name:     "wildcard but wrong suffix",
+			path:     "tasks[0].notebook_task",
+			pattern:  "tasks[*].spark_task",
+			expected: false,
+		},
+		{
+			name:     "pattern longer than path",
+			path:     "tasks[0]",
+			pattern:  "tasks[*].notebook_task",
+			expected: false,
+		},
+		{
+			name:     "no wildcard mismatch",
+			path:     "a.b.c",
+			pattern:  "a.x.c",
+			expected: false,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			path, err := ParsePath(tt.path)
+			require.NoError(t, err)
+
+			pattern, err := ParsePattern(tt.pattern)
+			require.NoError(t, err)
+
+			result := path.HasPatternPrefix(pattern)
+			assert.Equal(t, tt.expected, result, "path.HasPatternPrefix(pattern) where path=%q, pattern=%q", tt.path, tt.pattern)
+
+			// If the full pattern matches, every prefix of it must also match.
+			if tt.expected {
+				for n := range pattern.Len() + 1 {
+					prefix := pattern.Prefix(n)
+					assert.True(t, path.HasPatternPrefix(prefix),
+						"path=%q pattern=%q prefix_len=%d prefix=%s", tt.path, tt.pattern, n, prefix)
+				}
+			}
+		})
+	}
+}
+
 func TestPathNodeYAMLMarshal(t *testing.T) {
 	tests := []struct {
 		name     string