[Python] Initialize and normalize resources before/after PythonMutator (#2590)

## Changes
Initialize and normalize resources before/after `PythonMutator`.

Create `ResourceProcessor` that applies `initialize` and `normalize` for
resources. It should be used by mutators after they have added or
modified resources.

- when a new resource its normalized and then initialized
- each resource is initialized only once
- each time a resource is updated, it's normalized

Implementation takes a slice of bundle configuration containing
added/modified resources and runs a set of mutators on it. After that,
only resources are merged back. If initialize or normalize resource
mutators change bundle configuration outside if "resources", such
changes are discarded.

Stacked on top of #2637

## Why
Using `ResourceProcessor,` we fix problems with `PythonMutator` where
either input or output wasn't normalized, default configuration wasn't
applied, or it was impossible to override the default configuration
(e.g., remove the `[dev` prefix).

## Tests
Acceptance and unit tests

Author: kanterov

acceptance/bundle/debug/out.stderr.txt

@@ -34,30 +34,17 @@
 10:07:59 Debug: Apply pid=12345 mutator=PrependWorkspacePrefix
 10:07:59 Debug: Apply pid=12345 mutator=RewriteWorkspacePrefix
 10:07:59 Debug: Apply pid=12345 mutator=SetVariables
-10:07:59 Debug: Apply pid=12345 mutator=PythonMutator(load)
-10:07:59 Debug: Apply pid=12345 mutator=PythonMutator(init)
-10:07:59 Debug: Apply pid=12345 mutator=PythonMutator(load_resources)
-10:07:59 Debug: Apply pid=12345 mutator=PythonMutator(apply_mutators)
 10:07:59 Debug: Apply pid=12345 mutator=ResolveVariableReferences
 10:07:59 Debug: Apply pid=12345 mutator=ResolveResourceReferences
 10:07:59 Debug: Apply pid=12345 mutator=ResolveVariableReferences
-10:07:59 Debug: Apply pid=12345 mutator=ResolveVariableReferences(resources)
+10:07:59 Debug: Apply pid=12345 mutator=ApplyTargetMode
+10:07:59 Debug: Apply pid=12345 mutator=ProcessStaticResources
 10:07:59 Debug: Apply pid=12345 mutator=NormalizePaths
-10:07:59 Debug: Apply pid=12345 mutator=ExpandPipelineGlobPaths
-10:07:59 Debug: Apply pid=12345 mutator=MergeJobClusters
-10:07:59 Debug: Apply pid=12345 mutator=MergeJobParameters
-10:07:59 Debug: Apply pid=12345 mutator=MergeJobTasks
-10:07:59 Debug: Apply pid=12345 mutator=MergePipelineClusters
-10:07:59 Debug: Apply pid=12345 mutator=MergeApps
-10:07:59 Debug: Apply pid=12345 mutator=CaptureSchemaDependency
+10:07:59 Debug: Apply pid=12345 mutator=PythonMutator(load)
+10:07:59 Debug: Apply pid=12345 mutator=PythonMutator(init)
+10:07:59 Debug: Apply pid=12345 mutator=PythonMutator(load_resources)
+10:07:59 Debug: Apply pid=12345 mutator=PythonMutator(apply_mutators)
 10:07:59 Debug: Apply pid=12345 mutator=CheckPermissions
-10:07:59 Debug: Apply pid=12345 mutator=SetRunAs
-10:07:59 Debug: Apply pid=12345 mutator=OverrideCompute
-10:07:59 Debug: Apply pid=12345 mutator=ConfigureDashboardDefaults
-10:07:59 Debug: Apply pid=12345 mutator=ConfigureVolumeDefaults
-10:07:59 Debug: Apply pid=12345 mutator=ApplyTargetMode
-10:07:59 Debug: Apply pid=12345 mutator=ApplyPresets
-10:07:59 Debug: Apply pid=12345 mutator=DefaultQueueing
 10:07:59 Debug: Apply pid=12345 mutator=ConfigureWSFS
 10:07:59 Debug: Apply pid=12345 mutator=TranslatePaths
 10:07:59 Debug: Apply pid=12345 mutator=PythonWrapperWarning
@@ -66,8 +53,6 @@
 10:07:59 Debug: Apply pid=12345 mutator=apps.Validate
 10:07:59 Debug: Apply pid=12345 mutator=ValidateTargetMode
 10:07:59 Debug: Apply pid=12345 mutator=ValidateSharedRootPermissions
-10:07:59 Debug: Apply pid=12345 mutator=ApplyBundlePermissions
-10:07:59 Debug: Apply pid=12345 mutator=FilterCurrentUserFromPermissions
 10:07:59 Debug: Apply pid=12345 mutator=metadata.AnnotateJobs
 10:07:59 Debug: Apply pid=12345 mutator=metadata.AnnotatePipelines
 10:07:59 Debug: Apply pid=12345 mutator=terraform.Initialize

acceptance/bundle/override/job_tasks/output.txt

@@ -1,5 +1,6 @@
 {
   "name": "job",
+  "permissions": [],
   "queue": {
     "enabled": true
   },
@@ -35,6 +36,7 @@ Error: file test1.py not found
 Exit code: 1
 {
   "name": "job",
+  "permissions": [],
   "queue": {
     "enabled": true
   },

acceptance/bundle/python/resolve-variable/databricks.yml

@@ -13,12 +13,8 @@ variables:
     default: "abc"
   int_variable:
     default: 42
-  # FIXME doesn't work, resolve variables before running PythonMutator
-  #
-  # > cannot interpolate non-string value: ${var.int_variable}
-  #
-  # nested_variable:
-  #   default: ${var.string_variable} ${var.int_variable}
+  nested_variable:
+    default: ${var.string_variable} ${var.string_variable}
   bool_variable_true:
     default: true
   bool_variable_false:
@@ -31,12 +27,12 @@ variables:
     default:
       task_key: "abc"
       notebook_task:
-        notebook_path: "cde"
+        notebook_path: "/Workspace/cde"
   complex_list_variable:
     default:
       - task_key: "abc"
         notebook_task:
-          notebook_path: "cde"
+          notebook_path: "/Workspace/cde"
       - task_key: "def"
         notebook_task:
-          notebook_path: "ghi"
+          notebook_path: "/Workspace/ghi"

acceptance/bundle/python/resolve-variable/output.txt

@@ -22,7 +22,21 @@
         "queue": {
           "enabled": true
         },
-        "tags": {}
+        "tags": {},
+        "tasks": [
+          {
+            "notebook_task": {
+              "notebook_path": "/Workspace/cde"
+            },
+            "task_key": "abc"
+          },
+          {
+            "notebook_task": {
+              "notebook_path": "/Workspace/ghi"
+            },
+            "task_key": "def"
+          }
+        ]
       }
     }
   }

