diff --git a/.nextchanges/bundles/grants-before-dependents.md b/.nextchanges/bundles/grants-before-dependents.md new file mode 100644 index 00000000000..6e4386567e4 --- /dev/null +++ b/.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`. diff --git a/acceptance/bundle/resources/grants/schemas/pipeline_run_as/databricks.yml.tmpl b/acceptance/bundle/resources/grants/schemas/pipeline_run_as/databricks.yml.tmpl new file mode 100644 index 00000000000..f029e2d98a1 --- /dev/null +++ b/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 diff --git a/acceptance/bundle/resources/grants/schemas/pipeline_run_as/out.test.toml b/acceptance/bundle/resources/grants/schemas/pipeline_run_as/out.test.toml new file mode 100644 index 00000000000..a927a5fbc06 --- /dev/null +++ b/acceptance/bundle/resources/grants/schemas/pipeline_run_as/out.test.toml @@ -0,0 +1,2 @@ +Cloud = false +EnvMatrix.DMS = ["", "true"] diff --git a/acceptance/bundle/resources/grants/schemas/pipeline_run_as/output.txt b/acceptance/bundle/resources/grants/schemas/pipeline_run_as/output.txt new file mode 100644 index 00000000000..9eff49bbd33 --- /dev/null +++ b/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 diff --git a/acceptance/bundle/resources/grants/schemas/pipeline_run_as/script b/acceptance/bundle/resources/grants/schemas/pipeline_run_as/script new file mode 100644 index 00000000000..975cdfd9ba0 --- /dev/null +++ b/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 diff --git a/acceptance/bundle/resources/grants/schemas/pipeline_run_as/test.toml b/acceptance/bundle/resources/grants/schemas/pipeline_run_as/test.toml new file mode 100644 index 00000000000..10a7e8b41bb --- /dev/null +++ b/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 diff --git a/acceptance/bundle/resources/grants/schemas/pipeline_run_as/transformation.py b/acceptance/bundle/resources/grants/schemas/pipeline_run_as/transformation.py new file mode 100644 index 00000000000..7df869a15e7 --- /dev/null +++ b/acceptance/bundle/resources/grants/schemas/pipeline_run_as/transformation.py @@ -0,0 +1 @@ +print("Hello, World!") diff --git a/acceptance/bundle/resources/permissions/pipelines/depends_on/databricks.yml b/acceptance/bundle/resources/permissions/pipelines/depends_on/databricks.yml new file mode 100644 index 00000000000..2a17f078d0c --- /dev/null +++ b/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 diff --git a/acceptance/bundle/resources/permissions/pipelines/depends_on/out.test.toml b/acceptance/bundle/resources/permissions/pipelines/depends_on/out.test.toml new file mode 100644 index 00000000000..a927a5fbc06 --- /dev/null +++ b/acceptance/bundle/resources/permissions/pipelines/depends_on/out.test.toml @@ -0,0 +1,2 @@ +Cloud = false +EnvMatrix.DMS = ["", "true"] diff --git a/acceptance/bundle/resources/permissions/pipelines/depends_on/output.txt b/acceptance/bundle/resources/permissions/pipelines/depends_on/output.txt new file mode 100644 index 00000000000..ff917c2fc80 --- /dev/null +++ b/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" + } + ] +} diff --git a/acceptance/bundle/resources/permissions/pipelines/depends_on/script b/acceptance/bundle/resources/permissions/pipelines/depends_on/script new file mode 100644 index 00000000000..4979de31b5d --- /dev/null +++ b/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)' diff --git a/bundle/config/mutator/resourcemutator/capture_pipeline_dependencies.go b/bundle/config/mutator/resourcemutator/capture_pipeline_dependencies.go new file mode 100644 index 00000000000..98408eca92c --- /dev/null +++ b/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 +} diff --git a/bundle/config/mutator/resourcemutator/capture_pipeline_dependencies_test.go b/bundle/config/mutator/resourcemutator/capture_pipeline_dependencies_test.go new file mode 100644 index 00000000000..382aeab9cd8 --- /dev/null +++ b/bundle/config/mutator/resourcemutator/capture_pipeline_dependencies_test.go @@ -0,0 +1,200 @@ +package resourcemutator + +import ( + "reflect" + "testing" + + "github.com/databricks/cli/bundle" + "github.com/databricks/cli/bundle/config" + "github.com/databricks/cli/bundle/config/resources" + "github.com/databricks/databricks-sdk-go/service/catalog" + "github.com/databricks/databricks-sdk-go/service/jobs" + "github.com/databricks/databricks-sdk-go/service/pipelines" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestCapturePipelineDependencies(t *testing.T) { + grants := func(principal string) []catalog.PrivilegeAssignment { + return []catalog.PrivilegeAssignment{{Principal: principal, Privileges: []catalog.Privilege{catalog.PrivilegeCreateTable}}} + } + schema := func(grants []catalog.PrivilegeAssignment) *resources.Schema { + return &resources.Schema{CatalogName: "main", Name: "src", Grants: grants} + } + // A pipeline that refers to the schema. A constructor, because the mutator writes to the + // pipeline and a shared pointer would leak lifecycle.depends_on between cases. + pipeline := func() *resources.Pipeline { + return &resources.Pipeline{ + Catalog: "${resources.schemas.src.catalog_name}", + Schema: "${resources.schemas.src.name}", + } + } + // A job with permissions, referred to by name. + job := func(permissions []resources.JobPermission) *resources.Job { + return &resources.Job{ + Name: "job", + Permissions: permissions, + } + } + + tests := []struct { + name string + resources config.Resources + expected map[string][]resources.DependsOn + }{ + { + name: "pipeline that refers to a resource with grants depends on the grants", + resources: config.Resources{ + Schemas: map[string]*resources.Schema{"src": schema(grants("sp"))}, + Pipelines: map[string]*resources.Pipeline{"load": pipeline()}, + }, + expected: map[string][]resources.DependsOn{ + "pipelines.load": {{Resource: "schemas.src.grants"}}, + }, + }, + { + name: "empty grants and permissions add nothing", + resources: config.Resources{ + Schemas: map[string]*resources.Schema{"src": schema([]catalog.PrivilegeAssignment{})}, + Jobs: map[string]*resources.Job{"job": job([]resources.JobPermission{})}, + Pipelines: map[string]*resources.Pipeline{"load": {CreatePipeline: pipelines.CreatePipeline{Name: "${resources.jobs.job.name}", Catalog: "${resources.schemas.src.catalog_name}", Schema: "${resources.schemas.src.name}"}}}, + }, + expected: map[string][]resources.DependsOn{}, + }, + { + name: "null pipeline entry does not panic and other pipelines are still processed", + resources: config.Resources{ + Schemas: map[string]*resources.Schema{"src": schema(grants("sp"))}, + Pipelines: map[string]*resources.Pipeline{"empty": nil, "load": pipeline()}, + }, + expected: map[string][]resources.DependsOn{ + "pipelines.load": {{Resource: "schemas.src.grants"}}, + }, + }, + { + name: "pipeline waits for the schema's grants, not its catalog's", + resources: config.Resources{ + Catalogs: map[string]*resources.Catalog{"main": {CreateCatalog: catalog.CreateCatalog{Name: "main"}, Grants: grants("sp")}}, + Schemas: map[string]*resources.Schema{"src": { + CreateSchema: catalog.CreateSchema{CatalogName: "${resources.catalogs.main.name}", Name: "src"}, + Grants: grants("sp"), + }}, + Pipelines: map[string]*resources.Pipeline{"load": pipeline()}, + }, + expected: map[string][]resources.DependsOn{ + "pipelines.load": {{Resource: "schemas.src.grants"}}, + }, + }, + { + name: "non-pipeline resource that refers to a resource with grants gets nothing", + resources: config.Resources{ + Schemas: map[string]*resources.Schema{"src": schema(grants("sp"))}, + Jobs: map[string]*resources.Job{"job": {JobSettings: jobs.JobSettings{Name: "${resources.schemas.src.name}"}}}, + }, + expected: map[string][]resources.DependsOn{}, + }, + { + name: "grants that refer to an app that does not refer to the pipeline are added", + resources: config.Resources{ + Schemas: map[string]*resources.Schema{"src": schema(grants("${resources.apps.app.service_principal_client_id}"))}, + Apps: map[string]*resources.App{"app": {Name: "app", Description: "${resources.schemas.src.name}"}}, + Pipelines: map[string]*resources.Pipeline{"load": pipeline()}, + }, + expected: map[string][]resources.DependsOn{ + "pipelines.load": {{Resource: "schemas.src.grants"}}, + }, + }, + { + name: "grants that depend on an app that refers to the pipeline are skipped", + resources: config.Resources{ + Schemas: map[string]*resources.Schema{"src": schema(grants("${resources.apps.app.service_principal_client_id}"))}, + Apps: map[string]*resources.App{"app": {Name: "app", Description: "${resources.pipelines.load.name}"}}, + Pipelines: map[string]*resources.Pipeline{"load": pipeline()}, + }, + expected: map[string][]resources.DependsOn{}, + }, + { + name: "lifecycle.depends_on on a pipeline implies the grants of the resource", + resources: config.Resources{ + Schemas: map[string]*resources.Schema{"src": schema(grants("sp"))}, + Pipelines: map[string]*resources.Pipeline{"load": { + BaseResource: resources.BaseResource{Lifecycle: resources.Lifecycle{DependsOn: []resources.DependsOn{{Resource: "schemas.src"}}}}, + CreatePipeline: pipeline().CreatePipeline, + }}, + }, + expected: map[string][]resources.DependsOn{ + "pipelines.load": {{Resource: "schemas.src"}, {Resource: "schemas.src.grants"}}, + }, + }, + { + name: "grants already in lifecycle.depends_on", + resources: config.Resources{ + Schemas: map[string]*resources.Schema{"src": schema(grants("sp"))}, + Pipelines: map[string]*resources.Pipeline{"load": { + BaseResource: resources.BaseResource{Lifecycle: resources.Lifecycle{DependsOn: []resources.DependsOn{{Resource: "schemas.src.grants"}}}}, + CreatePipeline: pipeline().CreatePipeline, + }}, + }, + expected: map[string][]resources.DependsOn{ + "pipelines.load": {{Resource: "schemas.src.grants"}}, + }, + }, + { + name: "pipeline depends on the permissions of a resource it refers to", + resources: config.Resources{ + Jobs: map[string]*resources.Job{"job": job([]resources.JobPermission{{Level: "CAN_VIEW", GroupName: "users"}})}, + Pipelines: map[string]*resources.Pipeline{"load": { + CreatePipeline: pipelines.CreatePipeline{Name: "${resources.jobs.job.name}"}, + }}, + }, + expected: map[string][]resources.DependsOn{ + "pipelines.load": {{Resource: "jobs.job.permissions"}}, + }, + }, + { + name: "non-pipeline resource that refers to a resource with permissions gets nothing", + resources: config.Resources{ + Jobs: map[string]*resources.Job{ + "job": job([]resources.JobPermission{{Level: "CAN_VIEW", GroupName: "users"}}), + "reader": {JobSettings: jobs.JobSettings{Name: "${resources.jobs.job.name}"}}, + }, + }, + expected: map[string][]resources.DependsOn{}, + }, + { + name: "permissions that depend on the referring pipeline are skipped", + resources: config.Resources{ + Jobs: map[string]*resources.Job{"job": job([]resources.JobPermission{{Level: "CAN_VIEW", UserName: "${resources.pipelines.load.name}"}})}, + Pipelines: map[string]*resources.Pipeline{"load": { + CreatePipeline: pipelines.CreatePipeline{Name: "${resources.jobs.job.name}"}, + }}, + }, + expected: map[string][]resources.DependsOn{}, + }, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + b := &bundle.Bundle{Config: config.Root{Resources: tc.resources}} + + // Normalization runs more than once, so applying the mutator again must not add anything. + for range 2 { + diags := bundle.Apply(t.Context(), b, CapturePipelineDependencies()) + require.NoError(t, diags.Error()) + + actual := make(map[string][]resources.DependsOn) + for _, group := range b.Config.Resources.AllResources() { + for key, resource := range group.Resources { + if reflect.ValueOf(resource).IsNil() { + continue + } + if deps := resource.GetLifecycle().GetDependsOn(); len(deps) > 0 { + actual[group.Description.PluralName+"."+key] = deps + } + } + } + assert.Equal(t, tc.expected, actual) + } + }) + } +} diff --git a/bundle/config/mutator/resourcemutator/resource_mutator.go b/bundle/config/mutator/resourcemutator/resource_mutator.go index e00444cc773..16a997ebf24 100644 --- a/bundle/config/mutator/resourcemutator/resource_mutator.go +++ b/bundle/config/mutator/resourcemutator/resource_mutator.go @@ -183,6 +183,10 @@ func applyNormalizeMutators(ctx context.Context, b *bundle.Bundle) { // Translates implicit schema and catalog references across all UC resources to explicit syntax to capture dependencies CaptureUCDependencies(), + // Reads (dynamic and typed): resources.*.** (searches for ${resources.*} references), resources.*.*.{grants,permissions,lifecycle.depends_on} + // Updates (typed): resources.pipelines.*.lifecycle.depends_on (adds the grants and permissions of the resources a pipeline depends on) + CapturePipelineDependencies(), + // Reads (dynamic): resources.dashboards.*.file_path // Updates (dynamic): resources.dashboards.*.serialized_dashboard // Drops (dynamic): resources.dashboards.*.file_path