diff --git a/acceptance/bundle/dms/multiple-resources/output.txt b/acceptance/bundle/dms/multiple-resources/output.txt index 215a2fb2534..cdb857ee2ba 100644 --- a/acceptance/bundle/dms/multiple-resources/output.txt +++ b/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 diff --git a/acceptance/bundle/dms/multiple-resources/script b/acceptance/bundle/dms/multiple-resources/script index 0f0276ff357..55e21494154 100644 --- a/acceptance/bundle/dms/multiple-resources/script +++ b/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" diff --git a/acceptance/bundle/dms/operation-upload-fails-once/output.txt b/acceptance/bundle/dms/operation-upload-fails-once/output.txt index d17e9ee26fb..72c48dfb96c 100644 --- a/acceptance/bundle/dms/operation-upload-fails-once/output.txt +++ b/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 [ diff --git a/acceptance/bundle/dms/operation-upload-fails-once/script b/acceptance/bundle/dms/operation-upload-fails-once/script index 69dab957164..94db48614ef 100644 --- a/acceptance/bundle/dms/operation-upload-fails-once/script +++ b/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"])') diff --git a/bundle/direct/dstate/state.go b/bundle/direct/dstate/state.go index 384246913f4..326c3b8354c 100644 --- a/bundle/direct/dstate/state.go +++ b/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,6 +574,8 @@ 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 } @@ -569,6 +583,7 @@ func (db *DeploymentState) reset() { 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 diff --git a/bundle/phases/telemetry.go b/bundle/phases/telemetry.go index 2e6296be7f8..5fcd24db299 100644 --- a/bundle/phases/telemetry.go +++ b/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, diff --git a/libs/dms/operation_buffer.go b/libs/dms/operation_buffer.go index 7a77cdf83c8..a74f24903dd 100644 --- a/libs/dms/operation_buffer.go +++ b/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,7 +214,7 @@ 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. @@ -183,20 +222,30 @@ func (b *OperationBuffer) run(ctx context.Context) { } // 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. diff --git a/libs/dms/operation_buffer_test.go b/libs/dms/operation_buffer_test.go index 0e019284360..6aa6aea4943 100644 --- a/libs/dms/operation_buffer_test.go +++ b/libs/dms/operation_buffer_test.go @@ -25,8 +25,9 @@ import ( func queued() *OperationBuffer { return &OperationBuffer{ queue: make(chan string, bufferedOperations), - pending: make(map[string]OperationUpdate), + pending: make(map[string]*bufferedUpdate), latestState: make(map[string]OperationUpdate), + operations: make(map[string]*bufferedUpdate), } } @@ -46,17 +47,20 @@ func TestBufferCoalescesWhileAKeyIsPending(t *testing.T) { s.record("resources.jobs.foo", stateUpdate(t, "v2"), false, nil) assert.Len(t, s.queue, 1) - update, ok := s.take("resources.jobs.foo") + entry, ok := s.take("resources.jobs.foo") require.True(t, ok) - assert.JSONEq(t, `{"state":{"name":"v2"}}`, string(update.State)) + assert.JSONEq(t, `{"state":{"name":"v2"}}`, string(entry.update.State)) + first := entry // Taken means a request has it, and an in-flight request cannot be recalled, so the next // write gets its own slot rather than joining it. s.record("resources.jobs.foo", stateUpdate(t, "v3"), false, nil) assert.Len(t, s.queue, 2) - update, ok = s.take("resources.jobs.foo") + entry, ok = s.take("resources.jobs.foo") require.True(t, ok) - assert.JSONEq(t, `{"state":{"name":"v3"}}`, string(update.State)) + assert.JSONEq(t, `{"state":{"name":"v3"}}`, string(entry.update.State)) + assert.NotSame(t, first, entry) + assert.JSONEq(t, `{"state":{"name":"v2"}}`, string(first.update.State)) } func TestBufferFailsOnOversizedState(t *testing.T) { @@ -64,12 +68,15 @@ func TestBufferFailsOnOversizedState(t *testing.T) { // Failing here says so, where reporting nothing would leave DMS without the resource // and the next plan would create it again. s := queued() + s.stats.StagedOperationCount = 1 s.RecordOperation(t.Context(), "resources.jobs.foo", false, "id-1", json.RawMessage(strings.Repeat("x", maxStateSize+1))) assert.ErrorContains(t, s.Err(), "exceeds the 65536 byte limit") assert.Empty(t, s.queue) assert.Empty(t, s.pending) + assert.Equal(t, 1, s.Stats().OperationRecordingFailureCount) + assert.Zero(t, s.Stats().OperationRecordingPendingCount) } func TestDrainIsIdempotent(t *testing.T) { @@ -93,12 +100,12 @@ func TestRecordFailureKeepsTheStateTheLastWriteRecorded(t *testing.T) { // state), which it must not use: the write already recorded v2, which the service may hold. s.RecordFailure("resources.jobs.foo", "id-old", json.RawMessage(`{"state":{"name":"v1"}}`), errors.New("wait timed out")) - update, ok := s.take("resources.jobs.foo") + entry, ok := s.take("resources.jobs.foo") require.True(t, ok) - assert.JSONEq(t, `{"state":{"name":"v2"}}`, string(update.State), "keeps the state the write recorded") - assert.Equal(t, "id-new", update.ResourceID, "keeps the id the write recorded, not the pre-deploy id") - assert.Equal(t, bundledeployments.OperationStatusOperationStatusFailed, update.Status) - assert.Contains(t, update.ErrorMessage, "wait timed out") + assert.JSONEq(t, `{"state":{"name":"v2"}}`, string(entry.update.State), "keeps the state the write recorded") + assert.Equal(t, "id-new", entry.update.ResourceID, "keeps the id the write recorded, not the pre-deploy id") + assert.Equal(t, bundledeployments.OperationStatusOperationStatusFailed, entry.update.Status) + assert.Contains(t, entry.update.ErrorMessage, "wait timed out") } func TestRecordFailureUsesTheGivenStateWhenNoWriteRecorded(t *testing.T) { @@ -108,10 +115,10 @@ func TestRecordFailureUsesTheGivenStateWhenNoWriteRecorded(t *testing.T) { s.RecordFailure("resources.jobs.foo", "id-1", json.RawMessage(`{"state":{"name":"v1"}}`), errors.New("create failed")) - update, ok := s.take("resources.jobs.foo") + entry, ok := s.take("resources.jobs.foo") require.True(t, ok) - assert.JSONEq(t, `{"state":{"name":"v1"}}`, string(update.State)) - assert.Equal(t, bundledeployments.OperationStatusOperationStatusFailed, update.Status) + assert.JSONEq(t, `{"state":{"name":"v1"}}`, string(entry.update.State)) + assert.Equal(t, bundledeployments.OperationStatusOperationStatusFailed, entry.update.Status) } func TestHistoryQueueDoesNotBlockWhenWriterStops(t *testing.T) { @@ -119,7 +126,7 @@ func TestHistoryQueueDoesNotBlockWhenWriterStops(t *testing.T) { const operationCount = 1000 ctx, cancel := context.WithCancel(t.Context()) defer cancel() - b := StartOperationBuffer(ctx, "dep-1", 1, BufferOptions{BestEffort: true, OperationCount: operationCount}) + b := StartOperationBuffer(ctx, "dep-1", 1, BufferOptions{BestEffort: true, StagedOperationCount: operationCount}) synctest.Wait() cancel() <-b.done diff --git a/libs/telemetry/protos/bundle_deploy.go b/libs/telemetry/protos/bundle_deploy.go index 4464f36bca4..97deb98c7ac 100644 --- a/libs/telemetry/protos/bundle_deploy.go +++ b/libs/telemetry/protos/bundle_deploy.go @@ -38,6 +38,8 @@ type BundleDeployEvent struct { // Per-resource-type metadata (counts and state-size statistics). ResourcesMetadata *BundleResourcesMetadata `json:"resources_metadata,omitempty"` + DeploymentHistory *BundleDeploymentHistory `json:"deployment_history,omitempty"` + // Usage of Python support (PyDABs). Set only when the bundle declares // Python resource loaders or mutators. Pydabs *BundleDeployPydabs `json:"pydabs,omitempty"` @@ -45,6 +47,27 @@ type BundleDeployEvent struct { Experimental *BundleDeployExperimental `json:"experimental,omitempty"` } +// BundleDeploymentHistory measures history recording for one deployment version. +type BundleDeploymentHistory struct { + // Number of resource operations staged for this deployment version. + StagedOperationCount int64 `json:"staged_operation_count"` + // Peak number of queued resource keys, excluding the in-flight upload. + PeakPendingUpdateCount int64 `json:"peak_pending_update_count"` + // Resource operations still queued when Drain ends, excluding the active upload. + QueuedUpdateCountAtTimeout int64 `json:"queued_update_count_at_timeout"` + // Failed update-building or upload attempts. + // Later successful updates do not clear the count. + RecordingAttemptFailureCount int64 `json:"recording_attempt_failure_count"` + // Successful upload attempts. Repeated updates to a resource count separately. + RecordingAttemptSuccessCount int64 `json:"recording_attempt_success_count"` + // Resource operations whose final history update uploaded successfully. + OperationRecordingSuccessCount int64 `json:"operation_recording_success_count"` + // Resource operations whose final history update failed to build or upload. + OperationRecordingFailureCount int64 `json:"operation_recording_failure_count"` + // Staged resource operations with no completed final history update. + OperationRecordingPendingCount int64 `json:"operation_recording_pending_count"` +} + // These metrics are experimental and are often added in an adhoc manner. There // are no guarantees for these metrics and they maybe removed in the future without // any notice.