acceptance/bundle/python/resolve-variable/resources.py

@@ -22,9 +22,8 @@ def load_resources(bundle: Bundle) -> Resources:
     int_variable = bundle.resolve_variable(Variables.int_variable)
     assert int_variable == 42
 
-    # FIXME resolve variables before running PythonMutator
-    # nested_variable = bundle.resolve_variable(Variables.nested_variable)
-    # assert nested_variable == "abc 42"
+    nested_variable = bundle.resolve_variable(Variables.nested_variable)
+    assert nested_variable == "abc abc"
 
     bool_variable_true = bundle.resolve_variable(Variables.bool_variable_true)
     assert bool_variable_true
@@ -39,26 +38,24 @@ def load_resources(bundle: Bundle) -> Resources:
     assert dict_variable == {"a": 1, "b": 2}
 
     complex_variable = bundle.resolve_variable(Variables.complex_variable)
-    assert complex_variable == Task(task_key="abc", notebook_task=NotebookTask(notebook_path="cde"))
+    assert complex_variable == Task(
+        task_key="abc",
+        notebook_task=NotebookTask(notebook_path="/Workspace/cde"),
+    )
 
     complex_list_variable = bundle.resolve_variable(Variables.complex_list_variable)
     print(complex_list_variable)
     assert complex_list_variable == [
-        Task(task_key="abc", notebook_task=NotebookTask(notebook_path="cde")),
-        Task(task_key="def", notebook_task=NotebookTask(notebook_path="ghi")),
+        Task(task_key="abc", notebook_task=NotebookTask(notebook_path="/Workspace/cde")),
+        Task(task_key="def", notebook_task=NotebookTask(notebook_path="/Workspace/ghi")),
     ]
 
     resources = Resources()
     resources.add_job(
         "my_job",
         Job(
             name=Variables.string_variable,
-            # FIXME we can't output complex variables in place where we override locations
-            # with bundle root path. See loadOutput in python_mutator.go
-            #
-            # > failed to update locations: expected a sequence at "resources.jobs.my_job.tasks", found string
-            #
-            # tasks=Variables.complex_list_variable,
+            tasks=Variables.complex_list_variable,
         ),
     )
 

bundle/config/mutator/python/python_locations_test.go

