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!
 
@@ -72,12 +74,9 @@ Resources:
   }
 }
 
-=== the triggered run id is stored in state
->>> read_id.py my_run
-[MY_RUN_ID]
-
->>> read_id.py my_job
-[MY_JOB_ID]
+=== the triggered run id is stored in state, without result_state
+>>> read_state.py job_runs my_run id result_state
+job_runs my_run id='[MY_RUN_ID]' result_state=None
 
 >>> [CLI] bundle destroy --auto-approve
 The following resources will be deleted:

acceptance/bundle/resources/job_runs/basic/script

@@ -20,6 +20,8 @@ trace $CLI bundle summary
 title "exactly one run-now request was made"
 trace print_requests.py //jobs/run-now
 
-title "the triggered run id is stored in state"
-trace read_id.py my_run
-trace read_id.py my_job
+# result_state is a planning sentinel and is cleared before SaveState.
+title "the triggered run id is stored in state, without result_state"
+trace read_state.py job_runs my_run id result_state
+read_id.py my_run > /dev/null
+read_id.py my_job > /dev/null

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,104 @@
+
+=== 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 -o json
+{
+  "depends_on": [
+    {
+      "node": "resources.jobs.my_job",
+      "label": "${resources.jobs.my_job.id}"
+    }
+  ],
+  "action": "recreate",
+  "new_state": {
+    "value": {
+      "job_id": [MY_JOB_ID],
+      "result_state": "SUCCESS"
+    }
+  },
+  "remote_state": {
+    "job_id": [MY_JOB_ID],
+    "result_state": "FAILED",
+    "run_id": [MY_RUN_ID],
+    "run_name": "test-job-[UNIQUE_NAME]",
+    "run_page_url": "[RUN_URL]",
+    "run_type": "JOB_RUN",
+    "state": {
+      "life_cycle_state": "INTERNAL_ERROR",
+      "result_state": "FAILED",
+      "state_message": "Task main failed with message: Workload failed, see run output for details."
+    }
+  },
+  "changes": {
+    "result_state": {
+      "action": "recreate",
+      "reason": "immutable",
+      "new": "SUCCESS",
+      "remote": "FAILED"
+    }
+  }
+}
+
+=== 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,31 @@
+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 WaitAfterCreate, so FAILED vs 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 -o json | jq '.plan["resources.job_runs.my_run"]'
+
+# 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,31 @@
+# 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]'
+
+# [DATABRICKS_URL] is the recording proxy's host, not the workspace host the Jobs
+# API reports here.
+[[Repls]]
+Old = '"run_page_url": ".*"'
+New = '"run_page_url": "[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}