diff --git a/.nextchanges/bundles/job-runs-wait-for-completion.md b/.nextchanges/bundles/job-runs-wait-for-completion.md new file mode 100644 index 00000000000..d7052ab4b5a --- /dev/null +++ b/.nextchanges/bundles/job-runs-wait-for-completion.md @@ -0,0 +1 @@ +direct: the experimental `job_runs` resource now waits for the triggered run to finish, so other resources can reference its outcome (e.g. `${resources.job_runs.nightly.state.result_state}`). A run that does not succeed fails the deploy, naming the failed task, and is run again on the next deploy. diff --git a/acceptance/bundle/invariant/configs/job_run.yml.tmpl b/acceptance/bundle/invariant/configs/job_run.yml.tmpl index a09987da02a..2fab76ff020 100644 --- a/acceptance/bundle/invariant/configs/job_run.yml.tmpl +++ b/acceptance/bundle/invariant/configs/job_run.yml.tmpl @@ -6,14 +6,13 @@ resources: foo: name: test-job-$UNIQUE_NAME tasks: + # Deploying a job_run actually runs the job, so use a condition task, + # which needs no workspace files or compute. - task_key: only_task - notebook_task: - notebook_path: /Shared/notebook - new_cluster: - spark_version: $DEFAULT_SPARK_VERSION - node_type_id: $NODE_TYPE_ID - instance_pool_id: $TEST_INSTANCE_POOL_ID - num_workers: 1 + condition_task: + op: EQUAL_TO + left: "1" + right: "1" job_runs: foo_run: diff --git a/acceptance/bundle/refschema/out.fields.txt b/acceptance/bundle/refschema/out.fields.txt index ffec0976857..9fb0058ca7b 100644 --- a/acceptance/bundle/refschema/out.fields.txt +++ b/acceptance/bundle/refschema/out.fields.txt @@ -870,6 +870,7 @@ resources.job_runs.*.python_params[*] string ALL resources.job_runs.*.queue *jobs.QueueSettings ALL resources.job_runs.*.queue.enabled bool ALL resources.job_runs.*.resolved_job_id int64 INPUT +resources.job_runs.*.result_state jobs.RunResultState REMOTE STATE resources.job_runs.*.run_id int64 REMOTE resources.job_runs.*.run_name string REMOTE resources.job_runs.*.run_page_url string REMOTE diff --git a/acceptance/bundle/resources/job_runs/basic/output.txt b/acceptance/bundle/resources/job_runs/basic/output.txt index 14018b93ec6..67d2a32021b 100644 --- a/acceptance/bundle/resources/job_runs/basic/output.txt +++ b/acceptance/bundle/resources/job_runs/basic/output.txt @@ -34,6 +34,8 @@ Resources: >>> [CLI] bundle deploy Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/job-runs-basic/default/files... Deploying resources... +job run [MY_RUN_ID]: Run URL: [DATABRICKS_URL]/jobs/[MY_JOB_ID]/runs/[MY_RUN_ID]?o=[NUMID] +job run [MY_RUN_ID]: [TIMESTAMP] "my-job" TERMINATED SUCCESS Updating deployment state... Deployment complete! diff --git a/acceptance/bundle/resources/job_runs/failed_run/databricks.yml.tmpl b/acceptance/bundle/resources/job_runs/failed_run/databricks.yml.tmpl new file mode 100644 index 00000000000..a2b382f501a --- /dev/null +++ b/acceptance/bundle/resources/job_runs/failed_run/databricks.yml.tmpl @@ -0,0 +1,40 @@ +bundle: + name: job-runs-failed-run + +workspace: + root_path: ~/.bundle/$UNIQUE_NAME + +resources: + jobs: + my_job: + name: test-job-$UNIQUE_NAME + tasks: + # Serverless keeps the run to a few minutes: the Jobs API retries the + # failed task once before giving up on it. + - task_key: main + spark_python_task: + python_file: ./fail.py + environment_key: default + + environments: + - environment_key: default + spec: + environment_version: "2" + + # Reads the run's outcome, so the failing run aborts the deploy before this + # job is created. Separate from my_job, which my_run already depends on, to + # avoid a cycle. + downstream_job: + name: test-downstream-job-$UNIQUE_NAME + tags: + run_result: ${resources.job_runs.my_run.state.result_state} + tasks: + - task_key: main + condition_task: + op: EQUAL_TO + left: "1" + right: "1" + + job_runs: + my_run: + job_id: ${resources.jobs.my_job.id} diff --git a/acceptance/bundle/resources/job_runs/failed_run/fail.py b/acceptance/bundle/resources/job_runs/failed_run/fail.py new file mode 100644 index 00000000000..fa56481ece5 --- /dev/null +++ b/acceptance/bundle/resources/job_runs/failed_run/fail.py @@ -0,0 +1 @@ +raise RuntimeError("intentional failure") diff --git a/acceptance/bundle/resources/job_runs/failed_run/out.test.toml b/acceptance/bundle/resources/job_runs/failed_run/out.test.toml new file mode 100644 index 00000000000..fe4076cdf9b --- /dev/null +++ b/acceptance/bundle/resources/job_runs/failed_run/out.test.toml @@ -0,0 +1,4 @@ +Local = true +Cloud = true +RequiresUnityCatalog = true +EnvMatrix.DATABRICKS_BUNDLE_ENGINE = ["direct"] diff --git a/acceptance/bundle/resources/job_runs/failed_run/output.txt b/acceptance/bundle/resources/job_runs/failed_run/output.txt new file mode 100644 index 00000000000..894b5c48736 --- /dev/null +++ b/acceptance/bundle/resources/job_runs/failed_run/output.txt @@ -0,0 +1,74 @@ + +=== a run that finishes FAILED fails the deploy +>>> [CLI] bundle deploy +Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/[UNIQUE_NAME]/files... +Deploying resources... +job run [MY_RUN_ID]: Run URL: [RUN_URL] +job run [MY_RUN_ID]: [TIMESTAMP] "test-job-[UNIQUE_NAME]" INTERNAL_ERROR FAILED Task main failed with message: Workload failed, see run output for details. +Error: cannot create resources.job_runs.my_run: waiting after creating id=[MY_RUN_ID]: run did not succeed: FAILED: Task main failed with message: Workload failed, see run output for details. +task "main": RuntimeError: intentional failure +run page: [RUN_URL] + +Error: cannot create resources.jobs.downstream_job: dependency failed: resources.job_runs.my_run + +Updating deployment state... + +=== the failed run is recorded, and not having succeeded is drift +>>> read_id.py my_run +[MY_RUN_ID] + +>>> [CLI] bundle plan +recreate job_runs.my_run +create jobs.downstream_job + +Plan: 2 to add, 0 to change, 1 to delete, 1 unchanged + +=== so a redeploy runs the job again, and fails again +>>> [CLI] bundle deploy +Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/[UNIQUE_NAME]/files... +Deploying resources... +job run [MY_RUN_ID_2]: Run URL: [RUN_URL] +job run [MY_RUN_ID_2]: [TIMESTAMP] "test-job-[UNIQUE_NAME]" INTERNAL_ERROR FAILED Task main failed with message: Workload failed, see run output for details. +Error: cannot recreate resources.job_runs.my_run: waiting after creating id=[MY_RUN_ID_2]: run did not succeed: FAILED: Task main failed with message: Workload failed, see run output for details. +task "main": RuntimeError: intentional failure +run page: [RUN_URL] + +Error: cannot create resources.jobs.downstream_job: dependency failed: resources.job_runs.my_run + +Updating deployment state... + +=== run-now was issued once per deploy, and the recreate deleted the failed run +>>> print_requests.py --keep //jobs/run-now +{ + "method": "POST", + "path": "/api/2.2/jobs/run-now", + "body": { + "job_id": [MY_JOB_ID] + } +} +{ + "method": "POST", + "path": "/api/2.2/jobs/run-now", + "body": { + "job_id": [MY_JOB_ID] + } +} + +>>> print_requests.py //jobs/runs/delete +{ + "method": "POST", + "path": "/api/2.2/jobs/runs/delete", + "body": { + "run_id": [MY_RUN_ID] + } +} + +>>> [CLI] bundle destroy --auto-approve +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/[UNIQUE_NAME] + +Deleting files... +Destroy complete! diff --git a/acceptance/bundle/resources/job_runs/failed_run/script b/acceptance/bundle/resources/job_runs/failed_run/script new file mode 100644 index 00000000000..7c74b20a637 --- /dev/null +++ b/acceptance/bundle/resources/job_runs/failed_run/script @@ -0,0 +1,33 @@ +envsubst < databricks.yml.tmpl > databricks.yml + +cleanup() { + trace $CLI bundle destroy --auto-approve + rm -f out.requests.txt +} +trap cleanup EXIT + +# The error names the failed task and the message the workspace reported for it, +# and downstream_job is reported as a failed dependency because it reads the +# run's result_state. +title "a run that finishes FAILED fails the deploy" +musterr trace $CLI bundle deploy + +# The framework saves the run id before calling WaitAfterCreate, so the failed run +# stays recorded, and FAILED against the required SUCCESS is drift. +title "the failed run is recorded, and not having succeeded is drift" +trace read_id.py my_run +trace $CLI bundle plan + +read_id.py my_job > /dev/null + +# The recreate re-runs the job instead of accepting the recorded failure. +title "so a redeploy runs the job again, and fails again" +musterr trace $CLI bundle deploy +read_id.py my_run > /dev/null + +# The delete names [MY_RUN_ID], the run that failed first, rather than the +# [MY_RUN_ID_2] that replaced it: the recreate discards the failed run instead of +# leaving it in the workspace. +title "run-now was issued once per deploy, and the recreate deleted the failed run" +trace print_requests.py --keep //jobs/run-now +trace print_requests.py //jobs/runs/delete diff --git a/acceptance/bundle/resources/job_runs/failed_run/test.toml b/acceptance/bundle/resources/job_runs/failed_run/test.toml new file mode 100644 index 00000000000..2f30fb8d8a9 --- /dev/null +++ b/acceptance/bundle/resources/job_runs/failed_run/test.toml @@ -0,0 +1,27 @@ +# job_runs is a direct-engine-only resource; the Terraform provider has no +# equivalent, so restrict the matrix to direct. +EnvMatrix.DATABRICKS_BUNDLE_ENGINE = ["direct"] +RecordRequests = true + +# Runs the failing job for real, so the message the deploy names the task with is +# one a workspace reported rather than one the test server wrote. Serverless needs +# Unity Catalog. +Cloud = true +RequiresUnityCatalog = true + +# databricks.yml is rendered by the script, and the deploy fails mid-way, leaving +# local deployment state behind. +Ignore = [ + ".databricks", + "databricks.yml", +] + +# The host and the workspace selector in the run URL differ per workspace; the URL +# form itself is covered by libs/workspaceurls. +[[Repls]] +Old = 'Run URL: .*' +New = 'Run URL: [RUN_URL]' + +[[Repls]] +Old = 'run page: .*' +New = 'run page: [RUN_URL]' diff --git a/acceptance/bundle/resources/job_runs/job_parameters/output.txt b/acceptance/bundle/resources/job_runs/job_parameters/output.txt index bcf3f21e017..7424580640b 100644 --- a/acceptance/bundle/resources/job_runs/job_parameters/output.txt +++ b/acceptance/bundle/resources/job_runs/job_parameters/output.txt @@ -3,6 +3,8 @@ >>> [CLI] bundle deploy Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/job-runs-job-parameters/default/files... Deploying resources... +job run [MY_RUN_ID]: Run URL: [DATABRICKS_URL]/jobs/[MY_JOB_ID]/runs/[MY_RUN_ID]?o=[NUMID] +job run [MY_RUN_ID]: [TIMESTAMP] "my-job" TERMINATED SUCCESS Updating deployment state... Deployment complete! @@ -11,7 +13,7 @@ Deployment complete! "method": "POST", "path": "/api/2.2/jobs/run-now", "body": { - "job_id": [NUMID], + "job_id": [MY_JOB_ID], "job_parameters": { "env": "prod" } @@ -32,11 +34,11 @@ Resources: Job Runs: my_run: Name: - URL: [DATABRICKS_URL]/jobs/[NUMID]/runs/[NUMID]?w=[NUMID] + URL: [DATABRICKS_URL]/jobs/[MY_JOB_ID]/runs/[MY_RUN_ID]?w=[NUMID] Jobs: my_job: Name: my-job - URL: [DATABRICKS_URL]/jobs/[NUMID]?w=[NUMID] + URL: [DATABRICKS_URL]/jobs/[MY_JOB_ID]?w=[NUMID] >>> [CLI] bundle destroy --auto-approve The following resources will be deleted: diff --git a/acceptance/bundle/resources/job_runs/job_parameters/script b/acceptance/bundle/resources/job_runs/job_parameters/script index 7128b53ef68..16eedf5d7dc 100644 --- a/acceptance/bundle/resources/job_runs/job_parameters/script +++ b/acceptance/bundle/resources/job_runs/job_parameters/script @@ -6,6 +6,12 @@ trap cleanup EXIT title "deploy triggers the run with only the overridden parameter" trace $CLI bundle deploy + +# Name the ids so the request body and the URLs below say which number is the job +# and which is the run. +read_id.py my_job > /dev/null +read_id.py my_run > /dev/null + trace print_requests.py //jobs/run-now title "plan is stable: the resolved job default does not look like drift" diff --git a/acceptance/bundle/resources/job_runs/redeploy/output.txt b/acceptance/bundle/resources/job_runs/redeploy/output.txt index 6ec19be8bce..701104e7a36 100644 --- a/acceptance/bundle/resources/job_runs/redeploy/output.txt +++ b/acceptance/bundle/resources/job_runs/redeploy/output.txt @@ -3,6 +3,8 @@ >>> [CLI] bundle deploy Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/job-runs-redeploy/default/files... Deploying resources... +job run [MY_RUN_ID]: Run URL: [DATABRICKS_URL]/jobs/[MY_JOB_ID]/runs/[MY_RUN_ID]?o=[NUMID] +job run [MY_RUN_ID]: [TIMESTAMP] "my-job" TERMINATED SUCCESS Updating deployment state... Deployment complete! @@ -16,7 +18,7 @@ Resources: Job Runs: my_run: Name: - URL: [DATABRICKS_URL]/jobs/[MY_JOB_ID]/runs/[NUMID]?w=[NUMID] + URL: [DATABRICKS_URL]/jobs/[MY_JOB_ID]/runs/[MY_RUN_ID]?w=[NUMID] Jobs: my_job: Name: my-job @@ -54,7 +56,8 @@ Resources: "job_id": [MY_JOB_ID], "job_parameters": { "env": "prod" - } + }, + "result_state": "SUCCESS" } }, "remote_state": { @@ -62,12 +65,14 @@ Resources: "job_parameters": { "env": "dev" }, - "run_id": [NUMID], + "result_state": "SUCCESS", + "run_id": [MY_RUN_ID], "run_name": "my-job", - "run_page_url": "[DATABRICKS_URL]/?o=[NUMID]#job/[MY_JOB_ID]/run/[NUMID]", + "run_page_url": "[DATABRICKS_URL]/jobs/[MY_JOB_ID]/runs/[MY_RUN_ID]?o=[NUMID]", "run_type": "JOB_RUN", "state": { - "life_cycle_state": "RUNNING" + "life_cycle_state": "TERMINATED", + "result_state": "SUCCESS" } }, "changes": { @@ -84,6 +89,8 @@ Resources: >>> [CLI] bundle deploy Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/job-runs-redeploy/default/files... Deploying resources... +job run [MY_RUN_ID_2]: Run URL: [DATABRICKS_URL]/jobs/[MY_JOB_ID]/runs/[MY_RUN_ID_2]?o=[NUMID] +job run [MY_RUN_ID_2]: [TIMESTAMP] "my-job" TERMINATED SUCCESS Updating deployment state... Deployment complete! @@ -97,7 +104,7 @@ Resources: Job Runs: my_run: Name: - URL: [DATABRICKS_URL]/jobs/[MY_JOB_ID]/runs/[NUMID]?w=[NUMID] + URL: [DATABRICKS_URL]/jobs/[MY_JOB_ID]/runs/[MY_RUN_ID_2]?w=[NUMID] Jobs: my_job: Name: my-job @@ -109,7 +116,7 @@ Resources: "method": "POST", "path": "/api/2.2/jobs/runs/delete", "body": { - "run_id": [NUMID] + "run_id": [MY_RUN_ID] } } diff --git a/acceptance/bundle/resources/job_runs/redeploy/script b/acceptance/bundle/resources/job_runs/redeploy/script index 87d035b9eee..8ceb3a88052 100644 --- a/acceptance/bundle/resources/job_runs/redeploy/script +++ b/acceptance/bundle/resources/job_runs/redeploy/script @@ -8,12 +8,18 @@ title "initial deploy triggers the first run" trace $CLI bundle deploy trace $CLI bundle summary trace read_id.py my_job + +# Name the first run, so the second one below reads as [MY_RUN_ID_2] and the two +# are told apart in the URLs and in the delete request. +read_id.py my_run > /dev/null + trace print_requests.py //jobs/run-now title "change the run configuration and redeploy" trace update_file.py databricks.yml "env: dev" "env: prod" trace $CLI bundle plan -o json | jq '.plan["resources.job_runs.my_run"]' trace $CLI bundle deploy +read_id.py my_run > /dev/null trace $CLI bundle summary title "the config change deleted the previous run and triggered a second, different run" diff --git a/acceptance/bundle/resources/job_runs/wait/databricks.yml.tmpl b/acceptance/bundle/resources/job_runs/wait/databricks.yml.tmpl new file mode 100644 index 00000000000..6d1f58cea51 --- /dev/null +++ b/acceptance/bundle/resources/job_runs/wait/databricks.yml.tmpl @@ -0,0 +1,49 @@ +bundle: + name: job-runs-wait + +workspace: + root_path: ~/.bundle/$UNIQUE_NAME + +resources: + jobs: + my_job: + name: test-job-$UNIQUE_NAME + parameters: + - name: env + default: dev + - name: region + default: us + tasks: + # Serverless keeps the run to about a minute. + - task_key: main + spark_python_task: + python_file: ./hello.py + environment_key: default + + environments: + - environment_key: default + spec: + environment_version: "2" + + # Reads the run's outcome, so its tag shows the deploy waited for the run + # before creating what depends on it. Separate from my_job, which my_run + # already depends on, to avoid a cycle. + downstream_job: + name: test-downstream-job-$UNIQUE_NAME + tags: + run_result: ${resources.job_runs.my_run.state.result_state} + tasks: + - task_key: main + condition_task: + op: EQUAL_TO + left: "1" + right: "1" + + job_runs: + my_run: + job_id: ${resources.jobs.my_job.id} + # Override one of the job's two parameters: a real GetRun reports the full + # resolved set, including the region default, which ignore_remote_changes + # has to absorb. + job_parameters: + env: prod diff --git a/acceptance/bundle/resources/job_runs/wait/hello.py b/acceptance/bundle/resources/job_runs/wait/hello.py new file mode 100644 index 00000000000..93e0cef4a92 --- /dev/null +++ b/acceptance/bundle/resources/job_runs/wait/hello.py @@ -0,0 +1 @@ +print("hello from a job_run") diff --git a/acceptance/bundle/resources/job_runs/wait/out.test.toml b/acceptance/bundle/resources/job_runs/wait/out.test.toml new file mode 100644 index 00000000000..fe4076cdf9b --- /dev/null +++ b/acceptance/bundle/resources/job_runs/wait/out.test.toml @@ -0,0 +1,4 @@ +Local = true +Cloud = true +RequiresUnityCatalog = true +EnvMatrix.DATABRICKS_BUNDLE_ENGINE = ["direct"] diff --git a/acceptance/bundle/resources/job_runs/wait/output.txt b/acceptance/bundle/resources/job_runs/wait/output.txt new file mode 100644 index 00000000000..c842c9c667e --- /dev/null +++ b/acceptance/bundle/resources/job_runs/wait/output.txt @@ -0,0 +1,39 @@ + +=== the deploy waits for the run to finish +>>> [CLI] bundle deploy +Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/[UNIQUE_NAME]/files... +Deploying resources... +job run [MY_RUN_ID]: Run URL: [RUN_URL] +job run [MY_RUN_ID]: [TIMESTAMP] "test-job-[UNIQUE_NAME]" TERMINATED SUCCESS +Updating deployment state... +Deployment complete! + +>>> read_id.py my_run +[MY_RUN_ID] + +=== the downstream job was created with the run's result_state +>>> [CLI] jobs get [DOWNSTREAM_JOB_ID] -o json +SUCCESS + +=== nothing the run resolved is drift, so a redeploy starts no new run +>>> [CLI] bundle plan -o json +resources.job_runs.my_run skip +resources.jobs.downstream_job skip +resources.jobs.my_job skip + +>>> [CLI] bundle deploy +Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/[UNIQUE_NAME]/files... +Deploying resources... +Updating deployment state... +Deployment complete! + +>>> [CLI] bundle destroy --auto-approve +The following resources will be deleted: + delete resources.job_runs.my_run + delete resources.jobs.downstream_job + delete resources.jobs.my_job + +All files and directories at the following location will be deleted: /Workspace/Users/[USERNAME]/.bundle/[UNIQUE_NAME] + +Deleting files... +Destroy complete! diff --git a/acceptance/bundle/resources/job_runs/wait/script b/acceptance/bundle/resources/job_runs/wait/script new file mode 100644 index 00000000000..70cee1c9082 --- /dev/null +++ b/acceptance/bundle/resources/job_runs/wait/script @@ -0,0 +1,19 @@ +envsubst < databricks.yml.tmpl > databricks.yml + +cleanup() { + trace $CLI bundle destroy --auto-approve +} +trap cleanup EXIT + +title "the deploy waits for the run to finish" +trace $CLI bundle deploy + +trace read_id.py my_run + +title "the downstream job was created with the run's result_state" +downstream_id=$(read_id.py downstream_job) +trace $CLI jobs get $downstream_id -o json | jq -r ".settings.tags.run_result" + +title "nothing the run resolved is drift, so a redeploy starts no new run" +trace $CLI bundle plan -o json | jq -r '.plan | to_entries[] | "\(.key) \(.value.action)"' +trace $CLI bundle deploy diff --git a/acceptance/bundle/resources/job_runs/wait/test.toml b/acceptance/bundle/resources/job_runs/wait/test.toml new file mode 100644 index 00000000000..5c84ac24640 --- /dev/null +++ b/acceptance/bundle/resources/job_runs/wait/test.toml @@ -0,0 +1,20 @@ +# job_runs is a direct-engine-only resource; the Terraform provider has no +# equivalent, so restrict the matrix to direct. +EnvMatrix.DATABRICKS_BUNDLE_ENGINE = ["direct"] + +# Runs the job for real on cloud. Serverless needs Unity Catalog. +Cloud = true +RequiresUnityCatalog = true + +# The wait polls GetRun until the run is terminal, so the recorded requests +# depend on how long the run takes. +RecordRequests = false + +# databricks.yml is rendered by the script. +Ignore = ["databricks.yml"] + +# The host and the workspace selector in the run URL differ per workspace; the URL +# form itself is covered by libs/workspaceurls. +[[Repls]] +Old = 'Run URL: .*' +New = 'Run URL: [RUN_URL]' diff --git a/bundle/direct/bundle_plan_test.go b/bundle/direct/bundle_plan_test.go index 37605302874..0e93200f2ca 100644 --- a/bundle/direct/bundle_plan_test.go +++ b/bundle/direct/bundle_plan_test.go @@ -4,11 +4,14 @@ import ( "bytes" "testing" + "github.com/databricks/cli/bundle/config/resources" "github.com/databricks/cli/bundle/deployplan" "github.com/databricks/cli/bundle/direct/dresources" "github.com/databricks/cli/libs/dyn" "github.com/databricks/cli/libs/dyn/yamlloader" "github.com/databricks/cli/libs/structs/structpath" + "github.com/databricks/cli/libs/structs/structvar" + "github.com/databricks/databricks-sdk-go/service/jobs" "github.com/databricks/databricks-sdk-go/service/pipelines" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" @@ -284,3 +287,127 @@ func TestShouldSkipBackendDefault_MapDriftUsesBracketKeys(t *testing.T) { assert.True(t, ok) assert.Equal(t, deployplan.ReasonBackendDefault, reason) } + +const jobRunKey = "resources.job_runs.my_run" + +// The plan skips only a run that reached the required SUCCESS, so a reference +// served from the remote state cache reads a finished run. +func TestLookupReferencePreDeploy_FinishedJobRun(t *testing.T) { + b := bundleWithSkippedJobRun(t, &dresources.JobRunRemote{ + RunId: 123, + ResultState: jobs.RunResultStateSuccess, + State: &jobs.RunState{ + LifeCycleState: jobs.RunLifeCycleStateTerminated, + ResultState: jobs.RunResultStateSuccess, + }, + }) + + value, err := b.LookupReferencePreDeploy(t.Context(), structpath.MustParsePath(jobRunKey+".state.result_state")) + + require.NoError(t, err) + assert.Equal(t, jobs.RunResultStateSuccess, value) +} + +// A resource reads the run the workspace reports. These paths are absent from +// JobRunState, so they resolve from the remote, and the failure-shaped values +// below come back unrewritten. result_state is the exception: it resolves to the +// outcome PrepareState requires, which TestJobRunOutcomeIsDrift keeps from +// standing in for one the run never reached. +func TestLookupReferencePreDeploy_JobRunReferencesAreRaw(t *testing.T) { + b := bundleWithSkippedJobRun(t, &dresources.JobRunRemote{ + RunId: 123, + RunName: "my-job", + RunPageUrl: "https://myworkspace.databricks.test/jobs/456/runs/123", + ResultState: jobs.RunResultStateFailed, + State: &jobs.RunState{ + LifeCycleState: jobs.RunLifeCycleStateTerminated, + ResultState: jobs.RunResultStateFailed, + StateMessage: "task main failed", + }, + }) + + for field, want := range map[string]any{ + "state.result_state": jobs.RunResultStateFailed, + "state.life_cycle_state": jobs.RunLifeCycleStateTerminated, + "state.state_message": "task main failed", + "run_id": int64(123), + "run_name": "my-job", + "run_page_url": "https://myworkspace.databricks.test/jobs/456/runs/123", + } { + t.Run(field, func(t *testing.T) { + value, err := b.LookupReferencePreDeploy(t.Context(), structpath.MustParsePath(jobRunKey+"."+field)) + + require.NoError(t, err) + assert.Equal(t, want, value) + }) + } +} + +// References are served from the remote state cache only for a run the plan +// skips, and any outcome other than the required SUCCESS is a recreate. +func TestJobRunOutcomeIsDrift(t *testing.T) { + adapters, err := dresources.InitAll(nil) + require.NoError(t, err) + + for name, remote := range map[string]jobs.RunResultState{ + "FAILED": jobs.RunResultStateFailed, + "CANCELED": jobs.RunResultStateCanceled, + "TIMEDOUT": jobs.RunResultStateTimedout, + // A run that is still going reports no result at all. + "no result yet": "", + } { + t.Run(name, func(t *testing.T) { + // Old == New: a deploy records the outcome it asked for, and only the + // remote says the run missed it. + changes := deployplan.Changes{"result_state": &deployplan.ChangeDesc{ + Old: jobs.RunResultStateSuccess, + New: jobs.RunResultStateSuccess, + Remote: remote, + }} + + err := addPerFieldActions(t.Context(), adapters["job_runs"], changes, nil) + + require.NoError(t, err) + assert.Equal(t, deployplan.Recreate, changes["result_state"].Action) + }) + } +} + +// The run reached the required outcome, so the plan can skip it. +func TestJobRunSucceededOutcomeIsNotDrift(t *testing.T) { + adapters, err := dresources.InitAll(nil) + require.NoError(t, err) + + changes := deployplan.Changes{"result_state": &deployplan.ChangeDesc{ + Old: jobs.RunResultStateSuccess, + New: jobs.RunResultStateSuccess, + Remote: jobs.RunResultStateSuccess, + }} + + err = addPerFieldActions(t.Context(), adapters["job_runs"], changes, nil) + + require.NoError(t, err) + assert.Equal(t, deployplan.Skip, changes["result_state"].Action) +} + +// bundleWithSkippedJobRun sets up a deploy whose plan skips the run, so +// references to it resolve from the remote state cache. The state cache holds +// PrepareState's output, as a real plan does. +func bundleWithSkippedJobRun(t *testing.T, remote *dresources.JobRunRemote) *DeploymentBundle { + t.Helper() + + adapters, err := dresources.InitAll(nil) + require.NoError(t, err) + + plan := deployplan.NewPlanDirect() + plan.Plan[jobRunKey] = &deployplan.PlanEntry{ + Action: deployplan.Skip, + NewState: &structvar.StructVarJSON{}, + } + + b := &DeploymentBundle{Adapters: adapters, Plan: plan} + state := (&dresources.ResourceJobRun{}).PrepareState(&resources.JobRun{}) + b.StateCache.Store(jobRunKey, structvar.NewStructVar(state, nil)) + b.RemoteStateCache.Store(jobRunKey, remote) + return b +} diff --git a/bundle/direct/dresources/all_test.go b/bundle/direct/dresources/all_test.go index 54541d94ba9..d0e8b02de6d 100644 --- a/bundle/direct/dresources/all_test.go +++ b/bundle/direct/dresources/all_test.go @@ -1000,10 +1000,19 @@ func testCRUD(t *testing.T, group string, adapter *Adapter, client *databricks.W "unexpected differences between remappedState and remappedRemoteStateFromCreate") } + // Only a job run's state moves while WaitAfterCreate blocks, and its result_state + // fills in only then, so its field checks below need the settled state. + waitSettlesState := group == "job_runs" + remoteStateFromWaitCreate, err := adapter.WaitAfterCreate(ctx, createdID, newState) require.NoError(t, err) if remoteStateFromWaitCreate != nil { - require.Equal(t, remote, remoteStateFromWaitCreate) + if waitSettlesState { + remappedState, err = adapter.RemapState(remoteStateFromWaitCreate) + require.NoError(t, err) + } else { + require.Equal(t, remote, remoteStateFromWaitCreate) + } } if adapter.HasDoUpdate() { diff --git a/bundle/direct/dresources/job_run.go b/bundle/direct/dresources/job_run.go index 0a2ae0ea6af..63af7f47459 100644 --- a/bundle/direct/dresources/job_run.go +++ b/bundle/direct/dresources/job_run.go @@ -1,19 +1,36 @@ package dresources import ( + "cmp" "context" + "errors" "fmt" "strconv" + "strings" + "time" "github.com/databricks/cli/bundle/config/resources" + "github.com/databricks/cli/bundle/run/progress" + "github.com/databricks/cli/libs/cmdio" + "github.com/databricks/cli/libs/log" + "github.com/databricks/cli/libs/workspaceurls" "github.com/databricks/databricks-sdk-go" "github.com/databricks/databricks-sdk-go/marshal" + "github.com/databricks/databricks-sdk-go/retries" "github.com/databricks/databricks-sdk-go/service/jobs" ) -// JobRunState is what we persist for a triggered run: the RunNow request. +// jobRunTimeout matches the timeout `bundle run` allows a run (bundle/run/job.go). +const jobRunTimeout = 24 * time.Hour + +// JobRunState is what we persist for a triggered run: the RunNow request, plus +// the outcome the run is required to reach. type JobRunState struct { jobs.RunNow + + // Always SUCCESS, never read from the config: a run that did not succeed + // differs from the remote and is re-triggered by recreate_on_changes. + ResultState jobs.RunResultState `json:"result_state,omitempty"` } func (s *JobRunState) UnmarshalJSON(b []byte) error { @@ -29,6 +46,10 @@ func (s JobRunState) MarshalJSON() ([]byte, error) { type JobRunRemote struct { jobs.RunNow + // Repeats state.result_state at the path JobRunState uses, so the planner can + // compare the two. See "RemapState is a dumb copy" in README.md. + ResultState jobs.RunResultState `json:"result_state,omitempty"` + RunId int64 `json:"run_id,omitempty"` RunName string `json:"run_name,omitempty"` State *jobs.RunState `json:"state,omitempty"` @@ -58,7 +79,8 @@ func (*ResourceJobRun) New(client *databricks.WorkspaceClient) *ResourceJobRun { func (*ResourceJobRun) PrepareState(input *resources.JobRun) *JobRunState { return &JobRunState{ - RunNow: input.RunNow, + RunNow: input.RunNow, + ResultState: jobs.RunResultStateSuccess, } } @@ -97,17 +119,16 @@ func makeJobRunRemote(run *jobs.Run) *JobRunRemote { Queue: nil, ForceSendFields: nil, }, - RunId: run.RunId, - RunName: run.RunName, - State: run.State, - RunPageUrl: run.RunPageUrl, - RunType: run.RunType, + ResultState: run.State.ResultState, + RunId: run.RunId, + RunName: run.RunName, + State: run.State, + RunPageUrl: workspaceurls.ModernizeJobRunPageURL(run.RunPageUrl), + RunType: run.RunType, } } -// DoRead returns the run as GetRun reports it; a 404 lets the planner -// re-trigger. Root ignore_remote_changes suppresses all remote drift, so a run -// is recreated only on a local config change. +// DoRead returns the run as GetRun reports it; a 404 lets the planner re-trigger. func (r *ResourceJobRun) DoRead(ctx context.Context, id string) (*JobRunRemote, error) { runID, err := parseRunID(id) if err != nil { @@ -123,9 +144,10 @@ func (r *ResourceJobRun) DoRead(ctx context.Context, id string) (*JobRunRemote, return makeJobRunRemote(run), nil } -// RemapState extracts the embedded RunNow as the state used for diffing. +// RemapState extracts the fields used for diffing: the RunNow request and the +// outcome the run reached. func (*ResourceJobRun) RemapState(remote *JobRunRemote) *JobRunState { - return &JobRunState{RunNow: remote.RunNow} + return &JobRunState{RunNow: remote.RunNow, ResultState: remote.ResultState} } func (r *ResourceJobRun) DoCreate(ctx context.Context, config *JobRunState) (string, *JobRunRemote, error) { @@ -138,20 +160,202 @@ func (r *ResourceJobRun) DoCreate(ctx context.Context, config *JobRunState) (str return strconv.FormatInt(wait.RunId, 10), nil, nil } +// WaitAfterCreate blocks until the run finishes, so a resource referencing its +// output (e.g. state.result_state) sees a settled run. Only SUCCESS continues the deploy. +func (r *ResourceJobRun) WaitAfterCreate(ctx context.Context, id string, _ *JobRunState) (*JobRunRemote, error) { + runID, err := parseRunID(id) + if err != nil { + return nil, err + } + + // A run can take hours, so report progress like `bundle run` does. pageURL + // outlives the poll so an abandoned wait can still link the run. + var tracker progress.JobStateTracker + var pageURL string + // Polled here rather than through the SDK waiter, which halts with an error of + // its own on the INTERNAL_ERROR a run whose task failed reports, hiding the + // task that failed. + run, err := retries.Poll(ctx, jobRunTimeout, func() (*jobs.Run, *retries.Err) { + var req jobs.GetRunRequest + req.RunId = runID + run, err := r.client.Jobs.GetRun(ctx, req) + if err != nil { + return nil, retries.Halt(err) + } + pageURL = cmp.Or(pageURL, run.RunPageUrl) + logRunProgress(ctx, run, &tracker) + if !runIsTerminal(run.State.LifeCycleState) { + return nil, retries.Continues(run.State.StateMessage) + } + return run, nil + }) + if err != nil { + // The wait can end with the run still going (timeout, interrupt), so link + // the run page. + if ctx.Err() != nil { + // A cancelled context is reported as a timeout. + return nil, fmt.Errorf("interrupted while waiting for the run to finish: %w%s", ctx.Err(), runPageLine(pageURL)) + } + return nil, fmt.Errorf("%w%s", err, runPageLine(pageURL)) + } + if run.State.ResultState != jobs.RunResultStateSuccess { + return nil, r.runFailedError(ctx, run) + } + return makeJobRunRemote(run), nil +} + +// runFailedError reports why the run did not succeed, naming each failed task +// and the error it reported. +func (r *ResourceJobRun) runFailedError(ctx context.Context, run *jobs.Run) error { + outcome := string(run.State.ResultState) + if outcome == "" { + // A skipped run has no result_state; report the lifecycle state. + outcome = string(run.State.LifeCycleState) + } + var msg strings.Builder + // The framework already prefixes the resource key and the run id. + fmt.Fprintf(&msg, "run did not succeed: %s", outcome) + if run.State.StateMessage != "" { + fmt.Fprintf(&msg, ": %s", run.State.StateMessage) + } + for _, task := range lastFailedAttempts(run.Tasks) { + fmt.Fprintf(&msg, "\ntask %q: %s", task.TaskKey, r.taskError(ctx, task)) + } + msg.WriteString(runPageLine(run.RunPageUrl)) + return errors.New(msg.String()) +} + +// lastFailedAttempts returns the failed tasks in the order the run reports them, +// one per task key: a task the Jobs API retried is reported once per attempt, and +// only its last one says how the run ended up. +func lastFailedAttempts(tasks []jobs.RunTask) []jobs.RunTask { + latest := make(map[string]jobs.RunTask) + var keys []string + for _, task := range tasks { + if !taskFailed(task) { + continue + } + previous, seen := latest[task.TaskKey] + if !seen { + keys = append(keys, task.TaskKey) + } + if !seen || task.AttemptNumber > previous.AttemptNumber { + latest[task.TaskKey] = task + } + } + result := make([]jobs.RunTask, 0, len(keys)) + for _, key := range keys { + result = append(result, latest[key]) + } + return result +} + +// taskFailed reports whether a task is a cause of the run's failure. A task left +// SKIPPED or UPSTREAM_FAILED never ran, so it has no error to report. +func taskFailed(task jobs.RunTask) bool { + // State is deprecated in favour of Status, so it may be absent. + if task.State == nil { + return false + } + return task.State.LifeCycleState == jobs.RunLifeCycleStateInternalError || + task.State.ResultState == jobs.RunResultStateFailed || + task.State.ResultState == jobs.RunResultStateTimedout +} + +// taskError returns the message the task reported, from the same place `bundle +// run` reads it. Only reached for a task taskFailed accepted, so State is set. +func (r *ResourceJobRun) taskError(ctx context.Context, task jobs.RunTask) string { + var reported string + output, err := r.client.Jobs.GetRunOutput(ctx, jobs.GetRunOutputRequest{RunId: task.RunId}) + if err != nil { + log.Debugf(ctx, "could not read output of task %s: %v", task.TaskKey, err) + } else { + reported = output.Error + } + // Not every task type reports an error through GetRunOutput, so fall back to + // what the run itself says about the task. + return cmp.Or(reported, task.State.StateMessage, string(task.State.ResultState), string(task.State.LifeCycleState)) +} + +func runPageLine(rawURL string) string { + if rawURL == "" { + return "" + } + return "\nrun page: " + workspaceurls.ModernizeJobRunPageURL(rawURL) +} + +// logRunProgress logs every state change like `bundle run` does, but reports only +// the run page URL and the final state to the user: how many states a run passes +// through varies with how long its compute takes to start, and a deploy's output +// has to stay reproducible. +func logRunProgress(ctx context.Context, run *jobs.Run, tracker *progress.JobStateTracker) { + event, first := tracker.Poll(run) + if event == nil { + return + } + log.Info(ctx, event.String()) + if first && run.RunPageUrl != "" { + line := "Run URL: " + workspaceurls.ModernizeJobRunPageURL(run.RunPageUrl) + log.Info(ctx, line) + reportRunLine(ctx, run.RunId, line) + } + if runIsTerminal(run.State.LifeCycleState) { + reportRunLine(ctx, run.RunId, event.String()) + } +} + +// runIsTerminal reports whether a run has stopped, whatever it stopped as. +func runIsTerminal(state jobs.RunLifeCycleState) bool { + return state == jobs.RunLifeCycleStateTerminated || + state == jobs.RunLifeCycleStateSkipped || + state == jobs.RunLifeCycleStateInternalError +} + +// reportRunLine names the run, since resources deploy concurrently onto one stream. +func reportRunLine(ctx context.Context, runID int64, msg string) { + if cmdio.HasIO(ctx) { + cmdio.LogString(ctx, fmt.Sprintf("job run %d: %s", runID, msg)) + } +} + // DoUpdate is intentionally not implemented: a run can't be modified in place, // so any change recreates it (delete + a fresh RunNow). // DoDelete deletes the run via jobs/runs/delete, on both destroy and the -// recreate path. The API rejects a still-active run; this milestone doesn't -// await completion, so that error surfaces to the user. +// recreate path. The API rejects a still-active run, which an interrupted wait +// leaves behind, so cancel it first. func (r *ResourceJobRun) DoDelete(ctx context.Context, id string, _ *JobRunState) error { runID, err := parseRunID(id) if err != nil { return err } + remote, err := r.DoRead(ctx, id) + if err != nil { + return err + } + if !runIsTerminal(remote.State.LifeCycleState) { + err = r.cancelRun(ctx, runID) + if err != nil { + return err + } + } return r.client.Jobs.DeleteRunByRunId(ctx, runID) } +// cancelRun cancels a run and waits for it to settle. Cancellation is +// asynchronous, so a delete issued right after would still be rejected. +func (r *ResourceJobRun) cancelRun(ctx context.Context, runID int64) error { + waiter, err := r.client.Jobs.CancelRun(ctx, jobs.CancelRun{RunId: runID}) + if err != nil { + return fmt.Errorf("cancelling run %d before deleting it: %w", runID, err) + } + _, err = waiter.Get() + if err != nil { + return fmt.Errorf("waiting for run %d to be cancelled: %w", runID, err) + } + return nil +} + func parseRunID(id string) (int64, error) { result, err := strconv.ParseInt(id, 10, 64) if err != nil { diff --git a/bundle/direct/dresources/job_run_test.go b/bundle/direct/dresources/job_run_test.go new file mode 100644 index 00000000000..89f0c029f59 --- /dev/null +++ b/bundle/direct/dresources/job_run_test.go @@ -0,0 +1,372 @@ +package dresources + +import ( + "context" + "reflect" + "slices" + "strings" + "sync/atomic" + "testing" + "time" + + "github.com/databricks/cli/bundle/config/resources" + "github.com/databricks/cli/libs/structs/structpath" + "github.com/databricks/cli/libs/testserver" + "github.com/databricks/databricks-sdk-go" + "github.com/databricks/databricks-sdk-go/service/jobs" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// jobRunClientFor returns a client talking to server. Call it after the test +// registers its own handlers: first registration wins, so the defaults added here +// only fill the gaps. +func jobRunClientFor(t *testing.T, server *testserver.Server) *databricks.WorkspaceClient { + t.Helper() + testserver.AddDefaultHandlers(server) + + client, err := databricks.NewWorkspaceClient(&databricks.Config{ + Host: server.URL, + Token: "testtoken", + }) + require.NoError(t, err) + return client +} + +// jobRunServer returns a client whose runs/get is the given handler, so a wait can +// be driven without a real run. +func jobRunServer(t *testing.T, getRun testserver.HandlerFunc) *databricks.WorkspaceClient { + t.Helper() + server := testserver.New(t) + server.Handle("GET", "/api/2.2/jobs/runs/get", getRun) + return jobRunClientFor(t, server) +} + +// The Jobs API reports the run page in the legacy fragment form; errors and +// progress lines carry the path form it converts to. +const ( + testRunPageURL = "https://myworkspace.databricks.test/?o=900800700600#job/456/run/123" + testRunPageLink = "run page: https://myworkspace.databricks.test/jobs/456/runs/123?o=900800700600" +) + +// jobRunClient returns a client whose GetRun always reports the given run state. +func jobRunClient(t *testing.T, state *jobs.RunState) *databricks.WorkspaceClient { + t.Helper() + return jobRunServer(t, func(req testserver.Request) any { + return jobs.Run{RunId: 123, JobId: 456, State: state, RunPageUrl: testRunPageURL} + }) +} + +// waitForTestRun drives the framework hook, so it covers parsing the id the +// framework hands back from DoCreate along with the wait itself. +func waitForTestRun(t *testing.T, ctx context.Context, client *databricks.WorkspaceClient) (*JobRunRemote, error) { + t.Helper() + r := (&ResourceJobRun{}).New(client) + 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 TestJobRunWaitFailsOnFailedResult(t *testing.T) { + client := jobRunClient(t, &jobs.RunState{ + LifeCycleState: jobs.RunLifeCycleStateTerminated, + ResultState: jobs.RunResultStateFailed, + StateMessage: "task failed", + }) + + _, err := waitForTestRun(t, t.Context(), client) + + 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"`) +} + +func TestJobRunWaitFailsOnSkipped(t *testing.T) { + // A skipped run has no result_state, so the lifecycle state is reported. + client := jobRunClient(t, &jobs.RunState{ + LifeCycleState: jobs.RunLifeCycleStateSkipped, + }) + + _, err := waitForTestRun(t, t.Context(), client) + + require.ErrorContains(t, err, "did not succeed: SKIPPED") +} + +func TestJobRunWaitFailsOnInternalError(t *testing.T) { + client := jobRunClient(t, &jobs.RunState{ + LifeCycleState: jobs.RunLifeCycleStateInternalError, + }) + + _, err := waitForTestRun(t, t.Context(), client) + + require.ErrorContains(t, err, "run did not succeed: INTERNAL_ERROR") + 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, + 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: "main", RunId: 998, AttemptNumber: 0, State: failed}, + {TaskKey: "main", RunId: 999, AttemptNumber: 1, State: failed}, + }, + } + }) + server.Handle("GET", "/api/2.2/jobs/runs/get-output", func(req testserver.Request) any { + return jobs.RunOutput{Error: "output of run " + req.URL.Query().Get("run_id")} + }) + + _, err := waitForTestRun(t, t.Context(), jobRunClientFor(t, server)) + + // The Jobs API reports a retried task once per attempt; only the last one says + // how the run ended up. + require.ErrorContains(t, err, `task "main": output of run 999`) + assert.NotContains(t, err.Error(), "output of run 998") +} + +func TestJobRunWaitAbandonedLinksTheRun(t *testing.T) { + client := jobRunClient(t, &jobs.RunState{LifeCycleState: jobs.RunLifeCycleStateRunning}) + + ctx, cancel := context.WithTimeout(t.Context(), time.Millisecond) + defer cancel() + + _, err := waitForTestRun(t, ctx, client) + + // The run keeps going, so the error links to it and names the interrupt. + require.ErrorContains(t, err, "interrupted while waiting for the run to finish") + 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) +} + +// 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) { + for _, outcome := range []jobs.RunResultState{ + jobs.RunResultStateSuccess, + jobs.RunResultStateFailed, + // A run still going, and a SKIPPED one, report no result at all. + "", + } { + t.Run(string(outcome), func(t *testing.T) { + remote := &JobRunRemote{RunId: 123, ResultState: outcome} + + state := (&ResourceJobRun{}).RemapState(remote) + + assert.Equal(t, outcome, state.ResultState) + }) + } +} + +// resources.yml ignores remote drift on everything the RunNow request carries, +// since GetRun does not echo it back faithfully, and leaves result_state alone. +func TestJobRunIgnoresEveryRequestField(t *testing.T) { + adapters, err := InitAll(nil) + require.NoError(t, err) + ignored := adapters["job_runs"].ResourceConfig().IgnoreRemoteChanges + + for field := range reflect.TypeFor[jobs.RunNow]().Fields() { + name, _, _ := strings.Cut(field.Tag.Get("json"), ",") + if name == "" || name == "-" { + continue + } + assert.True(t, ignoresRemoteChanges(ignored, name), "jobs.RunNow field %q is not in job_runs ignore_remote_changes", name) + } + + assert.False(t, ignoresRemoteChanges(ignored, "result_state"), "result_state must stay comparable against the remote") +} + +// ignoresRemoteChanges reports whether the rules suppress remote drift on field. +func ignoresRemoteChanges(rules []FieldRule, field string) bool { + path := structpath.MustParsePath(field) + return slices.ContainsFunc(rules, func(r FieldRule) bool { return path.HasPatternPrefix(r.Field) }) +} + +// Reporting RUNNING for the first two polls exercises the poll loop. +func TestJobRunWaitPollsUntilTerminal(t *testing.T) { + var gets atomic.Int32 + client := jobRunServer(t, func(req testserver.Request) any { + if gets.Add(1) <= 2 { + return jobs.Run{RunId: 123, JobId: 456, State: &jobs.RunState{ + LifeCycleState: jobs.RunLifeCycleStateRunning, + }} + } + return jobs.Run{RunId: 123, JobId: 456, State: &jobs.RunState{ + LifeCycleState: jobs.RunLifeCycleStateTerminated, + ResultState: jobs.RunResultStateSuccess, + }} + }) + + remote, err := waitForTestRun(t, t.Context(), client) + require.NoError(t, err) + + // SUCCESS is only reachable by polling past the RUNNING reads. + require.NotNil(t, remote.State) + assert.Equal(t, jobs.RunResultStateSuccess, remote.State.ResultState) + assert.Equal(t, int32(3), gets.Load(), "expected the wait to poll past both RUNNING reads") +} + +// 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, + } + + 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} + }) + server.Handle("POST", "/api/2.2/jobs/runs/cancel", func(req testserver.Request) any { + deletion.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, + }) + + require.NoError(t, deleteTestRun(t, client)) + + assert.False(t, deletion.cancelled.Load(), "a run that already finished has nothing to cancel") +} diff --git a/bundle/direct/dresources/resources.yml b/bundle/direct/dresources/resources.yml index 6061db0fb04..9ccc9468040 100644 --- a/bundle/direct/dresources/resources.yml +++ b/bundle/direct/dresources/resources.yml @@ -139,11 +139,42 @@ resources: - field: job_clusters[*].new_cluster.data_security_mode job_runs: - # A run is immutable and fire-once: re-trigger only on a local config change, - # never on remote drift. Both rules omit `field` to match every field (root; - # see TestFieldRuleOmittedIsRoot). `field: ""` would instead match nothing. + # Every jobs.RunNow field: the request cannot drift, and GetRun does not echo + # it back faithfully (job_parameters come back resolved against the job's + # defaults, request-only fields not at all). TestJobRunIgnoresEveryRequestField + # keeps the list in step with the SDK. result_state is left out on purpose: + # comparing it against the remote re-triggers a run that did not succeed. ignore_remote_changes: - - reason: immutable + - field: dbt_commands + reason: immutable + - field: idempotency_token + reason: immutable + - field: jar_params + reason: immutable + - field: job_id + reason: immutable + - field: job_parameters + reason: immutable + - field: notebook_params + reason: immutable + - field: only + reason: immutable + - field: performance_target + reason: immutable + - field: pipeline_params + reason: immutable + - field: python_named_params + reason: immutable + - field: python_params + reason: immutable + - field: queue + reason: immutable + - field: spark_submit_params + reason: immutable + - field: sql_params + reason: immutable + # A run is immutable and fire-once, so any change recreates it. Omitting + # `field` matches every field (root; see TestFieldRuleOmittedIsRoot). recreate_on_changes: - reason: immutable diff --git a/bundle/internal/schema/annotations.yml b/bundle/internal/schema/annotations.yml index 4db1d40524d..88e4b7db396 100644 --- a/bundle/internal/schema/annotations.yml +++ b/bundle/internal/schema/annotations.yml @@ -968,6 +968,8 @@ resources: "job_runs": "description": |- The job run definitions for the bundle, where each key is the name of the job run. Each job run triggers a run of an existing job as part of bundle deployment. + + The deployment waits for the run to finish and fails if it does not succeed, so other resources can reference the run's outcome, for example `${resources.job_runs..state.result_state}`. A run that did not succeed is run again on the next deployment. "$fields": "lifecycle": "description": |- diff --git a/bundle/run/job.go b/bundle/run/job.go index 98f059e5784..38534cf544d 100644 --- a/bundle/run/job.go +++ b/bundle/run/job.go @@ -84,42 +84,25 @@ func (r *jobRunner) logFailedTasks(ctx context.Context, runId int64) { // jobRunMonitor tracks state for a single job run and provides callbacks // for monitoring progress. type jobRunMonitor struct { - ctx context.Context - prevState *jobs.RunState + ctx context.Context + tracker progress.JobStateTracker } // onProgress is the single callback that handles all state tracking and logging. func (m *jobRunMonitor) onProgress(info *jobs.Run) { - state := info.State - if state == nil { + event, first := m.tracker.Poll(info) + if event == nil { return } // First time we see this run. - if m.prevState == nil { + if first { runURL := workspaceurls.ModernizeJobRunPageURL(info.RunPageUrl) log.Infof(m.ctx, "Run available at %s", runURL) cmdio.Log(m.ctx, progress.NewJobRunUrlEvent(runURL)) } - // No state change: do not log. - if m.prevState != nil && - m.prevState.LifeCycleState == state.LifeCycleState && - m.prevState.ResultState == state.ResultState { - return - } - - // Capture current state as previous state for next call. - m.prevState = state - // Log progress event both to the terminal (in place or append), and to the logger. - event := &progress.JobProgressEvent{ - Timestamp: time.Now(), - JobId: info.JobId, - RunId: info.RunId, - RunName: info.RunName, - State: *info.State, - } cmdio.Log(m.ctx, event) log.Info(m.ctx, event.String()) } diff --git a/bundle/run/progress/job.go b/bundle/run/progress/job.go index 6deee451ad6..9a849e84918 100644 --- a/bundle/run/progress/job.go +++ b/bundle/run/progress/job.go @@ -16,6 +16,32 @@ type JobProgressEvent struct { State jobs.RunState `json:"state"` } +// JobStateTracker turns the polls of a job run into one event per state change. +type JobStateTracker struct { + prev *jobs.RunState +} + +// Poll returns the event to report, or nil when the state has not changed. first +// is true for the state the run is seen in initially, where callers also report +// the run page URL. +func (t *JobStateTracker) Poll(run *jobs.Run) (event *JobProgressEvent, first bool) { + if run.State == nil { + return nil, false + } + first = t.prev == nil + if !first && t.prev.LifeCycleState == run.State.LifeCycleState && t.prev.ResultState == run.State.ResultState { + return nil, false + } + t.prev = run.State + return &JobProgressEvent{ + Timestamp: time.Now(), + JobId: run.JobId, + RunId: run.RunId, + RunName: run.RunName, + State: *run.State, + }, first +} + func (event *JobProgressEvent) String() string { result := strings.Builder{} result.WriteString(event.Timestamp.Format("2006-01-02 15:04:05") + " ") diff --git a/bundle/run/progress/job_test.go b/bundle/run/progress/job_test.go index 31196520305..521c54f357a 100644 --- a/bundle/run/progress/job_test.go +++ b/bundle/run/progress/job_test.go @@ -22,3 +22,36 @@ func TestJobProgressEventString(t *testing.T) { } assert.Equal(t, "-0001-11-30 00:00:00 \"run_name\" TERMINATED SUCCESS state_message", event.String()) } + +func TestJobStateTrackerPoll(t *testing.T) { + running := &jobs.Run{RunId: 456, State: &jobs.RunState{LifeCycleState: jobs.RunLifeCycleStateRunning}} + terminated := &jobs.Run{RunId: 456, State: &jobs.RunState{ + LifeCycleState: jobs.RunLifeCycleStateTerminated, + ResultState: jobs.RunResultStateSuccess, + }} + + var tracker JobStateTracker + + event, first := tracker.Poll(running) + assert.True(t, first) + assert.Equal(t, jobs.RunLifeCycleStateRunning, event.State.LifeCycleState) + + // The same state again is not worth reporting, and is no longer the first one. + event, first = tracker.Poll(running) + assert.Nil(t, event) + assert.False(t, first) + + event, first = tracker.Poll(terminated) + assert.False(t, first) + assert.Equal(t, jobs.RunResultStateSuccess, event.State.ResultState) +} + +func TestJobStateTrackerPollWithoutState(t *testing.T) { + var tracker JobStateTracker + + event, first := tracker.Poll(&jobs.Run{RunId: 456}) + + // A run reported without a state has no progress to report. + assert.Nil(t, event) + assert.False(t, first) +} diff --git a/bundle/schema/jsonschema.json b/bundle/schema/jsonschema.json index 9c41d1a0b9b..1584dc3bcf3 100644 --- a/bundle/schema/jsonschema.json +++ b/bundle/schema/jsonschema.json @@ -3270,7 +3270,7 @@ "markdownDescription": "The instance pool definitions for the bundle, where each key is the name of the instance pool. See [instance_pools](https://docs.databricks.com/dev-tools/bundles/resources.html#instance_pools)." }, "job_runs": { - "description": "The job run definitions for the bundle, where each key is the name of the job run. Each job run triggers a run of an existing job as part of bundle deployment.", + "description": "The job run definitions for the bundle, where each key is the name of the job run. Each job run triggers a run of an existing job as part of bundle deployment.\n\nThe deployment waits for the run to finish and fails if it does not succeed, so other resources can reference the run's outcome, for example `${resources.job_runs.\u003cname\u003e.state.result_state}`. A run that did not succeed is run again on the next deployment.", "$ref": "#/$defs/map/github.com/databricks/cli/bundle/config/resources.JobRun" }, "jobs": { diff --git a/libs/testserver/jobs.go b/libs/testserver/jobs.go index 9097d43c086..4c99f7b624f 100644 --- a/libs/testserver/jobs.go +++ b/libs/testserver/jobs.go @@ -20,6 +20,39 @@ import ( const missingJobGitProviderMessage = "git_source.git_provider must be one of: github,gitlab,bitbucketcloud,gitlabenterpriseedition,bitbucketserver,azuredevopsservices,githubenterprise,awscodecommit" +// errNoCodeInWorkspace marks a task there is nothing to execute for, e.g. code +// uploaded as a snapshot zip this server never unpacks. The gap is here, not in +// the job, so the task succeeds. +var errNoCodeInWorkspace = errors.New("task code is not in the workspace") + +// taskFailureMessage is what a real workspace reports for a task that failed: a +// generic pointer at the run output, on both the task and the run. Measured on +// serverless against a spark_python_task that raises. The run-level message +// wraps it as "Task failed with message: ." and ends in a period, +// which the task-level one does not. +const taskFailureMessage = "Workload failed, see run output for details" + +// taskFailure splits a failed task's output the way jobs/runs/get-output does: +// error carries the exception, error_trace the traceback. Reporting the whole +// output as the error instead would leak this server's own wording into the +// message a deploy names the task with. +type taskFailure struct { + message string + trace string +} + +func (e *taskFailure) Error() string { + return e.message +} + +// newTaskFailure takes the exception from the last line of the task's output, +// where a Python traceback ends. +func newTaskFailure(output string) *taskFailure { + trimmed := strings.TrimRight(output, "\r\n") + lastLine := strings.TrimSpace(trimmed[strings.LastIndex(trimmed, "\n")+1:]) + return &taskFailure{message: lastLine, trace: trimmed} +} + // venvPython returns the path to the Python executable in a venv. // On Unix: venv/bin/python // On Windows: venv\Scripts\python.exe @@ -387,12 +420,20 @@ func (s *FakeWorkspace) JobsRunNow(req Request) Response { logs, err = s.executeSparkPythonTask(t) } - if err != nil { + switch { + case errors.Is(err, errNoCodeInWorkspace): + // Nothing ran, so the task keeps its SUCCESS state. + case err != nil: taskRun.State.ResultState = jobs.RunResultStateFailed - s.JobRunOutputs[taskRunId] = jobs.RunOutput{ - Error: err.Error(), + taskRun.State.StateMessage = taskFailureMessage + runOutput := jobs.RunOutput{Error: err.Error()} + // A task this server could not even start (e.g. uv failed) has no + // traceback to report. + if failure, ok := errors.AsType[*taskFailure](err); ok { + runOutput.ErrorTrace = failure.trace } - } else if logs != "" { + s.JobRunOutputs[taskRunId] = runOutput + case logs != "": s.JobRunOutputs[taskRunId] = jobs.RunOutput{ Logs: logs, } @@ -599,7 +640,7 @@ func (s *FakeWorkspace) executePythonWheelTask(jobSettings *jobs.JobSettings, ta } data := s.files[whlPath].Data if len(data) == 0 { - return "", fmt.Errorf("wheel file not found in workspace: %s", whlPath) + return "", fmt.Errorf("%w: wheel file not found in workspace: %s", errNoCodeInWorkspace, whlPath) } localPath := filepath.Join(env.dir, filepath.Base(whlPath)) if err := os.WriteFile(localPath, data, 0o644); err != nil { @@ -618,7 +659,7 @@ func (s *FakeWorkspace) executePythonWheelTask(jobSettings *jobs.JobSettings, ta } if len(env.installedLibs) == 0 { - return "", errors.New("no wheel libraries found in task") + return "", fmt.Errorf("%w: no wheel libraries found in task", errNoCodeInWorkspace) } // Run the entry point using runpy with sys.argv[0] set to the package name, @@ -638,7 +679,7 @@ func (s *FakeWorkspace) executePythonWheelTask(jobSettings *jobs.JobSettings, ta output, err := cmd.CombinedOutput() if err != nil { - return string(output), fmt.Errorf("wheel task execution failed: %s\n%s", err, output) + return string(output), newTaskFailure(string(output)) } // Normalize trailing newlines to match cloud behavior (exactly one trailing newline) @@ -664,7 +705,7 @@ func (s *FakeWorkspace) executeNotebookTask(task jobs.Task, notebookParams map[s notebookData = s.files[notebookPath+".py"].Data } if len(notebookData) == 0 { - return "", fmt.Errorf("notebook not found in workspace: %s (also tried .py)", notebookPath) + return "", fmt.Errorf("%w: notebook not found in workspace: %s (also tried .py)", errNoCodeInWorkspace, notebookPath) } // Create a temporary Python environment for notebook execution @@ -726,7 +767,7 @@ func (s *FakeWorkspace) executeNotebookTask(task jobs.Task, notebookParams map[s output, err := cmd.CombinedOutput() if err != nil { - return string(output), fmt.Errorf("notebook task execution failed: %s\n%s", err, output) + return string(output), newTaskFailure(string(output)) } // Normalize trailing newlines to match cloud behavior (exactly one trailing newline) @@ -750,7 +791,7 @@ func (s *FakeWorkspace) executeSparkPythonTask(task jobs.Task) (string, error) { pythonData := s.files[pythonPath].Data if len(pythonData) == 0 { - return "", fmt.Errorf("python file not found in workspace: %s", pythonPath) + return "", fmt.Errorf("%w: python file not found in workspace: %s", errNoCodeInWorkspace, pythonPath) } env, cleanup, err := s.getOrCreateClusterEnv(task) @@ -772,7 +813,7 @@ func (s *FakeWorkspace) executeSparkPythonTask(task jobs.Task) (string, error) { output, err := exec.Command(venvPython(env.venvDir), runArgs...).CombinedOutput() if err != nil { - return string(output), fmt.Errorf("spark python task execution failed: %s\n%s", err, output) + return string(output), newTaskFailure(string(output)) } // Normalize trailing newlines to match cloud behavior (exactly one trailing newline) @@ -848,6 +889,33 @@ func sparkVersionToPython(task jobs.Task) string { return "3.10" } +// terminateRun completes the run, rolling task outcomes up into the run-level +// state the way the Jobs API does: one failed task fails the whole run, and the +// run reports INTERNAL_ERROR in the deprecated life_cycle_state even though its +// tasks are TERMINATED (status.state is TERMINATED with RUN_EXECUTION_ERROR). +func terminateRun(run *jobs.Run) { + for i := range run.Tasks { + // Tasks that were never executed (jobs/runs/submit) are still running. + if run.Tasks[i].State.LifeCycleState != jobs.RunLifeCycleStateTerminated { + run.Tasks[i].State.LifeCycleState = jobs.RunLifeCycleStateTerminated + run.Tasks[i].State.ResultState = jobs.RunResultStateSuccess + } + } + + run.State = &jobs.RunState{ + LifeCycleState: jobs.RunLifeCycleStateTerminated, + ResultState: jobs.RunResultStateSuccess, + } + for _, task := range run.Tasks { + if task.State.ResultState != jobs.RunResultStateSuccess { + run.State.LifeCycleState = jobs.RunLifeCycleStateInternalError + run.State.ResultState = task.State.ResultState + run.State.StateMessage = fmt.Sprintf("Task %s failed with message: %s.", task.TaskKey, taskFailureMessage) + return + } + } +} + func (s *FakeWorkspace) JobsGetRun(req Request) Response { runId := req.URL.Query().Get("run_id") runIdInt, err := strconv.ParseInt(runId, 10, 64) @@ -865,19 +933,10 @@ func (s *FakeWorkspace) JobsGetRun(req Request) Response { return Response{StatusCode: 404} } - // Simulate cloud behavior: first poll returns RUNNING, next returns TERMINATED SUCCESS. + // Simulate cloud behavior: first poll returns RUNNING, next the terminal state. if run.State.LifeCycleState == jobs.RunLifeCycleStateRunning { // Transition stored state to TERMINATED for the next poll. - run.State = &jobs.RunState{ - LifeCycleState: jobs.RunLifeCycleStateTerminated, - ResultState: jobs.RunResultStateSuccess, - } - for i := range run.Tasks { - run.Tasks[i].State = &jobs.RunState{ - LifeCycleState: jobs.RunLifeCycleStateTerminated, - ResultState: jobs.RunResultStateSuccess, - } - } + terminateRun(&run) s.JobRuns[runIdInt] = run // Return RUNNING for this poll (before the transition). diff --git a/libs/testserver/jobs_test.go b/libs/testserver/jobs_test.go index 38e47b68b08..8a7886ed24b 100644 --- a/libs/testserver/jobs_test.go +++ b/libs/testserver/jobs_test.go @@ -4,6 +4,7 @@ import ( "encoding/json" "net/url" "strconv" + "strings" "testing" "github.com/databricks/databricks-sdk-go/service/jobs" @@ -87,6 +88,93 @@ func TestJobsSubmit_RunReachesTerminalStateOnPoll(t *testing.T) { assert.Equal(t, jobs.RunResultStateSuccess, second.State.ResultState) } +func createJob(t *testing.T, workspace *FakeWorkspace, tasks ...jobs.Task) int64 { + t.Helper() + body, err := json.Marshal(jobs.CreateJob{Name: "my-job", Tasks: tasks}) + require.NoError(t, err) + + response := workspace.JobsCreate(Request{Body: body}) + require.Equal(t, 0, response.StatusCode) + return response.Body.(jobs.CreateResponse).JobId +} + +func runNow(t *testing.T, workspace *FakeWorkspace, request jobs.RunNow) Response { + t.Helper() + body, err := json.Marshal(request) + require.NoError(t, err) + return workspace.JobsRunNow(Request{Body: body}) +} + +func terminatedTask(taskKey string, result jobs.RunResultState) jobs.RunTask { + return jobs.RunTask{ + TaskKey: taskKey, + State: &jobs.RunState{ + LifeCycleState: jobs.RunLifeCycleStateTerminated, + ResultState: result, + }, + } +} + +func TestTerminateRun_FailedTaskFailsTheRun(t *testing.T) { + run := jobs.Run{Tasks: []jobs.RunTask{ + terminatedTask("first", jobs.RunResultStateSuccess), + terminatedTask("second", jobs.RunResultStateFailed), + }} + + terminateRun(&run) + + // A failed run reports INTERNAL_ERROR, not TERMINATED, in life_cycle_state. + assert.Equal(t, jobs.RunLifeCycleStateInternalError, run.State.LifeCycleState) + assert.Equal(t, jobs.RunResultStateFailed, run.State.ResultState) + assert.Equal(t, "Task second failed with message: Workload failed, see run output for details.", run.State.StateMessage) +} + +func TestNewTaskFailure_ReportsTheExceptionAndTheTraceback(t *testing.T) { + output := "Traceback (most recent call last):\n File \"fail.py\", line 1\nRuntimeError: intentional failure\n" + + failure := newTaskFailure(output) + + assert.Equal(t, "RuntimeError: intentional failure", failure.Error()) + assert.Equal(t, strings.TrimRight(output, "\n"), failure.trace) +} + +func TestNewTaskFailure_SingleLineOutputIsTheException(t *testing.T) { + failure := newTaskFailure("RuntimeError: intentional failure\n") + + assert.Equal(t, "RuntimeError: intentional failure", failure.Error()) +} + +func TestTerminateRun_CompletesTasksThatAreStillRunning(t *testing.T) { + // jobs/runs/submit records its tasks as running: they are never executed. + run := jobs.Run{Tasks: []jobs.RunTask{ + {TaskKey: "main", State: &jobs.RunState{LifeCycleState: jobs.RunLifeCycleStateRunning}}, + }} + + terminateRun(&run) + + assert.Equal(t, jobs.RunResultStateSuccess, run.State.ResultState) + assert.Empty(t, run.State.StateMessage) + assert.Equal(t, jobs.RunResultStateSuccess, run.Tasks[0].State.ResultState) +} + +// See errNoCodeInWorkspace: a missing notebook is this server's gap, not a +// failure of the job. +func TestJobsGetRun_TaskWithoutCodeDoesNotFailTheRun(t *testing.T) { + workspace := NewFakeWorkspace("http://test", "dbapi123") + jobID := createJob(t, workspace, jobs.Task{ + TaskKey: "main", + NotebookTask: &jobs.NotebookTask{NotebookPath: "/missing-notebook"}, + }) + + response := runNow(t, workspace, jobs.RunNow{JobId: jobID}) + require.Equal(t, 0, response.StatusCode) + runID := response.Body.(jobs.RunNowResponse).RunId + + // The first poll reports RUNNING, the second the terminal state. + require.Equal(t, jobs.RunLifeCycleStateRunning, getRun(t, workspace, runID).State.LifeCycleState) + assert.Equal(t, jobs.RunResultStateSuccess, getRun(t, workspace, runID).State.ResultState) +} + func TestJobsSubmit_RejectsInvalidGitProvider(t *testing.T) { workspace := NewFakeWorkspace("http://test", "dbapi123")