@@ -153,16 +153,6 @@ func TestLoadOutput(t *testing.T) {
 	name, err := dyn.Get(value, "resources.jobs.my_job.name")
 	require.NoError(t, err)
 	require.Equal(t, []dyn.Location{location}, name.Locations())
-
-	// until we implement path normalization, we have to keep locations of values
-	// that change semantic depending on their location
-	//
-	// note: it's important to have absolute path including 'bundleRoot'
-	// because mutator pipeline already has expanded locations into absolute path
-	notebookPath, err := dyn.Get(value, "resources.jobs.my_job.tasks[0].notebook_task.notebook_path")
-	require.NoError(t, err)
-	require.Len(t, notebookPath.Locations(), 1)
-	require.Equal(t, filepath.Join(bundleRoot, generatedFileName), notebookPath.Locations()[0].File)
 }
 
 func TestParsePythonLocations_absolutePath(t *testing.T) {

bundle/config/mutator/python/python_mutator.go

@@ -13,9 +13,9 @@ import (
 	"reflect"
 	"strings"
 
-	"github.com/databricks/cli/libs/log"
+	"github.com/databricks/cli/bundle/config/mutator/resourcemutator"
 
-	"github.com/databricks/cli/bundle/config/mutator/paths"
+	"github.com/databricks/cli/libs/log"
 
 	"github.com/databricks/databricks-sdk-go/logger"
 	"github.com/fatih/color"
@@ -200,6 +200,7 @@ func (m *pythonMutator) Apply(ctx context.Context, b *bundle.Bundle) diag.Diagno
 
 	// mutateDiags is used because Mutate returns 'error' instead of 'diag.Diagnostics'
 	var mutateDiags diag.Diagnostics
+	var result applyPythonOutputResult
 	mutateDiagsHasError := errors.New("unexpected error")
 
 	err = b.Config.Mutate(func(leftRoot dyn.Value) (dyn.Value, error) {
@@ -224,9 +225,10 @@ func (m *pythonMutator) Apply(ctx context.Context, b *bundle.Bundle) diag.Diagno
 			return dyn.InvalidValue, mutateDiagsHasError
 		}
 
-		newRoot, result, err := applyPythonOutput(leftRoot, rightRoot)
+		newRoot, result0, err := applyPythonOutput(leftRoot, rightRoot)
+		result = result0
 		if err != nil {
-			return dyn.InvalidValue, err
+			return dyn.InvalidValue, fmt.Errorf("internal error when merging output of Python mutator: %w", err)
 		}
 
 		for _, resourceKey := range result.AddedResources.ToArray() {
@@ -261,10 +263,21 @@ func (m *pythonMutator) Apply(ctx context.Context, b *bundle.Bundle) diag.Diagno
 			panic("mutateDiags has no error, but error is expected")
 		}
 
+		return mutateDiags
+	} else {
+		mutateDiags = mutateDiags.Extend(diag.FromErr(err))
+	}
+
+	if mutateDiags.HasError() {
 		return mutateDiags
 	}
 
-	return mutateDiags.Extend(diag.FromErr(err))
+	mutateDiags = mutateDiags.Extend(resourcemutator.NormalizeAndInitializeResources(ctx, b, result.AddedResources))
+	if mutateDiags.HasError() {
+		return mutateDiags
+	}
+
+	return mutateDiags.Extend(resourcemutator.NormalizeResources(ctx, b, result.UpdatedResources))
 }
 
 func createCacheDir(ctx context.Context) (string, error) {
@@ -453,21 +466,6 @@ func loadOutput(rootPath string, outputFile io.Reader, locations *pythonLocation
 		return dyn.InvalidValue, diag.FromErr(fmt.Errorf("failed to parse output file: %w", err))
 	}
 
-	// paths are resolved relative to locations of their values, if we change location
-	// we have to update each path, until we simplify that, we don't update locations
-	// for such values, so we don't change how paths are resolved
-	//
-	// we can remove this once we:
-	// - add variable interpolation before and after PythonMutator
-	// - implement path normalization (aka path normal form)
-	_, err = paths.VisitJobPaths(generated, func(p dyn.Path, mode paths.TranslateMode, v dyn.Value) (dyn.Value, error) {
-		putPythonLocation(locations, p, v.Location())
-		return v, nil
-	})
-	if err != nil {
-		return dyn.InvalidValue, diag.FromErr(fmt.Errorf("failed to update locations: %w", err))
-	}
-
 	// generated has dyn.Location as if it comes from generated YAML file
 	// earlier we loaded locations.json with source locations in Python code
 	generatedWithLocations, err := mergePythonLocations(generated, locations)

…ndle/config/mutator/apply_target_mode.go …tor/resourcemutator/apply_target_mode.gobundle/config/mutator/apply_target_mode.go renamed to bundle/config/mutator/resourcemutator/apply_target_mode.go bundle/config/mutator/apply_target_mode.go renamed to bundle/config/mutator/resourcemutator/apply_target_mode.go

@@ -1,4 +1,4 @@
-package mutator
+package resourcemutator
 
 import (
 	"context"