From 20b98dd49efc8a5904ab61f5bc924d10a0f549ec Mon Sep 17 00:00:00 2001 From: Shreyas Goenka Date: Wed, 14 Oct 2026 15:40:11 +0000 Subject: [PATCH] Recover damaged DMS deployment-history markers --- .../bundle/dms/marker-writes/output.txt | 26 +++ acceptance/bundle/dms/marker-writes/script | 16 ++ .../bundle/dms/state-resolution/output.txt | 28 +-- acceptance/bundle/dms/state-resolution/script | 18 +- .../dms/version-never-created/output.txt | 3 + bundle/statemgmt/state_pull.go | 171 +++++++++++++++--- bundle/statemgmt/state_recovery_test.go | 161 +++++++++++++++++ cmd/bundle/utils/process.go | 44 +---- libs/dms/metadata.go | 36 ++++ libs/dms/metadata_fetch_test.go | 60 ++++++ 10 files changed, 482 insertions(+), 81 deletions(-) create mode 100644 bundle/statemgmt/state_recovery_test.go create mode 100644 libs/dms/metadata_fetch_test.go diff --git a/acceptance/bundle/dms/marker-writes/output.txt b/acceptance/bundle/dms/marker-writes/output.txt index cf32c690b0d..44071d45a1f 100644 --- a/acceptance/bundle/dms/marker-writes/output.txt +++ b/acceptance/bundle/dms/marker-writes/output.txt @@ -36,3 +36,29 @@ resources.json writes=0 >>> [CLI] bundle summary -o json {"job_id":"[FOO_ID]","version":2} + +=== 4. Corrupt remote marker: repair once, preserve ID, continue at version 3 +>>> [CLI] bundle deploy +Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/dms-marker-writes-[UNIQUE_NAME]/default/files... +Updated jobs.foo +Files: 1 uploaded, 0 deleted +Resources: 0 created, 1 changed, 0 deleted, 0 unchanged + +>>> resources_json_writes 1 +resources.json writes=1 + +>>> [CLI] bundle summary -o json +{"job_id":"[FOO_ID]","version":3} + +=== 5. Missing remote marker: repair once, preserve ID, continue at version 4 +>>> [CLI] bundle deploy +Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/dms-marker-writes-[UNIQUE_NAME]/default/files... +Updated jobs.foo +Files: 1 uploaded, 0 deleted +Resources: 0 created, 1 changed, 0 deleted, 0 unchanged + +>>> resources_json_writes 1 +resources.json writes=1 + +>>> [CLI] bundle summary -o json +{"job_id":"[FOO_ID]","version":4} diff --git a/acceptance/bundle/dms/marker-writes/script b/acceptance/bundle/dms/marker-writes/script index cd83ebef8bb..b55b015a9f1 100644 --- a/acceptance/bundle/dms/marker-writes/script +++ b/acceptance/bundle/dms/marker-writes/script @@ -29,5 +29,21 @@ trace $CLI bundle deploy 2>&1 | contains.py 'Resources: 0 created, 0 changed, 0 trace resources_json_writes 0 trace $CLI bundle summary -o json | jq -c '{job_id: .resources.jobs.foo.id, version: .bundle.deployment.latest_version_id}' +title "4. Corrupt remote marker: repair once, preserve ID, continue at version 3" +MSYS_NO_PATHCONV=1 $CLI workspace import "$state_path/resources.json" --file databricks.yml --format RAW --overwrite &> LOG.corrupt +rm -f "$OUT_REQUESTS" +update_file.py databricks.yml 'name: updated' 'name: repaired' +trace $CLI bundle deploy 2>&1 | contains.py 'Updated jobs.foo' '!Created ' '!Recovering deployment history state with a new lineage' +trace resources_json_writes 1 +trace $CLI bundle summary -o json | jq -c '{job_id: .resources.jobs.foo.id, version: .bundle.deployment.latest_version_id}' + +title "5. Missing remote marker: repair once, preserve ID, continue at version 4" +MSYS_NO_PATHCONV=1 $CLI workspace delete "$state_path/resources.json" &> LOG.missing +rm -f "$OUT_REQUESTS" +update_file.py databricks.yml 'name: repaired' 'name: restored' +trace $CLI bundle deploy 2>&1 | contains.py 'Updated jobs.foo' '!Created ' '!Recovering deployment history state with a new lineage' +trace resources_json_writes 1 +trace $CLI bundle summary -o json | jq -c '{job_id: .resources.jobs.foo.id, version: .bundle.deployment.latest_version_id}' + $CLI bundle destroy --auto-approve &> LOG.cleanup rm -f "$OUT_REQUESTS" diff --git a/acceptance/bundle/dms/state-resolution/output.txt b/acceptance/bundle/dms/state-resolution/output.txt index b2307d2f738..d5fe5b97672 100644 --- a/acceptance/bundle/dms/state-resolution/output.txt +++ b/acceptance/bundle/dms/state-resolution/output.txt @@ -2,8 +2,8 @@ === Setup (LOG.setup) Current: DMS enabled; remote reference snapshot saved. -=== 1. Stale cache -Expect: remote beats serial 999/wrong lineage; cache matches remote. +=== 1. Stale local copy +Expect: remote beats serial 999/wrong lineage; local copy matches remote. >>> jq {serial, lineage} .databricks/bundle/default/resources.json { @@ -15,11 +15,11 @@ Expect: remote beats serial 999/wrong lineage; cache matches remote. resources.json: remote direct state serial=1 lineage="[UUID]" >>> cmp .databricks/resources.remote.json .databricks/bundle/default/resources.json -PASS: stale cache replaced +PASS: stale local copy replaced -=== 2. Invalid cache -Input: YAML in JSON cache paths. -Expect: use remote state; overwrite broken JSON; remove Terraform cache; no resource changes. +=== 2. Invalid local copies +Input: YAML in JSON state paths. +Expect: use remote state; overwrite broken JSON; remove Terraform copy; no resource changes. >>> cp databricks.yml .databricks/bundle/default/resources.json @@ -33,18 +33,20 @@ Plan: 0 to add, 0 to change, 0 to delete, 1 unchanged >>> test ! -e .databricks/bundle/default/terraform/terraform.tfstate >>> cmp .databricks/resources.remote.json .databricks/bundle/default/resources.json -PASS: invalid caches handled +PASS: invalid local copies handled -=== 3. Missing remote -Expect: bad JSON fails; local cache bytes remain unchanged. +=== 3. Missing remote checkpoint and invalid local copy +Expect: recover confirmed DMS state with a new lineage and invalidate old plans. >>> MSYS_NO_PATHCONV=1 [CLI] workspace delete /Workspace/Users/[USERNAME]/.bundle/dms-state-resolution-[UNIQUE_NAME]/default/state/resources.json >>> cp databricks.yml .databricks/bundle/default/resources.json ->>> musterr [CLI] bundle plan -Error: invalid character 'b' looking for beginning of value +>>> [CLI] bundle plan +Warning: Recovering deployment history state with a new lineage; existing saved plans are no longer valid +Plan: 0 to add, 0 to change, 0 to delete, 1 unchanged ->>> cmp databricks.yml .databricks/bundle/default/resources.json -PASS: failure preserved cache +>>> jq -e --slurpfile original .databricks/resources.remote.json .lineage != $original[0].lineage and (.features | has("deployment_history")) and .state == {} .databricks/bundle/default/resources.json +true +PASS: DMS recovery replaced the invalid local copy diff --git a/acceptance/bundle/dms/state-resolution/script b/acceptance/bundle/dms/state-resolution/script index 284e212dd88..dab92add28d 100644 --- a/acceptance/bundle/dms/state-resolution/script +++ b/acceptance/bundle/dms/state-resolution/script @@ -7,31 +7,31 @@ title "Setup (LOG.setup)\nCurrent: DMS enabled; remote reference snapshot saved. trace cp .databricks/bundle/default/resources.json .databricks/resources.remote.json } &> LOG.setup -title "1. Stale cache\nExpect: remote beats serial 999/wrong lineage; cache matches remote.\n" +title "1. Stale local copy\nExpect: remote beats serial 999/wrong lineage; local copy matches remote.\n" { trace update_file.py .databricks/bundle/default/resources.json '"serial": 1,' '"serial": 999,' trace update_file.py .databricks/bundle/default/resources.json '"lineage": "' '"lineage": "stale-' -} &> LOG.stale-cache +} &> LOG.stale-copy trace jq '{serial, lineage}' .databricks/bundle/default/resources.json trace $CLI bundle debug states --force-pull 2>&1 | contains.py 'remote direct state' '!local ' '!stale-' '!serial=999' trace cmp .databricks/resources.remote.json .databricks/bundle/default/resources.json -echo "PASS: stale cache replaced" +echo "PASS: stale local copy replaced" -title "2. Invalid cache\nInput: YAML in JSON cache paths.\nExpect: use remote state; overwrite broken JSON; remove Terraform cache; no resource changes.\n" +title "2. Invalid local copies\nInput: YAML in JSON state paths.\nExpect: use remote state; overwrite broken JSON; remove Terraform copy; no resource changes.\n" trace cp databricks.yml .databricks/bundle/default/resources.json trace mkdir -p .databricks/bundle/default/terraform trace cp databricks.yml .databricks/bundle/default/terraform/terraform.tfstate trace $CLI bundle plan | contains.py 'Plan: 0 to add, 0 to change, 0 to delete, 1 unchanged' trace test ! -e .databricks/bundle/default/terraform/terraform.tfstate trace cmp .databricks/resources.remote.json .databricks/bundle/default/resources.json -echo "PASS: invalid caches handled" +echo "PASS: invalid local copies handled" -title "3. Missing remote\nExpect: bad JSON fails; local cache bytes remain unchanged.\n" +title "3. Missing remote checkpoint and invalid local copy\nExpect: recover confirmed DMS state with a new lineage and invalidate old plans.\n" trace MSYS_NO_PATHCONV=1 $CLI workspace delete "$state_path/resources.json" trace cp databricks.yml .databricks/bundle/default/resources.json -trace musterr $CLI bundle plan 2>&1 | contains.py 'Error' 'invalid character' -trace cmp databricks.yml .databricks/bundle/default/resources.json -echo "PASS: failure preserved cache" +trace $CLI bundle plan 2>&1 | contains.py 'Recovering deployment history state with a new lineage; existing saved plans are no longer valid' 'Plan: 0 to add, 0 to change, 0 to delete, 1 unchanged' +trace jq -e --slurpfile original .databricks/resources.remote.json '.lineage != $original[0].lineage and (.features | has("deployment_history")) and .state == {}' .databricks/bundle/default/resources.json +echo "PASS: DMS recovery replaced the invalid local copy" { trace MSYS_NO_PATHCONV=1 $CLI workspace import "$state_path/resources.json" --file .databricks/resources.remote.json --format AUTO --overwrite diff --git a/acceptance/bundle/dms/version-never-created/output.txt b/acceptance/bundle/dms/version-never-created/output.txt index 989744df818..e8a11547b16 100644 --- a/acceptance/bundle/dms/version-never-created/output.txt +++ b/acceptance/bundle/dms/version-never-created/output.txt @@ -12,9 +12,12 @@ API message: Internal error Files: 1 uploaded, 0 deleted >>> [CLI] bundle plan -o json +Warning: Recovering deployment history state with a new lineage; existing saved plans are no longer valid + { "plan_version": [PLAN_VERSION], "cli_version": "[CLI_VERSION]", + "lineage": "[UUID]", "features": { "deployment_history": {} } diff --git a/bundle/statemgmt/state_pull.go b/bundle/statemgmt/state_pull.go index 1fed9fe4980..c76eb073023 100644 --- a/bundle/statemgmt/state_pull.go +++ b/bundle/statemgmt/state_pull.go @@ -16,13 +16,17 @@ import ( "github.com/databricks/cli/bundle" "github.com/databricks/cli/bundle/config/engine" "github.com/databricks/cli/bundle/deploy" + "github.com/databricks/cli/bundle/direct/dstate" bundleenv "github.com/databricks/cli/bundle/env" "github.com/databricks/cli/libs/atomicfile" "github.com/databricks/cli/libs/diag" + "github.com/databricks/cli/libs/dms" "github.com/databricks/cli/libs/filer" "github.com/databricks/cli/libs/log" "github.com/databricks/cli/libs/logdiag" "github.com/databricks/cli/libs/shortid" + "github.com/databricks/databricks-sdk-go/service/bundledeployments" + "github.com/google/uuid" ) type AlwaysPull bool @@ -41,6 +45,10 @@ type StateDesc struct { IsLocal bool `json:"-"` AllStates []*StateDesc + + // Deployment retains a confirmed repair lookup for state initialization. + DeploymentID string `json:"-"` + Deployment *bundledeployments.Deployment `json:"-"` } func (s *StateDesc) String() string { @@ -84,8 +92,11 @@ func localRead(fullPath string, engine engine.EngineType) *StateDesc { state := &StateDesc{} if err != nil { state.readErr = fmt.Errorf("reading %s: %w", filepath.ToSlash(fullPath), err) - } else if err := json.Unmarshal(content, state); err != nil { - state.readErr = fmt.Errorf("parsing %s: %w", filepath.ToSlash(fullPath), err) + } else { + state, err = parseResourceState(content, engine) + if err != nil { + state.readErr = fmt.Errorf("parsing %s: %w", filepath.ToSlash(fullPath), err) + } } state.SourcePath = filepath.ToSlash(fullPath) @@ -96,7 +107,28 @@ func localRead(fullPath string, engine engine.EngineType) *StateDesc { return state } -func _filerRead(ctx context.Context, f filer.Filer, path string) (*StateDesc, error) { +// parseResourceState reads the state metadata while validating its JSON shape. +func parseResourceState(content []byte, engine engine.EngineType) (*StateDesc, error) { + state := &StateDesc{} + if !engine.IsDirect() { + return state, json.Unmarshal(content, state) + } + var data dstate.Database + err := json.Unmarshal(content, &data) + state.Serial = data.Serial + state.Lineage = data.Lineage + state.Features = data.Features + return state, err +} + +// isJSONStateError distinguishes damaged JSON from access and transport failures. +func isJSONStateError(err error) bool { + _, syntaxError := errors.AsType[*json.SyntaxError](err) + _, typeError := errors.AsType[*json.UnmarshalTypeError](err) + return syntaxError || typeError +} + +func _filerRead(ctx context.Context, f filer.Filer, path string, engine engine.EngineType) (*StateDesc, error) { r, err := f.Read(ctx, path) if err != nil { if errors.Is(err, fs.ErrNotExist) { @@ -110,25 +142,29 @@ func _filerRead(ctx context.Context, f filer.Filer, path string) (*StateDesc, er return nil, fmt.Errorf("reading data: %w", err) } - state := &StateDesc{} - err = json.Unmarshal(content, state) - if err != nil { - return nil, fmt.Errorf("parsing state: %w", err) - } + state, parseErr := parseResourceState(content, engine) // QQQ: would be nice to merge with filer root state.SourcePath = filepath.ToSlash(path) state.IsLocal = false state.Content = content + if parseErr != nil { + return state, fmt.Errorf("parsing state: %w", parseErr) + } return state, nil } func filerRead(ctx context.Context, f filer.Filer, path string, engine engine.EngineType) *StateDesc { - state, err := _filerRead(ctx, f, path) + state, err := _filerRead(ctx, f, path, engine) if err != nil { - logdiag.LogError(ctx, fmt.Errorf("reading %s: %w", path, err)) + if state == nil { + state = &StateDesc{SourcePath: path} + } + state.readErr = fmt.Errorf("reading %s: %w", path, err) } else if state != nil { log.Debugf(ctx, "read %s: %s", path, state.String()) + } + if state != nil { state.Engine = engine } return state @@ -148,15 +184,50 @@ func PullResourcesState(ctx context.Context, b *bundle.Bundle, alwaysPull Always configured = b.Config.Experimental.DeploymentHistory } historyEnabled := bundleenv.RecordsDeploymentHistory(ctx, configured) - states := readStates(ctx, b, AlwaysPull(bool(alwaysPull) || historyEnabled)) + states, readRemote := readStates(ctx, b, AlwaysPull(bool(alwaysPull) || historyEnabled)) if logdiag.HasError(ctx) { return nil } // DMS owns recorded resource state; a remote marker's serial is only a checkpoint. remoteRecorded := slices.ContainsFunc(states, func(state *StateDesc) bool { - return !state.IsLocal && state.Engine.IsDirect() && state.IsDMS() + return !state.IsLocal && state.Engine.IsDirect() && state.readErr == nil && state.IsDMS() }) - if historyEnabled || remoteRecorded { + for _, state := range states { + if state.IsLocal && (historyEnabled || remoteRecorded) { + continue + } + if state.readErr != nil && (!state.Engine.IsDirect() || !isJSONStateError(state.readErr)) { + logdiag.LogError(ctx, state.readErr) + } + } + if logdiag.HasError(ctx) { + return nil + } + repairCandidate := historyEnabled || slices.ContainsFunc(states, func(state *StateDesc) bool { + return state.Engine.IsDirect() && (isJSONStateError(state.readErr) || (state.IsLocal && state.IsDMS())) + }) + if readRemote && repairCandidate { + repaired, err := recoverDMSState(ctx, b, states) + if err != nil { + logdiag.LogError(ctx, err) + return nil + } + if repaired != nil { + states = []*StateDesc{repaired} + remoteRecorded = true + } + } + hasRemoteState := slices.ContainsFunc(states, func(state *StateDesc) bool { return !state.IsLocal }) + for _, state := range states { + if state.readErr == nil || (state.IsLocal && (remoteRecorded || (historyEnabled && hasRemoteState))) { + continue + } + logdiag.LogError(ctx, state.readErr) + } + if logdiag.HasError(ctx) { + return nil + } + if remoteRecorded || (historyEnabled && hasRemoteState) { states = slices.DeleteFunc(states, func(state *StateDesc) bool { if !state.IsLocal { return false @@ -169,12 +240,6 @@ func PullResourcesState(ctx context.Context, b *bundle.Bundle, alwaysPull Always return true }) } - for _, state := range states { - if state.readErr != nil { - logdiag.LogError(ctx, state.readErr) - } - } - if logdiag.HasError(ctx) { return nil } @@ -228,24 +293,25 @@ func PullResourcesState(ctx context.Context, b *bundle.Bundle, alwaysPull Always return winner } -func readStates(ctx context.Context, b *bundle.Bundle, alwaysPull AlwaysPull) []*StateDesc { +func readStates(ctx context.Context, b *bundle.Bundle, alwaysPull AlwaysPull) ([]*StateDesc, bool) { var states []*StateDesc remotePathDirect, localPathDirect := b.StateFilenameDirect(ctx) remotePathTerraform, localPathTerraform := b.StateFilenameTerraform(ctx) if logdiag.HasError(ctx) { - return nil + return nil, false } directLocalState := localRead(localPathDirect, engine.EngineDirect) terraformLocalState := localRead(localPathTerraform, engine.EngineTerraform) - if (directLocalState == nil && terraformLocalState == nil) || alwaysPull { + readRemote := (directLocalState == nil && terraformLocalState == nil) || bool(alwaysPull) + if readRemote { f, err := deploy.StateFiler(ctx, b) if err != nil { logdiag.LogError(ctx, err) - return nil + return nil, true } var wg sync.WaitGroup @@ -272,7 +338,64 @@ func readStates(ctx context.Context, b *bundle.Bundle, alwaysPull AlwaysPull) [] return a.Serial - b.Serial }) - return states + return states, readRemote +} + +// recoverDMSState reconstructs an unusable marker only for a confirmed recorded deployment. +func recoverDMSState(ctx context.Context, b *bundle.Bundle, states []*StateDesc) (*StateDesc, error) { + var surviving *StateDesc + for _, state := range states { + if state.readErr == nil { + if !state.Engine.IsDirect() || !state.IsDMS() || !state.IsLocal { + return nil, nil + } + surviving = state + } + } + + deploymentID, deployment, err := dms.FetchDeploymentFromStatePath(ctx, b.WorkspaceClient(ctx), b.Config.Workspace.StatePath) + if err != nil || deployment == nil { + return nil, err + } + + data := dstate.NewDatabase("", 0) + if surviving != nil { + var db dstate.DeploymentState + if err := db.Open(ctx, surviving.SourcePath, dstate.WithRecovery(false), dstate.WithWrite(false), dstate.WithDeploymentHistory(true), dstate.OpenDmsArgs{}); err != nil { + return nil, fmt.Errorf("recovering deployment history marker: %w", err) + } + data.Header = db.Data.Header + if len(db.Data.State) != 0 { + return nil, fmt.Errorf("state file for a recorded deployment carries %d resources, expected none", len(db.Data.State)) + } + if _, err := db.Finalize(ctx); err != nil { + return nil, err + } + } + if data.Lineage == "" { + data.Lineage = uuid.NewString() + logdiag.LogDiag(ctx, diag.Warningf("Recovering deployment history state with a new lineage; existing saved plans are no longer valid")[0]) + } + if data.Features == nil { + data.Features = make(map[string]struct{}) + } + data.Features[dstate.FeatureDeploymentHistory] = struct{}{} + content, err := json.Marshal(data) + if err != nil { + return nil, fmt.Errorf("reconstructing deployment history marker: %w", err) + } + remotePath, _ := b.StateFilenameDirect(ctx) + b.NeedsDMSStateUpload = true + return &StateDesc{ + Serial: data.Serial, + Lineage: data.Lineage, + Features: data.Features, + SourcePath: remotePath, + Content: content, + Engine: engine.EngineDirect, + DeploymentID: deploymentID, + Deployment: deployment, + }, nil } func validateStates(states []*StateDesc) error { diff --git a/bundle/statemgmt/state_recovery_test.go b/bundle/statemgmt/state_recovery_test.go new file mode 100644 index 00000000000..1ec526c37f5 --- /dev/null +++ b/bundle/statemgmt/state_recovery_test.go @@ -0,0 +1,161 @@ +package statemgmt_test + +import ( + "context" + "encoding/json" + "os" + "path" + "strings" + "testing" + + "github.com/databricks/cli/bundle" + "github.com/databricks/cli/bundle/config" + "github.com/databricks/cli/bundle/direct/dstate" + "github.com/databricks/cli/bundle/statemgmt" + "github.com/databricks/cli/libs/dms" + "github.com/databricks/cli/libs/logdiag" + "github.com/databricks/cli/libs/testserver" + "github.com/databricks/databricks-sdk-go" + "github.com/databricks/databricks-sdk-go/service/bundledeployments" + "github.com/databricks/databricks-sdk-go/service/workspace" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +const ( + recordedMarker = `{"state_version":3,"cli_version":"0.1.0","lineage":"recorded-lineage","serial":2,"features":{"deployment_history":{}},"state":{}}` + directSnapshot = `{"state_version":3,"lineage":"direct-lineage","serial":99,"state":{"resources.jobs.foo":{"__id__":"456","state":{"name":"foo"}}}}` + statePath = "/Workspace/state" +) + +// newStateTestBundle isolates local files and diagnostics for state I/O tests. +func newStateTestBundle(t *testing.T) (context.Context, *bundle.Bundle, *testserver.Server) { + t.Helper() + server := testserver.New(t) + client, err := databricks.NewWorkspaceClient(&databricks.Config{Host: server.URL, Token: "testtoken"}) + require.NoError(t, err) + b := &bundle.Bundle{ + BundleRootPath: t.TempDir(), + Config: config.Root{ + Bundle: config.Bundle{Target: "test"}, + Workspace: config.Workspace{StatePath: statePath}, + Experimental: &config.Experimental{DeploymentHistory: new(false)}, + }, + } + b.SetWorkpaceClient(client) + ctx := logdiag.InitContext(t.Context()) + logdiag.SetCollect(ctx, true) + return ctx, b, server +} + +func TestPullResourcesStateRepair(t *testing.T) { + tests := []struct { + name string + remote string + local string + node bool + historyEnabled bool + noLookup bool + remoteError bool + recordError bool + wal bool + wantErr string + wantRepair bool + wantRecorded bool + wantNew bool + }{ + {name: "new direct deployment", noLookup: true}, + {name: "local recorded marker without backend", local: recordedMarker, wantRecorded: true}, + {name: "syntax error", remote: "{", node: true, wantRepair: true, wantNew: true}, + {name: "header type error", remote: `{"serial":"invalid"}`, node: true, wantRepair: true, wantNew: true}, + {name: "resource type error", remote: `{"state":[]}`, node: true, wantRepair: true, wantNew: true}, + {name: "missing marker", node: true, historyEnabled: true, wantRepair: true, wantNew: true}, + {name: "surviving recorded header", remote: "{", local: recordedMarker, node: true, wantRepair: true}, + {name: "missing marker with surviving header", local: recordedMarker, node: true, wantRepair: true}, + {name: "both copies malformed", remote: "{", local: "{", node: true, wantRepair: true, wantNew: true}, + {name: "corrupt direct without deployment", remote: "{", wantErr: "parsing state"}, + {name: "local corruption without deployment", local: "{", wantErr: "parsing"}, + {name: "node without record", remote: "{", node: true, recordError: true, wantErr: "record not found"}, + {name: "remote permission failure", node: true, remoteError: true, wantErr: "access denied"}, + {name: "valid remote direct snapshot", remote: directSnapshot, node: true}, + {name: "valid local direct snapshot", local: directSnapshot, node: true}, + {name: "corrupt remote with direct local resources", remote: "{", local: directSnapshot, node: true, wantErr: "parsing state"}, + {name: "newer surviving schema", remote: "{", local: strings.Replace(recordedMarker, `"state_version":3`, `"state_version":99`, 1), node: true, wantErr: "newer than supported"}, + {name: "unknown surviving feature", remote: "{", local: strings.Replace(recordedMarker, `"deployment_history":{}`, `"deployment_history":{},"future_feature":{}`, 1), node: true, wantErr: "features this CLI does not support"}, + {name: "surviving WAL", remote: "{", local: recordedMarker, node: true, wal: true, wantErr: "unexpected WAL"}, + {name: "surviving resources are not a marker", remote: "{", local: strings.Replace(directSnapshot, `"serial":99`, `"serial":99,"features":{"deployment_history":{}}`, 1), node: true, wantErr: "expected none"}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + ctx, b, server := newStateTestBundle(t) + b.Config.Experimental.DeploymentHistory = new(tt.historyEnabled) + _, localPath := b.StateFilenameDirect(ctx) + if tt.local != "" { + writeLocalState(t, localPath, tt.local) + } + if tt.wal { + writeLocalState(t, localPath+".wal", "{}\n") + } + server.Handle("GET", "/api/2.0/workspace/get-status", func(req testserver.Request) any { + requestedPath := req.URL.Query().Get("path") + switch requestedPath { + case path.Join(statePath, "resources.json"): + if tt.remoteError { + return testserver.Response{StatusCode: 403, Body: map[string]string{"error_code": "PERMISSION_DENIED", "message": "access denied"}} + } + if tt.remote != "" { + return workspace.ObjectInfo{Path: requestedPath, ObjectType: workspace.ObjectTypeFile} + } + case path.Join(statePath, dms.DeploymentNodeName): + assert.False(t, tt.noLookup, "fresh direct state must not query DMS") + if tt.node { + return workspace.ObjectInfo{Path: requestedPath, ObjectId: 123} + } + } + return testserver.Response{StatusCode: 404, Body: map[string]string{"error_code": "RESOURCE_DOES_NOT_EXIST", "message": "not found"}} + }) + server.Handle("GET", "/api/2.0/workspace/export", func(req testserver.Request) any { + return []byte(tt.remote) + }) + server.Handle("GET", "/api/2.0/bundle/deployments/123", func(req testserver.Request) any { + if tt.recordError { + return testserver.Response{StatusCode: 404, Body: map[string]string{"error_code": "RESOURCE_DOES_NOT_EXIST", "message": "record not found"}} + } + return bundledeployments.Deployment{Name: dms.DeploymentName("123"), LastVersionId: "7"} + }) + + state := statemgmt.PullResourcesState(ctx, b, statemgmt.AlwaysPull(true)) + assert.Equal(t, tt.wantRepair, b.NeedsDMSStateUpload) + if tt.wantErr != "" { + require.True(t, logdiag.HasError(ctx)) + assert.Contains(t, logdiag.GetFirstErrorSummary(ctx), tt.wantErr) + return + } + require.False(t, logdiag.HasError(ctx), "%v", logdiag.FlushCollected(ctx)) + require.NotNil(t, state) + if !tt.wantRepair { + assert.Equal(t, tt.wantRecorded, state.IsDMS()) + return + } + assert.True(t, state.IsDMS()) + assert.Equal(t, "123", state.DeploymentID) + require.NotNil(t, state.Deployment) + assert.Equal(t, "7", state.Deployment.LastVersionId) + content, err := os.ReadFile(localPath) + require.NoError(t, err) + var reconstructed dstate.Database + require.NoError(t, json.Unmarshal(content, &reconstructed)) + assert.Empty(t, reconstructed.State) + if tt.wantNew { + assert.NotEmpty(t, state.Lineage) + assert.NotEqual(t, "recorded-lineage", state.Lineage) + assert.Equal(t, 1, logdiag.NumWarnings(ctx)) + } else { + assert.Equal(t, "recorded-lineage", state.Lineage) + assert.Equal(t, 2, state.Serial) + assert.Equal(t, "0.1.0", reconstructed.CLIVersion) + assert.Zero(t, logdiag.NumWarnings(ctx)) + } + }) + } +} diff --git a/cmd/bundle/utils/process.go b/cmd/bundle/utils/process.go index 489f1ad9685..9416ea807ee 100644 --- a/cmd/bundle/utils/process.go +++ b/cmd/bundle/utils/process.go @@ -4,7 +4,6 @@ import ( "context" "errors" "fmt" - "path" "path/filepath" "strconv" "time" @@ -34,8 +33,6 @@ import ( "github.com/databricks/cli/libs/sync" "github.com/databricks/cli/libs/telemetry" "github.com/databricks/cli/libs/telemetry/protos" - "github.com/databricks/databricks-sdk-go" - "github.com/databricks/databricks-sdk-go/apierr" "github.com/databricks/databricks-sdk-go/service/bundledeployments" "github.com/databricks/databricks-sdk-go/useragent" "github.com/spf13/cobra" @@ -379,7 +376,10 @@ func ProcessBundleRet(cmd *cobra.Command, opts ProcessOptions) (b *bundle.Bundle // Repeated migrations may already have a DMS deployment even when the state lacks its history marker. if needDirectState && (stateDesc.IsDMS() || b.MigratingToDMS) { var err error - dmsDeploymentID, dmsDeployment, err = fetchDeploymentFromStatePath(ctx, b.WorkspaceClient(ctx), b.Config.Workspace.StatePath) + dmsDeploymentID, dmsDeployment = stateDesc.DeploymentID, stateDesc.Deployment + if dmsDeployment == nil { + dmsDeploymentID, dmsDeployment, err = dms.FetchDeploymentFromStatePath(ctx, b.WorkspaceClient(ctx), b.Config.Workspace.StatePath) + } if err != nil { logdiag.LogError(ctx, err) return b, stateDesc, root.ErrAlreadyPrinted @@ -597,36 +597,6 @@ func ValidateEngineSetting(ctx context.Context, b *bundle.Bundle) error { return nil } -// Lookup and return the deployment object from ${workspace.state_path}/resources.deployment.json -// -// TODO: a deployment is only usable when both the node and the service's record exist, and a -// half-created one - node present, record missing - blocks the bundle here even though -// CreateDeployment already recovers from it. Move this behind a dms.ReadDeployment(ctx, statePath) -// that returns an empty id and version unless both halves are there, leaving the caller to call -// CreateDeployment to create or finalize it. -// -// TODO: ask the service for a lookup by state path, so this is one round trip rather than two - a -// workspace lookup to turn the node into an id, then a get by that id. -func fetchDeploymentFromStatePath(ctx context.Context, w *databricks.WorkspaceClient, statePath string) (string, *bundledeployments.Deployment, error) { - nodePath := path.Join(statePath, dms.DeploymentNodeName) - - obj, err := w.Workspace.GetStatusByPath(ctx, nodePath) - if errors.Is(err, apierr.ErrNotFound) || errors.Is(err, apierr.ErrResourceDoesNotExist) { - return "", nil, nil - } - if err != nil { - return "", nil, fmt.Errorf("looking up deployment at %s: %w", nodePath, err) - } - deploymentID := strconv.FormatInt(obj.ObjectId, 10) - deployment, err := w.BundleDeployments.GetDeployment(ctx, bundledeployments.GetDeploymentRequest{ - Name: dms.DeploymentName(deploymentID), - }) - if err != nil { - return "", nil, err - } - return deploymentID, deployment, nil -} - // parseLastVersionID parses the deployment's last recorded version, which the service reports // as a string. It returns 0 when the deployment does not exist yet or has no recorded version. func parseLastVersionID(dmsDeployment *bundledeployments.Deployment) (int, error) { @@ -652,7 +622,11 @@ func OpenDirectStateForRead(ctx context.Context, b *bundle.Bundle, stateDesc *st return enforceDeploymentHistorySetting(ctx, b, stateDesc, false) } - dmsDeploymentID, dmsDeployment, err := fetchDeploymentFromStatePath(ctx, b.WorkspaceClient(ctx), b.Config.Workspace.StatePath) + dmsDeploymentID, dmsDeployment := stateDesc.DeploymentID, stateDesc.Deployment + var err error + if dmsDeployment == nil { + dmsDeploymentID, dmsDeployment, err = dms.FetchDeploymentFromStatePath(ctx, b.WorkspaceClient(ctx), b.Config.Workspace.StatePath) + } if err != nil { return err } diff --git a/libs/dms/metadata.go b/libs/dms/metadata.go index f9e23a67db0..16092d66033 100644 --- a/libs/dms/metadata.go +++ b/libs/dms/metadata.go @@ -1,11 +1,16 @@ package dms import ( + "context" + "errors" "fmt" + "path" "strconv" "strings" "github.com/databricks/cli/libs/structs/structdiff" + "github.com/databricks/databricks-sdk-go" + "github.com/databricks/databricks-sdk-go/apierr" "github.com/databricks/databricks-sdk-go/service/bundledeployments" ) @@ -85,6 +90,37 @@ func NextVersion(lastVersionID string) (int, error) { // Keep it equal to the service's DEPLOYMENT_NODE_NAME. const DeploymentNodeName = "resources.deployment.json" +// FetchDeploymentFromStatePath looks up the deployment node and its service record. +// Lookup and return the deployment object from ${workspace.state_path}/resources.deployment.json +// +// TODO: a deployment is only usable when both the node and the service's record exist, and a +// half-created one - node present, record missing - blocks the bundle here even though +// CreateDeployment already recovers from it. Move this behind a dms.ReadDeployment(ctx, statePath) +// that returns an empty id and version unless both halves are there, leaving the caller to call +// CreateDeployment to create or finalize it. +// +// TODO: ask the service for a lookup by state path, so this is one round trip rather than two - a +// workspace lookup to turn the node into an id, then a get by that id. +func FetchDeploymentFromStatePath(ctx context.Context, w *databricks.WorkspaceClient, statePath string) (string, *bundledeployments.Deployment, error) { + nodePath := path.Join(statePath, DeploymentNodeName) + + obj, err := w.Workspace.GetStatusByPath(ctx, nodePath) + if errors.Is(err, apierr.ErrNotFound) || errors.Is(err, apierr.ErrResourceDoesNotExist) { + return "", nil, nil + } + if err != nil { + return "", nil, fmt.Errorf("looking up deployment at %s: %w", nodePath, err) + } + deploymentID := strconv.FormatInt(obj.ObjectId, 10) + deployment, err := w.BundleDeployments.GetDeployment(ctx, bundledeployments.GetDeploymentRequest{ + Name: DeploymentName(deploymentID), + }) + if err != nil { + return "", nil, err + } + return deploymentID, deployment, nil +} + // DeploymentName, VersionName and OperationName build the resource names the // service uses, so callers pass only ids. func DeploymentName(deploymentID string) string { diff --git a/libs/dms/metadata_fetch_test.go b/libs/dms/metadata_fetch_test.go new file mode 100644 index 00000000000..457e85b168b --- /dev/null +++ b/libs/dms/metadata_fetch_test.go @@ -0,0 +1,60 @@ +package dms_test + +import ( + "testing" + + "github.com/databricks/cli/libs/dms" + "github.com/databricks/databricks-sdk-go/apierr" + "github.com/databricks/databricks-sdk-go/experimental/mocks" + "github.com/databricks/databricks-sdk-go/service/bundledeployments" + "github.com/databricks/databricks-sdk-go/service/workspace" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/mock" + "github.com/stretchr/testify/require" +) + +func TestFetchDeploymentFromStatePath(t *testing.T) { + tests := []struct { + name string + nodeError error + recordError error + wantError bool + }{ + {name: "confirmed deployment"}, + {name: "missing node", nodeError: apierr.ErrNotFound}, + {name: "absent node", nodeError: apierr.ErrResourceDoesNotExist}, + {name: "node permission denied", nodeError: apierr.ErrPermissionDenied, wantError: true}, + {name: "missing service record", recordError: apierr.ErrNotFound, wantError: true}, + {name: "service permission denied", recordError: apierr.ErrPermissionDenied, wantError: true}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + m := mocks.NewMockWorkspaceClient(t) + m.GetMockWorkspaceAPI().EXPECT().GetStatusByPath(mock.Anything, "/Workspace/state/"+dms.DeploymentNodeName). + Return(&workspace.ObjectInfo{ObjectId: 123}, tt.nodeError).Once() + deployment := &bundledeployments.Deployment{Name: dms.DeploymentName("123"), LastVersionId: "7"} + if tt.nodeError == nil { + m.GetMockBundleDeploymentsAPI().EXPECT().GetDeployment(mock.Anything, bundledeployments.GetDeploymentRequest{Name: deployment.Name}). + Return(deployment, tt.recordError).Once() + } + id, found, err := dms.FetchDeploymentFromStatePath(t.Context(), m.WorkspaceClient, "/Workspace/state") + if tt.wantError { + require.Error(t, err) + if tt.nodeError != nil { + assert.ErrorIs(t, err, tt.nodeError) + } else { + assert.ErrorIs(t, err, tt.recordError) + } + return + } + require.NoError(t, err) + if tt.nodeError != nil { + assert.Empty(t, id) + assert.Nil(t, found) + return + } + assert.Equal(t, "123", id) + assert.Same(t, deployment, found) + }) + } +}