Job runs wait for completion

.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.

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:

acceptance/bundle/invariant/data/job_run.py

@@ -0,0 +1 @@
+print("hello from a job_run")

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

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!
 

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}

acceptance/bundle/resources/job_runs/failed_run/fail.py

@@ -0,0 +1 @@
+raise RuntimeError("intentional failure")

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!

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

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]'

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}

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!

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

acceptance/bundle/resources/job_runs/interrupted_run/test.toml

@@ -0,0 +1,8 @@
+# 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"]
+
+# The interruption is staged by injecting a fault into the testserver, so this
+# stays off cloud.
+RecordRequests = true
+Ignore = ["tmp.plan.json"]