bundle: add deployment history buffer telemetry

acceptance/bundle/dms/multiple-resources/output.txt

@@ -10,6 +10,19 @@ Created jobs.two
 Files: 2 uploaded, 0 deleted
 Resources: 5 created, 0 changed, 0 deleted, 0 unchanged
 
+=== History telemetry survives version completion and counts all five staged operations
+>>> cat [OUTPUT_DIR]/out.requests.txt
+{
+  "staged_operation_count": 5,
+  "operation_recording_success_count": 5,
+  "operation_recording_failure_count": 0,
+  "operation_recording_pending_count": 0,
+  "queued_update_count_at_timeout": 0,
+  "recording_attempt_failure_count": 0,
+  "recording_attempt_success_count": 5,
+  "queue_size_in_range": true
+}
+
 >>> print_requests.py --dms //versions/1/operations --sort --del-body state --oneline
 {"method": "PATCH", "path": "/api/2.0/bundle/deployments/[DEPLOYMENT_ID]/versions/1/operations/jobs.five", "q": {"update_mask": "state,error_message,resource_id,status"}, "body": {"error_message": "", "resource_id": "[FIVE_ID]", "sequence_id": 0, "status": "OPERATION_STATUS_SUCCEEDED"}}
 {"method": "PATCH", "path": "/api/2.0/bundle/deployments/[DEPLOYMENT_ID]/versions/1/operations/jobs.four", "q": {"update_mask": "state,error_message,resource_id,status"}, "body": {"error_message": "", "resource_id": "[FOUR_ID]", "sequence_id": 0, "status": "OPERATION_STATUS_SUCCEEDED"}}
@@ -172,6 +185,10 @@ Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/dms-multiple-resou
 Files: 0 uploaded, 0 deleted
 Resources: 0 created, 0 changed, 0 deleted, 5 unchanged
 
+=== A deploy with no version omits operation buffer telemetry
+>>> cat [OUTPUT_DIR]/out.requests.txt
+false
+
 >>> print_requests.py --dms //api/2.0/bundle --sort --oneline
 
 === The plan records the deployment_history feature

acceptance/bundle/dms/multiple-resources/script

@@ -2,6 +2,8 @@ envsubst < databricks.yml.tmpl > databricks.yml
 
 title "Deploy several resources: operations are uploaded from background workers, so exactly one is recorded per resource no matter what order the uploads finish in. The serialized state is dropped from the output here; other bundle/dms tests cover it."
 trace $CLI bundle deploy
+title "History telemetry survives version completion and counts all five staged operations"
+trace cat "$OUT_REQUESTS" | jq 'select(.path? == "/telemetry-ext") | .body.protoLogs[] | fromjson | .entry.databricks_cli_log.bundle_deploy_event.deployment_history | {staged_operation_count, operation_recording_success_count, operation_recording_failure_count, operation_recording_pending_count, queued_update_count_at_timeout, recording_attempt_failure_count, recording_attempt_success_count, queue_size_in_range: (.peak_pending_update_count >= 1 and .peak_pending_update_count <= 5)}'
 replace_ids.py
 trace print_requests.py --dms //versions/1/operations --sort --del-body state --oneline
 
@@ -18,6 +20,8 @@ trace print_state.py --no-dms
 
 title "Redeploy with no changes: nothing is applied, so no version is created at all - the deployment's version tracks state changes, like serial"
 trace $CLI bundle deploy
+title "A deploy with no version omits operation buffer telemetry"
+trace cat "$OUT_REQUESTS" | jq 'select(.path? == "/telemetry-ext") | .body.protoLogs[] | fromjson | .entry.databricks_cli_log.bundle_deploy_event | has("deployment_history")'
 trace print_requests.py --dms //api/2.0/bundle --sort --oneline
 
 title "The plan records the deployment_history feature"

acceptance/bundle/dms/operation-upload-fails-once/output.txt

