Wait for grants and permissions before creating pipelines

.nextchanges/bundles/grants-before-dependents.md

@@ -0,0 +1 @@
+* Pipelines now wait for the grants and permissions of the resources they refer to or list in `lifecycle.depends_on`, so a pipeline whose run_as principal is granted access to its schema in the same bundle no longer fails with `User does not have CREATE TABLE on Schema`.

acceptance/bundle/resources/grants/schemas/pipeline_run_as/databricks.yml.tmpl

@@ -0,0 +1,34 @@
+bundle:
+  name: schema-grants-pipeline-$UNIQUE_NAME
+
+resources:
+  catalogs:
+    pipeline_catalog:
+      name: catalog_grants_pipeline_$UNIQUE_NAME
+      grants:
+        - principal: $RUN_AS_SP
+          privileges:
+            - USE_CATALOG
+
+  schemas:
+    src:
+      catalog_name: ${resources.catalogs.pipeline_catalog.name}
+      name: src
+      grants:
+        - principal: $RUN_AS_SP
+          privileges:
+            - USE_SCHEMA
+            - CREATE_TABLE
+            - CREATE_MATERIALIZED_VIEW
+
+  pipelines:
+    load:
+      name: schema-grants-pipeline-$UNIQUE_NAME
+      catalog: ${resources.schemas.src.catalog_name}
+      schema: ${resources.schemas.src.name}
+      serverless: true
+      libraries:
+        - file:
+            path: ./transformation.py
+      run_as:
+        service_principal_name: $RUN_AS_SP

acceptance/bundle/resources/grants/schemas/pipeline_run_as/out.test.toml

@@ -0,0 +1,2 @@
+Cloud = false
+EnvMatrix.DMS = ["", "true"]

acceptance/bundle/resources/grants/schemas/pipeline_run_as/output.txt

@@ -0,0 +1,93 @@
+
+=== The grants of a referenced schema become an implicit lifecycle.depends_on
+
+>>> [CLI] bundle validate -o json
+{
+  "catalogs": {
+    "pipeline_catalog": null
+  },
+  "pipelines": {
+    "load": [
+      {
+        "resource": "schemas.src.grants"
+      }
+    ]
+  },
+  "schemas": {
+    "src": null
+  }
+}
+
+=== The plan orders dependents after the grants
+
+>>> [CLI] bundle plan -o json
+{
+  "resources.catalogs.pipeline_catalog": null,
+  "resources.catalogs.pipeline_catalog.grants": [
+    {
+      "node": "resources.catalogs.pipeline_catalog",
+      "label": "${resources.catalogs.pipeline_catalog.id}"
+    }
+  ],
+  "resources.pipelines.load": [
+    {
+      "node": "resources.schemas.src",
+      "label": "${resources.schemas.src.catalog_name}"
+    },
+    {
+      "node": "resources.schemas.src",
+      "label": "${resources.schemas.src.name}"
+    },
+    {
+      "node": "resources.schemas.src.grants",
+      "label": "depends_on"
+    }
+  ],
+  "resources.schemas.src": [
+    {
+      "node": "resources.catalogs.pipeline_catalog",
+      "label": "${resources.catalogs.pipeline_catalog.name}"
+    }
+  ],
+  "resources.schemas.src.grants": [
+    {
+      "node": "resources.schemas.src",
+      "label": "${resources.schemas.src.id}"
+    }
+  ]
+}
+
+>>> [CLI] bundle deploy
+Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/schema-grants-pipeline-[UNIQUE_NAME]/default/files...
+Created catalogs.pipeline_catalog
+Created catalogs.pipeline_catalog.grants
+Created pipelines.load
+Created schemas.src
+Created schemas.src.grants
+Files: 3 uploaded, 0 deleted
+Resources: 5 created, 0 changed, 0 deleted, 0 unchanged
+
+>>> print_requests.py --summary //unity-catalog/schemas //unity-catalog/permissions/schema //pipelines
+POST /api/2.1/unity-catalog/schemas
+PATCH /api/2.1/unity-catalog/permissions/schema/catalog_grants_pipeline_[UNIQUE_NAME].src
+POST /api/2.0/pipelines
+
+>>> [CLI] bundle plan
+Plan: 0 to add, 0 to change, 0 to delete, 5 unchanged
+
+>>> [CLI] bundle destroy --auto-approve
+The following resources will be deleted:
+  delete resources.catalogs.pipeline_catalog
+  delete resources.pipelines.load
+  delete resources.schemas.src
+
+This action will result in the deletion of the following UC schemas. Any underlying data may be lost:
+  delete resources.schemas.src
+
+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.load
+
+All files and directories at the following location will be deleted: /Workspace/Users/[USERNAME]/.bundle/schema-grants-pipeline-[UNIQUE_NAME]/default
+
+Destroy: 3 deleted

