diff --git a/acceptance/bundle/resources/job_runs/destroy_unfinished_run/databricks.yml b/acceptance/bundle/resources/job_runs/destroy_unfinished_run/databricks.yml new file mode 100644 index 0000000000..177a9ca222 --- /dev/null +++ b/acceptance/bundle/resources/job_runs/destroy_unfinished_run/databricks.yml @@ -0,0 +1,15 @@ +bundle: + name: job-runs-destroy-unfinished-run + +resources: + jobs: + my_job: + name: my-job + tasks: + - task_key: main + notebook_task: + notebook_path: /Workspace/test + + job_runs: + my_run: + job_id: ${resources.jobs.my_job.id} diff --git a/acceptance/bundle/resources/job_runs/destroy_unfinished_run/out.test.toml b/acceptance/bundle/resources/job_runs/destroy_unfinished_run/out.test.toml new file mode 100644 index 0000000000..0938e67898 --- /dev/null +++ b/acceptance/bundle/resources/job_runs/destroy_unfinished_run/out.test.toml @@ -0,0 +1,2 @@ +Cloud = false +EnvMatrix.DATABRICKS_BUNDLE_ENGINE = ["direct"] diff --git a/acceptance/bundle/resources/job_runs/destroy_unfinished_run/output.txt b/acceptance/bundle/resources/job_runs/destroy_unfinished_run/output.txt new file mode 100644 index 0000000000..1b1fd310b0 --- /dev/null +++ b/acceptance/bundle/resources/job_runs/destroy_unfinished_run/output.txt @@ -0,0 +1,44 @@ + +=== the deploy stops waiting before the run finishes +>>> errcode [CLI] bundle deploy +Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/job-runs-destroy-unfinished-run/default/files... +Error: cannot create resources.job_runs.my_run: waiting after creating id=[MY_RUN_ID]: Fault injected by test. (403 INJECTED) + +Endpoint: GET [DATABRICKS_URL]/api/2.2/jobs/runs/get?run_id=[MY_RUN_ID] +HTTP Status: 403 Forbidden +API error_code: INJECTED +API message: Fault injected by test. + +Files: 4 uploaded, 0 deleted + +Exit code: 1 + +>>> read_id.py my_run +[MY_RUN_ID] + +=== destroy cancels the run before deleting it +>>> [CLI] bundle destroy --auto-approve +Warn: planning resources.job_runs.my_run: reading resources.job_runs.my_run id="[MY_RUN_ID]": Fault injected by test. +The following resources will be deleted: + delete resources.job_runs.my_run + delete resources.jobs.my_job + +All files and directories at the following location will be deleted: /Workspace/Users/[USERNAME]/.bundle/job-runs-destroy-unfinished-run/default + +Destroy: 2 deleted + +>>> print_requests.py //jobs/runs +{ + "method": "POST", + "path": "/api/2.2/jobs/runs/cancel", + "body": { + "run_id": [MY_RUN_ID] + } +} +{ + "method": "POST", + "path": "/api/2.2/jobs/runs/delete", + "body": { + "run_id": [MY_RUN_ID] + } +} diff --git a/acceptance/bundle/resources/job_runs/destroy_unfinished_run/script b/acceptance/bundle/resources/job_runs/destroy_unfinished_run/script new file mode 100644 index 0000000000..61074d97ee --- /dev/null +++ b/acceptance/bundle/resources/job_runs/destroy_unfinished_run/script @@ -0,0 +1,17 @@ +cleanup() { + rm -f out.requests.txt +} +trap cleanup EXIT + +# Fake workspace settles a run on the first successful poll, so both the deploy +# wait and destroy's refresh are faulted: delete sees an in-progress run. +fault.py "GET /api/2.2/jobs/runs/get" 403 0 2 + +title "the deploy stops waiting before the run finishes" +trace errcode $CLI bundle deploy +trace read_id.py my_run + +# jobs/runs/delete rejects an active run, so the run is cancelled first. +title "destroy cancels the run before deleting it" +trace $CLI bundle destroy --auto-approve +trace print_requests.py //jobs/runs diff --git a/bundle/direct/dresources/job_run_test.go b/bundle/direct/dresources/job_run_test.go index cef8cadecc..0ffc1c4804 100644 --- a/bundle/direct/dresources/job_run_test.go +++ b/bundle/direct/dresources/job_run_test.go @@ -10,9 +10,6 @@ import ( "testing" "time" - "github.com/databricks/cli/bundle/config/resources" - "github.com/databricks/cli/bundle/deployplan" - "github.com/databricks/cli/libs/cmdio" "github.com/databricks/cli/libs/structs/structpath" "github.com/databricks/cli/libs/testserver" "github.com/databricks/databricks-sdk-go" @@ -68,28 +65,6 @@ func waitForTestRun(t *testing.T, ctx context.Context, client *databricks.Worksp return r.WaitAfterCreate(ctx, "123", &JobRunState{}) } -func TestJobRunWaitSucceeds(t *testing.T) { - client := jobRunClient(t, &jobs.RunState{ - LifeCycleState: jobs.RunLifeCycleStateTerminated, - ResultState: jobs.RunResultStateSuccess, - }) - - remote, err := waitForTestRun(t, t.Context(), client) - - require.NoError(t, err) - require.NotNil(t, remote.State) - assert.Equal(t, jobs.RunResultStateSuccess, remote.State.ResultState) -} - -func TestReportRunLineIncludesResourceKey(t *testing.T) { - ctx, stderr := cmdio.NewTestContextWithStderr(t.Context()) - ctx = WithResourceKey(ctx, "job_runs.my_run") - - reportRunLine(ctx, 123, "SUCCESS") - - assert.Equal(t, "Output from job_runs.my_run: id=123: SUCCESS\n", stderr.String()) -} - func TestJobRunWaitFailsOnFailedResult(t *testing.T) { client := jobRunClient(t, &jobs.RunState{ LifeCycleState: jobs.RunLifeCycleStateTerminated, @@ -102,36 +77,6 @@ func TestJobRunWaitFailsOnFailedResult(t *testing.T) { require.ErrorContains(t, err, "did not succeed: FAILED: task failed") } -func TestJobRunWaitReportsFailedTask(t *testing.T) { - failed := &jobs.RunState{ - LifeCycleState: jobs.RunLifeCycleStateTerminated, - ResultState: jobs.RunResultStateFailed, - } - server := testserver.New(t) - server.Handle("GET", "/api/2.2/jobs/runs/get", func(req testserver.Request) any { - return jobs.Run{ - RunId: 123, - JobId: 456, - State: failed, - Tasks: []jobs.RunTask{ - {TaskKey: "ok", RunId: 998, State: &jobs.RunState{ - LifeCycleState: jobs.RunLifeCycleStateTerminated, - ResultState: jobs.RunResultStateSuccess, - }}, - {TaskKey: "main", RunId: 999, State: failed}, - }, - } - }) - server.Handle("GET", "/api/2.2/jobs/runs/get-output", func(req testserver.Request) any { - return jobs.RunOutput{Error: "notebook not found"} - }) - - _, err := waitForTestRun(t, t.Context(), jobRunClientFor(t, server)) - - require.ErrorContains(t, err, `task "main": notebook not found`) - assert.NotContains(t, err.Error(), `task "ok"`) -} - // Without the deprecated per-task state, a failed task is told apart from a // skipped one by its termination details. func TestJobRunWaitReportsFailedTaskWithoutDeprecatedState(t *testing.T) { @@ -257,39 +202,6 @@ func TestJobRunWaitFailsOnInternalError(t *testing.T) { require.ErrorContains(t, err, testRunPageLink) } -// A real workspace reports a run whose task failed as INTERNAL_ERROR in the -// deprecated life_cycle_state. The failing task still has to be named. -func TestJobRunWaitReportsFailedTaskOfInternalErrorRun(t *testing.T) { - server := testserver.New(t) - server.Handle("GET", "/api/2.2/jobs/runs/get", func(req testserver.Request) any { - return jobs.Run{ - RunId: 123, - JobId: 456, - RunPageUrl: testRunPageURL, - State: &jobs.RunState{ - LifeCycleState: jobs.RunLifeCycleStateInternalError, - ResultState: jobs.RunResultStateFailed, - StateMessage: "Task main failed with message: Workload failed, see run output for details.", - }, - Tasks: []jobs.RunTask{ - {TaskKey: "main", RunId: 999, State: &jobs.RunState{ - LifeCycleState: jobs.RunLifeCycleStateTerminated, - ResultState: jobs.RunResultStateFailed, - }}, - }, - } - }) - server.Handle("GET", "/api/2.2/jobs/runs/get-output", func(req testserver.Request) any { - return jobs.RunOutput{Error: "RuntimeError: intentional failure"} - }) - - _, err := waitForTestRun(t, t.Context(), jobRunClientFor(t, server)) - - require.ErrorContains(t, err, "run did not succeed: FAILED") - require.ErrorContains(t, err, `task "main": RuntimeError: intentional failure`) - require.ErrorContains(t, err, testRunPageLink) -} - func TestJobRunWaitReportsOnlyTheLastAttemptOfATask(t *testing.T) { failed := &jobs.RunState{ LifeCycleState: jobs.RunLifeCycleStateTerminated, @@ -332,86 +244,6 @@ func TestJobRunWaitAbandonedLinksTheRun(t *testing.T) { require.ErrorContains(t, err, testRunPageLink) } -// An abandoned wait leaves the run going with its id recorded, so the next deploy -// reads an empty outcome, which result_state drift catches. -func TestJobRunReadOfUnfinishedRunReportsNoResult(t *testing.T) { - client := jobRunClient(t, &jobs.RunState{LifeCycleState: jobs.RunLifeCycleStateRunning}) - - remote, err := (&ResourceJobRun{}).New(client).DoRead(t.Context(), "123") - - require.NoError(t, err) - require.NotNil(t, remote.State) - assert.Equal(t, jobs.RunLifeCycleStateRunning, remote.State.LifeCycleState) - assert.Empty(t, remote.ResultState) -} - -// PrepareState records the outcome the run must reach, the same for every run, -// so the planner has something to compare the remote against. -func TestJobRunPrepareStateRequiresSuccess(t *testing.T) { - state := (&ResourceJobRun{}).PrepareState(&resources.JobRun{RunNow: jobs.RunNow{JobId: 456}}) - - assert.Equal(t, jobs.RunResultStateSuccess, state.ResultState) -} - -func TestJobRunPrepareStateOnBundleDeploy(t *testing.T) { - t.Run("unset", func(t *testing.T) { - state := (&ResourceJobRun{}).PrepareState(&resources.JobRun{}) - assert.Nil(t, state.Lifecycle) - }) - - t.Run("armed", func(t *testing.T) { - on := true - input := &resources.JobRun{ - Lifecycle: &resources.JobRunLifecycle{ - Triggers: []resources.JobRunTrigger{{OnBundleDeploy: &on}}, - }, - } - first := (&ResourceJobRun{}).PrepareState(input) - require.NotNil(t, first.Lifecycle) - require.NotNil(t, first.Lifecycle.Triggers) - assert.NotEmpty(t, first.Lifecycle.Triggers.OnBundleDeploy) - - second := (&ResourceJobRun{}).PrepareState(input) - assert.NotEqual(t, first.Lifecycle.Triggers.OnBundleDeploy, second.Lifecycle.Triggers.OnBundleDeploy) - }) -} - -func TestJobRunOverrideChangeDescTriggerRemoved(t *testing.T) { - r := &ResourceJobRun{} - - t.Run("clearing lifecycle downgrades to skip", func(t *testing.T) { - change := &ChangeDesc{ - Action: deployplan.Recreate, - Old: &JobRunLifecycleState{Triggers: &JobRunTriggersState{OnBundleDeploy: "old"}}, - New: nil, - } - require.NoError(t, r.OverrideChangeDesc(t.Context(), structpath.MustParsePath("lifecycle"), change, nil)) - assert.Equal(t, deployplan.Skip, change.Action) - assert.Equal(t, "trigger removed", change.Reason) - }) - - t.Run("clearing on_bundle_deploy leaf downgrades to skip", func(t *testing.T) { - change := &ChangeDesc{ - Action: deployplan.Recreate, - Old: "old", - New: "", - } - require.NoError(t, r.OverrideChangeDesc(t.Context(), structpath.MustParsePath("lifecycle.triggers.on_bundle_deploy"), change, nil)) - assert.Equal(t, deployplan.Skip, change.Action) - assert.Equal(t, "trigger removed", change.Reason) - }) - - t.Run("fresh fingerprint still recreates", func(t *testing.T) { - change := &ChangeDesc{ - Action: deployplan.Recreate, - Old: "old", - New: "new", - } - require.NoError(t, r.OverrideChangeDesc(t.Context(), structpath.MustParsePath("lifecycle.triggers.on_bundle_deploy"), change, nil)) - assert.Equal(t, deployplan.Recreate, change.Action) - }) -} - // The planner diffs RemapState(remote) against PrepareState(config), so a run // that did not end in SUCCESS has to surface as a difference on result_state. func TestJobRunRemapStateCarriesTheOutcome(t *testing.T) { @@ -503,68 +335,25 @@ func TestJobRunCreateSendsAFreshIdempotencyToken(t *testing.T) { assert.Empty(t, config.IdempotencyToken) } -// jobRunDeletion records what the fake workspace saw while a run was deleted. -type jobRunDeletion struct { - cancelled atomic.Bool - settled atomic.Bool - settledAtDelete atomic.Bool -} - -// jobRunDeleteClient returns a client for a run in the given state, whose cancel -// settles one poll late the way the API's asynchronous cancellation does. -func jobRunDeleteClient(t *testing.T, state *jobs.RunState) (*databricks.WorkspaceClient, *jobRunDeletion) { - t.Helper() - var deletion jobRunDeletion - cancelled := &jobs.RunState{ - LifeCycleState: jobs.RunLifeCycleStateTerminated, - ResultState: jobs.RunResultStateCanceled, - } - +func TestJobRunDeleteLeavesFinishedRunAlone(t *testing.T) { + var cancelled atomic.Bool server := testserver.New(t) server.Handle("GET", "/api/2.2/jobs/runs/get", func(req testserver.Request) any { - current := state - switch { - case deletion.settled.Load(): - current = cancelled - case deletion.cancelled.Load(): - // Report the run's old state once more, then settle on the next poll. - deletion.settled.Store(true) - } - return jobs.Run{RunId: 123, JobId: 456, State: current} + return jobs.Run{RunId: 123, JobId: 456, State: &jobs.RunState{ + LifeCycleState: jobs.RunLifeCycleStateTerminated, + ResultState: jobs.RunResultStateSuccess, + }} }) server.Handle("POST", "/api/2.2/jobs/runs/cancel", func(req testserver.Request) any { - deletion.cancelled.Store(true) + cancelled.Store(true) return testserver.Response{} }) server.Handle("POST", "/api/2.2/jobs/runs/delete", func(req testserver.Request) any { - deletion.settledAtDelete.Store(deletion.settled.Load()) return testserver.Response{} }) - return jobRunClientFor(t, server), &deletion -} - -func deleteTestRun(t *testing.T, client *databricks.WorkspaceClient) error { - t.Helper() - return (&ResourceJobRun{}).New(client).DoDelete(t.Context(), "123", &JobRunState{}) -} - -func TestJobRunDeleteCancelsUnfinishedRun(t *testing.T) { - // An interrupted wait leaves the run going, and jobs/runs/delete rejects it. - client, deletion := jobRunDeleteClient(t, &jobs.RunState{LifeCycleState: jobs.RunLifeCycleStateRunning}) - - require.NoError(t, deleteTestRun(t, client)) - - assert.True(t, deletion.cancelled.Load(), "expected the run to be cancelled") - assert.True(t, deletion.settledAtDelete.Load(), "expected the delete to wait for the cancellation to settle") -} - -func TestJobRunDeleteLeavesFinishedRunAlone(t *testing.T) { - client, deletion := jobRunDeleteClient(t, &jobs.RunState{ - LifeCycleState: jobs.RunLifeCycleStateTerminated, - ResultState: jobs.RunResultStateSuccess, - }) + r := (&ResourceJobRun{}).New(jobRunClientFor(t, server)) - require.NoError(t, deleteTestRun(t, client)) + require.NoError(t, r.DoDelete(t.Context(), "123", &JobRunState{})) - assert.False(t, deletion.cancelled.Load(), "a run that already finished has nothing to cancel") + assert.False(t, cancelled.Load(), "a run that already finished has nothing to cancel") }