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 the run's outcome (e.g. `${resources.job_runs.nightly.state.result_state}`). The deploy reports the run page URL while it waits, and a run that does not succeed fails the deploy, naming the failed task and the message it reported. A failed run stays recorded, so the next deploy does not run the job again. Destroying a run that has not finished, which is what interrupting a deploy mid-wait leaves behind, cancels it first.

acceptance/bundle/invariant/configs/job_run.yml.tmpl

@@ -6,14 +6,13 @@ resources:
     foo:
       name: test-job-$UNIQUE_NAME
       tasks:
+        # Deploying a job_run actually runs the job, so use a condition task,
+        # which needs no workspace files or compute.
         - task_key: only_task
-          notebook_task:
-            notebook_path: /Shared/notebook
-          new_cluster:
-            spark_version: $DEFAULT_SPARK_VERSION
-            node_type_id: $NODE_TYPE_ID
-            instance_pool_id: $TEST_INSTANCE_POOL_ID
-            num_workers: 1
+          condition_task:
+            op: EQUAL_TO
+            left: "1"
+            right: "1"
 
   job_runs:
     foo_run:

acceptance/bundle/invariant/test.toml

@@ -83,6 +83,12 @@ no_alert_on_cloud = ["CONFIG_Cloud=true", "INPUT_CONFIG=alert.yml.tmpl"]
 # so this config is local-only (the mock server stores it verbatim).
 no_run_job_ref_on_cloud = ["CONFIG_Cloud=true", "INPUT_CONFIG=job_run_job_ref.yml.tmpl"]
 
+# Deploying a job_run waits for the run to succeed, and a real workspace reports a
+# run of condition tasks alone as SKIPPED. A task that does succeed would run in
+# every variant of this suite, so resources/job_runs/wait_cloud covers the real run
+# on cloud instead. Still exercised locally here.
+no_job_run_on_cloud = ["CONFIG_Cloud=true", "INPUT_CONFIG=job_run.yml.tmpl"]
+
 # Postgres resources only work on AWS
 no_postgres_project_on_cloud = ["CONFIG_Cloud=true", "INPUT_CONFIG=postgres_project.yml.tmpl"]
 no_postgres_branch_on_cloud = ["CONFIG_Cloud=true", "INPUT_CONFIG=postgres_branch.yml.tmpl"]

acceptance/bundle/resources/job_runs/basic/output.txt

@@ -34,6 +34,8 @@ Resources:
 >>> [CLI] bundle deploy
 Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/job-runs-basic/default/files...
 Deploying resources...
+job run [MY_RUN_ID]: Run URL: [DATABRICKS_URL]/jobs/[MY_JOB_ID]/runs/[MY_RUN_ID]?o=[NUMID]
+job run [MY_RUN_ID]: [TIMESTAMP] "my-job" TERMINATED SUCCESS
 Updating deployment state...
 Deployment complete!
 

acceptance/bundle/resources/job_runs/failed_cloud/databricks.yml.tmpl

@@ -0,0 +1,26 @@
+bundle:
+  name: job-runs-failed-cloud
+
+workspace:
+  root_path: ~/.bundle/$UNIQUE_NAME
+
+resources:
+  jobs:
+    my_job:
+      name: test-job-$UNIQUE_NAME
+      tasks:
+        # Serverless keeps the run to about a minute; a job cluster would take
+        # several.
+        - task_key: main
+          spark_python_task:
+            python_file: ./fail.py
+          environment_key: default
+
+      environments:
+        - environment_key: default
+          spec:
+            environment_version: "2"
+
+  job_runs:
+    my_run:
+      job_id: ${resources.jobs.my_job.id}

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

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

acceptance/bundle/resources/job_runs/failed_cloud/output.txt

