Track bundle resource counts and state sizes in telemetry

Adds a typed `resources_metadata` field to the bundle deploy telemetry
event, capturing per resource type:
- count of resources declared in the bundle configuration (replacing the
  deprecated DatabricksBundleDeployEvent.resource_*_count fields, still
  populated during the transition)
- max / mean / median state size in bytes across resources of that type

plus the whole simulated state file size and the effective deploy engine
("direct" or "terraform").

State sizes are computed from the bundle configuration, not by reading the
on-disk state file. Each resource's typed config is run through the direct
engine's adapter.PrepareState -- the same transformation direct uses to
derive the value it persists to resources.json -- and marshaled with the
indented encoding dstate.SaveState uses. Two consequences:
- the numbers are engine-independent, so direct and terraform deploys of the
  same logical bundle report identical sizes (tfstate is never read);
- for direct deploys each per-resource size equals len(entry.State) on disk
  byte-for-byte, and the file size matches resources.json up to the
  Lineage/Serial fields.

The feature is one isolated module (bundle/phases/resources_metadata.go) plus
one line at the telemetry-emission call site. Telemetry never fails a deploy:
all parse/adapter errors are logged at debug level and treated as missing
data.

The universe proto (DatabricksBundleDeployEvent.resources_metadata,
BundleResourcesMetadata, ResourceMetadata) already exists, so this is ingested
rather than dropped.

Co-authored-by: Isaac

Author: shreyas-goenka

acceptance/bundle/telemetry/deploy/out.telemetry.txt

@@ -35,6 +35,26 @@
           "[UUID]",
           "[UUID]"
         ],