acceptance/bundle/resources/grants/schemas/pipeline_run_as/script

@@ -0,0 +1,29 @@
+# Creating the pipeline fails unless its run_as principal can already create tables in the
+# schema, so the pipeline must be deployed after the grants it relies on.
+# See https://github.com/databricks/cli/issues/4573
+if [ -z "$CLOUD_ENV" ]; then
+    export TEST_SP_APPLICATION_ID="22222222-2222-2222-2222-222222222222"
+fi
+
+add_repl "$TEST_SP_APPLICATION_ID" TEST_SP_APPLICATION_ID
+
+export RUN_AS_SP="$TEST_SP_APPLICATION_ID"
+
+envsubst < databricks.yml.tmpl > databricks.yml
+
+cleanup() {
+    trace $CLI bundle destroy --auto-approve
+}
+trap cleanup EXIT
+
+title "The grants of a referenced schema become an implicit lifecycle.depends_on"
+trace $CLI bundle validate -o json | jq '.resources | map_values(map_values(.lifecycle.depends_on))'
+
+title "The plan orders dependents after the grants"
+trace $CLI bundle plan -o json | jq '.plan | map_values(.depends_on)'
+
+trace $CLI bundle deploy
+# The catalog PATCH is excluded because its order relative to the schema is not fixed.
+trace print_requests.py --summary //unity-catalog/schemas //unity-catalog/permissions/schema //pipelines
+
+trace $CLI bundle plan

acceptance/bundle/resources/grants/schemas/pipeline_run_as/test.toml

@@ -0,0 +1,3 @@
+# Local only: on cloud the deploying identity lacks the servicePrincipal.user role on
+# TEST_SP_APPLICATION_ID, so it cannot set that service principal as the pipeline's run_as.
+Cloud = false

acceptance/bundle/resources/grants/schemas/pipeline_run_as/transformation.py

@@ -0,0 +1 @@
+print("Hello, World!")

acceptance/bundle/resources/permissions/pipelines/depends_on/databricks.yml

@@ -0,0 +1,15 @@
+bundle:
+  name: pipeline-depends-on-permissions
+
+resources:
+  jobs:
+    notify:
+      name: notify
+      permissions:
+        - level: CAN_VIEW
+          group_name: users
+
+  pipelines:
+    load:
+      # The pipeline refers to the job by name, so it waits for the job's permissions.
+      name: ${resources.jobs.notify.name}-load

acceptance/bundle/resources/permissions/pipelines/depends_on/out.test.toml

@@ -0,0 +1,2 @@
+Cloud = false
+EnvMatrix.DMS = ["", "true"]

acceptance/bundle/resources/permissions/pipelines/depends_on/output.txt

@@ -0,0 +1,23 @@
+
+=== A pipeline that refers to a job waits for the job's permissions
+
+>>> [CLI] bundle plan -o json
+{
+  "resources.jobs.notify": null,
+  "resources.jobs.notify.permissions": [
+    {
+      "node": "resources.jobs.notify",
+      "label": "${resources.jobs.notify.id}"
+    }
+  ],
+  "resources.pipelines.load": [
+    {
+      "node": "resources.jobs.notify",
+      "label": "${resources.jobs.notify.name}"
+    },
+    {
+      "node": "resources.jobs.notify.permissions",
+      "label": "depends_on"
+    }
+  ]
+}

acceptance/bundle/resources/permissions/pipelines/depends_on/script

@@ -0,0 +1,2 @@
+title "A pipeline that refers to a job waits for the job's permissions"
+trace $CLI bundle plan -o json | jq '.plan | map_values(.depends_on)'

bundle/config/mutator/resourcemutator/capture_pipeline_dependencies.go