@@ -0,0 +1,20 @@
+
+=== a run that fails on a real workspace fails the deploy
+>>> contains.py run did not succeed: FAILED run page: http !task "main": FAILED
+
+>>> grep -cE task "main": .+ deploy.log
+1
+
+=== the failed run is recorded, so it is destroyed rather than left behind
+>>> read_id.py my_run
+[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_cloud/script

@@ -0,0 +1,19 @@
+envsubst < databricks.yml.tmpl > databricks.yml
+
+cleanup() {
+    trace $CLI bundle destroy --auto-approve
+}
+trap cleanup EXIT
+
+# A real task's traceback is text we do not control, so keep the deploy output out
+# of the golden and assert the parts the CLI produces.
+title "a run that fails on a real workspace fails the deploy"
+musterr $CLI bundle deploy > deploy.log 2>&1
+trace contains.py 'run did not succeed: FAILED' 'run page: http' '!task "main": FAILED' < deploy.log > /dev/null
+
+# A non-empty message means the task reported one: the negative assertion above
+# rules out the fallback to the states the run itself reports.
+trace grep -cE 'task "main": .+' deploy.log
+
+title "the failed run is recorded, so it is destroyed rather than left behind"
+trace read_id.py my_run

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

@@ -0,0 +1,21 @@
+# 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 failure counterpart of wait_cloud: the error the test server reports is one it
+# writes itself, so only a real workspace shows whether a failed task reports a
+# message the deploy can name. Serverless needs Unity Catalog.
+Cloud = true
+RequiresUnityCatalog = true
+
+# A real workspace is not proxied, so there are no recorded requests to assert on.
+RecordRequests = false
+
+# The deploy fails mid-way, leaving local deployment state behind.
+Ignore = [
+    ".databricks",
+    "databricks.yml",
+    "databricks.yml.tmpl",
+    "fail.py",
+    "deploy.log",
+]

acceptance/bundle/resources/job_runs/failed_run/databricks.yml

@@ -0,0 +1,34 @@
+bundle:
+  name: job-runs-failed-run
+
+resources:
+  jobs:
+    my_job:
+      name: my-job
+      tasks:
+        # The test server runs this locally: it exits non-zero, failing the task
+        # and with it the run.
+        - task_key: main
+          spark_python_task:
+            python_file: ./fail.py
+          environment_key: default
+
+      environments:
+        - environment_key: default
+          spec:
+            client: "2"
+
+    # Depends on my_run's result_state, so the failing run aborts the deploy
+    # before this job is created.
+    downstream_job:
+      name: downstream-job
+      tags:
+        run_result: ${resources.job_runs.my_run.state.result_state}
+      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/failed_run/fail.py

@@ -0,0 +1,4 @@
+import sys
+
+print("intentional failure", file=sys.stderr)
+sys.exit(1)

acceptance/bundle/resources/job_runs/failed_run/output.txt

@@ -0,0 +1,56 @@
+
+=== a run that finishes FAILED fails the deploy
+>>> [CLI] bundle deploy
+Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/job-runs-failed-run/default/files...
+Deploying resources...
+job run [MY_RUN_ID]: Run URL: [DATABRICKS_URL]/jobs/[NUMID]/runs/[MY_RUN_ID]?o=[NUMID]
+job run [MY_RUN_ID]: [TIMESTAMP] "my-job" TERMINATED FAILED task main failed
+Error: cannot create resources.job_runs.my_run: waiting after creating id=[MY_RUN_ID]: run did not succeed: FAILED: task main failed
+task "main": spark python task execution failed: exit status 1
+intentional failure
+
+run page: [DATABRICKS_URL]/jobs/[NUMID]/runs/[MY_RUN_ID]?o=[NUMID]
+
+Error: cannot create resources.jobs.downstream_job: dependency failed: resources.job_runs.my_run
+
+Updating deployment state...
+
+=== the failed run stays recorded, so a redeploy starts no new run
+>>> read_id.py my_run
+[MY_RUN_ID]
+
+>>> [CLI] bundle plan
+create jobs.downstream_job
+
+Plan: 1 to add, 0 to change, 0 to delete, 2 unchanged
+
+>>> [CLI] bundle deploy
+Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/job-runs-failed-run/default/files...
+Deploying resources...
+Updating deployment state...
+Deployment complete!
+
+=== downstream_job resolved its tag from the failed run
+>>> jq -r select(.path == "/api/2.2/jobs/create" and .body.name == "downstream-job") | .body.tags.run_result out.requests.txt
+FAILED
+
+=== run-now was issued once
+>>> print_requests.py //jobs/run-now
+{
+  "method": "POST",
+  "path": "/api/2.2/jobs/run-now",
+  "body": {
+    "job_id": [NUMID]
+  }
+}
+
+>>> [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/job-runs-failed-run/default
+
+Deleting files...
+Destroy complete!

acceptance/bundle/resources/job_runs/failed_run/script

@@ -0,0 +1,24 @@
+cleanup() {
+    trace $CLI bundle destroy --auto-approve
+    rm -f out.requests.txt
+}
+trap cleanup EXIT
+
+# The run finishes FAILED, so the deploy aborts: the error names the failed task
+# and the message it reported, 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 a run that
+# failed stays recorded and an unchanged config plans no second run.
+title "the failed run stays recorded, so a redeploy starts no new run"
+trace read_id.py my_run
+trace $CLI bundle plan
+trace $CLI bundle deploy
+
+title "downstream_job resolved its tag from the failed run"
+trace jq -r 'select(.path == "/api/2.2/jobs/create" and .body.name == "downstream-job") | .body.tags.run_result' out.requests.txt
+
+title "run-now was issued once"
+trace print_requests.py //jobs/run-now

acceptance/bundle/resources/job_runs/failed_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"]
+RecordRequests = true
+
+# The deploy fails mid-way, leaving local deployment state behind.
+Ignore = [".databricks"]

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 [NUMID]: Run URL: [DATABRICKS_URL]/jobs/[NUMID]/runs/[NUMID]?o=[NUMID]
+job run [NUMID]: [TIMESTAMP] "my-job" TERMINATED SUCCESS
 Updating deployment state...
 Deployment complete!
 

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 [NUMID]: Run URL: [DATABRICKS_URL]/jobs/[MY_JOB_ID]/runs/[NUMID]?o=[NUMID]
+job run [NUMID]: [TIMESTAMP] "my-job" TERMINATED SUCCESS
 Updating deployment state...
 Deployment complete!
 
@@ -64,10 +66,11 @@ Resources:
     },
     "run_id": [NUMID],
     "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/[NUMID]?o=[NUMID]",
     "run_type": "JOB_RUN",
     "state": {
-      "life_cycle_state": "RUNNING"
+      "life_cycle_state": "TERMINATED",
+      "result_state": "SUCCESS"
     }
   },
   "changes": {
@@ -84,6 +87,8 @@ Resources:
 >>> [CLI] bundle deploy
 Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/job-runs-redeploy/default/files...
 Deploying resources...
+job run [NUMID]: Run URL: [DATABRICKS_URL]/jobs/[MY_JOB_ID]/runs/[NUMID]?o=[NUMID]
+job run [NUMID]: [TIMESTAMP] "my-job" TERMINATED SUCCESS
 Updating deployment state...
 Deployment complete!
 

acceptance/bundle/resources/job_runs/wait_cloud/databricks.yml.tmpl

@@ -0,0 +1,49 @@
+bundle:
+  name: job-runs-wait-cloud
+
+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; a job cluster would take
+        # several.
+        - 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 the tag it is created with shows whether the
+    # deploy waited for the run before creating resources that depend on it.
+    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 to avoid perpetual drift.
+      job_parameters:
+        env: prod

acceptance/bundle/resources/job_runs/wait_cloud/hello.py

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