internal: Parse permissions & grants out of terraform state (#3963)

## Changes
- Update ExportedResourcesMap to be a flat map resourceKey ->
resourceEntry
- Extract permissions and grants from terraform state.

## Why
Need to extract the state to migrate it in #3847

Author: denik

bundle/deploy/terraform/check_dashboards_modified_remotely.go

@@ -3,6 +3,7 @@ package terraform
 import (
 	"context"
 	"fmt"
+	"strings"
 
 	"github.com/databricks/cli/bundle"
 	"github.com/databricks/cli/bundle/config/engine"
@@ -30,7 +31,18 @@ func collectDashboardsFromState(ctx context.Context, b *bundle.Bundle, directDep
 	}
 
 	var dashboards []dashboardState
-	for resourceName, instance := range state["dashboards"] {
+	for resourceKey, instance := range state {
+		// Check if this is a dashboard resource key
+		if !strings.HasPrefix(resourceKey, "resources.dashboards.") {
+			continue
+		}
+		// Extract dashboard name from "resources.dashboards.name"
+		parts := strings.Split(resourceKey, ".")
+		if len(parts) != 3 {
+			continue
+		}
+		resourceName := parts[2]
+
 		dashboards = append(dashboards, dashboardState{
 			Name: resourceName,
 			ID:   instance.ID,

bundle/deploy/terraform/check_running_resources.go

@@ -6,6 +6,7 @@ import (
 	"strconv"
 
 	"github.com/databricks/cli/bundle"
+	"github.com/databricks/cli/bundle/config"
 	"github.com/databricks/cli/libs/diag"
 	"github.com/databricks/databricks-sdk-go"
 	"github.com/databricks/databricks-sdk-go/service/jobs"
@@ -53,42 +54,41 @@ func CheckRunningResource() *checkRunningResources {
 func checkAnyResourceRunning(ctx context.Context, w *databricks.WorkspaceClient, state ExportedResourcesMap) error {
 	errs, errCtx := errgroup.WithContext(ctx)
 
-	for _, jobAttrs := range state["jobs"] {
-		id := jobAttrs.ID
+	for resourceKey, attrs := range state {
+		id := attrs.ID
 		if id == "" {
 			continue
 		}
 
-		errs.Go(func() error {
-			isRunning, err := IsJobRunning(errCtx, w, id)
-			// If there's an error retrieving the job, we assume it's not running
-			if err != nil {
-				return err
-			}
-			if isRunning {
-				return &ErrResourceIsRunning{resourceType: "job", resourceId: id}
-			}
-			return nil
-		})
-	}
-
-	for _, pipelineAttrs := range state["pipelines"] {
-		id := pipelineAttrs.ID
-		if id == "" {
-			continue
+		resourceType := config.GetResourceTypeFromKey(resourceKey)
+
+		if resourceType == "jobs" {
+			errs.Go(func() error {
+				isRunning, err := IsJobRunning(errCtx, w, id)
+				// If there's an error retrieving the job, we assume it's not running
+				if err != nil {
+					return err
+				}
+				if isRunning {
+					return &ErrResourceIsRunning{resourceType: "job", resourceId: id}
+				}
+				return nil
+			})
 		}
 
-		errs.Go(func() error {
-			isRunning, err := IsPipelineRunning(errCtx, w, id)
-			// If there's an error retrieving the pipeline, we assume it's not running
-			if err != nil {
+		if resourceType == "pipelines" {
+			errs.Go(func() error {
+				isRunning, err := IsPipelineRunning(errCtx, w, id)
+				// If there's an error retrieving the pipeline, we assume it's not running
+				if err != nil {
+					return nil
+				}
+				if isRunning {
+					return &ErrResourceIsRunning{resourceType: "pipeline", resourceId: id}
+				}
 				return nil
-			}
-			if isRunning {
-				return &ErrResourceIsRunning{resourceType: "pipeline", resourceId: id}
-			}
-			return nil
-		})
+			})
+		}
 	}
 
 	return errs.Wait()

bundle/deploy/terraform/check_running_resources_test.go

@@ -21,9 +21,7 @@ func TestIsAnyResourceRunningWithEmptyState(t *testing.T) {
 func TestIsAnyResourceRunningWithJob(t *testing.T) {
 	m := mocks.NewMockWorkspaceClient(t)
 	resources := ExportedResourcesMap{
-		"jobs": map[string]ResourceState{
-			"job1": {ID: "123"},
-		},
+		"resources.jobs.job1": {ID: "123"},
 	}
 
 	jobsApi := m.GetMockJobsAPI()
@@ -49,9 +47,7 @@ func TestIsAnyResourceRunningWithJob(t *testing.T) {
 func TestIsAnyResourceRunningWithPipeline(t *testing.T) {
 	m := mocks.NewMockWorkspaceClient(t)
 	resources := ExportedResourcesMap{
-		"pipelines": map[string]ResourceState{
-			"pipeline1": {ID: "123"},
-		},
+		"resources.pipelines.pipeline1": {ID: "123"},
 	}
 
 	pipelineApi := m.GetMockPipelinesAPI()
@@ -79,9 +75,7 @@ func TestIsAnyResourceRunningWithAPIFailure(t *testing.T) {
 	m := mocks.NewMockWorkspaceClient(t)
 
 	resources := ExportedResourcesMap{
-		"pipelines": map[string]ResourceState{
-			"pipeline1": {ID: "123"},
-		},
+		"resources.pipelines.pipeline1": {ID: "123"},
 	}
 
 	pipelineApi := m.GetMockPipelinesAPI()

bundle/deploy/terraform/util.go

@@ -49,7 +49,7 @@ type stateInstanceAttributes struct {
 	ETag string `json:"etag,omitempty"`
 }
 
-// Returns a mapping group -> name -> stateInstanceAttributes
+// Returns a mapping resourceKey -> stateInstanceAttributes
 func parseResourcesState(ctx context.Context, path string) (ExportedResourcesMap, error) {
 	rawState, err := os.ReadFile(path)
 	if err != nil {
@@ -76,34 +76,48 @@ func parseResourcesState(ctx context.Context, path string) (ExportedResourcesMap
 		}
 		for _, instance := range resource.Instances {
 			groupName, ok := TerraformToGroupName[resource.Type]
+
 			if !ok {
-				// permissions, grants, secret_acls
+				// secret_acls
 				continue
 			}
 
-			group, present := result[groupName]
-			if !present {
-				group = make(map[string]ResourceState)
-				result[groupName] = group
-			}
+			var resourceKey string
+			var resourceState ResourceState
 
 			switch groupName {
 			case "apps":
-				group[resource.Name] = ResourceState{ID: instance.Attributes.Name}
+				resourceKey = "resources." + groupName + "." + resource.Name
+				resourceState = ResourceState{ID: instance.Attributes.Name}
 			case "secret_scopes":
-				group[resource.Name] = ResourceState{ID: instance.Attributes.Name}
+				resourceKey = "resources." + groupName + "." + resource.Name
+				resourceState = ResourceState{ID: instance.Attributes.Name}
 			case "dashboards":
-				group[resource.Name] = ResourceState{ID: instance.Attributes.ID, ETag: instance.Attributes.ETag}
+				resourceKey = "resources." + groupName + "." + resource.Name
+				resourceState = ResourceState{ID: instance.Attributes.ID, ETag: instance.Attributes.ETag}
+			case "permissions":
+				resourceKey = convertPermissionsResourceNameToKey(resource.Name)
+				resourceState = ResourceState{ID: instance.Attributes.ID}
+			case "grants":
+				resourceKey = convertGrantsResourceNameToKey(resource.Name)
+				resourceState = ResourceState{ID: instance.Attributes.ID}
 			default:
-				group[resource.Name] = ResourceState{ID: instance.Attributes.ID}
+				resourceKey = "resources." + groupName + "." + resource.Name
+				resourceState = ResourceState{ID: instance.Attributes.ID}
 			}
+
+			if resourceKey == "" {
+				return nil, fmt.Errorf("cannot calculate resource key for type=%q name=%q id=%q", resource.Type, resource.Name, instance.Attributes.ID)
+			}
+
+			result[resourceKey] = resourceState
 		}
 	}
 
 	return result, nil
 }
 
-// Returns a mapping group -> name -> stateInstanceAttributes
+// Returns a mapping resourceKey -> stateInstanceAttributes
 func ParseResourcesState(ctx context.Context, b *bundle.Bundle) (ExportedResourcesMap, error) {
 	cacheDir, err := Dir(ctx, b)
 	if err != nil {

bundle/deploy/terraform/util_test.go

@@ -91,9 +91,7 @@ func TestParseResourcesStateWithExistingStateFile(t *testing.T) {
 	state, err := parseResourcesState(ctx, localPath)
 	assert.NoError(t, err)
 	expected := ExportedResourcesMap{
-		"pipelines": map[string]ResourceState{
-			"test_pipeline": {ID: "123"},
-		},
+		"resources.pipelines.test_pipeline": {ID: "123"},
 	}
 	assert.Equal(t, expected, state)
 }

bundle/direct/dstate/state.go

@@ -5,7 +5,6 @@ import (
 	"encoding/json"
 	"fmt"
 	"os"
-	"strings"
 	"sync"
 
 	"github.com/databricks/cli/bundle/config/resources"
@@ -30,22 +29,6 @@ type ResourceEntry struct {
 	State any    `json:"state"`
 }
 
-// splitKey extracts group and name from the key: 'resources.jobs.foo' -> 'jobs', 'foo', true
-// For sub-resources like permissions it returns "", "", false
-// Note we don't use group/name anywhere in bundle/direct, this is only for ExportState
-// which makes ID available to other parts of DABs
-func splitKey(key string) (group, name string, ok bool) {
-	items := strings.Split(key, ".")
-	if len(items) != 3 {
-		// e.g. resources.jobs.foo.permissions
-		return "", "", false
-	}
-	if items[0] != "resources" {
-		return "", "", false
-	}
-	return items[1], items[2], true
-}
-
 func (db *DeploymentState) SaveState(key, newID string, state any) error {
 	db.AssertOpened()
 	db.mu.Lock()
@@ -140,15 +123,6 @@ func (db *DeploymentState) AssertOpened() {
 func (db *DeploymentState) ExportState(ctx context.Context) resourcestate.ExportedResourcesMap {
 	result := make(resourcestate.ExportedResourcesMap)
 	for key, entry := range db.Data.State {
-		groupName, resourceName, ok := splitKey(key)
-		if !ok {
-			continue
-		}
-		resultGroup, ok := result[groupName]
-		if !ok {
-			resultGroup = make(map[string]resourcestate.ResourceState)
-			result[groupName] = resultGroup
-		}
 		// Extract etag for dashboards.
 		var etag string
 		switch dashboard := entry.State.(type) {
@@ -166,7 +140,7 @@ func (db *DeploymentState) ExportState(ctx context.Context) resourcestate.Export
 			etag = dashboard.Etag
 		}
 
-		resultGroup[resourceName] = resourcestate.ResourceState{
+		result[key] = resourcestate.ResourceState{
 			ID:   entry.ID,
 			ETag: etag,
 		}

bundle/statemgmt/resourcestate/resourcestate.go

@@ -10,5 +10,5 @@ type ResourceState struct {
 }
 
 // ExportedResourcesMap stores relevant attributes from terraform/direct state for all resources
-// Maps group (e.g. "jobs") -> resource name -> ResourceState
-type ExportedResourcesMap map[string]map[string]ResourceState
+// Maps resource key (e.g. "resources.jobs.foo", "resources.jobs.foo.permissions") -> ResourceState
+type ExportedResourcesMap map[string]ResourceState

bundle/statemgmt/state_load.go

@@ -5,6 +5,7 @@ import (
 	"errors"
 	"fmt"
 	"slices"
+	"strings"
 
 	"github.com/databricks/cli/bundle"
 	"github.com/databricks/cli/bundle/config"
@@ -63,8 +64,19 @@ func (l *load) Apply(ctx context.Context, b *bundle.Bundle) diag.Diagnostics {
 	}
 
 	// Merge dashboard etags into configuration.
-	for k, dstate := range state["dashboards"] {
-		dconfig, ok := b.Config.Resources.Dashboards[k]
+	for resourceKey, dstate := range state {
+		// Check if this is a dashboard resource key
+		if !strings.HasPrefix(resourceKey, "resources.dashboards.") {
+			continue
+		}
+		// Extract dashboard name from "resources.dashboards.name"
+		parts := strings.Split(resourceKey, ".")
+		if len(parts) != 3 {
+			continue
+		}
+		dashboardName := parts[2]
+
+		dconfig, ok := b.Config.Resources.Dashboards[dashboardName]
 
 		// Case: A dashboard is defined in state but not in configuration.
 		// In this case the dashboard has been deleted and we do not need to load the etag.
@@ -98,31 +110,43 @@ func StateToBundle(ctx context.Context, state ExportedResourcesMap, config *conf
 			return v, err
 		}
 
-		for groupName, group := range state {
+		for resourceKey, attrs := range state {
+			// Parse resource key like "resources.jobs.foo" or "resources.jobs.foo.permissions"
+			parts := strings.Split(resourceKey, ".")
+			if len(parts) < 3 || parts[0] != "resources" {
+				continue // Skip invalid resource keys
+			}
+
+			groupName := parts[1]
+			resourceName := parts[2]
+
+			// Skip permissions for now as they are sub-resources
+			if len(parts) > 3 {
+				continue
+			}
+
 			var err error
 			v, err = ensureMap(v, dyn.Path{dyn.Key("resources"), dyn.Key(groupName)})
 			if err != nil {
 				return v, err
 			}
 
-			for resourceName, attrs := range group {
-				path := dyn.Path{dyn.Key("resources"), dyn.Key(groupName), dyn.Key(resourceName)}
-				resource, err := dyn.GetByPath(v, path)
-				if !resource.IsValid() {
-					m := dyn.NewMapping()
-					m.SetLoc("id", nil, dyn.V(attrs.ID))
-					m.SetLoc("modified_status", nil, dyn.V(resources.ModifiedStatusDeleted))
-					v, err = dyn.SetByPath(v, path, dyn.V(m))
-					if err != nil {
-						return dyn.InvalidValue, err
-					}
-				} else if err != nil {
+			path := dyn.Path{dyn.Key("resources"), dyn.Key(groupName), dyn.Key(resourceName)}
+			resource, err := dyn.GetByPath(v, path)
+			if !resource.IsValid() {
+				m := dyn.NewMapping()
+				m.SetLoc("id", nil, dyn.V(attrs.ID))
+				m.SetLoc("modified_status", nil, dyn.V(resources.ModifiedStatusDeleted))
+				v, err = dyn.SetByPath(v, path, dyn.V(m))
+				if err != nil {
+					return dyn.InvalidValue, err
+				}
+			} else if err != nil {
+				return dyn.InvalidValue, err
+			} else {
+				v, err = dyn.SetByPath(v, dyn.Path{dyn.Key("resources"), dyn.Key(groupName), dyn.Key(resourceName), dyn.Key("id")}, dyn.V(attrs.ID))
+				if err != nil {
 					return dyn.InvalidValue, err
-				} else {
-					v, err = dyn.SetByPath(v, dyn.Path{dyn.Key("resources"), dyn.Key(groupName), dyn.Key(resourceName), dyn.Key("id")}, dyn.V(attrs.ID))
-					if err != nil {
-						return dyn.InvalidValue, err
-					}
 				}
 			}
 		}