@@ -0,0 +1,135 @@
+package resourcemutator
+
+import (
+	"context"
+	"fmt"
+	"maps"
+	"reflect"
+	"slices"
+	"strings"
+
+	"github.com/databricks/cli/bundle"
+	"github.com/databricks/cli/bundle/config"
+	"github.com/databricks/cli/bundle/config/resources"
+	"github.com/databricks/cli/libs/diag"
+	"github.com/databricks/cli/libs/dyn"
+	"github.com/databricks/cli/libs/dyn/dynvar"
+)
+
+type capturePipelineDependencies struct{}
+
+// CapturePipelineDependencies makes a pipeline wait for the grants and permissions of the resources
+// it refers to, directly or through lifecycle.depends_on, so a pipeline is not created before the
+// privileges it depends on are in place. A dependency that would create a cycle is skipped, e.g.
+// when an app refers to the pipeline and its service principal is granted on the pipeline's schema.
+func CapturePipelineDependencies() bundle.Mutator {
+	return &capturePipelineDependencies{}
+}
+
+func (m *capturePipelineDependencies) Name() string {
+	return "CapturePipelineDependencies"
+}
+
+func (m *capturePipelineDependencies) Apply(ctx context.Context, b *bundle.Bundle) diag.Diagnostics {
+	graph, err := dependencyGraph(b)
+	if err != nil {
+		return diag.FromErr(err)
+	}
+
+	// Pipelines are visited in a fixed order, so the same cycle-forming dependency is skipped on every run.
+	for _, key := range slices.Sorted(maps.Keys(b.Config.Resources.Pipelines)) {
+		p := b.Config.Resources.Pipelines[key]
+		node := "resources.pipelines." + key
+		for _, dep := range graph[node] {
+			for _, sub := range []string{".grants", ".permissions"} {
+				target := dep + sub
+				if _, ok := graph[target]; !ok || slices.Contains(graph[node], target) || reaches(graph, target, node) {
+					continue
+				}
+				graph[node] = append(graph[node], target)
+				p.Lifecycle.DependsOn = append(p.Lifecycle.DependsOn, resources.DependsOn{Resource: strings.TrimPrefix(target, "resources.")})
+			}
+		}
+	}
+	return nil
+}
+
+// dependencyGraph returns, for every resource and its grants or permissions, the nodes it depends
+// on: those it refers to and those in its lifecycle.depends_on. Nodes are keyed as in the deployment
+// plan, e.g. "resources.schemas.foo.grants"; a node without dependencies maps to nil.
+func dependencyGraph(b *bundle.Bundle) (map[string][]string, error) {
+	graph := make(map[string][]string)
+
+	err := dyn.WalkReadOnly(b.Config.Value(), func(p dyn.Path, v dyn.Value) error {
+		node, resourceType := config.GetNodeAndType(p)
+		if resourceType == "" {
+			return nil
+		}
+
+		src := node.String()
+		if len(p) == len(node) {
+			// Grants and permissions without entries are not deployed, so nothing can depend on them.
+			if seq, ok := v.AsSequence(); len(node) == 4 && (!ok || len(seq) == 0) {
+				return nil
+			}
+			if _, ok := graph[src]; !ok {
+				graph[src] = nil
+			}
+			return nil
+		}
+
+		ref, ok := dynvar.NewRef(v)
+		if !ok {
+			return nil
+		}
+		for _, r := range ref.References() {
+			path, err := dyn.NewPathFromString(r)
+			if err != nil {
+				return fmt.Errorf("parsing %q: %w", r, err)
+			}
+			if target, targetType := config.GetNodeAndType(path); targetType != "" {
+				graph[src] = append(graph[src], target.String())
+			}
+		}
+		return nil
+	})
+	if err != nil {
+		return nil, err
+	}
+
+	for _, group := range b.Config.Resources.AllResources() {
+		for key, resource := range group.Resources {
+			// Python-generated configs can contain null resource entries.
+			if reflect.ValueOf(resource).IsNil() {
+				continue
+			}
+			src := "resources." + group.Description.PluralName + "." + key
+			for _, dep := range resource.GetLifecycle().GetDependsOn() {
+				// An entry that does not resolve is reported by validate.LifecycleDependsOn.
+				if target, err := b.Config.Resources.ResolveDependsOn(dep.Resource); err == nil {
+					graph[src] = append(graph[src], target)
+				}
+			}
+		}
+	}
+	return graph, nil
+}
+
+// reaches reports whether from is to or depends on it, directly or through other nodes.
+func reaches(graph map[string][]string, from, to string) bool {
+	visited := make(map[string]bool)
+	stack := []string{from}
+	for len(stack) > 0 {
+		n := stack[len(stack)-1]
+		stack = stack[:len(stack)-1]
+		if n == to {
+			return true
+		}
+		if visited[n] {
+			continue
+		}
+		visited[n] = true
+		stack = append(stack, graph[n]...)
+	}
+	return false
+}