@@ -68,6 +68,18 @@ Warn: Failed to record deployment history: recording operation for resources.sch
 Files: 2 uploaded, 0 deleted
 Resources: 1 created, 0 changed, 1 deleted, 0 unchanged
 
+=== A failed history upload is counted separately from the queued operations
+>>> cat [OUTPUT_DIR]/out.requests.txt
+{
+  "staged_operation_count": 1,
+  "operation_recording_success_count": 1,
+  "operation_recording_failure_count": 0,
+  "operation_recording_pending_count": 0,
+  "recording_attempt_failure_count": 1,
+  "recording_attempt_success_count": 1,
+  "queued_update_count_at_timeout": 0
+}
+
 === The service lists the schema with the failure
 >>> MSYS_NO_PATHCONV=1 [CLI] bundle-deployments list-resources deployments/[DEPLOYMENT_ID] -o json
 [

acceptance/bundle/dms/operation-upload-fails-once/script

@@ -23,6 +23,8 @@ $CLI catalogs create "$other_catalog" &> LOG.catalog
 trap '$CLI catalogs delete "$other_catalog" --force >> LOG.catalog 2>&1' EXIT
 trace update_file.py databricks.yml "catalog_name: main" "catalog_name: ${other_catalog}"
 trace $CLI bundle deploy --auto-approve
+title "A failed history upload is counted separately from the queued operations"
+trace cat "$OUT_REQUESTS" | jq 'select(.path? == "/telemetry-ext") | .body.protoLogs[] | fromjson | .entry.databricks_cli_log.bundle_deploy_event.deployment_history | {staged_operation_count, operation_recording_success_count, operation_recording_failure_count, operation_recording_pending_count, recording_attempt_failure_count, recording_attempt_success_count, queued_update_count_at_timeout}'
 
 title "The service lists the schema with the failure"
 deployment_id=$(MSYS_NO_PATHCONV=1 $CLI workspace get-status "/Workspace/Users/${CURRENT_USER_NAME}/.bundle/dms-upload-fails-once-${UNIQUE_NAME}/default/state/resources.deployment.json" | python3 -c 'import sys, json; print(json.load(sys.stdin)["object_id"])')

bundle/direct/dstate/state.go

@@ -105,7 +105,8 @@ type DeploymentState struct {
 
 	// operationBuffer records each state write with DMS. Nil unless the bundle records deployment
 	// history, in which case InitializeOperationBuffer installs it once the version exists.
-	operationBuffer *dms.OperationBuffer
+	operationBuffer      *dms.OperationBuffer
+	operationBufferStats *dms.OperationBufferStats
 
 	// versionCompleted makes CompleteVersion a no-op after the first call, so a deferred safety-net
 	// completion after an explicit one does nothing.
@@ -184,8 +185,8 @@ func (db *DeploymentState) InitializeOperationBuffer(ctx context.Context, deploy
 	db.mu.Lock()
 	defer db.mu.Unlock()
 	options := dms.BufferOptions{
-		BestEffort:     db.mode == WritingDeploymentHistoryShadow,
-		OperationCount: operationCount,
+		BestEffort:           db.mode == WritingDeploymentHistoryShadow,
+		StagedOperationCount: operationCount,
 	}
 	db.operationBuffer = dms.StartOperationBuffer(ctx, deploymentID, versionID, options)
 	db.DeploymentID = deploymentID
@@ -199,6 +200,17 @@ func (db *DeploymentState) getOperationBuffer() *dms.OperationBuffer {
 	return db.operationBuffer
 }
 
+// OperationBufferStats returns history measurements, including after version completion.
+func (db *DeploymentState) OperationBufferStats() *dms.OperationBufferStats {
+	db.mu.Lock()
+	defer db.mu.Unlock()
+	if db.operationBuffer != nil {
+		stats := db.operationBuffer.Stats()
+		return &stats
+	}
+	return db.operationBufferStats
+}
+
 // RecordingError reports whether recording state writes to the service has failed, so the apply
 // stops touching resources once the service is no longer keeping up. File-backed history modes
 // always return nil so history recording errors do not block the deployment; resource state
@@ -562,13 +574,16 @@ func (db *DeploymentState) reset() {
 	// Telemetry and destroy still need the mode and IDs after version completion.
 	// Clear the recording buffer once CompleteVersion has copied it.
 	if db.versionCompleted {
+		stats := db.operationBuffer.Stats()
+		db.operationBufferStats = &stats
 		db.operationBuffer = nil
 		db.versionCompleted = false
 	}
 }
 
 func (db *DeploymentState) unlockedOpen(ctx context.Context, path string, mode Mode, withRecovery WithRecovery, dmsDeployment OpenDmsArgs) error {
 	db.Path = path
+	db.operationBufferStats = nil
 	db.mode = mode
 	db.DeploymentID = dmsDeployment.DeploymentID
 	db.VersionID = dmsDeployment.LastVersionID

bundle/phases/telemetry.go

@@ -404,6 +404,20 @@ func LogDeployTelemetry(ctx context.Context, b *bundle.Bundle, errMsg string) {
 		uploadFileSizers[i] = f
 	}
 
+	var deploymentHistory *protos.BundleDeploymentHistory
+	if stats := b.DeploymentBundle.StateDB.OperationBufferStats(); stats != nil {
+		deploymentHistory = &protos.BundleDeploymentHistory{
+			StagedOperationCount:           int64(stats.StagedOperationCount),
+			PeakPendingUpdateCount:         int64(stats.PeakPendingUpdateCount),
+			QueuedUpdateCountAtTimeout:     int64(stats.QueuedUpdateCountAtTimeout),
+			RecordingAttemptFailureCount:   int64(stats.RecordingAttemptFailureCount),
+			RecordingAttemptSuccessCount:   int64(stats.RecordingAttemptSuccessCount),
+			OperationRecordingSuccessCount: int64(stats.OperationRecordingSuccessCount),
+			OperationRecordingFailureCount: int64(stats.OperationRecordingFailureCount),
+			OperationRecordingPendingCount: int64(stats.OperationRecordingPendingCount),
+		}
+	}
+
 	telemetry.Log(ctx, protos.DatabricksCliLog{
 		BundleDeployEvent: &protos.BundleDeployEvent{
 			BundleUuid:   bundleUuid,
@@ -430,6 +444,7 @@ func LogDeployTelemetry(ctx context.Context, b *bundle.Bundle, errMsg string) {
 			ResourceDashboardIDs: dashboardIds,
 
 			ResourcesMetadata: collectResourcesMetadata(ctx, b),
+			DeploymentHistory: deploymentHistory,
 
 			Pydabs: pydabsSection,
 

libs/dms/operation_buffer.go

@@ -17,12 +17,37 @@ const bufferedOperations = 10
 
 const historyDrainTimeoutPerOperation = 5 * time.Second
 
+type recordingState uint8
+
+const (
+	recordingPending recordingState = iota
+	recordingSucceeded
+	recordingFailed
+)
+
+type bufferedUpdate struct {
+	update OperationUpdate
+	state  recordingState
+}
+
 // BufferOptions selects optional deployment history shadow recording behavior.
 type BufferOptions struct {
 	BestEffort bool
-	// OperationCount counts staged resource keys. All recorded keys must be staged.
+	// StagedOperationCount counts staged resource keys. All recorded keys must be staged.
 	// File history uses it for queue capacity and the drain timeout.
-	OperationCount int
+	StagedOperationCount int
+}
+
+// OperationBufferStats describes one deployment version's history uploads.
+type OperationBufferStats struct {
+	StagedOperationCount           int
+	PeakPendingUpdateCount         int
+	QueuedUpdateCountAtTimeout     int
+	RecordingAttemptFailureCount   int
+	RecordingAttemptSuccessCount   int
+	OperationRecordingSuccessCount int
+	OperationRecordingFailureCount int
+	OperationRecordingPendingCount int
 }
 
 // OperationBuffer records each state write with DMS for one deployment version, off the apply
@@ -56,7 +81,10 @@ type OperationBuffer struct {
 	mu sync.Mutex
 
 	// pending holds the newest update per resource key, absent once the writer takes it.
-	pending map[string]OperationUpdate
+	pending map[string]*bufferedUpdate
+	stats   OperationBufferStats
+	// Each operation points to its latest update; older uploads retain their own entry.
+	operations map[string]*bufferedUpdate
 
 	// latestState is the newest state-bearing update per resource key, kept for the buffer's
 	// whole life - unlike pending, which the writer removes on take. RecordFailure merges its
@@ -77,7 +105,7 @@ func StartOperationBuffer(ctx context.Context, deploymentID string, versionNum i
 		// Updates coalesce, so each staged key occupies at most one queue slot.
 		// The worker frees that slot before the key can be queued again.
 		// One slot per staged key therefore avoids deployment backpressure.
-		capacity = options.OperationCount
+		capacity = options.StagedOperationCount
 	}
 	b := &OperationBuffer{
 		deploymentID: deploymentID,
@@ -86,10 +114,12 @@ func StartOperationBuffer(ctx context.Context, deploymentID string, versionNum i
 		done:         make(chan struct{}),
 		cancel:       cancel,
 		bestEffort:   options.BestEffort,
-		drainTimeout: time.Duration(options.OperationCount) * historyDrainTimeoutPerOperation,
-		pending:      make(map[string]OperationUpdate),
+		drainTimeout: time.Duration(options.StagedOperationCount) * historyDrainTimeoutPerOperation,
+		pending:      make(map[string]*bufferedUpdate),
 		latestState:  make(map[string]OperationUpdate),
 		sequenceIDs:  make(map[string]int64),
+		stats:        OperationBufferStats{StagedOperationCount: options.StagedOperationCount},
+		operations:   make(map[string]*bufferedUpdate, options.StagedOperationCount),
 	}
 	b.stopQueue = sync.OnceFunc(func() { close(b.queue) })
 	go b.run(ctx)
@@ -101,6 +131,12 @@ func StartOperationBuffer(ctx context.Context, deploymentID string, versionNum i
 func (b *OperationBuffer) RecordOperation(ctx context.Context, resourceKey string, inProgress bool, resourceID string, state json.RawMessage) {
 	update, err := NewStateUpdate(resourceID, state, inProgress)
 	if err != nil {
+		b.mu.Lock()
+		b.stats.RecordingAttemptFailureCount++
+		if !inProgress {
+			b.operations[resourceKey] = &bufferedUpdate{state: recordingFailed}
+		}
+		b.mu.Unlock()
 		b.setErr(fmt.Errorf("recording operation for %s: %w", resourceKey, err))
 		return
 	}
@@ -130,9 +166,12 @@ func (b *OperationBuffer) record(resourceKey string, update OperationUpdate, kee
 	}
 	waiting, queued := b.pending[resourceKey]
 	if queued {
-		update = waiting.Merge(update)
+		update = waiting.update.Merge(update)
 	}
-	b.pending[resourceKey] = update
+	entry := &bufferedUpdate{update: update, state: recordingPending}
+	b.pending[resourceKey] = entry
+	b.operations[resourceKey] = entry
+	b.stats.PeakPendingUpdateCount = max(b.stats.PeakPendingUpdateCount, len(b.pending))
 	b.mu.Unlock()
 
 	// when already queued: the writer reads the map when it gets to the key,
@@ -145,13 +184,13 @@ func (b *OperationBuffer) record(resourceKey string, update OperationUpdate, kee
 }
 
 // take claims the update waiting for resourceKey.
-func (b *OperationBuffer) take(resourceKey string) (OperationUpdate, bool) {
+func (b *OperationBuffer) take(resourceKey string) (*bufferedUpdate, bool) {
 	b.mu.Lock()
 	defer b.mu.Unlock()
 
-	update, ok := b.pending[resourceKey]
+	entry, ok := b.pending[resourceKey]
 	delete(b.pending, resourceKey)
-	return update, ok
+	return entry, ok
 }
 
 func (b *OperationBuffer) run(ctx context.Context) {
@@ -175,28 +214,38 @@ func (b *OperationBuffer) run(ctx context.Context) {
 		if !ok {
 			return
 		}
-		update, ok := b.take(resourceKey)
+		entry, ok := b.take(resourceKey)
 		if !ok {
 			// Unreachable: a key is queued only when nothing was waiting for it. Guard so
 			// a stray key could never write a zero-valued update.
 			continue
 		}
 
 		// Keep going after a failure, so one bad write does not drop everything behind it.
-		if err := b.write(ctx, resourceKey, update); err != nil {
+		if err := b.write(ctx, resourceKey, entry); err != nil {
 			b.setErr(fmt.Errorf("recording operation for %s: %w", resourceKey, err))
 		}
 	}
 }
 
 // write sends one update, at the sequence id the resource is at.
-func (b *OperationBuffer) write(ctx context.Context, key string, update OperationUpdate) error {
+func (b *OperationBuffer) write(ctx context.Context, key string, entry *bufferedUpdate) error {
 	w := cmdctx.WorkspaceClient(ctx)
 	result, err := w.BundleDeployments.UpdateOperation(ctx, bundledeployments.UpdateOperationRequest{
 		Name:       OperationName(b.deploymentID, b.versionNum, key),
-		Operation:  update.operation(b.sequenceIDs[key]),
-		UpdateMask: update.Fields.Mask(),
+		Operation:  entry.update.operation(b.sequenceIDs[key]),
+		UpdateMask: entry.update.Fields.Mask(),
 	})
+	b.mu.Lock()
+	defer b.mu.Unlock()
+	state := recordingSucceeded
+	if err != nil {
+		b.stats.RecordingAttemptFailureCount++
+		state = recordingFailed
+	} else {
+		b.stats.RecordingAttemptSuccessCount++
+	}
+	entry.state = state
 	if err != nil {
 		return err
 	}
@@ -230,10 +279,38 @@ func (b *OperationBuffer) Drain(ctx context.Context) error {
 		return b.Err()
 	case <-ctx.Done():
 		// Deadline or cancellation: stop the upload so deployment can finish promptly.
-		b.setErr(ctx.Err())
+		b.mu.Lock()
+		defer b.mu.Unlock()
+		b.stats.QueuedUpdateCountAtTimeout = len(b.queue)
+		if b.err == nil {
+			b.err = ctx.Err()
+		}
 		b.cancel()
-		return b.Err()
+		return b.err
+	}
+}
+
+// Stats returns a snapshot of the history measurements.
+func (b *OperationBuffer) Stats() OperationBufferStats {
+	b.mu.Lock()
+	defer b.mu.Unlock()
+	stats := b.stats
+	for _, entry := range b.operations {
+		// A recreation's delete update is not the operation's final result.
+		if entry.update.Status == StatusPending {
+			continue
+		}
+		switch entry.state {
+		case recordingPending:
+			continue
+		case recordingSucceeded:
+			stats.OperationRecordingSuccessCount++
+		case recordingFailed:
+			stats.OperationRecordingFailureCount++
+		}
 	}
+	stats.OperationRecordingPendingCount = stats.StagedOperationCount - stats.OperationRecordingSuccessCount - stats.OperationRecordingFailureCount
+	return stats
 }
 
 // setErr keeps the first recording error for Drain to report.