From 55902586c920f827e7d5883f940ea243d2a1a7b6 Mon Sep 17 00:00:00 2001 From: Shreyas Goenka Date: Fri, 24 Jul 2026 01:41:10 +0200 Subject: [PATCH 1/2] bundle: record and read deployment state via DMS with server-generated IDs Wire the direct engine into the Deployment Metadata Service (DMS) so that a `record_deployment_history`-enabled bundle records each deploy/destroy as a version and can read its resource state back from DMS. The deployment ID is now assigned by the server: the first deploy calls CreateDeployment with an empty ID, reads the assigned ID back from the response, and persists it in the direct-engine state header (Header.DeploymentID). Later deploys pass the stored ID back, so a bundle maps one-to-one to a DMS deployment even after the local cache is deleted (the ID rides along in the workspace-synced state file). - libs/dms: Recorder creates the deployment (server-assigned ID) + version, heartbeats the lease, completes it, and deletes the deployment on destroy. - bundle/direct: operationRecorder reports each applied resource operation; the wire resource_key drops the CLI-internal "resources." prefix. - bundle/direct/dstate: Open takes a DMS client and overlays DMS resource state when DMS holds a successful version; deployment ID persisted in the header. - bundle/phases: create the version after plan approval, complete it under the lock, record operations during apply. - libs/testserver: stateful fake DMS (deployments/versions/operations/resources) with server-generated IDs; acceptance test covers deploy, cache-loss redeploy, and destroy. Co-authored-by: Isaac --- acceptance/bundle/dms/record/databricks.yml | 10 + acceptance/bundle/dms/record/out.test.toml | 3 + acceptance/bundle/dms/record/output.txt | 140 +++++++++++ acceptance/bundle/dms/record/script | 15 ++ acceptance/bundle/dms/test.toml | 13 + bundle/configsync/diff.go | 2 +- bundle/configsync/variables.go | 2 +- bundle/direct/bind.go | 12 +- bundle/direct/bundle_apply.go | 13 + bundle/direct/dstate/dms.go | 104 ++++++++ bundle/direct/dstate/state.go | 43 +++- bundle/direct/dstate/state_test.go | 45 +++- bundle/direct/oprecorder.go | 107 ++++++++ bundle/direct/oprecorder_test.go | 84 +++++++ bundle/direct/pkg.go | 5 + bundle/phases/deploy.go | 32 +++ bundle/phases/destroy.go | 23 ++ bundle/phases/dms.go | 37 +++ cmd/bundle/generate/dashboard.go | 2 +- cmd/bundle/generate/genie_space.go | 2 +- cmd/bundle/utils/process.go | 12 +- libs/dms/recorder.go | 255 ++++++++++++++++++++ libs/dms/recorder_test.go | 165 +++++++++++++ libs/testserver/bundle.go | 228 +++++++++++++++++ libs/testserver/fake_workspace.go | 5 + libs/testserver/handlers.go | 29 +++ 26 files changed, 1363 insertions(+), 25 deletions(-) create mode 100644 acceptance/bundle/dms/record/databricks.yml create mode 100644 acceptance/bundle/dms/record/out.test.toml create mode 100644 acceptance/bundle/dms/record/output.txt create mode 100644 acceptance/bundle/dms/record/script create mode 100644 acceptance/bundle/dms/test.toml create mode 100644 bundle/direct/dstate/dms.go create mode 100644 bundle/direct/oprecorder.go create mode 100644 bundle/direct/oprecorder_test.go create mode 100644 bundle/phases/dms.go create mode 100644 libs/dms/recorder.go create mode 100644 libs/dms/recorder_test.go create mode 100644 libs/testserver/bundle.go diff --git a/acceptance/bundle/dms/record/databricks.yml b/acceptance/bundle/dms/record/databricks.yml new file mode 100644 index 00000000000..b20e6274310 --- /dev/null +++ b/acceptance/bundle/dms/record/databricks.yml @@ -0,0 +1,10 @@ +bundle: + name: dms-record + +experimental: + record_deployment_history: true + +resources: + jobs: + foo: + name: foo diff --git a/acceptance/bundle/dms/record/out.test.toml b/acceptance/bundle/dms/record/out.test.toml new file mode 100644 index 00000000000..e90b6d5d1ba --- /dev/null +++ b/acceptance/bundle/dms/record/out.test.toml @@ -0,0 +1,3 @@ +Local = true +Cloud = false +EnvMatrix.DATABRICKS_BUNDLE_ENGINE = ["direct"] diff --git a/acceptance/bundle/dms/record/output.txt b/acceptance/bundle/dms/record/output.txt new file mode 100644 index 00000000000..5c0317f38cc --- /dev/null +++ b/acceptance/bundle/dms/record/output.txt @@ -0,0 +1,140 @@ + +=== Deploy: the server assigns the deployment ID, and a version + create operation are recorded +>>> [CLI] bundle deploy +Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/dms-record/default/files... +Deploying resources... +Updating deployment state... +Deployment complete! + +>>> print_requests.py //api/2.0/bundle --sort +{ + "method": "POST", + "path": "/api/2.0/bundle/deployments", + "body": { + "target_name": "default" + } +} +{ + "method": "POST", + "path": "/api/2.0/bundle/deployments/[UUID]/versions", + "q": { + "version_id": "1" + }, + "body": { + "cli_version": "[CLI_VERSION]", + "target_name": "default", + "version_type": "VERSION_TYPE_DEPLOY" + } +} +{ + "method": "POST", + "path": "/api/2.0/bundle/deployments/[UUID]/versions/1/complete", + "body": { + "completion_reason": "VERSION_COMPLETE_SUCCESS" + } +} +{ + "method": "POST", + "path": "/api/2.0/bundle/deployments/[UUID]/versions/1/operations", + "q": { + "resource_key": "jobs.foo" + }, + "body": { + "action_type": "OPERATION_ACTION_TYPE_CREATE", + "resource_id": "[NUMID]", + "resource_key": "jobs.foo", + "state": { + "deployment": { + "kind": "BUNDLE", + "metadata_file_path": "/Workspace/Users/[USERNAME]/.bundle/dms-record/default/state/metadata.json" + }, + "edit_mode": "UI_LOCKED", + "format": "MULTI_TASK", + "max_concurrent_runs": 1, + "name": "foo", + "queue": { + "enabled": true + } + }, + "status": "OPERATION_STATUS_SUCCEEDED" + } +} + +=== The server-assigned deployment ID is persisted in the local state file +>>> jq .deployment_id .databricks/bundle/default/resources.json +"[UUID]" + +=== Redeploy after deleting the local cache: the deployment ID is recovered from remote state, the same deployment is reused, and the version increments (no new CreateDeployment) +>>> [CLI] bundle deploy +Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/dms-record/default/files... +Deploying resources... +Updating deployment state... +Deployment complete! + +>>> print_requests.py //api/2.0/bundle --sort +{ + "method": "POST", + "path": "/api/2.0/bundle/deployments/[UUID]/versions", + "q": { + "version_id": "2" + }, + "body": { + "cli_version": "[CLI_VERSION]", + "target_name": "default", + "version_type": "VERSION_TYPE_DEPLOY" + } +} +{ + "method": "POST", + "path": "/api/2.0/bundle/deployments/[UUID]/versions/2/complete", + "body": { + "completion_reason": "VERSION_COMPLETE_SUCCESS" + } +} + +=== Destroy: a destroy version and delete operation are recorded, then the deployment is deleted +>>> [CLI] bundle destroy --auto-approve +The following resources will be deleted: + delete resources.jobs.foo + +All files and directories at the following location will be deleted: /Workspace/Users/[USERNAME]/.bundle/dms-record/default + +Deleting files... +Destroy complete! + +>>> print_requests.py //api/2.0/bundle --sort +{ + "method": "DELETE", + "path": "/api/2.0/bundle/deployments/[UUID]" +} +{ + "method": "POST", + "path": "/api/2.0/bundle/deployments/[UUID]/versions", + "q": { + "version_id": "3" + }, + "body": { + "cli_version": "[CLI_VERSION]", + "target_name": "default", + "version_type": "VERSION_TYPE_DESTROY" + } +} +{ + "method": "POST", + "path": "/api/2.0/bundle/deployments/[UUID]/versions/3/complete", + "body": { + "completion_reason": "VERSION_COMPLETE_SUCCESS" + } +} +{ + "method": "POST", + "path": "/api/2.0/bundle/deployments/[UUID]/versions/3/operations", + "q": { + "resource_key": "jobs.foo" + }, + "body": { + "action_type": "OPERATION_ACTION_TYPE_DELETE", + "resource_key": "jobs.foo", + "status": "OPERATION_STATUS_SUCCEEDED" + } +} diff --git a/acceptance/bundle/dms/record/script b/acceptance/bundle/dms/record/script new file mode 100644 index 00000000000..ab59d38afb4 --- /dev/null +++ b/acceptance/bundle/dms/record/script @@ -0,0 +1,15 @@ +title "Deploy: the server assigns the deployment ID, and a version + create operation are recorded" +trace $CLI bundle deploy +trace print_requests.py //api/2.0/bundle --sort + +title "The server-assigned deployment ID is persisted in the local state file" +trace jq .deployment_id .databricks/bundle/default/resources.json + +title "Redeploy after deleting the local cache: the deployment ID is recovered from remote state, the same deployment is reused, and the version increments (no new CreateDeployment)" +rm -rf .databricks +trace $CLI bundle deploy +trace print_requests.py //api/2.0/bundle --sort + +title "Destroy: a destroy version and delete operation are recorded, then the deployment is deleted" +trace $CLI bundle destroy --auto-approve +trace print_requests.py //api/2.0/bundle --sort diff --git a/acceptance/bundle/dms/test.toml b/acceptance/bundle/dms/test.toml new file mode 100644 index 00000000000..24ce9756629 --- /dev/null +++ b/acceptance/bundle/dms/test.toml @@ -0,0 +1,13 @@ +Local = true +Cloud = false + +# Deployment Metadata Service (DMS) recording is only meaningful in the direct +# engine, where the deployment ID is stored in and read from the direct-engine +# state. +EnvMatrix.DATABRICKS_BUNDLE_ENGINE = ["direct"] + +RecordRequests = true + +Ignore = [ + '.databricks', +] diff --git a/bundle/configsync/diff.go b/bundle/configsync/diff.go index ea45903508b..ca5b2c9410b 100644 --- a/bundle/configsync/diff.go +++ b/bundle/configsync/diff.go @@ -149,7 +149,7 @@ func OpenDeploymentState(ctx context.Context, b *bundle.Bundle, engine engine.En deployBundle := &direct.DeploymentBundle{} _, statePath := b.StateFilenameConfigSnapshot(ctx) - if err := deployBundle.StateDB.Open(ctx, statePath, dstate.WithRecovery(true), dstate.WithWrite(false)); err != nil { + if err := deployBundle.StateDB.Open(ctx, statePath, dstate.WithRecovery(true), dstate.WithWrite(false), nil); err != nil { return nil, fmt.Errorf("failed to open state: %w", err) } return deployBundle, nil diff --git a/bundle/configsync/variables.go b/bundle/configsync/variables.go index 055a47dc934..433b607a037 100644 --- a/bundle/configsync/variables.go +++ b/bundle/configsync/variables.go @@ -147,7 +147,7 @@ func resourceIDLookup(ctx context.Context, b *bundle.Bundle) func(string) string } _, statePath := b.StateFilenameConfigSnapshot(ctx) db := &dstate.DeploymentState{} - if err := db.Open(ctx, statePath, dstate.WithRecovery(false), dstate.WithWrite(false)); err != nil { + if err := db.Open(ctx, statePath, dstate.WithRecovery(false), dstate.WithWrite(false), nil); err != nil { log.Debugf(ctx, "variable restoration: failed to open state DB at %s: %v", statePath, err) return nil } diff --git a/bundle/direct/bind.go b/bundle/direct/bind.go index 9760ce95666..ec910b2734e 100644 --- a/bundle/direct/bind.go +++ b/bundle/direct/bind.go @@ -62,7 +62,7 @@ type BindResult struct { func (b *DeploymentBundle) Bind(ctx context.Context, client *databricks.WorkspaceClient, configRoot *config.Root, statePath, resourceKey, resourceID string) (*BindResult, error) { // Check if the resource is already managed (bound to a different ID) var checkStateDB dstate.DeploymentState - if err := checkStateDB.Open(ctx, statePath, dstate.WithRecovery(true), dstate.WithWrite(false)); err == nil { + if err := checkStateDB.Open(ctx, statePath, dstate.WithRecovery(true), dstate.WithWrite(false), nil); err == nil { existingID := checkStateDB.GetResourceID(resourceKey) if _, err := checkStateDB.Finalize(ctx); err != nil { log.Warnf(ctx, "failed to finalize state: %v", err) @@ -86,7 +86,7 @@ func (b *DeploymentBundle) Bind(ctx context.Context, client *databricks.Workspac } // Open temp state - err := b.StateDB.Open(ctx, tmpStatePath, dstate.WithRecovery(false), dstate.WithWrite(true)) + err := b.StateDB.Open(ctx, tmpStatePath, dstate.WithRecovery(false), dstate.WithWrite(true), nil) if err != nil { os.Remove(tmpStatePath) return nil, err @@ -109,7 +109,7 @@ func (b *DeploymentBundle) Bind(ctx context.Context, client *databricks.Workspac log.Infof(ctx, "Bound %s to id=%s (in temp state)", resourceKey, resourceID) // First plan + update: populate state with resolved config - err = b.StateDB.Open(ctx, tmpStatePath, dstate.WithRecovery(true), dstate.WithWrite(false)) + err = b.StateDB.Open(ctx, tmpStatePath, dstate.WithRecovery(true), dstate.WithWrite(false), nil) if err != nil { os.Remove(tmpStatePath) return nil, err @@ -145,7 +145,7 @@ func (b *DeploymentBundle) Bind(ctx context.Context, client *databricks.Workspac } } - err = b.StateDB.Open(ctx, tmpStatePath, dstate.WithRecovery(true), dstate.WithWrite(true)) + err = b.StateDB.Open(ctx, tmpStatePath, dstate.WithRecovery(true), dstate.WithWrite(true), nil) if err != nil { os.Remove(tmpStatePath) return nil, err @@ -165,7 +165,7 @@ func (b *DeploymentBundle) Bind(ctx context.Context, client *databricks.Workspac } // Second plan: this is the plan to present to the user (change between remote resource and config) - err = b.StateDB.Open(ctx, tmpStatePath, dstate.WithRecovery(true), dstate.WithWrite(false)) + err = b.StateDB.Open(ctx, tmpStatePath, dstate.WithRecovery(true), dstate.WithWrite(false), nil) if err != nil { os.Remove(tmpStatePath) return nil, err @@ -215,7 +215,7 @@ func (result *BindResult) Cancel() { // Unbind removes a resource from direct engine state without deleting // the workspace resource. Also removes associated permissions/grants entries. func (b *DeploymentBundle) Unbind(ctx context.Context, statePath, resourceKey string) error { - err := b.StateDB.Open(ctx, statePath, dstate.WithRecovery(true), dstate.WithWrite(true)) + err := b.StateDB.Open(ctx, statePath, dstate.WithRecovery(true), dstate.WithWrite(true), nil) if err != nil { return err } diff --git a/bundle/direct/bundle_apply.go b/bundle/direct/bundle_apply.go index c4178c4e601..afef2367e5b 100644 --- a/bundle/direct/bundle_apply.go +++ b/bundle/direct/bundle_apply.go @@ -88,6 +88,11 @@ func (b *DeploymentBundle) Apply(ctx context.Context, client *databricks.Workspa logdiag.LogError(ctx, fmt.Errorf("%s: %w", errorPrefix, err)) return false } + // Record the delete with DMS. State is nil: the resource is gone. + if err := b.recordOperation(ctx, resourceKey, action, "", nil); err != nil { + logdiag.LogError(ctx, fmt.Errorf("%s: %w", errorPrefix, err)) + return false + } return true } @@ -116,6 +121,14 @@ func (b *DeploymentBundle) Apply(ctx context.Context, client *databricks.Workspa logdiag.LogError(ctx, fmt.Errorf("%s: %w", errorPrefix, err)) return false } + + // Record the operation with DMS. The resource ID and applied config + // (sv.Value) come from the write just performed; GetResourceID reads + // the ID assigned by Deploy. + if err := b.recordOperation(ctx, resourceKey, action, b.StateDB.GetResourceID(resourceKey), sv.Value); err != nil { + logdiag.LogError(ctx, fmt.Errorf("%s: %w", errorPrefix, err)) + return false + } } // TODO: Note, we only really need remote state if there are remote references. diff --git a/bundle/direct/dstate/dms.go b/bundle/direct/dstate/dms.go new file mode 100644 index 00000000000..1d19d1fe214 --- /dev/null +++ b/bundle/direct/dstate/dms.go @@ -0,0 +1,104 @@ +package dstate + +import ( + "context" + "encoding/json" + "errors" + "fmt" + + "github.com/databricks/databricks-sdk-go/apierr" + "github.com/databricks/databricks-sdk-go/service/bundledeployments" +) + +// overlayDMSState replaces the file-derived resource state with the state +// recorded in the deployment metadata service (DMS), when DMS owns this +// deployment. Once DMS is authoritative its resource set is trusted even when +// empty (a successful deploy with no resources); the file's resources are only +// used when DMS has no successful version, or when the user opts out of +// recording deployment history. The caller holds db.mu and has already +// populated db.Data from the file, including the DeploymentID. +func (db *DeploymentState) overlayDMSState(ctx context.Context, client bundledeployments.BundleDeploymentsInterface) error { + authoritative, err := deploymentHasSuccessfulVersion(ctx, client, db.Data.DeploymentID) + if err != nil { + return err + } + if !authoritative { + // DMS has no completed version for this deployment: a prior direct deploy + // that has not yet successfully recorded to DMS. Keep the file state. + return nil + } + + resources, err := fetchDeploymentResources(ctx, client, db.Data.DeploymentID) + if err != nil { + return err + } + + db.Data.State = resources + db.stateIDs = make(map[string]string, len(resources)) + for key, entry := range resources { + db.stateIDs[key] = entry.ID + } + return nil +} + +// deploymentHasSuccessfulVersion reports whether DMS holds a successfully +// completed version for the deployment. It is the signal that DMS owns the +// state: if the deployment was never recorded to DMS, or its initial DMS deploy +// did not complete successfully, DMS state is absent or partial and Open keeps +// the local file's resources instead. +func deploymentHasSuccessfulVersion(ctx context.Context, client bundledeployments.BundleDeploymentsInterface, deploymentID string) (bool, error) { + // Versions are listed newest-first and fetched page by page, and we stop at + // the first successful one, so a deployment with a long version history does + // not require reading the whole list (typically just the first page). + it := client.ListVersions(ctx, bundledeployments.ListVersionsRequest{ + Parent: "deployments/" + deploymentID, + }) + for it.HasNext(ctx) { + v, err := it.Next(ctx) + if err != nil { + // A deployment that was never recorded to DMS is not an error here: it + // just means DMS is not (yet) the source of truth. + if errors.Is(err, apierr.ErrNotFound) { + return false, nil + } + return false, fmt.Errorf("listing versions from deployment metadata service: %w", err) + } + if v.Status == bundledeployments.VersionStatusVersionStatusCompleted && + v.CompletionReason == bundledeployments.VersionCompleteVersionCompleteSuccess { + return true, nil + } + } + return false, nil +} + +// fetchDeploymentResources lists every resource recorded for the deployment in +// DMS and maps them into state entries keyed by the fully-qualified resource key. +func fetchDeploymentResources(ctx context.Context, client bundledeployments.BundleDeploymentsInterface, deploymentID string) (map[string]ResourceEntry, error) { + it := client.ListResources(ctx, bundledeployments.ListResourcesRequest{ + Parent: "deployments/" + deploymentID, + }) + + out := make(map[string]ResourceEntry) + for it.HasNext(ctx) { + res, err := it.Next(ctx) + if err != nil { + return nil, fmt.Errorf("listing resources from deployment metadata service: %w", err) + } + + // DMS reports resource keys without the "resources." prefix (e.g. + // "jobs.foo"), but the state DB keys are fully qualified + // ("resources.jobs.foo"), so prepend it here. + key := "resources." + res.ResourceKey + + var state json.RawMessage + if res.State != nil { + state = *res.State + } + + out[key] = ResourceEntry{ + ID: res.ResourceId, + State: state, + } + } + return out, nil +} diff --git a/bundle/direct/dstate/state.go b/bundle/direct/dstate/state.go index f6c8fc8ba3c..64fc050bdc0 100644 --- a/bundle/direct/dstate/state.go +++ b/bundle/direct/dstate/state.go @@ -19,6 +19,7 @@ import ( "github.com/databricks/cli/libs/dyn" "github.com/databricks/cli/libs/log" "github.com/databricks/cli/libs/structs/structwalk" + "github.com/databricks/databricks-sdk-go/service/bundledeployments" "github.com/google/uuid" ) @@ -80,6 +81,13 @@ type Header struct { Lineage string `json:"lineage"` Serial int `json:"serial"` + // DeploymentID is the ID the deployment metadata service (DMS) assigned to + // this deployment. Unlike Lineage (a locally generated identifier for the + // state file), it is minted server-side by CreateDeployment and stored here so + // later deploys can find the same DMS deployment record and read its state. + // Empty/omitted until the bundle first records to DMS. + DeploymentID string `json:"deployment_id,omitempty"` + // Features maps each feature flag this state depends on to a (currently empty) // value. This CLI writes no features; it only reads the field to detect a state // that depends on features it lacks and refuse it (see migrateState). It is a @@ -209,6 +217,25 @@ func (db *DeploymentState) GetOrInitLineage() string { return db.Data.Lineage } +// GetDeploymentID returns the DMS deployment ID recorded in the state, or an +// empty string if this bundle has not yet recorded a deployment to DMS. +func (db *DeploymentState) GetDeploymentID() string { + db.mu.Lock() + defer db.mu.Unlock() + return db.Data.DeploymentID +} + +// SetDeploymentID stores the DMS-assigned deployment ID in the in-memory state +// header. It is set during deploy, after CreateDeployment returns the +// server-generated ID, and persisted to the state file by Finalize. Storing it +// on db.Data (not the WAL header, which is written before the ID is known) +// means the subsequent state write carries it forward. +func (db *DeploymentState) SetDeploymentID(id string) { + db.mu.Lock() + defer db.mu.Unlock() + db.Data.DeploymentID = id +} + type ( // If true, then Open reads the WAL and merges it in the state. If false, and WAL is present, Open returns an error. WithRecovery bool @@ -218,7 +245,15 @@ type ( WithWrite bool ) -func (db *DeploymentState) Open(ctx context.Context, path string, withRecovery WithRecovery, withWrite WithWrite) error { +// Open reads the deployment state from disk (and recovers the WAL when +// withRecovery is set). When dmsClient is non-nil, the deployment metadata +// service is the source of truth for resource state: if DMS holds a +// successfully completed version for this deployment, the resources read from +// the file are replaced with the ones recorded in DMS. The local identity +// (lineage, serial, and deployment ID) always comes from the file, since that +// is what the write path increments and carries forward. A nil dmsClient keeps +// the behavior file-only. +func (db *DeploymentState) Open(ctx context.Context, path string, withRecovery WithRecovery, withWrite WithWrite, dmsClient bundledeployments.BundleDeploymentsInterface) error { db.mu.Lock() defer db.mu.Unlock() @@ -266,6 +301,12 @@ func (db *DeploymentState) Open(ctx context.Context, path string, withRecovery W return fmt.Errorf("migrating state %s: %w", path, err) } + if dmsClient != nil && db.Data.DeploymentID != "" { + if err := db.overlayDMSState(ctx, dmsClient); err != nil { + return err + } + } + if withWrite { if err := os.MkdirAll(filepath.Dir(walPath), 0o755); err != nil { return fmt.Errorf("failed to create state directory: %w", err) diff --git a/bundle/direct/dstate/state_test.go b/bundle/direct/dstate/state_test.go index 11589944472..e95ad1b0224 100644 --- a/bundle/direct/dstate/state_test.go +++ b/bundle/direct/dstate/state_test.go @@ -20,24 +20,43 @@ func TestOpenSaveFinalizeRoundTrip(t *testing.T) { path := filepath.Join(t.TempDir(), "state.json") var db DeploymentState - require.NoError(t, db.Open(t.Context(), path, WithRecovery(true), WithWrite(true))) + require.NoError(t, db.Open(t.Context(), path, WithRecovery(true), WithWrite(true), nil)) require.NoError(t, db.SaveState("jobs.my_job", "123", map[string]string{"key": "val"}, nil)) mustFinalize(t, &db) // Re-open and verify persisted data. var db2 DeploymentState - require.NoError(t, db2.Open(t.Context(), path, WithRecovery(false), WithWrite(false))) + require.NoError(t, db2.Open(t.Context(), path, WithRecovery(false), WithWrite(false), nil)) assert.Equal(t, 1, db2.Data.Serial) assert.Equal(t, "123", db2.GetResourceID("jobs.my_job")) mustFinalize(t, &db2) } +func TestDeploymentIDPersistsAcrossOpen(t *testing.T) { + path := filepath.Join(t.TempDir(), "state.json") + + var db DeploymentState + require.NoError(t, db.Open(t.Context(), path, WithRecovery(true), WithWrite(true), nil)) + assert.Empty(t, db.GetDeploymentID()) + + // The deployment ID is set during deploy (after CreateDeployment) and + // persisted by Finalize even though it is not part of the WAL header. + db.SetDeploymentID("server-assigned-id") + require.NoError(t, db.SaveState("jobs.my_job", "123", map[string]string{}, nil)) + mustFinalize(t, &db) + + var reopened DeploymentState + require.NoError(t, reopened.Open(t.Context(), path, WithRecovery(false), WithWrite(false), nil)) + assert.Equal(t, "server-assigned-id", reopened.GetDeploymentID()) + mustFinalize(t, &reopened) +} + func TestFinalizeWithNoEntriesDoesNotWriteStateFile(t *testing.T) { path := filepath.Join(t.TempDir(), "state.json") var db DeploymentState - require.NoError(t, db.Open(t.Context(), path, WithRecovery(true), WithWrite(true))) + require.NoError(t, db.Open(t.Context(), path, WithRecovery(true), WithWrite(true), nil)) mustFinalize(t, &db) _, err := os.Stat(path) @@ -93,10 +112,10 @@ func TestPanicOnDoubleOpen(t *testing.T) { path := filepath.Join(t.TempDir(), "state.json") var db DeploymentState - require.NoError(t, db.Open(t.Context(), path, WithRecovery(true), WithWrite(true))) + require.NoError(t, db.Open(t.Context(), path, WithRecovery(true), WithWrite(true), nil)) assert.Panics(t, func() { - _ = db.Open(t.Context(), path, WithRecovery(true), WithWrite(true)) + _ = db.Open(t.Context(), path, WithRecovery(true), WithWrite(true), nil) }) mustFinalize(t, &db) } @@ -107,12 +126,12 @@ func TestHeaderOnlyWALRecoveryDoesNotAdvanceSerial(t *testing.T) { // Commit serial 1 with one resource. var db DeploymentState - require.NoError(t, db.Open(t.Context(), path, WithRecovery(true), WithWrite(true))) + require.NoError(t, db.Open(t.Context(), path, WithRecovery(true), WithWrite(true), nil)) require.NoError(t, db.SaveState("jobs.my_job", "123", map[string]string{}, nil)) mustFinalize(t, &db) var committed DeploymentState - require.NoError(t, committed.Open(t.Context(), path, WithRecovery(false), WithWrite(false))) + require.NoError(t, committed.Open(t.Context(), path, WithRecovery(false), WithWrite(false), nil)) lineage := committed.Data.Lineage require.Equal(t, 1, committed.Data.Serial) mustFinalize(t, &committed) @@ -128,7 +147,7 @@ func TestHeaderOnlyWALRecoveryDoesNotAdvanceSerial(t *testing.T) { require.NoError(t, os.WriteFile(walPath, append(headerLine, '\n'), 0o600)) var recovered DeploymentState - require.NoError(t, recovered.Open(t.Context(), path, WithRecovery(true), WithWrite(false))) + require.NoError(t, recovered.Open(t.Context(), path, WithRecovery(true), WithWrite(false), nil)) assert.Equal(t, 1, recovered.Data.Serial) assert.Equal(t, "123", recovered.GetResourceID("jobs.my_job")) assert.NoFileExists(t, walPath) @@ -171,17 +190,17 @@ func TestDeleteState(t *testing.T) { path := filepath.Join(t.TempDir(), "state.json") var db DeploymentState - require.NoError(t, db.Open(t.Context(), path, WithRecovery(true), WithWrite(true))) + require.NoError(t, db.Open(t.Context(), path, WithRecovery(true), WithWrite(true), nil)) require.NoError(t, db.SaveState("jobs.my_job", "123", map[string]string{}, nil)) mustFinalize(t, &db) var db2 DeploymentState - require.NoError(t, db2.Open(t.Context(), path, WithRecovery(true), WithWrite(true))) + require.NoError(t, db2.Open(t.Context(), path, WithRecovery(true), WithWrite(true), nil)) require.NoError(t, db2.DeleteState("jobs.my_job")) mustFinalize(t, &db2) var db3 DeploymentState - require.NoError(t, db3.Open(t.Context(), path, WithRecovery(false), WithWrite(false))) + require.NoError(t, db3.Open(t.Context(), path, WithRecovery(false), WithWrite(false), nil)) assert.Equal(t, 2, db3.Data.Serial) assert.Empty(t, db3.GetResourceID("jobs.my_job")) mustFinalize(t, &db3) @@ -193,7 +212,7 @@ func TestGetOrInitLineageReadableBeforeWriteAndPersisted(t *testing.T) { // Fresh state opened read-only, as the deploy does before planning: no // lineage yet. var db DeploymentState - require.NoError(t, db.Open(t.Context(), path, WithRecovery(true), WithWrite(false))) + require.NoError(t, db.Open(t.Context(), path, WithRecovery(true), WithWrite(false), nil)) require.Empty(t, db.Data.Lineage) // GetOrInitLineage initializes the lineage and makes it readable before any @@ -210,7 +229,7 @@ func TestGetOrInitLineageReadableBeforeWriteAndPersisted(t *testing.T) { // Re-open: the persisted lineage matches the one read before the write. var reopened DeploymentState - require.NoError(t, reopened.Open(t.Context(), path, WithRecovery(false), WithWrite(false))) + require.NoError(t, reopened.Open(t.Context(), path, WithRecovery(false), WithWrite(false), nil)) assert.Equal(t, lineage, reopened.Data.Lineage) mustFinalize(t, &reopened) } diff --git a/bundle/direct/oprecorder.go b/bundle/direct/oprecorder.go new file mode 100644 index 00000000000..467f8ac648c --- /dev/null +++ b/bundle/direct/oprecorder.go @@ -0,0 +1,107 @@ +package direct + +import ( + "context" + "encoding/json" + "fmt" + "strings" + + "github.com/databricks/cli/bundle/deployplan" + "github.com/databricks/databricks-sdk-go/service/bundledeployments" +) + +// opRecorder records a resource operation with the deployment metadata service +// (DMS) after it has been applied to the workspace. state is the serialized +// local config after the operation and must be nil for delete operations. +type opRecorder interface { + record(ctx context.Context, resourceKey string, action deployplan.ActionType, resourceID string, state any) error +} + +// recordOperation reports an applied resource operation to DMS. It is a no-op +// unless the bundle opted into recording deployment history (OpRec is set). +// state is the serialized local config after the operation and must be nil for +// delete operations. +func (b *DeploymentBundle) recordOperation(ctx context.Context, resourceKey string, action deployplan.ActionType, resourceID string, state any) error { + if b.OpRec == nil { + return nil + } + return b.OpRec.record(ctx, resourceKey, action, resourceID, state) +} + +// operationRecorder records operations via the DMS CreateOperation API. +type operationRecorder struct { + client bundledeployments.BundleDeploymentsInterface + // parent is the version the operations are recorded under, formatted as + // "deployments/{deployment_id}/versions/{version_id}". + parent string +} + +// NewOperationRecorder returns an opRecorder backed by the DMS CreateOperation +// API. deploymentID and version identify the deployment version assigned by DMS +// that the operations are recorded under. +func NewOperationRecorder(client bundledeployments.BundleDeploymentsInterface, deploymentID string, version int64) opRecorder { + return &operationRecorder{ + client: client, + parent: fmt.Sprintf("deployments/%s/versions/%d", deploymentID, version), + } +} + +func (r *operationRecorder) record(ctx context.Context, resourceKey string, action deployplan.ActionType, resourceID string, state any) error { + actionType, err := deployActionToSDK(action) + if err != nil { + return err + } + + // DMS resource keys are unprefixed (e.g. "jobs.foo"), while the CLI's state + // keys carry a leading "resources." (e.g. "resources.jobs.foo"). Strip it on + // the way out; the read path re-adds it (see dstate.fetchDeploymentResources). + dmsKey := strings.TrimPrefix(resourceKey, "resources.") + + op := bundledeployments.Operation{ + ActionType: actionType, + ResourceId: resourceID, + ResourceKey: dmsKey, + Status: bundledeployments.OperationStatusOperationStatusSucceeded, + } + + // The DMS Operation.State field carries the serialized config so the backend + // can serve it as resource state. It is intentionally left unset for delete, + // where the resource no longer exists. + if state != nil { + raw, err := json.Marshal(state) + if err != nil { + return fmt.Errorf("serializing state: %w", err) + } + msg := json.RawMessage(raw) + op.State = &msg + } + + _, err = r.client.CreateOperation(ctx, bundledeployments.CreateOperationRequest{ + Parent: r.parent, + ResourceKey: dmsKey, + Operation: op, + }) + return err +} + +// deployActionToSDK maps a deployplan action to its DMS operation action type. +// Only actions that mutate a resource are recordable; Skip and Undefined never +// reach a recorder and are rejected rather than silently coerced. +func deployActionToSDK(a deployplan.ActionType) (bundledeployments.OperationActionType, error) { + switch a { + case deployplan.Create: + return bundledeployments.OperationActionTypeOperationActionTypeCreate, nil + case deployplan.Update: + return bundledeployments.OperationActionTypeOperationActionTypeUpdate, nil + case deployplan.UpdateWithID: + return bundledeployments.OperationActionTypeOperationActionTypeUpdateWithId, nil + case deployplan.Recreate: + return bundledeployments.OperationActionTypeOperationActionTypeRecreate, nil + case deployplan.Resize: + return bundledeployments.OperationActionTypeOperationActionTypeResize, nil + case deployplan.Delete: + return bundledeployments.OperationActionTypeOperationActionTypeDelete, nil + default: + return "", fmt.Errorf("cannot record operation: unsupported action %q", a) + } +} diff --git a/bundle/direct/oprecorder_test.go b/bundle/direct/oprecorder_test.go new file mode 100644 index 00000000000..56d860de3e6 --- /dev/null +++ b/bundle/direct/oprecorder_test.go @@ -0,0 +1,84 @@ +package direct + +import ( + "context" + "testing" + + "github.com/databricks/cli/bundle/deployplan" + "github.com/databricks/databricks-sdk-go/service/bundledeployments" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +type fakeOpClient struct { + bundledeployments.BundleDeploymentsInterface + requests []bundledeployments.CreateOperationRequest +} + +func (f *fakeOpClient) CreateOperation(ctx context.Context, req bundledeployments.CreateOperationRequest) (*bundledeployments.Operation, error) { + f.requests = append(f.requests, req) + return &bundledeployments.Operation{}, nil +} + +func TestOperationRecorderStripsResourcePrefix(t *testing.T) { + f := &fakeOpClient{} + r := NewOperationRecorder(f, "dep-1", 2) + + err := r.record(t.Context(), "resources.jobs.foo", deployplan.Create, "job-123", map[string]string{"name": "foo"}) + require.NoError(t, err) + + require.Len(t, f.requests, 1) + req := f.requests[0] + // The wire key drops the CLI-internal "resources." prefix, both in the query + // param and the operation body. + assert.Equal(t, "jobs.foo", req.ResourceKey) + assert.Equal(t, "jobs.foo", req.Operation.ResourceKey) + assert.Equal(t, "deployments/dep-1/versions/2", req.Parent) + assert.Equal(t, bundledeployments.OperationActionTypeOperationActionTypeCreate, req.Operation.ActionType) + assert.Equal(t, "job-123", req.Operation.ResourceId) + require.NotNil(t, req.Operation.State) +} + +func TestOperationRecorderDeleteHasNoState(t *testing.T) { + f := &fakeOpClient{} + r := NewOperationRecorder(f, "dep-1", 3) + + err := r.record(t.Context(), "resources.jobs.foo", deployplan.Delete, "", nil) + require.NoError(t, err) + + require.Len(t, f.requests, 1) + assert.Equal(t, bundledeployments.OperationActionTypeOperationActionTypeDelete, f.requests[0].Operation.ActionType) + // Delete operations carry no serialized state. + assert.Nil(t, f.requests[0].Operation.State) +} + +func TestDeployActionToSDK(t *testing.T) { + cases := []struct { + action deployplan.ActionType + want bundledeployments.OperationActionType + }{ + {deployplan.Create, bundledeployments.OperationActionTypeOperationActionTypeCreate}, + {deployplan.Update, bundledeployments.OperationActionTypeOperationActionTypeUpdate}, + {deployplan.UpdateWithID, bundledeployments.OperationActionTypeOperationActionTypeUpdateWithId}, + {deployplan.Recreate, bundledeployments.OperationActionTypeOperationActionTypeRecreate}, + {deployplan.Resize, bundledeployments.OperationActionTypeOperationActionTypeResize}, + {deployplan.Delete, bundledeployments.OperationActionTypeOperationActionTypeDelete}, + } + for _, c := range cases { + got, err := deployActionToSDK(c.action) + require.NoError(t, err) + assert.Equal(t, c.want, got) + } + + // Skip and Undefined never reach a recorder and are rejected. + _, err := deployActionToSDK(deployplan.Skip) + assert.Error(t, err) + _, err = deployActionToSDK(deployplan.Undefined) + assert.Error(t, err) +} + +func TestRecordOperationNoOpWithoutRecorder(t *testing.T) { + b := &DeploymentBundle{} + // No OpRec set: recording is a no-op. + assert.NoError(t, b.recordOperation(t.Context(), "resources.jobs.foo", deployplan.Create, "id", struct{}{})) +} diff --git a/bundle/direct/pkg.go b/bundle/direct/pkg.go index 48a9c5a2ff7..f95b515f726 100644 --- a/bundle/direct/pkg.go +++ b/bundle/direct/pkg.go @@ -44,6 +44,11 @@ type DeploymentBundle struct { Plan *deployplan.Plan RemoteStateCache sync.Map StateCache structvar.Cache + + // OpRec records each applied resource operation with the deployment metadata + // service (DMS). It is nil unless the bundle opts into recording deployment + // history, in which case the phases package sets it after CreateVersion. + OpRec opRecorder } // SetRemoteState updates the remote state with type validation and marks as fresh. diff --git a/bundle/phases/deploy.go b/bundle/phases/deploy.go index f65e50a940e..792c016f963 100644 --- a/bundle/phases/deploy.go +++ b/bundle/phases/deploy.go @@ -17,6 +17,7 @@ import ( "github.com/databricks/cli/bundle/deploy/snapshot" "github.com/databricks/cli/bundle/deploy/terraform" "github.com/databricks/cli/bundle/deployplan" + "github.com/databricks/cli/bundle/direct" "github.com/databricks/cli/bundle/libraries" "github.com/databricks/cli/bundle/metrics" "github.com/databricks/cli/bundle/permissions" @@ -24,6 +25,7 @@ import ( "github.com/databricks/cli/bundle/statemgmt" "github.com/databricks/cli/libs/agent" "github.com/databricks/cli/libs/cmdio" + "github.com/databricks/cli/libs/dms" "github.com/databricks/cli/libs/log" "github.com/databricks/cli/libs/logdiag" "github.com/databricks/cli/libs/sync" @@ -161,7 +163,17 @@ func Deploy(ctx context.Context, b *bundle.Bundle, outputHandler sync.OutputHand } // lock is acquired here + // + // Set up DMS recording of this deployment as a version. The version is not + // created until the plan is approved (below), so a cancelled deploy records + // nothing; the deferred CompleteVersion is a no-op until CreateVersion runs. + // CompleteVersion is deferred before lock.Release so it runs while the lock + // is still held (defers run last-in-first-out). + recorder := newDeploymentRecorder(ctx, b, stateEngine, dms.VersionTypeDeploy) defer func() { + if err := recorder.CompleteVersion(ctx, !logdiag.HasError(ctx)); err != nil { + logdiag.LogError(ctx, err) + } bundle.ApplyContext(ctx, b, lock.Release(lock.GoalDeploy)) }() @@ -255,6 +267,26 @@ func Deploy(ctx context.Context, b *bundle.Bundle, outputHandler sync.OutputHand return } if haveApproval { + // Record the DMS version now that the plan is approved and the state WAL + // has been opened. CreateVersion requests version_id == last_version_id + 1; + // the server returns ABORTED if a concurrent deploy advanced the deployment + // since the plan was computed, so a stale plan is not applied. + if err := recorder.CreateVersion(ctx); err != nil { + logdiag.LogError(ctx, err) + return + } + if recorder != nil { + // On a first deploy the server assigned the deployment ID; persist it in + // state (Finalize writes it to disk) so later deploys reuse the record. + // Record operations under the version just created so DMS holds the + // deployed resource state. + b.DeploymentBundle.StateDB.SetDeploymentID(recorder.DeploymentID()) + b.DeploymentBundle.OpRec = direct.NewOperationRecorder( + b.WorkspaceClient(ctx).BundleDeployments, + recorder.DeploymentID(), + recorder.Version(), + ) + } deployCore(ctx, b, plan, stateEngine, requestedEngine) } else { cmdio.LogString(ctx, "Deployment cancelled!") diff --git a/bundle/phases/destroy.go b/bundle/phases/destroy.go index 2496c7033ad..244f593476f 100644 --- a/bundle/phases/destroy.go +++ b/bundle/phases/destroy.go @@ -13,8 +13,10 @@ import ( "github.com/databricks/cli/bundle/deploy/lock" "github.com/databricks/cli/bundle/deploy/terraform" "github.com/databricks/cli/bundle/deployplan" + "github.com/databricks/cli/bundle/direct" "github.com/databricks/cli/libs/cmdio" "github.com/databricks/cli/libs/diag" + "github.com/databricks/cli/libs/dms" "github.com/databricks/cli/libs/log" "github.com/databricks/cli/libs/logdiag" "github.com/databricks/databricks-sdk-go/apierr" @@ -131,7 +133,15 @@ func Destroy(ctx context.Context, b *bundle.Bundle, engine engine.EngineType) { return } + // Set up DMS recording of this destroy as a version. The version is not + // created until the destroy is approved (below), so a cancelled destroy + // records nothing; the deferred CompleteVersion is a no-op until then. It is + // deferred before lock.Release so it runs while the lock is still held. + recorder := newDeploymentRecorder(ctx, b, engine, dms.VersionTypeDestroy) defer func() { + if err := recorder.CompleteVersion(ctx, !logdiag.HasError(ctx)); err != nil { + logdiag.LogError(ctx, err) + } bundle.ApplyContext(ctx, b, lock.Release(lock.GoalDestroy)) }() @@ -188,6 +198,19 @@ func Destroy(ctx context.Context, b *bundle.Bundle, engine engine.EngineType) { return } } + // Record the DMS version now that the destroy is approved and the state WAL + // has been opened, then record each delete operation under it. + if err := recorder.CreateVersion(ctx); err != nil { + logdiag.LogError(ctx, err) + return + } + if recorder != nil { + b.DeploymentBundle.OpRec = direct.NewOperationRecorder( + b.WorkspaceClient(ctx).BundleDeployments, + recorder.DeploymentID(), + recorder.Version(), + ) + } destroyCore(ctx, b, plan, engine) } else { cmdio.LogString(ctx, "Destroy cancelled!") diff --git a/bundle/phases/dms.go b/bundle/phases/dms.go new file mode 100644 index 00000000000..667ef8627aa --- /dev/null +++ b/bundle/phases/dms.go @@ -0,0 +1,37 @@ +package phases + +import ( + "context" + + "github.com/databricks/cli/bundle" + "github.com/databricks/cli/bundle/config/engine" + "github.com/databricks/cli/libs/dms" +) + +// newDeploymentRecorder returns a dms.Recorder for the current deployment, or +// nil when DMS recording does not apply. A nil recorder is a no-op, so callers +// do not need to branch on it. +// +// Recording is enabled only when experimental.record_deployment_history is set +// AND the engine is direct: DMS resource state is tracked per direct-engine +// deployment, and only the direct engine opens the state DB where the +// deployment ID is stored. Returning nil for terraform leaves those deployments +// untouched. +// +// The deployment ID passed to the recorder is the one persisted in state from a +// previous deploy; it is empty on a bundle's first recorded deploy, in which +// case the recorder creates the deployment and the server assigns the ID. +func newDeploymentRecorder(ctx context.Context, b *bundle.Bundle, eng engine.EngineType, versionType dms.VersionType) *dms.Recorder { + if b.Config.Experimental == nil || !b.Config.Experimental.RecordDeploymentHistory { + return nil + } + if !eng.IsDirect() { + return nil + } + return dms.NewRecorder( + b.WorkspaceClient(ctx).BundleDeployments, + b.DeploymentBundle.StateDB.GetDeploymentID(), + b.Config.Bundle.Target, + versionType, + ) +} diff --git a/cmd/bundle/generate/dashboard.go b/cmd/bundle/generate/dashboard.go index 086ec1d600a..2b286bcad3d 100644 --- a/cmd/bundle/generate/dashboard.go +++ b/cmd/bundle/generate/dashboard.go @@ -404,7 +404,7 @@ func (d *dashboard) runForResource(ctx context.Context, b *bundle.Bundle) { var state statemgmt.ExportedResourcesMap if stateDesc.Engine.IsDirect() { _, localPath := b.StateFilenameDirect(ctx) - if err := b.DeploymentBundle.StateDB.Open(ctx, localPath, dstate.WithRecovery(true), dstate.WithWrite(false)); err != nil { + if err := b.DeploymentBundle.StateDB.Open(ctx, localPath, dstate.WithRecovery(true), dstate.WithWrite(false), nil); err != nil { logdiag.LogError(ctx, err) return } diff --git a/cmd/bundle/generate/genie_space.go b/cmd/bundle/generate/genie_space.go index 6d938c5e03d..48ecc92a6cd 100644 --- a/cmd/bundle/generate/genie_space.go +++ b/cmd/bundle/generate/genie_space.go @@ -322,7 +322,7 @@ func (g *genieSpace) runForResource(ctx context.Context, b *bundle.Bundle) { var state statemgmt.ExportedResourcesMap if stateDesc.Engine.IsDirect() { _, localPath := b.StateFilenameDirect(ctx) - if err := b.DeploymentBundle.StateDB.Open(ctx, localPath, dstate.WithRecovery(true), dstate.WithWrite(false)); err != nil { + if err := b.DeploymentBundle.StateDB.Open(ctx, localPath, dstate.WithRecovery(true), dstate.WithWrite(false), nil); err != nil { logdiag.LogError(ctx, err) return } diff --git a/cmd/bundle/utils/process.go b/cmd/bundle/utils/process.go index e4f232605ce..2815f591b22 100644 --- a/cmd/bundle/utils/process.go +++ b/cmd/bundle/utils/process.go @@ -25,6 +25,7 @@ import ( "github.com/databricks/cli/libs/logdiag" "github.com/databricks/cli/libs/sync" "github.com/databricks/cli/libs/telemetry/protos" + "github.com/databricks/databricks-sdk-go/service/bundledeployments" "github.com/spf13/cobra" ) @@ -211,7 +212,16 @@ func ProcessBundleRet(cmd *cobra.Command, opts ProcessOptions) (b *bundle.Bundle needDirectState := stateDesc.Engine.IsDirect() && (opts.InitIDs || opts.ErrorOnEmptyState || opts.Deploy || opts.ReadPlanPath != "" || opts.PreDeployChecks || opts.PostStateFunc != nil) if needDirectState { _, localPath := b.StateFilenameDirect(ctx) - if err := b.DeploymentBundle.StateDB.Open(ctx, localPath, dstate.WithRecovery(true), dstate.WithWrite(false)); err != nil { + + // When the bundle records deployment history, the deployment metadata + // service owns resource state, so hand Open its client to overlay DMS + // state on top of the local identity (lineage/serial/deployment ID). + // Reads open the state write-disabled, so no lineage is minted here. + var dmsClient bundledeployments.BundleDeploymentsInterface + if b.Config.Experimental != nil && b.Config.Experimental.RecordDeploymentHistory { + dmsClient = b.WorkspaceClient(ctx).BundleDeployments + } + if err := b.DeploymentBundle.StateDB.Open(ctx, localPath, dstate.WithRecovery(true), dstate.WithWrite(false), dmsClient); err != nil { logdiag.LogError(ctx, err) return b, stateDesc, root.ErrAlreadyPrinted } diff --git a/libs/dms/recorder.go b/libs/dms/recorder.go new file mode 100644 index 00000000000..eed8485f2c2 --- /dev/null +++ b/libs/dms/recorder.go @@ -0,0 +1,255 @@ +// Package dms records bundle deployments as versions with the Deployment +// Metadata Service (DMS). +// +// It is intentionally independent of the deployment lock: a Recorder does not +// acquire or hold any lock. Callers are responsible for serializing concurrent +// deployments (today via the workspace-filesystem lock). The server-side +// version counter — CreateVersion only succeeds when the requested version is +// last_version_id + 1 — provides the concurrency control for the records +// themselves. +package dms + +import ( + "context" + "errors" + "fmt" + "net/http" + "strconv" + "strings" + "time" + + "github.com/databricks/cli/internal/build" + "github.com/databricks/cli/libs/log" + "github.com/databricks/databricks-sdk-go/apierr" + "github.com/databricks/databricks-sdk-go/service/bundledeployments" +) + +// The server expires a version's lease if it does not receive a heartbeat +// within a 2-minute TTL; we heartbeat well inside that window. +const defaultHeartbeatInterval = 30 * time.Second + +// VersionType identifies the kind of deployment a version records. +type VersionType = bundledeployments.VersionType + +const ( + VersionTypeDeploy VersionType = bundledeployments.VersionTypeVersionTypeDeploy + VersionTypeDestroy VersionType = bundledeployments.VersionTypeVersionTypeDestroy +) + +// Recorder records a single deploy/destroy as a version with DMS. +// +// The deployment ID is assigned by the server on the first deploy: NewRecorder +// is given the ID persisted in state (empty on a bundle's first-ever recorded +// deploy), and CreateVersion creates the deployment record when that ID is +// empty and exposes the server-assigned ID via DeploymentID so the caller can +// persist it. Later deploys pass the stored ID back in and reuse the record. +type Recorder struct { + svc bundledeployments.BundleDeploymentsInterface + deploymentID string + targetName string + versionType VersionType + + // populated by CreateVersion + versionNum int64 + stopHeartbeat context.CancelFunc +} + +// NewRecorder returns a Recorder for the given deployment. deploymentID is the +// DMS deployment ID persisted in state, or empty if this bundle has not yet +// recorded a deployment (the server assigns one during CreateVersion). +func NewRecorder(svc bundledeployments.BundleDeploymentsInterface, deploymentID, targetName string, versionType VersionType) *Recorder { + return &Recorder{ + svc: svc, + deploymentID: deploymentID, + targetName: targetName, + versionType: versionType, + } +} + +// DeploymentID returns the DMS deployment ID this recorder is bound to. It is +// empty until CreateVersion has created the deployment record (on a first +// deploy) and non-empty afterwards, so callers persist it once CreateVersion +// succeeds. +func (r *Recorder) DeploymentID() string { + if r == nil { + return "" + } + return r.deploymentID +} + +// Version returns the version number claimed by CreateVersion. It is zero until +// CreateVersion has run; callers use it to parent operations under the version. +func (r *Recorder) Version() int64 { + if r == nil { + return 0 + } + return r.versionNum +} + +// CreateVersion registers a new version with DMS, claiming it for the duration +// of the deployment. A nil Recorder is a no-op, so callers can leave it nil +// when recording is disabled. +func (r *Recorder) CreateVersion(ctx context.Context) error { + if r == nil { + return nil + } + + versionID, err := r.createDeploymentVersion(ctx) + if err != nil { + return err + } + + versionNum, err := strconv.ParseInt(versionID, 10, 64) + if err != nil { + return fmt.Errorf("failed to parse version ID %q: %w", versionID, err) + } + r.versionNum = versionNum + r.stopHeartbeat = startHeartbeat(ctx, r.svc, r.deploymentID, versionID) + return nil +} + +// CompleteVersion finalizes the version created by CreateVersion. A nil +// Recorder, or one whose CreateVersion never ran, is a no-op. +func (r *Recorder) CompleteVersion(ctx context.Context, success bool) error { + if r == nil || r.stopHeartbeat == nil { + return nil + } + + r.stopHeartbeat() + + versionIDStr := strconv.FormatInt(r.versionNum, 10) + versionName := fmt.Sprintf("deployments/%s/versions/%s", r.deploymentID, versionIDStr) + + reason := bundledeployments.VersionCompleteVersionCompleteSuccess + if !success { + reason = bundledeployments.VersionCompleteVersionCompleteFailure + } + + _, err := r.svc.CompleteVersion(ctx, bundledeployments.CompleteVersionRequest{ + Name: versionName, + CompletionReason: reason, + }) + if err != nil { + return err + } + log.Infof(ctx, "Completed deployment version: deployment=%s version=%s reason=%s", r.deploymentID, versionIDStr, reason) + + // For destroy operations, delete the deployment record after the version + // completes successfully. + if success && r.versionType == VersionTypeDestroy { + err = r.svc.DeleteDeployment(ctx, bundledeployments.DeleteDeploymentRequest{ + Name: "deployments/" + r.deploymentID, + }) + if err != nil { + return fmt.Errorf("failed to delete deployment: %w", err) + } + } + + return nil +} + +// createDeploymentVersion ensures the deployment record exists, then creates a +// new version under it. On a first deploy (no stored deployment ID) it creates +// the deployment and lets the server assign the ID; otherwise it reads the +// existing deployment to compute the next version number. +func (r *Recorder) createDeploymentVersion(ctx context.Context) (versionID string, err error) { + if r.deploymentID == "" { + // First deploy: create the deployment with an empty ID so the server + // assigns one, then start at version 1. + dep, createErr := r.svc.CreateDeployment(ctx, bundledeployments.CreateDeploymentRequest{ + Deployment: bundledeployments.Deployment{ + TargetName: r.targetName, + }, + }) + if createErr != nil { + return "", fmt.Errorf("failed to create deployment: %w", createErr) + } + id, idErr := deploymentIDFromName(dep.Name) + if idErr != nil { + return "", idErr + } + r.deploymentID = id + versionID = "1" + } else { + // Existing deployment: read it to compute the next version number. + dep, getErr := r.svc.GetDeployment(ctx, bundledeployments.GetDeploymentRequest{ + Name: "deployments/" + r.deploymentID, + }) + if getErr != nil { + return "", fmt.Errorf("failed to get deployment: %w", getErr) + } + lastVersion, parseErr := strconv.ParseInt(dep.LastVersionId, 10, 64) + if parseErr != nil { + return "", fmt.Errorf("failed to parse last_version_id %q: %w", dep.LastVersionId, parseErr) + } + versionID = strconv.FormatInt(lastVersion+1, 10) + } + + // The server validates that versionID equals last_version_id + 1 and returns + // ABORTED otherwise (e.g. a concurrent deploy already created this version). + version, versionErr := r.svc.CreateVersion(ctx, bundledeployments.CreateVersionRequest{ + Parent: "deployments/" + r.deploymentID, + VersionId: versionID, + Version: bundledeployments.Version{ + CliVersion: build.GetInfo().Version, + VersionType: r.versionType, + TargetName: r.targetName, + }, + }) + if versionErr != nil { + return "", fmt.Errorf("failed to create deployment version: %w", versionErr) + } + + log.Infof(ctx, "Created deployment version: deployment=%s version=%s", r.deploymentID, version.VersionId) + return versionID, nil +} + +// deploymentIDFromName extracts the deployment ID from a DMS resource name of +// the form "deployments/{deployment_id}". +func deploymentIDFromName(name string) (string, error) { + id, ok := strings.CutPrefix(name, "deployments/") + if !ok || id == "" { + return "", fmt.Errorf("unexpected deployment name %q from deployment metadata service", name) + } + return id, nil +} + +// startHeartbeat starts a background goroutine that sends heartbeats to keep +// the deployment version's lease alive. Returns a cancel function to stop it. +func startHeartbeat(ctx context.Context, svc bundledeployments.BundleDeploymentsInterface, deploymentID, versionID string) context.CancelFunc { + ctx, cancel := context.WithCancel(ctx) + versionName := fmt.Sprintf("deployments/%s/versions/%s", deploymentID, versionID) + + go func() { + ticker := time.NewTicker(defaultHeartbeatInterval) + defer ticker.Stop() + + for { + select { + case <-ctx.Done(): + return + case <-ticker.C: + _, err := svc.Heartbeat(ctx, bundledeployments.HeartbeatRequest{Name: versionName}) + if err != nil { + // A 409 ABORTED is expected if the version was completed + // between the ticker firing and the heartbeat. + if isAbortedErr(err) { + log.Debugf(ctx, "Heartbeat stopped: version already completed") + return + } + log.Warnf(ctx, "Failed to send deployment heartbeat: %v", err) + } else { + log.Debugf(ctx, "Deployment heartbeat sent: deployment=%s version=%s", deploymentID, versionID) + } + } + } + }() + + return cancel +} + +// isAbortedErr reports whether err is an HTTP 409 ABORTED from the DMS API. +func isAbortedErr(err error) bool { + apiErr, ok := errors.AsType[*apierr.APIError](err) + return ok && apiErr.StatusCode == http.StatusConflict && apiErr.ErrorCode == "ABORTED" +} diff --git a/libs/dms/recorder_test.go b/libs/dms/recorder_test.go new file mode 100644 index 00000000000..91848f74a70 --- /dev/null +++ b/libs/dms/recorder_test.go @@ -0,0 +1,165 @@ +package dms + +import ( + "context" + "testing" + + "github.com/databricks/databricks-sdk-go/service/bundledeployments" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// fakeDMS records the calls the recorder makes and lets a test script the +// server-side responses. It embeds the SDK interface so it satisfies it while +// only overriding the methods the recorder uses. +type fakeDMS struct { + bundledeployments.BundleDeploymentsInterface + + // scripted behavior + getDeployment func(id string) (*bundledeployments.Deployment, error) + + // assigned deployment ID for CreateDeployment (server-generated flow) + assignedID string + + // captured requests + created []bundledeployments.CreateDeploymentRequest + versions []bundledeployments.CreateVersionRequest + completed []bundledeployments.CompleteVersionRequest + deleted []string +} + +func (f *fakeDMS) CreateDeployment(ctx context.Context, req bundledeployments.CreateDeploymentRequest) (*bundledeployments.Deployment, error) { + f.created = append(f.created, req) + id := req.DeploymentId + if id == "" { + id = f.assignedID + } + return &bundledeployments.Deployment{Name: "deployments/" + id}, nil +} + +func (f *fakeDMS) GetDeployment(ctx context.Context, req bundledeployments.GetDeploymentRequest) (*bundledeployments.Deployment, error) { + id := req.Name[len("deployments/"):] + return f.getDeployment(id) +} + +func (f *fakeDMS) CreateVersion(ctx context.Context, req bundledeployments.CreateVersionRequest) (*bundledeployments.Version, error) { + f.versions = append(f.versions, req) + return &bundledeployments.Version{VersionId: req.VersionId}, nil +} + +func (f *fakeDMS) CompleteVersion(ctx context.Context, req bundledeployments.CompleteVersionRequest) (*bundledeployments.Version, error) { + f.completed = append(f.completed, req) + return &bundledeployments.Version{}, nil +} + +func (f *fakeDMS) DeleteDeployment(ctx context.Context, req bundledeployments.DeleteDeploymentRequest) error { + f.deleted = append(f.deleted, req.Name) + return nil +} + +func (f *fakeDMS) Heartbeat(ctx context.Context, req bundledeployments.HeartbeatRequest) (*bundledeployments.HeartbeatResponse, error) { + return &bundledeployments.HeartbeatResponse{}, nil +} + +func TestRecorderFirstDeployCreatesDeploymentWithServerAssignedID(t *testing.T) { + f := &fakeDMS{assignedID: "server-generated-id"} + // A first deploy has no stored deployment ID. + r := NewRecorder(f, "", "dev", VersionTypeDeploy) + + require.NoError(t, r.CreateVersion(t.Context())) + + // The deployment was created with an empty ID so the server assigns one, and + // the recorder exposes the assigned ID for the caller to persist. + require.Len(t, f.created, 1) + assert.Empty(t, f.created[0].DeploymentId) + assert.Equal(t, "server-generated-id", r.DeploymentID()) + + // The first version is 1, parented under the assigned deployment. + require.Len(t, f.versions, 1) + assert.Equal(t, "1", f.versions[0].VersionId) + assert.Equal(t, "deployments/server-generated-id", f.versions[0].Parent) + assert.Equal(t, int64(1), r.Version()) + + require.NoError(t, r.CompleteVersion(t.Context(), true)) + require.Len(t, f.completed, 1) + assert.Equal(t, bundledeployments.VersionCompleteVersionCompleteSuccess, f.completed[0].CompletionReason) + assert.Empty(t, f.deleted) +} + +func TestRecorderSubsequentDeployReusesDeploymentAndIncrementsVersion(t *testing.T) { + f := &fakeDMS{ + getDeployment: func(id string) (*bundledeployments.Deployment, error) { + return &bundledeployments.Deployment{Name: "deployments/" + id, LastVersionId: "4"}, nil + }, + } + // A subsequent deploy passes the stored deployment ID. + r := NewRecorder(f, "stored-id", "dev", VersionTypeDeploy) + + require.NoError(t, r.CreateVersion(t.Context())) + + // No new deployment is created; the version increments to last_version_id + 1. + assert.Empty(t, f.created) + require.Len(t, f.versions, 1) + assert.Equal(t, "5", f.versions[0].VersionId) + assert.Equal(t, "stored-id", r.DeploymentID()) +} + +func TestRecorderDestroyDeletesDeploymentOnSuccess(t *testing.T) { + f := &fakeDMS{ + getDeployment: func(id string) (*bundledeployments.Deployment, error) { + return &bundledeployments.Deployment{Name: "deployments/" + id, LastVersionId: "2"}, nil + }, + } + r := NewRecorder(f, "stored-id", "dev", VersionTypeDestroy) + + require.NoError(t, r.CreateVersion(t.Context())) + assert.Equal(t, bundledeployments.VersionTypeVersionTypeDestroy, f.versions[0].Version.VersionType) + + require.NoError(t, r.CompleteVersion(t.Context(), true)) + // A successful destroy deletes the deployment record. + require.Equal(t, []string{"deployments/stored-id"}, f.deleted) +} + +func TestRecorderFailedDestroyKeepsDeployment(t *testing.T) { + f := &fakeDMS{ + getDeployment: func(id string) (*bundledeployments.Deployment, error) { + return &bundledeployments.Deployment{Name: "deployments/" + id, LastVersionId: "2"}, nil + }, + } + r := NewRecorder(f, "stored-id", "dev", VersionTypeDestroy) + + require.NoError(t, r.CreateVersion(t.Context())) + require.NoError(t, r.CompleteVersion(t.Context(), false)) + + assert.Equal(t, bundledeployments.VersionCompleteVersionCompleteFailure, f.completed[0].CompletionReason) + // A failed destroy leaves the deployment in place. + assert.Empty(t, f.deleted) +} + +func TestNilRecorderIsNoOp(t *testing.T) { + var r *Recorder + assert.NoError(t, r.CreateVersion(t.Context())) + assert.NoError(t, r.CompleteVersion(t.Context(), true)) + assert.Empty(t, r.DeploymentID()) + assert.Zero(t, r.Version()) +} + +func TestRecorderCompleteVersionNoOpWithoutCreateVersion(t *testing.T) { + f := &fakeDMS{} + r := NewRecorder(f, "stored-id", "dev", VersionTypeDeploy) + // CompleteVersion before CreateVersion is a no-op (nothing was claimed). + require.NoError(t, r.CompleteVersion(t.Context(), true)) + assert.Empty(t, f.completed) +} + +func TestDeploymentIDFromName(t *testing.T) { + id, err := deploymentIDFromName("deployments/abc-123") + require.NoError(t, err) + assert.Equal(t, "abc-123", id) + + _, err = deploymentIDFromName("abc-123") + assert.Error(t, err) + + _, err = deploymentIDFromName("deployments/") + assert.Error(t, err) +} diff --git a/libs/testserver/bundle.go b/libs/testserver/bundle.go new file mode 100644 index 00000000000..5f2c7e0cfd1 --- /dev/null +++ b/libs/testserver/bundle.go @@ -0,0 +1,228 @@ +package testserver + +import ( + "encoding/json" + "slices" + "strconv" + + "github.com/databricks/databricks-sdk-go/service/bundledeployments" +) + +// Handlers for the Deployment Metadata Service (DMS) API under /api/2.0/bundle. +// State is kept in FakeWorkspace.dmsDeployments, keyed by deployment ID. + +// dmsDeployment holds a deployment record together with the versions and +// resources recorded under it, so the read APIs (ListVersions/ListResources) +// can serve back what deploys wrote. +type dmsDeployment struct { + deployment bundledeployments.Deployment + versions map[string]*bundledeployments.Version + // resources is the latest resource state per resource key, updated as + // operations are recorded. + resources map[string]bundledeployments.Resource +} + +func (s *FakeWorkspace) CreateDeployment(req Request) Response { + // The client either supplies the deployment ID or, in the server-generated + // flow, leaves it empty for the server to mint one. + deploymentID := req.URL.Query().Get("deployment_id") + if deploymentID == "" { + deploymentID = nextUUID() + } + + var dep bundledeployments.Deployment + if err := json.Unmarshal(req.Body, &dep); err != nil { + return Response{StatusCode: 400, Body: map[string]string{"message": err.Error()}} + } + + defer s.LockUnlock()() + + dep.Name = "deployments/" + deploymentID + dep.Status = bundledeployments.DeploymentStatusDeploymentStatusActive + s.dmsDeployments[deploymentID] = &dmsDeployment{ + deployment: dep, + versions: map[string]*bundledeployments.Version{}, + resources: map[string]bundledeployments.Resource{}, + } + return Response{Body: dep} +} + +func (s *FakeWorkspace) GetDeployment(deploymentID string) Response { + defer s.LockUnlock()() + + d, ok := s.dmsDeployments[deploymentID] + if !ok { + return dmsNotFound("deployment " + deploymentID) + } + return Response{Body: d.deployment} +} + +func (s *FakeWorkspace) DeleteDeployment(deploymentID string) Response { + defer s.LockUnlock()() + + delete(s.dmsDeployments, deploymentID) + return Response{Body: map[string]any{}} +} + +func (s *FakeWorkspace) CreateVersion(req Request, deploymentID string) Response { + versionID := req.URL.Query().Get("version_id") + + var version bundledeployments.Version + if err := json.Unmarshal(req.Body, &version); err != nil { + return Response{StatusCode: 400, Body: map[string]string{"message": err.Error()}} + } + + defer s.LockUnlock()() + + d, ok := s.dmsDeployments[deploymentID] + if !ok { + return dmsNotFound("deployment " + deploymentID) + } + + // Mirror the server-side optimistic concurrency check: the new version must + // be exactly last_version_id + 1. + want := "1" + if d.deployment.LastVersionId != "" { + last, _ := strconv.ParseInt(d.deployment.LastVersionId, 10, 64) + want = strconv.FormatInt(last+1, 10) + } + if versionID != want { + return dmsAborted("expected version " + want + ", got " + versionID) + } + + d.deployment.LastVersionId = versionID + version.Name = "deployments/" + deploymentID + "/versions/" + versionID + version.VersionId = versionID + version.Status = bundledeployments.VersionStatusVersionStatusInProgress + d.versions[versionID] = &version + return Response{Body: version} +} + +func (s *FakeWorkspace) CompleteVersion(req Request, deploymentID, versionID string) Response { + var completeReq bundledeployments.CompleteVersionRequest + if err := json.Unmarshal(req.Body, &completeReq); err != nil { + return Response{StatusCode: 400, Body: map[string]string{"message": err.Error()}} + } + + defer s.LockUnlock()() + + d, ok := s.dmsDeployments[deploymentID] + if !ok { + return dmsNotFound("deployment " + deploymentID) + } + v, ok := d.versions[versionID] + if !ok { + return dmsNotFound("version " + versionID) + } + + v.Status = bundledeployments.VersionStatusVersionStatusCompleted + v.CompletionReason = completeReq.CompletionReason + return Response{Body: *v} +} + +func (s *FakeWorkspace) Heartbeat() Response { + return Response{Body: bundledeployments.HeartbeatResponse{}} +} + +func (s *FakeWorkspace) CreateOperation(req Request, deploymentID, versionID string) Response { + resourceKey := req.URL.Query().Get("resource_key") + + var op bundledeployments.Operation + if err := json.Unmarshal(req.Body, &op); err != nil { + return Response{StatusCode: 400, Body: map[string]string{"message": err.Error()}} + } + + defer s.LockUnlock()() + + d, ok := s.dmsDeployments[deploymentID] + if !ok { + return dmsNotFound("deployment " + deploymentID) + } + + op.Name = "deployments/" + deploymentID + "/versions/" + versionID + "/operations/" + resourceKey + op.ResourceKey = resourceKey + + // Reflect the operation onto the deployment-level resource set the way the + // backend does: a delete removes the resource, anything else upserts it. + if op.ActionType == bundledeployments.OperationActionTypeOperationActionTypeDelete { + delete(d.resources, resourceKey) + } else { + d.resources[resourceKey] = bundledeployments.Resource{ + Name: "deployments/" + deploymentID + "/resources/" + resourceKey, + ResourceKey: resourceKey, + ResourceId: op.ResourceId, + ResourceType: op.ResourceType, + LastActionType: op.ActionType, + LastVersionId: versionID, + State: op.State, + } + } + return Response{Body: op} +} + +func (s *FakeWorkspace) ListVersions(deploymentID string) Response { + defer s.LockUnlock()() + + d, ok := s.dmsDeployments[deploymentID] + if !ok { + return dmsNotFound("deployment " + deploymentID) + } + + // The API returns versions newest-first (descending version_id). + ids := make([]int64, 0, len(d.versions)) + for id := range d.versions { + n, _ := strconv.ParseInt(id, 10, 64) + ids = append(ids, n) + } + slices.SortFunc(ids, func(a, b int64) int { return int(b - a) }) + + versions := make([]bundledeployments.Version, 0, len(ids)) + for _, id := range ids { + versions = append(versions, *d.versions[strconv.FormatInt(id, 10)]) + } + return Response{Body: bundledeployments.ListVersionsResponse{Versions: versions}} +} + +func (s *FakeWorkspace) ListResources(deploymentID string) Response { + defer s.LockUnlock()() + + d, ok := s.dmsDeployments[deploymentID] + if !ok { + return dmsNotFound("deployment " + deploymentID) + } + + // Sort by resource key so the response order is deterministic. + keys := make([]string, 0, len(d.resources)) + for key := range d.resources { + keys = append(keys, key) + } + slices.Sort(keys) + + resources := make([]bundledeployments.Resource, 0, len(keys)) + for _, key := range keys { + resources = append(resources, d.resources[key]) + } + return Response{Body: bundledeployments.ListResourcesResponse{Resources: resources}} +} + +// dmsNotFound returns the RESOURCE_DOES_NOT_EXIST error shape the DMS API uses, +// which the SDK maps to apierr.ErrNotFound. +func dmsNotFound(what string) Response { + return Response{ + StatusCode: 404, + Body: map[string]string{ + "error_code": "RESOURCE_DOES_NOT_EXIST", + "message": what + " does not exist", + }, + } +} + +// dmsAborted returns the 409 ABORTED error the server uses for the version +// optimistic-concurrency check. +func dmsAborted(message string) Response { + return Response{ + StatusCode: 409, + Headers: map[string][]string{"Content-Type": {"application/json"}}, + Body: map[string]string{"error_code": "ABORTED", "message": message}, + } +} diff --git a/libs/testserver/fake_workspace.go b/libs/testserver/fake_workspace.go index 8d6e8ee0dd3..a3c4519ccc4 100644 --- a/libs/testserver/fake_workspace.go +++ b/libs/testserver/fake_workspace.go @@ -227,6 +227,10 @@ type FakeWorkspace struct { // clusterVenvs caches Python venvs per existing cluster ID, // matching cloud behavior where libraries are cached on running clusters. clusterVenvs map[string]*clusterEnv + + // dmsDeployments holds Deployment Metadata Service (DMS) records, keyed by + // deployment ID. Each record carries its versions and latest resource state. + dmsDeployments map[string]*dmsDeployment } func (s *FakeWorkspace) LockUnlock() func() { @@ -378,6 +382,7 @@ func NewFakeWorkspace(url, token string) *FakeWorkspace { postgresImplicitBranches: map[string]bool{}, postgresImplicitEndpoints: map[string]bool{}, clusterVenvs: map[string]*clusterEnv{}, + dmsDeployments: map[string]*dmsDeployment{}, Alerts: map[string]sql.AlertV2{}, Experiments: map[string]ml.GetExperimentResponse{}, ModelRegistryModels: map[string]ml.Model{}, diff --git a/libs/testserver/handlers.go b/libs/testserver/handlers.go index 1d534c47431..39fe6fbb057 100644 --- a/libs/testserver/handlers.go +++ b/libs/testserver/handlers.go @@ -266,6 +266,35 @@ func AddDefaultHandlers(server *Server) { return req.Workspace.JobsCreate(req) }) + // Deployment Metadata Service (DMS) endpoints. + server.Handle("POST", "/api/2.0/bundle/deployments", func(req Request) any { + return req.Workspace.CreateDeployment(req) + }) + server.Handle("GET", "/api/2.0/bundle/deployments/{deployment_id}", func(req Request) any { + return req.Workspace.GetDeployment(req.Vars["deployment_id"]) + }) + server.Handle("DELETE", "/api/2.0/bundle/deployments/{deployment_id}", func(req Request) any { + return req.Workspace.DeleteDeployment(req.Vars["deployment_id"]) + }) + server.Handle("GET", "/api/2.0/bundle/deployments/{deployment_id}/versions", func(req Request) any { + return req.Workspace.ListVersions(req.Vars["deployment_id"]) + }) + server.Handle("POST", "/api/2.0/bundle/deployments/{deployment_id}/versions", func(req Request) any { + return req.Workspace.CreateVersion(req, req.Vars["deployment_id"]) + }) + server.Handle("POST", "/api/2.0/bundle/deployments/{deployment_id}/versions/{version_id}/complete", func(req Request) any { + return req.Workspace.CompleteVersion(req, req.Vars["deployment_id"], req.Vars["version_id"]) + }) + server.Handle("POST", "/api/2.0/bundle/deployments/{deployment_id}/versions/{version_id}/heartbeat", func(req Request) any { + return req.Workspace.Heartbeat() + }) + server.Handle("POST", "/api/2.0/bundle/deployments/{deployment_id}/versions/{version_id}/operations", func(req Request) any { + return req.Workspace.CreateOperation(req, req.Vars["deployment_id"], req.Vars["version_id"]) + }) + server.Handle("GET", "/api/2.0/bundle/deployments/{deployment_id}/resources", func(req Request) any { + return req.Workspace.ListResources(req.Vars["deployment_id"]) + }) + server.Handle("POST", "/api/2.2/jobs/delete", func(req Request) any { var request jobs.DeleteJob if err := json.Unmarshal(req.Body, &request); err != nil { From da1ed9dfc00a9d6171b2fbeb60d9f17a3e900b26 Mon Sep 17 00:00:00 2001 From: Shreyas Goenka Date: Fri, 24 Jul 2026 16:49:32 +0200 Subject: [PATCH 2/2] bundle: read DMS authority from last_successful_version_id MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The read overlay decided whether DMS owns a deployment's state by listing versions and scanning for a successful one. The deployment now exposes last_successful_version_id directly, so a single GetDeployment answers the same question — no version listing. The field is still stage:DEVELOPMENT in the proto and therefore stripped from the generated SDK, so this reads the deployment via a raw GET into a local struct as a temporary stub. Once the field is promoted to PRIVATE_PREVIEW and regenerated, the raw call collapses to client.GetDeployment(...). LastSuccessfulVersionId and the threaded config argument goes away (see the TODO in deploymentHasSuccessfulVersion). The testserver's GetDeployment now serves last_successful_version_id (tracked on version completion), and the now-unused ListVersions fake is removed. Co-authored-by: Isaac --- bundle/configsync/diff.go | 2 +- bundle/configsync/variables.go | 2 +- bundle/direct/bind.go | 12 +++--- bundle/direct/dstate/dms.go | 64 +++++++++++++++++++----------- bundle/direct/dstate/state.go | 9 ++++- bundle/direct/dstate/state_test.go | 30 +++++++------- cmd/bundle/generate/dashboard.go | 2 +- cmd/bundle/generate/genie_space.go | 2 +- cmd/bundle/utils/process.go | 7 +++- libs/testserver/bundle.go | 45 ++++++++++----------- libs/testserver/handlers.go | 3 -- 11 files changed, 100 insertions(+), 78 deletions(-) diff --git a/bundle/configsync/diff.go b/bundle/configsync/diff.go index ca5b2c9410b..17ed3b30d5e 100644 --- a/bundle/configsync/diff.go +++ b/bundle/configsync/diff.go @@ -149,7 +149,7 @@ func OpenDeploymentState(ctx context.Context, b *bundle.Bundle, engine engine.En deployBundle := &direct.DeploymentBundle{} _, statePath := b.StateFilenameConfigSnapshot(ctx) - if err := deployBundle.StateDB.Open(ctx, statePath, dstate.WithRecovery(true), dstate.WithWrite(false), nil); err != nil { + if err := deployBundle.StateDB.Open(ctx, statePath, dstate.WithRecovery(true), dstate.WithWrite(false), nil, nil); err != nil { return nil, fmt.Errorf("failed to open state: %w", err) } return deployBundle, nil diff --git a/bundle/configsync/variables.go b/bundle/configsync/variables.go index 433b607a037..be3e536f37f 100644 --- a/bundle/configsync/variables.go +++ b/bundle/configsync/variables.go @@ -147,7 +147,7 @@ func resourceIDLookup(ctx context.Context, b *bundle.Bundle) func(string) string } _, statePath := b.StateFilenameConfigSnapshot(ctx) db := &dstate.DeploymentState{} - if err := db.Open(ctx, statePath, dstate.WithRecovery(false), dstate.WithWrite(false), nil); err != nil { + if err := db.Open(ctx, statePath, dstate.WithRecovery(false), dstate.WithWrite(false), nil, nil); err != nil { log.Debugf(ctx, "variable restoration: failed to open state DB at %s: %v", statePath, err) return nil } diff --git a/bundle/direct/bind.go b/bundle/direct/bind.go index ec910b2734e..ccfbcf788ab 100644 --- a/bundle/direct/bind.go +++ b/bundle/direct/bind.go @@ -62,7 +62,7 @@ type BindResult struct { func (b *DeploymentBundle) Bind(ctx context.Context, client *databricks.WorkspaceClient, configRoot *config.Root, statePath, resourceKey, resourceID string) (*BindResult, error) { // Check if the resource is already managed (bound to a different ID) var checkStateDB dstate.DeploymentState - if err := checkStateDB.Open(ctx, statePath, dstate.WithRecovery(true), dstate.WithWrite(false), nil); err == nil { + if err := checkStateDB.Open(ctx, statePath, dstate.WithRecovery(true), dstate.WithWrite(false), nil, nil); err == nil { existingID := checkStateDB.GetResourceID(resourceKey) if _, err := checkStateDB.Finalize(ctx); err != nil { log.Warnf(ctx, "failed to finalize state: %v", err) @@ -86,7 +86,7 @@ func (b *DeploymentBundle) Bind(ctx context.Context, client *databricks.Workspac } // Open temp state - err := b.StateDB.Open(ctx, tmpStatePath, dstate.WithRecovery(false), dstate.WithWrite(true), nil) + err := b.StateDB.Open(ctx, tmpStatePath, dstate.WithRecovery(false), dstate.WithWrite(true), nil, nil) if err != nil { os.Remove(tmpStatePath) return nil, err @@ -109,7 +109,7 @@ func (b *DeploymentBundle) Bind(ctx context.Context, client *databricks.Workspac log.Infof(ctx, "Bound %s to id=%s (in temp state)", resourceKey, resourceID) // First plan + update: populate state with resolved config - err = b.StateDB.Open(ctx, tmpStatePath, dstate.WithRecovery(true), dstate.WithWrite(false), nil) + err = b.StateDB.Open(ctx, tmpStatePath, dstate.WithRecovery(true), dstate.WithWrite(false), nil, nil) if err != nil { os.Remove(tmpStatePath) return nil, err @@ -145,7 +145,7 @@ func (b *DeploymentBundle) Bind(ctx context.Context, client *databricks.Workspac } } - err = b.StateDB.Open(ctx, tmpStatePath, dstate.WithRecovery(true), dstate.WithWrite(true), nil) + err = b.StateDB.Open(ctx, tmpStatePath, dstate.WithRecovery(true), dstate.WithWrite(true), nil, nil) if err != nil { os.Remove(tmpStatePath) return nil, err @@ -165,7 +165,7 @@ func (b *DeploymentBundle) Bind(ctx context.Context, client *databricks.Workspac } // Second plan: this is the plan to present to the user (change between remote resource and config) - err = b.StateDB.Open(ctx, tmpStatePath, dstate.WithRecovery(true), dstate.WithWrite(false), nil) + err = b.StateDB.Open(ctx, tmpStatePath, dstate.WithRecovery(true), dstate.WithWrite(false), nil, nil) if err != nil { os.Remove(tmpStatePath) return nil, err @@ -215,7 +215,7 @@ func (result *BindResult) Cancel() { // Unbind removes a resource from direct engine state without deleting // the workspace resource. Also removes associated permissions/grants entries. func (b *DeploymentBundle) Unbind(ctx context.Context, statePath, resourceKey string) error { - err := b.StateDB.Open(ctx, statePath, dstate.WithRecovery(true), dstate.WithWrite(true), nil) + err := b.StateDB.Open(ctx, statePath, dstate.WithRecovery(true), dstate.WithWrite(true), nil, nil) if err != nil { return err } diff --git a/bundle/direct/dstate/dms.go b/bundle/direct/dstate/dms.go index 1d19d1fe214..659d7c9dd0b 100644 --- a/bundle/direct/dstate/dms.go +++ b/bundle/direct/dstate/dms.go @@ -5,8 +5,12 @@ import ( "encoding/json" "errors" "fmt" + "net/http" + "github.com/databricks/cli/libs/auth" "github.com/databricks/databricks-sdk-go/apierr" + "github.com/databricks/databricks-sdk-go/client" + sdkconfig "github.com/databricks/databricks-sdk-go/config" "github.com/databricks/databricks-sdk-go/service/bundledeployments" ) @@ -17,8 +21,11 @@ import ( // used when DMS has no successful version, or when the user opts out of // recording deployment history. The caller holds db.mu and has already // populated db.Data from the file, including the DeploymentID. -func (db *DeploymentState) overlayDMSState(ctx context.Context, client bundledeployments.BundleDeploymentsInterface) error { - authoritative, err := deploymentHasSuccessfulVersion(ctx, client, db.Data.DeploymentID) +// +// cfg is threaded in only for the temporary raw read in +// deploymentHasSuccessfulVersion; see the TODO there. +func (db *DeploymentState) overlayDMSState(ctx context.Context, client bundledeployments.BundleDeploymentsInterface, cfg *sdkconfig.Config) error { + authoritative, err := deploymentHasSuccessfulVersion(ctx, cfg, db.Data.DeploymentID) if err != nil { return err } @@ -46,29 +53,40 @@ func (db *DeploymentState) overlayDMSState(ctx context.Context, client bundledep // state: if the deployment was never recorded to DMS, or its initial DMS deploy // did not complete successfully, DMS state is absent or partial and Open keeps // the local file's resources instead. -func deploymentHasSuccessfulVersion(ctx context.Context, client bundledeployments.BundleDeploymentsInterface, deploymentID string) (bool, error) { - // Versions are listed newest-first and fetched page by page, and we stop at - // the first successful one, so a deployment with a long version history does - // not require reading the whole list (typically just the first page). - it := client.ListVersions(ctx, bundledeployments.ListVersionsRequest{ - Parent: "deployments/" + deploymentID, - }) - for it.HasNext(ctx) { - v, err := it.Next(ctx) - if err != nil { - // A deployment that was never recorded to DMS is not an error here: it - // just means DMS is not (yet) the source of truth. - if errors.Is(err, apierr.ErrNotFound) { - return false, nil - } - return false, fmt.Errorf("listing versions from deployment metadata service: %w", err) - } - if v.Status == bundledeployments.VersionStatusVersionStatusCompleted && - v.CompletionReason == bundledeployments.VersionCompleteVersionCompleteSuccess { - return true, nil +// +// The deployment carries last_successful_version_id, which the server advances +// only when a version completes successfully (unlike last_version_id, which +// also advances on failure). So a non-empty value is exactly the "DMS owns the +// state" signal, readable in a single GetDeployment. +// +// TODO(DMS): this reads the deployment via a raw GET into a local struct +// because last_successful_version_id is still stage:DEVELOPMENT in the proto +// and therefore stripped from the generated SDK. Once the field is promoted to +// PRIVATE_PREVIEW and regenerated, replace the raw call with +// client.GetDeployment(...).LastSuccessfulVersionId and drop the cfg argument +// (revert overlayDMSState/Open back to taking only the typed client). +func deploymentHasSuccessfulVersion(ctx context.Context, cfg *sdkconfig.Config, deploymentID string) (bool, error) { + apiClient, err := client.New(cfg) + if err != nil { + return false, fmt.Errorf("creating API client for deployment metadata service: %w", err) + } + + // Mirrors the SDK's GetDeployment path (/api/2.0/bundle/{name} with + // name=deployments/{id}); we unmarshal into a local struct so we can read + // last_successful_version_id, which the typed SDK response drops. + var dep struct { + LastSuccessfulVersionID string `json:"last_successful_version_id"` + } + err = apiClient.Do(ctx, http.MethodGet, "/api/2.0/bundle/deployments/"+deploymentID, auth.WorkspaceIDHeaders(cfg), nil, nil, &dep) + if err != nil { + // A deployment that was never recorded to DMS is not an error here: it + // just means DMS is not (yet) the source of truth. + if errors.Is(err, apierr.ErrNotFound) { + return false, nil } + return false, fmt.Errorf("reading deployment from deployment metadata service: %w", err) } - return false, nil + return dep.LastSuccessfulVersionID != "", nil } // fetchDeploymentResources lists every resource recorded for the deployment in diff --git a/bundle/direct/dstate/state.go b/bundle/direct/dstate/state.go index 64fc050bdc0..2c969667c9e 100644 --- a/bundle/direct/dstate/state.go +++ b/bundle/direct/dstate/state.go @@ -19,6 +19,7 @@ import ( "github.com/databricks/cli/libs/dyn" "github.com/databricks/cli/libs/log" "github.com/databricks/cli/libs/structs/structwalk" + sdkconfig "github.com/databricks/databricks-sdk-go/config" "github.com/databricks/databricks-sdk-go/service/bundledeployments" "github.com/google/uuid" ) @@ -253,7 +254,11 @@ type ( // (lineage, serial, and deployment ID) always comes from the file, since that // is what the write path increments and carries forward. A nil dmsClient keeps // the behavior file-only. -func (db *DeploymentState) Open(ctx context.Context, path string, withRecovery WithRecovery, withWrite WithWrite, dmsClient bundledeployments.BundleDeploymentsInterface) error { +// +// dmsCfg accompanies dmsClient (both come from the same workspace client) and +// is used only for a temporary raw read of last_successful_version_id; see the +// TODO in deploymentHasSuccessfulVersion. +func (db *DeploymentState) Open(ctx context.Context, path string, withRecovery WithRecovery, withWrite WithWrite, dmsClient bundledeployments.BundleDeploymentsInterface, dmsCfg *sdkconfig.Config) error { db.mu.Lock() defer db.mu.Unlock() @@ -302,7 +307,7 @@ func (db *DeploymentState) Open(ctx context.Context, path string, withRecovery W } if dmsClient != nil && db.Data.DeploymentID != "" { - if err := db.overlayDMSState(ctx, dmsClient); err != nil { + if err := db.overlayDMSState(ctx, dmsClient, dmsCfg); err != nil { return err } } diff --git a/bundle/direct/dstate/state_test.go b/bundle/direct/dstate/state_test.go index e95ad1b0224..16066bf81f8 100644 --- a/bundle/direct/dstate/state_test.go +++ b/bundle/direct/dstate/state_test.go @@ -20,14 +20,14 @@ func TestOpenSaveFinalizeRoundTrip(t *testing.T) { path := filepath.Join(t.TempDir(), "state.json") var db DeploymentState - require.NoError(t, db.Open(t.Context(), path, WithRecovery(true), WithWrite(true), nil)) + require.NoError(t, db.Open(t.Context(), path, WithRecovery(true), WithWrite(true), nil, nil)) require.NoError(t, db.SaveState("jobs.my_job", "123", map[string]string{"key": "val"}, nil)) mustFinalize(t, &db) // Re-open and verify persisted data. var db2 DeploymentState - require.NoError(t, db2.Open(t.Context(), path, WithRecovery(false), WithWrite(false), nil)) + require.NoError(t, db2.Open(t.Context(), path, WithRecovery(false), WithWrite(false), nil, nil)) assert.Equal(t, 1, db2.Data.Serial) assert.Equal(t, "123", db2.GetResourceID("jobs.my_job")) mustFinalize(t, &db2) @@ -37,7 +37,7 @@ func TestDeploymentIDPersistsAcrossOpen(t *testing.T) { path := filepath.Join(t.TempDir(), "state.json") var db DeploymentState - require.NoError(t, db.Open(t.Context(), path, WithRecovery(true), WithWrite(true), nil)) + require.NoError(t, db.Open(t.Context(), path, WithRecovery(true), WithWrite(true), nil, nil)) assert.Empty(t, db.GetDeploymentID()) // The deployment ID is set during deploy (after CreateDeployment) and @@ -47,7 +47,7 @@ func TestDeploymentIDPersistsAcrossOpen(t *testing.T) { mustFinalize(t, &db) var reopened DeploymentState - require.NoError(t, reopened.Open(t.Context(), path, WithRecovery(false), WithWrite(false), nil)) + require.NoError(t, reopened.Open(t.Context(), path, WithRecovery(false), WithWrite(false), nil, nil)) assert.Equal(t, "server-assigned-id", reopened.GetDeploymentID()) mustFinalize(t, &reopened) } @@ -56,7 +56,7 @@ func TestFinalizeWithNoEntriesDoesNotWriteStateFile(t *testing.T) { path := filepath.Join(t.TempDir(), "state.json") var db DeploymentState - require.NoError(t, db.Open(t.Context(), path, WithRecovery(true), WithWrite(true), nil)) + require.NoError(t, db.Open(t.Context(), path, WithRecovery(true), WithWrite(true), nil, nil)) mustFinalize(t, &db) _, err := os.Stat(path) @@ -112,10 +112,10 @@ func TestPanicOnDoubleOpen(t *testing.T) { path := filepath.Join(t.TempDir(), "state.json") var db DeploymentState - require.NoError(t, db.Open(t.Context(), path, WithRecovery(true), WithWrite(true), nil)) + require.NoError(t, db.Open(t.Context(), path, WithRecovery(true), WithWrite(true), nil, nil)) assert.Panics(t, func() { - _ = db.Open(t.Context(), path, WithRecovery(true), WithWrite(true), nil) + _ = db.Open(t.Context(), path, WithRecovery(true), WithWrite(true), nil, nil) }) mustFinalize(t, &db) } @@ -126,12 +126,12 @@ func TestHeaderOnlyWALRecoveryDoesNotAdvanceSerial(t *testing.T) { // Commit serial 1 with one resource. var db DeploymentState - require.NoError(t, db.Open(t.Context(), path, WithRecovery(true), WithWrite(true), nil)) + require.NoError(t, db.Open(t.Context(), path, WithRecovery(true), WithWrite(true), nil, nil)) require.NoError(t, db.SaveState("jobs.my_job", "123", map[string]string{}, nil)) mustFinalize(t, &db) var committed DeploymentState - require.NoError(t, committed.Open(t.Context(), path, WithRecovery(false), WithWrite(false), nil)) + require.NoError(t, committed.Open(t.Context(), path, WithRecovery(false), WithWrite(false), nil, nil)) lineage := committed.Data.Lineage require.Equal(t, 1, committed.Data.Serial) mustFinalize(t, &committed) @@ -147,7 +147,7 @@ func TestHeaderOnlyWALRecoveryDoesNotAdvanceSerial(t *testing.T) { require.NoError(t, os.WriteFile(walPath, append(headerLine, '\n'), 0o600)) var recovered DeploymentState - require.NoError(t, recovered.Open(t.Context(), path, WithRecovery(true), WithWrite(false), nil)) + require.NoError(t, recovered.Open(t.Context(), path, WithRecovery(true), WithWrite(false), nil, nil)) assert.Equal(t, 1, recovered.Data.Serial) assert.Equal(t, "123", recovered.GetResourceID("jobs.my_job")) assert.NoFileExists(t, walPath) @@ -190,17 +190,17 @@ func TestDeleteState(t *testing.T) { path := filepath.Join(t.TempDir(), "state.json") var db DeploymentState - require.NoError(t, db.Open(t.Context(), path, WithRecovery(true), WithWrite(true), nil)) + require.NoError(t, db.Open(t.Context(), path, WithRecovery(true), WithWrite(true), nil, nil)) require.NoError(t, db.SaveState("jobs.my_job", "123", map[string]string{}, nil)) mustFinalize(t, &db) var db2 DeploymentState - require.NoError(t, db2.Open(t.Context(), path, WithRecovery(true), WithWrite(true), nil)) + require.NoError(t, db2.Open(t.Context(), path, WithRecovery(true), WithWrite(true), nil, nil)) require.NoError(t, db2.DeleteState("jobs.my_job")) mustFinalize(t, &db2) var db3 DeploymentState - require.NoError(t, db3.Open(t.Context(), path, WithRecovery(false), WithWrite(false), nil)) + require.NoError(t, db3.Open(t.Context(), path, WithRecovery(false), WithWrite(false), nil, nil)) assert.Equal(t, 2, db3.Data.Serial) assert.Empty(t, db3.GetResourceID("jobs.my_job")) mustFinalize(t, &db3) @@ -212,7 +212,7 @@ func TestGetOrInitLineageReadableBeforeWriteAndPersisted(t *testing.T) { // Fresh state opened read-only, as the deploy does before planning: no // lineage yet. var db DeploymentState - require.NoError(t, db.Open(t.Context(), path, WithRecovery(true), WithWrite(false), nil)) + require.NoError(t, db.Open(t.Context(), path, WithRecovery(true), WithWrite(false), nil, nil)) require.Empty(t, db.Data.Lineage) // GetOrInitLineage initializes the lineage and makes it readable before any @@ -229,7 +229,7 @@ func TestGetOrInitLineageReadableBeforeWriteAndPersisted(t *testing.T) { // Re-open: the persisted lineage matches the one read before the write. var reopened DeploymentState - require.NoError(t, reopened.Open(t.Context(), path, WithRecovery(false), WithWrite(false), nil)) + require.NoError(t, reopened.Open(t.Context(), path, WithRecovery(false), WithWrite(false), nil, nil)) assert.Equal(t, lineage, reopened.Data.Lineage) mustFinalize(t, &reopened) } diff --git a/cmd/bundle/generate/dashboard.go b/cmd/bundle/generate/dashboard.go index 2b286bcad3d..4866f27c5b3 100644 --- a/cmd/bundle/generate/dashboard.go +++ b/cmd/bundle/generate/dashboard.go @@ -404,7 +404,7 @@ func (d *dashboard) runForResource(ctx context.Context, b *bundle.Bundle) { var state statemgmt.ExportedResourcesMap if stateDesc.Engine.IsDirect() { _, localPath := b.StateFilenameDirect(ctx) - if err := b.DeploymentBundle.StateDB.Open(ctx, localPath, dstate.WithRecovery(true), dstate.WithWrite(false), nil); err != nil { + if err := b.DeploymentBundle.StateDB.Open(ctx, localPath, dstate.WithRecovery(true), dstate.WithWrite(false), nil, nil); err != nil { logdiag.LogError(ctx, err) return } diff --git a/cmd/bundle/generate/genie_space.go b/cmd/bundle/generate/genie_space.go index 48ecc92a6cd..b5dbeed6c56 100644 --- a/cmd/bundle/generate/genie_space.go +++ b/cmd/bundle/generate/genie_space.go @@ -322,7 +322,7 @@ func (g *genieSpace) runForResource(ctx context.Context, b *bundle.Bundle) { var state statemgmt.ExportedResourcesMap if stateDesc.Engine.IsDirect() { _, localPath := b.StateFilenameDirect(ctx) - if err := b.DeploymentBundle.StateDB.Open(ctx, localPath, dstate.WithRecovery(true), dstate.WithWrite(false), nil); err != nil { + if err := b.DeploymentBundle.StateDB.Open(ctx, localPath, dstate.WithRecovery(true), dstate.WithWrite(false), nil, nil); err != nil { logdiag.LogError(ctx, err) return } diff --git a/cmd/bundle/utils/process.go b/cmd/bundle/utils/process.go index 2815f591b22..f556c2b3450 100644 --- a/cmd/bundle/utils/process.go +++ b/cmd/bundle/utils/process.go @@ -25,6 +25,7 @@ import ( "github.com/databricks/cli/libs/logdiag" "github.com/databricks/cli/libs/sync" "github.com/databricks/cli/libs/telemetry/protos" + sdkconfig "github.com/databricks/databricks-sdk-go/config" "github.com/databricks/databricks-sdk-go/service/bundledeployments" "github.com/spf13/cobra" ) @@ -217,11 +218,15 @@ func ProcessBundleRet(cmd *cobra.Command, opts ProcessOptions) (b *bundle.Bundle // service owns resource state, so hand Open its client to overlay DMS // state on top of the local identity (lineage/serial/deployment ID). // Reads open the state write-disabled, so no lineage is minted here. + // dmsCfg accompanies the client for a temporary raw read (see the TODO + // in dstate.deploymentHasSuccessfulVersion). var dmsClient bundledeployments.BundleDeploymentsInterface + var dmsCfg *sdkconfig.Config if b.Config.Experimental != nil && b.Config.Experimental.RecordDeploymentHistory { dmsClient = b.WorkspaceClient(ctx).BundleDeployments + dmsCfg = b.WorkspaceClient(ctx).Config } - if err := b.DeploymentBundle.StateDB.Open(ctx, localPath, dstate.WithRecovery(true), dstate.WithWrite(false), dmsClient); err != nil { + if err := b.DeploymentBundle.StateDB.Open(ctx, localPath, dstate.WithRecovery(true), dstate.WithWrite(false), dmsClient, dmsCfg); err != nil { logdiag.LogError(ctx, err) return b, stateDesc, root.ErrAlreadyPrinted } diff --git a/libs/testserver/bundle.go b/libs/testserver/bundle.go index 5f2c7e0cfd1..34003a507a5 100644 --- a/libs/testserver/bundle.go +++ b/libs/testserver/bundle.go @@ -20,6 +20,12 @@ type dmsDeployment struct { // resources is the latest resource state per resource key, updated as // operations are recorded. resources map[string]bundledeployments.Resource + // lastSuccessfulVersionID is the highest version that completed + // successfully. The server advances last_successful_version_id only on + // success (unlike last_version_id), and the read path treats a non-empty + // value as "DMS owns the state". Tracked separately because the SDK + // Deployment struct does not yet carry the field (still stage:DEVELOPMENT). + lastSuccessfulVersionID string } func (s *FakeWorkspace) CreateDeployment(req Request) Response { @@ -54,7 +60,18 @@ func (s *FakeWorkspace) GetDeployment(deploymentID string) Response { if !ok { return dmsNotFound("deployment " + deploymentID) } - return Response{Body: d.deployment} + + // The SDK Deployment struct does not yet carry last_successful_version_id + // (still stage:DEVELOPMENT, so stripped from generation), but the read path + // reads it off the raw JSON. Serve it as an extra field alongside the typed + // deployment so the overlay behaves as it will against the real server. + return Response{Body: struct { + bundledeployments.Deployment + LastSuccessfulVersionID string `json:"last_successful_version_id,omitempty"` + }{ + Deployment: d.deployment, + LastSuccessfulVersionID: d.lastSuccessfulVersionID, + }} } func (s *FakeWorkspace) DeleteDeployment(deploymentID string) Response { @@ -117,6 +134,9 @@ func (s *FakeWorkspace) CompleteVersion(req Request, deploymentID, versionID str v.Status = bundledeployments.VersionStatusVersionStatusCompleted v.CompletionReason = completeReq.CompletionReason + if completeReq.CompletionReason == bundledeployments.VersionCompleteVersionCompleteSuccess { + d.lastSuccessfulVersionID = versionID + } return Response{Body: *v} } @@ -160,29 +180,6 @@ func (s *FakeWorkspace) CreateOperation(req Request, deploymentID, versionID str return Response{Body: op} } -func (s *FakeWorkspace) ListVersions(deploymentID string) Response { - defer s.LockUnlock()() - - d, ok := s.dmsDeployments[deploymentID] - if !ok { - return dmsNotFound("deployment " + deploymentID) - } - - // The API returns versions newest-first (descending version_id). - ids := make([]int64, 0, len(d.versions)) - for id := range d.versions { - n, _ := strconv.ParseInt(id, 10, 64) - ids = append(ids, n) - } - slices.SortFunc(ids, func(a, b int64) int { return int(b - a) }) - - versions := make([]bundledeployments.Version, 0, len(ids)) - for _, id := range ids { - versions = append(versions, *d.versions[strconv.FormatInt(id, 10)]) - } - return Response{Body: bundledeployments.ListVersionsResponse{Versions: versions}} -} - func (s *FakeWorkspace) ListResources(deploymentID string) Response { defer s.LockUnlock()() diff --git a/libs/testserver/handlers.go b/libs/testserver/handlers.go index 39fe6fbb057..3cbfa16c7e1 100644 --- a/libs/testserver/handlers.go +++ b/libs/testserver/handlers.go @@ -276,9 +276,6 @@ func AddDefaultHandlers(server *Server) { server.Handle("DELETE", "/api/2.0/bundle/deployments/{deployment_id}", func(req Request) any { return req.Workspace.DeleteDeployment(req.Vars["deployment_id"]) }) - server.Handle("GET", "/api/2.0/bundle/deployments/{deployment_id}/versions", func(req Request) any { - return req.Workspace.ListVersions(req.Vars["deployment_id"]) - }) server.Handle("POST", "/api/2.0/bundle/deployments/{deployment_id}/versions", func(req Request) any { return req.Workspace.CreateVersion(req, req.Vars["deployment_id"]) })