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..1c18d9bfefb --- /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. If a deploy is interrupted while waiting, the next one resumes waiting on the same run. diff --git a/acceptance/bundle/invariant/configs/job_run.yml.tmpl b/acceptance/bundle/invariant/configs/job_run.yml.tmpl index a09987da02a..f6cd068ff13 100644 --- a/acceptance/bundle/invariant/configs/job_run.yml.tmpl +++ b/acceptance/bundle/invariant/configs/job_run.yml.tmpl @@ -6,14 +6,17 @@ resources: foo: name: test-job-$UNIQUE_NAME tasks: + # Deploying a job_run actually runs the job, and the wait needs a task a + # workspace will execute. Serverless keeps this one to about a minute. - 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 + spark_python_task: + python_file: ./job_run.py + environment_key: default + + environments: + - environment_key: default + spec: + environment_version: "2" job_runs: foo_run: diff --git a/acceptance/bundle/invariant/data/job_run.py b/acceptance/bundle/invariant/data/job_run.py new file mode 100644 index 00000000000..93e0cef4a92 --- /dev/null +++ b/acceptance/bundle/invariant/data/job_run.py @@ -0,0 +1 @@ +print("hello from a job_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..0cd99d8ca8d 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]: 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..6c0c656a9b8 --- /dev/null +++ b/acceptance/bundle/resources/job_runs/failed_run/output.txt @@ -0,0 +1,72 @@ + +=== 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] +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] +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..e8a532f0157 --- /dev/null +++ b/acceptance/bundle/resources/job_runs/failed_run/script @@ -0,0 +1,32 @@ +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 + +# Name the job too, so the run-now bodies below say which number it is. +read_id.py my_job > /dev/null + +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, not the [MY_RUN_ID_2] +# that replaced it: the recreate discards the run it replaces. +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..afe7590b5ad --- /dev/null +++ b/acceptance/bundle/resources/job_runs/failed_run/test.toml @@ -0,0 +1,25 @@ +# 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 failing job for real, so the message the deploy names the task with is +# one a workspace reported. 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/interrupted_run/databricks.yml b/acceptance/bundle/resources/job_runs/interrupted_run/databricks.yml new file mode 100644 index 00000000000..506daeccca8 --- /dev/null +++ b/acceptance/bundle/resources/job_runs/interrupted_run/databricks.yml @@ -0,0 +1,15 @@ +bundle: + name: job-runs-interrupted-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/interrupted_run/out.test.toml b/acceptance/bundle/resources/job_runs/interrupted_run/out.test.toml new file mode 100644 index 00000000000..e90b6d5d1ba --- /dev/null +++ b/acceptance/bundle/resources/job_runs/interrupted_run/out.test.toml @@ -0,0 +1,3 @@ +Local = true +Cloud = false +EnvMatrix.DATABRICKS_BUNDLE_ENGINE = ["direct"] diff --git a/acceptance/bundle/resources/job_runs/interrupted_run/output.txt b/acceptance/bundle/resources/job_runs/interrupted_run/output.txt new file mode 100644 index 00000000000..b9358c210c5 --- /dev/null +++ b/acceptance/bundle/resources/job_runs/interrupted_run/output.txt @@ -0,0 +1,92 @@ + +=== the deploy stops waiting before the run finishes +>>> errcode [CLI] bundle deploy +Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/job-runs-interrupted-run/default/files... +Deploying resources... +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. + +Updating deployment state... + +Exit code: 1 + +=== the abandoned run is still tracked +>>> read_id.py my_job +[MY_JOB_ID] + +>>> read_id.py my_run +[MY_RUN_ID] + +>>> print_requests.py //jobs/run-now +{ + "method": "POST", + "path": "/api/2.2/jobs/run-now", + "body": { + "job_id": [MY_JOB_ID] + } +} + +=== a run that is still going is an update +>>> jq .plan["resources.job_runs.my_run"] tmp.plan.json +{ + "depends_on": [ + { + "node": "resources.jobs.my_job", + "label": "${resources.jobs.my_job.id}" + } + ], + "action": "update", + "new_state": { + "value": { + "job_id": [MY_JOB_ID], + "result_state": "SUCCESS" + } + }, + "remote_state": { + "job_id": [MY_JOB_ID], + "run_id": [MY_RUN_ID], + "run_name": "my-job", + "run_page_url": "[DATABRICKS_URL]/jobs/[MY_JOB_ID]/runs/[MY_RUN_ID]?o=[NUMID]", + "run_type": "JOB_RUN", + "state": { + "life_cycle_state": "RUNNING" + } + }, + "changes": { + "result_state": { + "action": "update", + "reason": "run in progress", + "old": "SUCCESS", + "new": "SUCCESS" + } + } +} + +=== the deploy adopts that run and waits for it to finish +>>> [CLI] bundle deploy --plan tmp.plan.json +Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/job-runs-interrupted-run/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]: SUCCESS +Updating deployment state... +Deployment complete! + +=== the same run finished, and nothing was cancelled, deleted or triggered +>>> read_state.py job_runs my_run id +job_runs my_run id='[MY_RUN_ID]' + +>>> print_requests.py //jobs + +>>> [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/job-runs-interrupted-run/default + +Deleting files... +Destroy complete! diff --git a/acceptance/bundle/resources/job_runs/interrupted_run/script b/acceptance/bundle/resources/job_runs/interrupted_run/script new file mode 100644 index 00000000000..8b25b74360f --- /dev/null +++ b/acceptance/bundle/resources/job_runs/interrupted_run/script @@ -0,0 +1,31 @@ +cleanup() { + trace $CLI bundle destroy --auto-approve + rm -f out.requests.txt +} +trap cleanup EXIT + +# The fault hits the poll, not the run: it ends the deploy's wait while leaving the +# run going, which is what an interrupted deploy leaves behind: the run id is +# recorded in the state, but no outcome was ever observed. +fault.py "GET /api/2.2/jobs/runs/get" 403 0 1 + +title "the deploy stops waiting before the run finishes" +trace errcode $CLI bundle deploy + +title "the abandoned run is still tracked" +trace read_id.py my_job +trace read_id.py my_run +trace print_requests.py //jobs/run-now + +title "a run that is still going is an update" +$CLI bundle plan -o json > tmp.plan.json +trace jq '.plan["resources.job_runs.my_run"]' tmp.plan.json + +# The testserver settles a run on its first poll, so the plan above is replayed +# here: the second read of the run reports it finished. +title "the deploy adopts that run and waits for it to finish" +trace $CLI bundle deploy --plan tmp.plan.json + +title "the same run finished, and nothing was cancelled, deleted or triggered" +trace read_state.py job_runs my_run id +trace print_requests.py //jobs diff --git a/acceptance/bundle/resources/job_runs/interrupted_run/test.toml b/acceptance/bundle/resources/job_runs/interrupted_run/test.toml new file mode 100644 index 00000000000..da4bb643490 --- /dev/null +++ b/acceptance/bundle/resources/job_runs/interrupted_run/test.toml @@ -0,0 +1,7 @@ +# 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"] + +# Local only: the interruption is staged by injecting a fault into the testserver. +RecordRequests = true +Ignore = ["tmp.plan.json"] diff --git a/acceptance/bundle/resources/job_runs/job_parameters/output.txt b/acceptance/bundle/resources/job_runs/job_parameters/output.txt index bcf3f21e017..cceb94e5b85 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]: 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..840e80bbd04 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]: 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]: 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..b96624932a1 --- /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]: 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..288618f3399 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,117 @@ 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) +} + +// jobRunResultStateAction classifies the result_state drift of a run in the given +// state. Old == New because a deploy records the outcome it asked for, and only +// the remote says whether the run reached it. +func jobRunResultStateAction(t *testing.T, state *jobs.RunState) *deployplan.ChangeDesc { + t.Helper() + adapters, err := dresources.InitAll(nil) + require.NoError(t, err) + + remote := &dresources.JobRunRemote{RunId: 123, ResultState: state.ResultState, State: state} + changes := deployplan.Changes{"result_state": &deployplan.ChangeDesc{ + Old: jobs.RunResultStateSuccess, + New: jobs.RunResultStateSuccess, + Remote: state.ResultState, + }} + + require.NoError(t, addPerFieldActions(t.Context(), adapters["job_runs"], changes, remote)) + return changes["result_state"] +} + +// References are served from the remote state cache only for a run the plan +// skips, so a run that stopped without succeeding has to be re-triggered. +func TestJobRunFinishedWithoutSuccessIsRecreate(t *testing.T) { + for name, state := range map[string]*jobs.RunState{ + "FAILED": { + LifeCycleState: jobs.RunLifeCycleStateTerminated, + ResultState: jobs.RunResultStateFailed, + }, + "CANCELED": { + LifeCycleState: jobs.RunLifeCycleStateTerminated, + ResultState: jobs.RunResultStateCanceled, + }, + "TIMEDOUT": { + LifeCycleState: jobs.RunLifeCycleStateTerminated, + ResultState: jobs.RunResultStateTimedout, + }, + // A skipped run never ran, so it reports no result_state at all — the same + // as a run still going. Only the lifecycle state tells the two apart. + "SKIPPED": {LifeCycleState: jobs.RunLifeCycleStateSkipped}, + } { + t.Run(name, func(t *testing.T) { + assert.Equal(t, deployplan.Recreate, jobRunResultStateAction(t, state).Action) + }) + } +} + +// A run that has not stopped yet may still succeed, so the deploy adopts it and +// waits for it. +func TestJobRunInProgressIsUpdate(t *testing.T) { + for _, lifeCycleState := range []jobs.RunLifeCycleState{ + jobs.RunLifeCycleStatePending, + jobs.RunLifeCycleStateRunning, + jobs.RunLifeCycleStateTerminating, + } { + t.Run(string(lifeCycleState), func(t *testing.T) { + change := jobRunResultStateAction(t, &jobs.RunState{LifeCycleState: lifeCycleState}) + + assert.Equal(t, deployplan.Update, change.Action) + assert.Equal(t, "run in progress", change.Reason) + }) + } +} + +// The run reached the required outcome, so the plan can skip it. +func TestJobRunSucceededOutcomeIsNotDrift(t *testing.T) { + change := jobRunResultStateAction(t, &jobs.RunState{ + LifeCycleState: jobs.RunLifeCycleStateTerminated, + ResultState: jobs.RunResultStateSuccess, + }) + + assert.Equal(t, deployplan.Skip, change.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..bdbbb862015 100644 --- a/bundle/direct/dresources/all_test.go +++ b/bundle/direct/dresources/all_test.go @@ -1003,7 +1003,15 @@ func testCRUD(t *testing.T, group string, adapter *Adapter, client *databricks.W remoteStateFromWaitCreate, err := adapter.WaitAfterCreate(ctx, createdID, newState) require.NoError(t, err) if remoteStateFromWaitCreate != nil { - require.Equal(t, remote, remoteStateFromWaitCreate) + // 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. + if group == "job_runs" { + 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..86e7c524b4d 100644 --- a/bundle/direct/dresources/job_run.go +++ b/bundle/direct/dresources/job_run.go @@ -1,19 +1,38 @@ package dresources import ( + "cmp" "context" + "errors" "fmt" + "slices" "strconv" + "strings" + "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/log" + "github.com/databricks/cli/libs/structs/structpath" + "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: it is the outcome the remote is + // compared against, which OverrideChangeDesc turns into a recreate or a wait. + ResultState jobs.RunResultState `json:"result_state,omitempty"` } func (s *JobRunState) UnmarshalJSON(b []byte) error { @@ -29,6 +48,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 +81,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 +121,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 +146,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 +162,222 @@ func (r *ResourceJobRun) DoCreate(ctx context.Context, config *JobRunState) (str return strconv.FormatInt(wait.RunId, 10), nil, nil } -// DoUpdate is intentionally not implemented: a run can't be modified in place, -// so any change recreates it (delete + a fresh RunNow). +// 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) { + return r.waitForRun(ctx, id) +} + +// waitForRun polls the run until it stops, and fails unless it succeeded. +func (r *ResourceJobRun) waitForRun(ctx context.Context, id string) (*JobRunRemote, error) { + runID, err := parseRunID(id) + if err != nil { + return nil, err + } + + // A run can take hours, so report the run page as soon as it is known. pageURL + // outlives the poll so an abandoned wait can still link the run. + var pageURL string + // Every state is logged, but only the outcome is reported: how many states a + // run passes through varies with its compute, and output has to stay reproducible. + var logged jobs.RunState + // Not the SDK's WaitGetRunJobTerminatedOrSkipped: it discards the run on + // INTERNAL_ERROR, the state a run whose task failed lands in, leaving + // runFailedError nothing to name the failing task from. This poll returns the + // run in whatever state it settles in. + 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) + } + if pageURL == "" && run.RunPageUrl != "" { + pageURL = run.RunPageUrl + reportRunLine(ctx, runID, "Run URL: "+workspaceurls.ModernizeJobRunPageURL(pageURL)) + } + if run.State.LifeCycleState != logged.LifeCycleState || run.State.ResultState != logged.ResultState { + logged = *run.State + log.Info(ctx, strings.TrimSpace(fmt.Sprintf("job run %d: %s %s", runID, logged.LifeCycleState, logged.ResultState))) + } + 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 { + // The error names the outcome. + return nil, r.runFailedError(ctx, run) + } + reportRunLine(ctx, runID, string(run.State.ResultState)) + 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 { + // A skipped run has no result_state; report the lifecycle state. + outcome := cmp.Or(string(run.State.ResultState), 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 { + var result []jobs.RunTask + for _, task := range tasks { + if !taskFailed(task) { + continue + } + i := slices.IndexFunc(result, func(seen jobs.RunTask) bool { return seen.TaskKey == task.TaskKey }) + switch { + case i < 0: + result = append(result, task) + case task.AttemptNumber > result[i].AttemptNumber: + result[i] = task + } + } + 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 { + if task.State != nil { + return task.State.LifeCycleState == jobs.RunLifeCycleStateInternalError || + task.State.ResultState == jobs.RunResultStateFailed || + task.State.ResultState == jobs.RunResultStateTimedout + } + // State is deprecated in favour of Status, so a workspace may stop setting it. + if task.Status == nil || task.Status.TerminationDetails == nil { + return false + } + details := task.Status.TerminationDetails + // SKIPPED is also the code for a task an upstream failure skipped. + return details.Type != jobs.TerminationTypeTypeSuccess && + details.Code != jobs.TerminationCodeCodeSkipped +} + +// taskError returns the message the task reported, from the same place `bundle +// run` reads it. +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. + return cmp.Or(reported, taskMessage(task)) +} + +// taskMessage reads the message off whichever field taskFailed accepted the task +// on, so the field it reads is always set. +func taskMessage(task jobs.RunTask) string { + if task.State != nil { + return task.State.StateMessage + } + return task.Status.TerminationDetails.Message +} + +func runPageLine(rawURL string) string { + if rawURL == "" { + return "" + } + return "\nrun page: " + workspaceurls.ModernizeJobRunPageURL(rawURL) +} + +// 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 adopts a run that is still going. A run can't be modified, so the +// only work an update does is finish the wait an interrupted deploy abandoned, +// which is the only reason the plan asks for one. +func (r *ResourceJobRun) DoUpdate(ctx context.Context, id string, _ *JobRunState, _ *PlanEntry) (*JobRunRemote, error) { + return r.waitForRun(ctx, id) +} + +// OverrideChangeDesc downgrades result_state drift to an update while the run is +// still going, so a run that may yet succeed is adopted and waited on. A run that +// stopped without succeeding keeps its recreate. A SKIPPED run reports no +// result_state either, so the lifecycle state is what tells the two apart. +func (*ResourceJobRun) OverrideChangeDesc(_ context.Context, path *structpath.PathNode, change *ChangeDesc, remote *JobRunRemote) error { + // The planner passes no remote state when the run could not be read. + if path.String() != "result_state" || remote == nil || runIsTerminal(remote.State.LifeCycleState) { + return nil + } + change.Action = deployplan.Update + change.Reason = "run in progress" + return nil +} // 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, since cancellation is +// asynchronous and the delete that follows needs a settled run. +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..4138ab18a66 --- /dev/null +++ b/bundle/direct/dresources/job_run_test.go @@ -0,0 +1,475 @@ +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"`) +} + +// 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) { + 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: &jobs.RunState{ + LifeCycleState: jobs.RunLifeCycleStateTerminated, + ResultState: jobs.RunResultStateFailed, + }, + Tasks: []jobs.RunTask{ + {TaskKey: "main", RunId: 999, Status: &jobs.RunStatus{ + State: jobs.RunLifecycleStateV2StateTerminated, + TerminationDetails: &jobs.TerminationDetails{ + Type: jobs.TerminationTypeTypeClientError, + Code: jobs.TerminationCodeCodeRunExecutionError, + Message: "Workload failed, see run output for details", + }, + }}, + // Never ran, so it has no error of its own to report. + {TaskKey: "downstream", RunId: 1000, Status: &jobs.RunStatus{ + State: jobs.RunLifecycleStateV2StateTerminated, + TerminationDetails: &jobs.TerminationDetails{ + Type: jobs.TerminationTypeTypeClientError, + Code: jobs.TerminationCodeCodeSkipped, + }, + }}, + }, + } + }) + 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, `task "main": RuntimeError: intentional failure`) + assert.NotContains(t, err.Error(), `task "downstream"`) +} + +// When the run output carries no error, the termination message stands in. +func TestJobRunWaitFallsBackToTheTerminationMessage(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, + State: &jobs.RunState{ + LifeCycleState: jobs.RunLifeCycleStateTerminated, + ResultState: jobs.RunResultStateFailed, + }, + Tasks: []jobs.RunTask{ + {TaskKey: "main", RunId: 999, Status: &jobs.RunStatus{ + State: jobs.RunLifecycleStateV2StateTerminated, + TerminationDetails: &jobs.TerminationDetails{ + Type: jobs.TerminationTypeTypeCloudFailure, + Code: jobs.TerminationCodeCodeCloudFailure, + Message: "the cloud provider ran out of capacity", + }, + }}, + }, + } + }) + server.Handle("GET", "/api/2.2/jobs/runs/get-output", func(req testserver.Request) any { + return jobs.RunOutput{} + }) + + _, err := waitForTestRun(t, t.Context(), jobRunClientFor(t, server)) + + require.ErrorContains(t, err, `task "main": the cloud provider ran out of capacity`) +} + +// The same fallback for a task still reported through the deprecated state. +func TestJobRunWaitFallsBackToTheTaskMessage(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, + State: &jobs.RunState{ + LifeCycleState: jobs.RunLifeCycleStateTerminated, + ResultState: jobs.RunResultStateFailed, + }, + Tasks: []jobs.RunTask{ + {TaskKey: "main", RunId: 999, State: &jobs.RunState{ + LifeCycleState: jobs.RunLifeCycleStateTerminated, + ResultState: jobs.RunResultStateFailed, + StateMessage: "Workload failed, see run output for details", + }}, + }, + } + }) + server.Handle("GET", "/api/2.2/jobs/runs/get-output", func(req testserver.Request) any { + return jobs.RunOutput{} + }) + + _, err := waitForTestRun(t, t.Context(), jobRunClientFor(t, server)) + + require.ErrorContains(t, err, `task "main": Workload failed, see run output for details`) +} + +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..d096ea5a93e 100644 --- a/bundle/direct/dresources/resources.yml +++ b/bundle/direct/dresources/resources.yml @@ -139,11 +139,46 @@ 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 is listed, so nothing the request carries is drift. + # TestJobRunIgnoresEveryRequestField keeps the list in step with the SDK. + # result_state is not in the list: comparing it against the remote is what + # makes a run that has not succeeded visible to the plan. ignore_remote_changes: - - reason: immutable + - field: job_id + reason: immutable + # GetRun reports job_parameters resolved against the job's defaults, and + # nests the rest under overriding_parameters. + - field: dbt_commands + reason: effective_vs_requested + - field: jar_params + reason: effective_vs_requested + - field: job_parameters + reason: effective_vs_requested + - field: notebook_params + reason: effective_vs_requested + - field: pipeline_params + reason: effective_vs_requested + - field: python_named_params + reason: effective_vs_requested + - field: python_params + reason: effective_vs_requested + - field: spark_submit_params + reason: effective_vs_requested + - field: sql_params + reason: effective_vs_requested + # Fields the run-now request carries that GetRun never reports back. + - field: idempotency_token + reason: not_returned_by_api + - field: only + reason: not_returned_by_api + - field: performance_target + reason: not_returned_by_api + - field: queue + reason: not_returned_by_api + # A run is immutable and fire-once, so any change recreates it. Omitting + # `field` matches every field (root; see TestFieldRuleOmittedIsRoot). + # `field: ""` would instead match nothing. The one exception is a run that is + # still going, which ResourceJobRun.OverrideChangeDesc downgrades to an update. recreate_on_changes: - reason: immutable diff --git a/bundle/internal/schema/annotations.yml b/bundle/internal/schema/annotations.yml index 03f1fdb45c4..afa7c937842 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/schema/jsonschema.json b/bundle/schema/jsonschema.json index 44c1c262481..8f9d128a3ac 100644 --- a/bundle/schema/jsonschema.json +++ b/bundle/schema/jsonschema.json @@ -3328,7 +3328,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/handlers.go b/libs/testserver/handlers.go index a47dfc63425..32a0692c04a 100644 --- a/libs/testserver/handlers.go +++ b/libs/testserver/handlers.go @@ -322,6 +322,10 @@ func AddDefaultHandlers(server *Server) { return req.Workspace.JobsGetRun(req) }) + server.Handle("POST", "/api/2.2/jobs/runs/cancel", func(req Request) any { + return req.Workspace.JobsCancelRun(req) + }) + server.Handle("POST", "/api/2.2/jobs/runs/delete", func(req Request) any { return req.Workspace.JobsDeleteRun(req) }) diff --git a/libs/testserver/jobs.go b/libs/testserver/jobs.go index 9097d43c086..7213e27077a 100644 --- a/libs/testserver/jobs.go +++ b/libs/testserver/jobs.go @@ -20,6 +20,36 @@ 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: ." with a trailing period. +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. +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, or err if the task wrote nothing. +func newTaskFailure(err error, output string) *taskFailure { + trimmed := strings.TrimRight(output, "\r\n") + lastLine := strings.TrimSpace(trimmed[strings.LastIndex(trimmed, "\n")+1:]) + return &taskFailure{message: cmp.Or(lastLine, err.Error()), 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 +417,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 +637,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 +656,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 +676,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(err, string(output)) } // Normalize trailing newlines to match cloud behavior (exactly one trailing newline) @@ -664,7 +702,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 +764,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(err, string(output)) } // Normalize trailing newlines to match cloud behavior (exactly one trailing newline) @@ -750,7 +788,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 +810,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(err, string(output)) } // Normalize trailing newlines to match cloud behavior (exactly one trailing newline) @@ -848,6 +886,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 +930,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). @@ -891,6 +947,36 @@ func (s *FakeWorkspace) JobsGetRun(req Request) Response { return Response{Body: run} } +// JobsCancelRun settles a run that is still going. The real API cancels +// asynchronously; the caller polls runs/get either way. +func (s *FakeWorkspace) JobsCancelRun(req Request) Response { + var request jobs.CancelRun + if err := json.Unmarshal(req.Body, &request); err != nil { + return Response{ + StatusCode: 400, + Body: fmt.Sprintf("request parsing error: %s", err), + } + } + + defer s.LockUnlock()() + + run, ok := s.JobRuns[request.RunId] + if !ok { + return Response{StatusCode: 404} + } + + // A run that already finished keeps the outcome it reached. + if run.State.LifeCycleState == jobs.RunLifeCycleStateRunning { + run.State = &jobs.RunState{ + LifeCycleState: jobs.RunLifeCycleStateTerminated, + ResultState: jobs.RunResultStateCanceled, + } + s.JobRuns[request.RunId] = run + } + + return Response{} +} + func (s *FakeWorkspace) JobsDeleteRun(req Request) Response { var request jobs.DeleteRun if err := json.Unmarshal(req.Body, &request); err != nil { diff --git a/libs/testserver/jobs_test.go b/libs/testserver/jobs_test.go index 38e47b68b08..878bfe3786e 100644 --- a/libs/testserver/jobs_test.go +++ b/libs/testserver/jobs_test.go @@ -2,8 +2,10 @@ package testserver import ( "encoding/json" + "errors" "net/url" "strconv" + "strings" "testing" "github.com/databricks/databricks-sdk-go/service/jobs" @@ -87,6 +89,138 @@ 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(errors.New("exit status 1"), 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(errors.New("exit status 1"), "RuntimeError: intentional failure\n") + + assert.Equal(t, "RuntimeError: intentional failure", failure.Error()) +} + +// A task can exit non-zero without writing anything, e.g. sys.exit(1). +func TestNewTaskFailure_SilentFailureIsReportedAsTheExitError(t *testing.T) { + failure := newTaskFailure(errors.New("exit status 1"), "") + + assert.Equal(t, "exit status 1", 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) +} + +// An interrupted deploy leaves the run going, and the CLI cancels it before +// deleting it, since jobs/runs/delete rejects an active run. +func TestJobsCancelRun_SettlesAnActiveRun(t *testing.T) { + workspace := NewFakeWorkspace("http://test", "dbapi123") + jobID := createJob(t, workspace, jobs.Task{ + TaskKey: "main", + NotebookTask: &jobs.NotebookTask{NotebookPath: "/missing-notebook"}, + }) + runID := runNow(t, workspace, jobs.RunNow{JobId: jobID}).Body.(jobs.RunNowResponse).RunId + + body, err := json.Marshal(jobs.CancelRun{RunId: runID}) + require.NoError(t, err) + require.Equal(t, 0, workspace.JobsCancelRun(Request{Body: body}).StatusCode) + + run := getRun(t, workspace, runID) + assert.Equal(t, jobs.RunLifeCycleStateTerminated, run.State.LifeCycleState) + assert.Equal(t, jobs.RunResultStateCanceled, run.State.ResultState) +} + +func TestJobsCancelRun_LeavesAFinishedRunAlone(t *testing.T) { + workspace := NewFakeWorkspace("http://test", "dbapi123") + jobID := createJob(t, workspace, jobs.Task{ + TaskKey: "main", + NotebookTask: &jobs.NotebookTask{NotebookPath: "/missing-notebook"}, + }) + runID := runNow(t, workspace, jobs.RunNow{JobId: jobID}).Body.(jobs.RunNowResponse).RunId + + // The first poll settles the run, the way a completed deploy leaves it. + getRun(t, workspace, runID) + require.Equal(t, jobs.RunResultStateSuccess, getRun(t, workspace, runID).State.ResultState) + + body, err := json.Marshal(jobs.CancelRun{RunId: runID}) + require.NoError(t, err) + require.Equal(t, 0, workspace.JobsCancelRun(Request{Body: body}).StatusCode) + + assert.Equal(t, jobs.RunResultStateSuccess, getRun(t, workspace, runID).State.ResultState) +} + func TestJobsSubmit_RejectsInvalidGitProvider(t *testing.T) { workspace := NewFakeWorkspace("http://test", "dbapi123")