+        "resources_metadata": {
+          "state_engine": "[ENGINE]",
+          "state_file_size_bytes": SMALL_INT,
+          "resources": [
+            {
+              "resource_type": "jobs",
+              "count": 3,
+              "state_size_max_bytes": SMALL_INT,
+              "state_size_mean_bytes": SMALL_INT,
+              "state_size_median_bytes": SMALL_INT
+            },
+            {
+              "resource_type": "pipelines",
+              "count": 2,
+              "state_size_max_bytes": SMALL_INT,
+              "state_size_mean_bytes": SMALL_INT,
+              "state_size_median_bytes": SMALL_INT
+            }
+          ]
+        },
         "experimental": {
           "configuration_file_count": 1,
           "variable_count": 0,

acceptance/bundle/telemetry/test.toml

@@ -20,3 +20,28 @@ New = '[OS]'
 [[Repls]]
 Old = '"local_cache_measurements_ms": \[[^\]]*\]'
 New = '"local_cache_measurements_ms": [...redacted...]'
+
+# Normalize the deployment engine name in resources_metadata so the same
+# fixture covers both DATABRICKS_BUNDLE_ENGINE=direct and =terraform runs.
+[[Repls]]
+Old = '"state_engine": "(direct|terraform)"'
+New = '"state_engine": "[ENGINE]"'
+
+# Normalize byte-size measurements in resources_metadata to placeholders.
+# Exact byte counts depend on resource state JSON formatting and would
+# make these golden files brittle across SDK changes.
+[[Repls]]
+Old = '"state_file_size_bytes": \d+'
+New = '"state_file_size_bytes": SMALL_INT'
+
+[[Repls]]
+Old = '"state_size_max_bytes": \d+'
+New = '"state_size_max_bytes": SMALL_INT'
+
+[[Repls]]
+Old = '"state_size_mean_bytes": \d+'
+New = '"state_size_mean_bytes": SMALL_INT'
+
+[[Repls]]
+Old = '"state_size_median_bytes": \d+'
+New = '"state_size_median_bytes": SMALL_INT'

bundle/phases/resources_metadata.go

@@ -0,0 +1,221 @@
+package phases
+
+import (
+	"cmp"
+	"context"
+	"encoding/json"
+	"fmt"
+	"slices"
+
+	"github.com/databricks/cli/bundle"
+	"github.com/databricks/cli/bundle/config/engine"
+	"github.com/databricks/cli/bundle/direct/dresources"
+	"github.com/databricks/cli/bundle/direct/dstate"
+	"github.com/databricks/cli/libs/dyn"
+	"github.com/databricks/cli/libs/log"
+	"github.com/databricks/cli/libs/telemetry/protos"
+)
+
+// collectResourcesMetadata builds a BundleResourcesMetadata for the deploy.
+//
+// State sizes are computed by running each resource's typed config through
+// the direct engine's adapter.PrepareState — the same transformation direct
+// uses to derive the value it persists to resources.json — and marshaling
+// each entry with dstate.SaveState's encoding (MarshalIndent("  ", " ")).
+// The whole-file size is then computed by assembling those entries into a
+// dstate.Database and marshaling it the way DeploymentState.unlockedSave
+// writes it (MarshalIndent("", " ")). So:
+//
+//   - Under DATABRICKS_BUNDLE_ENGINE=direct, per-resource sizes equal
+//     len(entry.State) on disk byte-for-byte, and state_file_size_bytes
+//     matches the resources.json file size to within a few bytes (only
+//     Lineage and Serial may differ, which we set to "" / 0 here).
+//   - Under =terraform, the same computation runs against the bundle config,
+//     producing identical numbers for the same logical bundle. tfstate is
+//     never read.
+//
+// Returns nil when the bundle declares no resources.
+func collectResourcesMetadata(ctx context.Context, b *bundle.Bundle) *protos.BundleResourcesMetadata {
+	counts, sizesByType, fileSize := collectResourceCountsAndSizes(ctx, b)
+	if len(counts) == 0 {
+		return nil
+	}
+
+	types := unionKeys(counts, sizesByType)
+	slices.Sort(types)
+
+	resources := make([]protos.ResourceMetadata, 0, len(types))
+	for _, t := range types {
+		sizes := sizesByType[t]
+		slices.SortFunc(sizes, func(a, b int64) int { return cmp.Compare(a, b) })
+		resources = append(resources, protos.ResourceMetadata{
+			ResourceType:         t,
+			Count:                counts[t],
+			StateSizeMaxBytes:    statMax(sizes),
+			StateSizeMeanBytes:   statMean(sizes),
+			StateSizeMedianBytes: statMedian(sizes),
+		})
+	}
+
+	return &protos.BundleResourcesMetadata{
+		StateEngine:        resolveDeployEngine(ctx, b),
+		StateFileSizeBytes: fileSize,
+		Resources:          resources,
+	}
+}
+
+// collectResourceCountsAndSizes walks the bundle config and assembles a
+// dstate.Database with each resource's PrepareState'd value, then marshals
+// that database the way direct writes resources.json. Returns per-type
+// counts, per-type per-resource byte lengths, and the byte length of the
+// whole simulated state file.
+func collectResourceCountsAndSizes(ctx context.Context, b *bundle.Bundle) (map[string]int64, map[string][]int64, int64) {
+	counts := make(map[string]int64)
+	sizesByType := make(map[string][]int64)
+
+	adapters := getAdapters(ctx, b)
+	db := dstate.NewDatabase("", 0)
+
+	pattern := dyn.NewPattern(dyn.Key("resources"), dyn.AnyKey(), dyn.AnyKey())
+	_, err := dyn.MapByPattern(b.Config.Value(), pattern, func(p dyn.Path, v dyn.Value) (dyn.Value, error) {
+		if len(p) < 3 {
+			return v, nil
+		}
+		resourceType := p[1].Key()
+		counts[resourceType]++
+
+		stateBytes, err := resourceStateBytes(b, adapters, p, resourceType)
+		if err != nil {
+			log.Debugf(ctx, "resources-metadata telemetry: %s: %s", p, err)
+			return v, nil
+		}
+		sizesByType[resourceType] = append(sizesByType[resourceType], int64(len(stateBytes)))
+		db.State[p.String()] = dstate.ResourceEntry{
+			ID:    extractResourceID(v),
+			State: stateBytes,
+		}
+		return v, nil
+	})
+	if err != nil {
+		log.Debugf(ctx, "resources-metadata telemetry: failed to walk config resources: %s", err)
+	}
+
+	var fileSize int64
+	if len(db.State) > 0 {
+		raw, mErr := json.MarshalIndent(db, "", " ")
+		if mErr != nil {
+			log.Debugf(ctx, "resources-metadata telemetry: failed to marshal database envelope: %s", mErr)
+		} else {
+			fileSize = int64(len(raw))
+		}
+	}
+	return counts, sizesByType, fileSize
+}
+
+// resourceStateBytes derives the bytes direct would store for one resource:
+// GetResourceConfig (typed) → adapter.PrepareState → MarshalIndent with the
+// same prefix/indent direct uses in dstate.SaveState. Falls back to marshaling
+// the typed config when no adapter is registered for the resource type
+// (e.g., a type the direct engine doesn't yet support).
+func resourceStateBytes(b *bundle.Bundle, adapters map[string]*dresources.Adapter, p dyn.Path, resourceType string) ([]byte, error) {
+	cfg, err := b.Config.GetResourceConfig(p.String())
+	if err != nil {
+		return nil, fmt.Errorf("get config: %w", err)
+	}
+
+	target := cfg
+	if adapter, ok := adapters[resourceType]; ok {
+		state, err := adapter.PrepareState(cfg)
+		if err != nil {
+			return nil, fmt.Errorf("prepare state: %w", err)
+		}
+		target = state
+	}
+
+	// dstate.SaveState writes resource state with MarshalIndent using these
+	// exact prefix/indent arguments; matching them here means each resource's
+	// byte length equals len(entry.State) on disk for direct deploys.
+	raw, err := json.MarshalIndent(target, "  ", " ")
+	if err != nil {
+		return nil, fmt.Errorf("marshal: %w", err)
+	}
+	return raw, nil
+}
+
+// extractResourceID returns the resource's ID string from its dyn.Value entry,
+// or "" if not yet set. Each resources.<type>.<name> entry has an "id" field
+// populated post-deploy (via BaseResource.ID).
+func extractResourceID(v dyn.Value) string {
+	idVal, err := dyn.Get(v, "id")
+	if err != nil || idVal.Kind() != dyn.KindString {
+		return ""
+	}
+	return idVal.MustString()
+}
+
+// getAdapters returns adapters initialized for PrepareState. If the bundle
+// already has them initialized (direct engine path), reuse them. Otherwise,
+// build a fresh set with a nil workspace client — PrepareState is a pure
+// transformation that doesn't touch the client.
+func getAdapters(ctx context.Context, b *bundle.Bundle) map[string]*dresources.Adapter {
+	if b.DeploymentBundle.Adapters != nil {
+		return b.DeploymentBundle.Adapters
+	}
+	adapters, err := dresources.InitAll(nil)
+	if err != nil {
+		log.Debugf(ctx, "resources-metadata telemetry: failed to init adapters: %s", err)
+		return nil
+	}
+	return adapters
+}
+
+// resolveDeployEngine returns the effective deploy engine ("direct" or
+// "terraform"). Mirrors cmd/bundle/utils.ResolveEngineSetting but is inlined
+// here to avoid a layering import (bundle/phases must not depend on cmd/).
+func resolveDeployEngine(ctx context.Context, b *bundle.Bundle) string {
+	if b.Config.Bundle.Engine != engine.EngineNotSet {
+		return string(b.Config.Bundle.Engine.ThisOrDefault())
+	}
+	envEngine, _ := engine.FromEnv(ctx)
+	return string(envEngine.ThisOrDefault())
+}
+
+func unionKeys(a map[string]int64, b map[string][]int64) []string {
+	seen := make(map[string]struct{}, len(a)+len(b))
+	for k := range a {
+		seen[k] = struct{}{}
+	}
+	for k := range b {
+		seen[k] = struct{}{}
+	}
+	out := make([]string, 0, len(seen))
+	for k := range seen {
+		out = append(out, k)
+	}
+	return out
+}
+
+func statMax(sortedSizes []int64) int64 {
+	if len(sortedSizes) == 0 {
+		return 0
+	}
+	return sortedSizes[len(sortedSizes)-1]
+}
+
+func statMean(sortedSizes []int64) int64 {
+	if len(sortedSizes) == 0 {
+		return 0
+	}
+	var total int64
+	for _, s := range sortedSizes {
+		total += s
+	}
+	return total / int64(len(sortedSizes))
+}
+
+func statMedian(sortedSizes []int64) int64 {
+	if len(sortedSizes) == 0 {
+		return 0
+	}
+	return sortedSizes[(len(sortedSizes)-1)/2]
+}