diff --git a/.nextchanges/bundles/job-runs-on-value-change.md b/.nextchanges/bundles/job-runs-on-value-change.md new file mode 100644 index 00000000000..dc031aabce5 --- /dev/null +++ b/.nextchanges/bundles/job-runs-on-value-change.md @@ -0,0 +1 @@ +* Add `lifecycle.triggers.on_value_change` for `job_runs`: re-fire the run when a watched expression's resolved value changes. ([#6341](https://github.com/databricks/cli/pull/6341)) diff --git a/acceptance/bundle/python/job_runs-on_value_change/databricks.yml b/acceptance/bundle/python/job_runs-on_value_change/databricks.yml new file mode 100644 index 00000000000..dca8ab672c0 --- /dev/null +++ b/acceptance/bundle/python/job_runs-on_value_change/databricks.yml @@ -0,0 +1,32 @@ +bundle: + name: test-bundle + +sync: {paths: []} # don't need to copy files + +variables: + x: + default: x1 + y: + default: y1 + +python: + resources: + - "resources:load_resources" + mutators: + - "mutators:watch_y" + +resources: + jobs: + my_job: + name: my-job + tasks: + - task_key: main + notebook_task: + notebook_path: /Workspace/test + + job_runs: + yaml_run: + job_id: ${resources.jobs.my_job.id} + lifecycle: + triggers: + - on_value_change: ${var.x} diff --git a/acceptance/bundle/python/job_runs-on_value_change/mutators.py b/acceptance/bundle/python/job_runs-on_value_change/mutators.py new file mode 100644 index 00000000000..15ef3dc2f58 --- /dev/null +++ b/acceptance/bundle/python/job_runs-on_value_change/mutators.py @@ -0,0 +1,12 @@ +from dataclasses import replace + +from databricks.bundles.core import job_run_mutator +from databricks.bundles.job_runs import JobRun, JobRunLifecycle, JobRunTrigger + + +@job_run_mutator +def watch_y(run: JobRun) -> JobRun: + lifecycle = run.lifecycle or JobRunLifecycle() + triggers = [*lifecycle.triggers, JobRunTrigger(on_value_change="${var.y}")] + + return replace(run, lifecycle=replace(lifecycle, triggers=triggers)) diff --git a/acceptance/bundle/python/job_runs-on_value_change/out.test.toml b/acceptance/bundle/python/job_runs-on_value_change/out.test.toml new file mode 100644 index 00000000000..aeddf3e619e --- /dev/null +++ b/acceptance/bundle/python/job_runs-on_value_change/out.test.toml @@ -0,0 +1,3 @@ +Cloud = false +EnvMatrix.DMS = ["", "true"] +EnvMatrix.PYDAB_VERSION = ["current"] diff --git a/acceptance/bundle/python/job_runs-on_value_change/output.txt b/acceptance/bundle/python/job_runs-on_value_change/output.txt new file mode 100644 index 00000000000..4374e23aa7f --- /dev/null +++ b/acceptance/bundle/python/job_runs-on_value_change/output.txt @@ -0,0 +1,84 @@ + +=== validate resolves watches from YAML, from Python and from a Python mutator +>>> uv run [UV_ARGS] -q [CLI] bundle validate --output json +{ + "py_run": { + "job_id": "${resources.jobs.my_job.id}", + "lifecycle": { + "triggers": [ + { + "on_value_change": "x1" + }, + { + "on_value_change": "y1" + } + ], + "triggers_state": { + "on_value_change": [ + "x1", + "y1" + ] + } + } + }, + "yaml_run": { + "job_id": "${resources.jobs.my_job.id}", + "lifecycle": { + "triggers": [ + { + "on_value_change": "x1" + }, + { + "on_value_change": "y1" + } + ], + "triggers_state": { + "on_value_change": [ + "x1", + "y1" + ] + } + } + } +} + +=== first deploy runs both +>>> uv run [UV_ARGS] -q [CLI] bundle deploy +Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/test-bundle/default/files... +Created job_runs.py_run +Created job_runs.yaml_run +Created jobs.my_job +Files: 0 uploaded, 0 deleted +Resources: 3 created, 0 changed, 0 deleted, 0 unchanged + +=== redeploy with unchanged values plans nothing +>>> uv run [UV_ARGS] -q [CLI] bundle plan +Plan: 0 to add, 0 to change, 0 to delete, 3 unchanged + +=== changing the variable watched in YAML and Python re-fires both runs +>>> uv run [UV_ARGS] -q [CLI] bundle plan +recreate job_runs.py_run +recreate job_runs.yaml_run + +Plan: 2 to add, 0 to change, 2 to delete, 1 unchanged + +>>> uv run [UV_ARGS] -q [CLI] bundle deploy +Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/test-bundle/default/files... +Recreated job_runs.py_run +Recreated job_runs.yaml_run +Files: 0 uploaded, 0 deleted +Resources: 2 created, 0 changed, 2 deleted, 1 unchanged + +=== changing the variable the Python mutator watches re-fires both runs +>>> uv run [UV_ARGS] -q [CLI] bundle plan +recreate job_runs.py_run +recreate job_runs.yaml_run + +Plan: 2 to add, 0 to change, 2 to delete, 1 unchanged + +>>> uv run [UV_ARGS] -q [CLI] bundle deploy +Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/test-bundle/default/files... +Recreated job_runs.py_run +Recreated job_runs.yaml_run +Files: 0 uploaded, 0 deleted +Resources: 2 created, 0 changed, 2 deleted, 1 unchanged diff --git a/acceptance/bundle/python/job_runs-on_value_change/resources.py b/acceptance/bundle/python/job_runs-on_value_change/resources.py new file mode 100644 index 00000000000..73c4f39944f --- /dev/null +++ b/acceptance/bundle/python/job_runs-on_value_change/resources.py @@ -0,0 +1,15 @@ +from databricks.bundles.core import Resources + + +def load_resources() -> Resources: + resources = Resources() + + resources.add_job_run( + "py_run", + { + "job_id": "${resources.jobs.my_job.id}", + "lifecycle": {"triggers": [{"on_value_change": "${var.x}"}]}, + }, + ) + + return resources diff --git a/acceptance/bundle/python/job_runs-on_value_change/script b/acceptance/bundle/python/job_runs-on_value_change/script new file mode 100644 index 00000000000..bb4aa13a287 --- /dev/null +++ b/acceptance/bundle/python/job_runs-on_value_change/script @@ -0,0 +1,20 @@ +title "validate resolves watches from YAML, from Python and from a Python mutator" +trace uv run $UV_ARGS -q $CLI bundle validate --output json | jq .resources.job_runs + +title "first deploy runs both" +trace uv run $UV_ARGS -q $CLI bundle deploy + +title "redeploy with unchanged values plans nothing" +trace uv run $UV_ARGS -q $CLI bundle plan + +title "changing the variable watched in YAML and Python re-fires both runs" +update_file.py databricks.yml "default: x1" "default: x2" +trace uv run $UV_ARGS -q $CLI bundle plan +trace uv run $UV_ARGS -q $CLI bundle deploy + +title "changing the variable the Python mutator watches re-fires both runs" +update_file.py databricks.yml "default: y1" "default: y2" +trace uv run $UV_ARGS -q $CLI bundle plan +trace uv run $UV_ARGS -q $CLI bundle deploy + +rm -fr .databricks __pycache__ diff --git a/acceptance/bundle/python/job_runs-on_value_change/test.toml b/acceptance/bundle/python/job_runs-on_value_change/test.toml new file mode 100644 index 00000000000..f99876d6945 --- /dev/null +++ b/acceptance/bundle/python/job_runs-on_value_change/test.toml @@ -0,0 +1,7 @@ +Cloud = false # tests don't interact with APIs + +# job_runs are only supported in the current version of the wheel +EnvMatrix.PYDAB_VERSION = ["current"] + +# Both runs fire in parallel, so their progress lines interleave in any order. +SuppressLines.run_output = '^Output from job_runs\.' diff --git a/acceptance/bundle/refschema/output/job_runs.txt b/acceptance/bundle/refschema/output/job_runs.txt index c613f750f7b..8cca9edd2e0 100644 --- a/acceptance/bundle/refschema/output/job_runs.txt +++ b/acceptance/bundle/refschema/output/job_runs.txt @@ -19,10 +19,13 @@ resources.job_runs.*.lifecycle.triggers []resources.JobRunTrigger INPUT resources.job_runs.*.lifecycle.triggers[*] resources.JobRunTrigger INPUT resources.job_runs.*.lifecycle.triggers[*].on_bundle_deploy *bool INPUT resources.job_runs.*.lifecycle.triggers[*].on_file_change *string INPUT +resources.job_runs.*.lifecycle.triggers[*].on_value_change *string INPUT resources.job_runs.*.lifecycle.triggers_state *resources.JobRunTriggersState INPUT STATE resources.job_runs.*.lifecycle.triggers_state.on_bundle_deploy string INPUT STATE resources.job_runs.*.lifecycle.triggers_state.on_file_change map[string]string INPUT STATE resources.job_runs.*.lifecycle.triggers_state.on_file_change.* string INPUT STATE +resources.job_runs.*.lifecycle.triggers_state.on_value_change []string INPUT STATE +resources.job_runs.*.lifecycle.triggers_state.on_value_change[*] string INPUT STATE resources.job_runs.*.modified_status string INPUT resources.job_runs.*.notebook_params map[string]string ALL resources.job_runs.*.notebook_params.* string ALL diff --git a/acceptance/bundle/resources/job_runs/on_value_change/databricks.yml b/acceptance/bundle/resources/job_runs/on_value_change/databricks.yml new file mode 100644 index 00000000000..d46e21e06ef --- /dev/null +++ b/acceptance/bundle/resources/job_runs/on_value_change/databricks.yml @@ -0,0 +1,25 @@ +bundle: + name: job-runs-on-value-change + +variables: + stamp: + default: v1 + other: + default: a1 + number: + default: 42 + +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} + lifecycle: + triggers: [{on_file_change: watched.txt}, {on_value_change: "${var.stamp}"}, {on_value_change: "${var.other}"}, {on_value_change: "${var.number}"}, {on_value_change: "${workspace.root_path}"}] diff --git a/acceptance/bundle/resources/job_runs/on_value_change/out.test.toml b/acceptance/bundle/resources/job_runs/on_value_change/out.test.toml new file mode 100644 index 00000000000..a927a5fbc06 --- /dev/null +++ b/acceptance/bundle/resources/job_runs/on_value_change/out.test.toml @@ -0,0 +1,2 @@ +Cloud = false +EnvMatrix.DMS = ["", "true"] diff --git a/acceptance/bundle/resources/job_runs/on_value_change/output.txt b/acceptance/bundle/resources/job_runs/on_value_change/output.txt new file mode 100644 index 00000000000..510452d70c3 --- /dev/null +++ b/acceptance/bundle/resources/job_runs/on_value_change/output.txt @@ -0,0 +1,205 @@ + +=== first deploy triggers a run +>>> [CLI] bundle deploy +Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/job-runs-on-value-change/default/files... +Created jobs.my_job +Output from job_runs.my_run: id=[MY_RUN_ID]: Run URL: [DATABRICKS_URL]/jobs/[MY_JOB_ID]/runs/[MY_RUN_ID]?o=[WORKSPACE_ID] +Output from job_runs.my_run: id=[MY_RUN_ID]: SUCCESS +Created job_runs.my_run +Files: 2 uploaded, 0 deleted +Resources: 2 created, 0 changed, 0 deleted, 0 unchanged + +>>> read_id.py my_job +[MY_JOB_ID] + +>>> print_requests.py //jobs/run-now +{ + "method": "POST", + "path": "/api/2.2/jobs/run-now", + "body": { + "idempotency_token": "[UUID]", + "job_id": [MY_JOB_ID] + } +} + +=== resolved values are stored in state in trigger order +>>> print_state.py +{ + "on_file_change": { + "watched.txt": "[FILE_HASH][0]" + }, + "on_value_change": [ + "v1", + "a1", + "42", + "/Workspace/Users/[USERNAME]/.bundle/job-runs-on-value-change/default" + ] +} + +=== redeploy with unchanged values plans nothing +>>> [CLI] bundle plan +Plan: 0 to add, 0 to change, 0 to delete, 2 unchanged + +>>> [CLI] bundle deploy +Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/job-runs-on-value-change/default/files... +Files: 0 uploaded, 0 deleted +Resources: 0 created, 0 changed, 0 deleted, 2 unchanged + +>>> print_requests.py //jobs/run-now + +=== changing one value re-fires +>>> changes +{ + "lifecycle.triggers_state.on_value_change[0]": { + "action": "recreate", + "reason": "immutable", + "old": "v1", + "new": "v2" + }, + "result_state": { + "action": "skip", + "reason": "remote_already_set", + "new": "SUCCESS", + "remote": "SUCCESS" + } +} + +>>> [CLI] bundle deploy +Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/job-runs-on-value-change/default/files... +Output from job_runs.my_run: id=[MY_RUN_ID_2]: Run URL: [DATABRICKS_URL]/jobs/[MY_JOB_ID]/runs/[MY_RUN_ID_2]?o=[WORKSPACE_ID] +Output from job_runs.my_run: id=[MY_RUN_ID_2]: SUCCESS +Recreated job_runs.my_run +Files: 1 uploaded, 0 deleted +Resources: 1 created, 0 changed, 1 deleted, 1 unchanged + +>>> print_requests.py --keep //jobs/runs/delete + +>>> print_requests.py //jobs/run-now +{ + "method": "POST", + "path": "/api/2.2/jobs/run-now", + "body": { + "idempotency_token": "[UUID]", + "job_id": [MY_JOB_ID] + } +} + +=== removing the last on_value_change through a saved plan does not run +>>> [CLI] bundle plan +Plan: 0 to add, 0 to change, 0 to delete, 2 unchanged + +>>> [CLI] bundle deploy --plan tmp.plan.json +Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/job-runs-on-value-change/default/files... +Files: 2 uploaded, 0 deleted +Resources: 0 created, 0 changed, 0 deleted, 2 unchanged + +>>> print_requests.py //jobs/run-now + +=== re-adding it with an unchanged value does not re-fire +>>> [CLI] bundle plan +Plan: 0 to add, 0 to change, 0 to delete, 2 unchanged + +>>> [CLI] bundle deploy +Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/job-runs-on-value-change/default/files... +Files: 1 uploaded, 0 deleted +Resources: 0 created, 0 changed, 0 deleted, 2 unchanged + +>>> print_requests.py //jobs/run-now + +=== removing an on_value_change that is not the last shifts the rest and re-fires +>>> changes +{ + "lifecycle.triggers_state.on_value_change": { + "action": "recreate", + "reason": "immutable", + "old": [ + "v2", + "a1", + "42", + "/Workspace/Users/[USERNAME]/.bundle/job-runs-on-value-change/default" + ], + "new": [ + "v2", + "42", + "/Workspace/Users/[USERNAME]/.bundle/job-runs-on-value-change/default" + ] + }, + "result_state": { + "action": "skip", + "reason": "remote_already_set", + "new": "SUCCESS", + "remote": "SUCCESS" + } +} + +>>> [CLI] bundle deploy +Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/job-runs-on-value-change/default/files... +Output from job_runs.my_run: id=[MY_RUN_ID_3]: Run URL: [DATABRICKS_URL]/jobs/[MY_JOB_ID]/runs/[MY_RUN_ID_3]?o=[WORKSPACE_ID] +Output from job_runs.my_run: id=[MY_RUN_ID_3]: SUCCESS +Recreated job_runs.my_run +Files: 1 uploaded, 0 deleted +Resources: 1 created, 0 changed, 1 deleted, 1 unchanged + +>>> print_requests.py //jobs/run-now +{ + "method": "POST", + "path": "/api/2.2/jobs/run-now", + "body": { + "idempotency_token": "[UUID]", + "job_id": [MY_JOB_ID] + } +} + +=== removing the remaining on_value_change entries does not run +>>> [CLI] bundle plan +Plan: 0 to add, 0 to change, 0 to delete, 2 unchanged + +>>> [CLI] bundle deploy +Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/job-runs-on-value-change/default/files... +Files: 1 uploaded, 0 deleted +Resources: 0 created, 0 changed, 0 deleted, 2 unchanged + +>>> print_requests.py //jobs/run-now + +=== file changes still re-fire after value triggers are removed +>>> changes +{ + "lifecycle.triggers_state.on_file_change['watched.txt']": { + "action": "recreate", + "reason": "immutable", + "old": "[FILE_HASH][0]", + "new": "[FILE_HASH][1]" + }, + "result_state": { + "action": "skip", + "reason": "remote_already_set", + "new": "SUCCESS", + "remote": "SUCCESS" + } +} + +>>> [CLI] bundle deploy +Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/job-runs-on-value-change/default/files... +Output from job_runs.my_run: id=[MY_RUN_ID_4]: Run URL: [DATABRICKS_URL]/jobs/[MY_JOB_ID]/runs/[MY_RUN_ID_4]?o=[WORKSPACE_ID] +Output from job_runs.my_run: id=[MY_RUN_ID_4]: SUCCESS +Recreated job_runs.my_run +Files: 1 uploaded, 0 deleted +Resources: 1 created, 0 changed, 1 deleted, 1 unchanged + +>>> print_requests.py //jobs/run-now +{ + "method": "POST", + "path": "/api/2.2/jobs/run-now", + "body": { + "idempotency_token": "[UUID]", + "job_id": [MY_JOB_ID] + } +} + +>>> [CLI] bundle destroy --auto-approve +The following resources will be deleted: + delete resources.jobs.my_job + +All files and directories at the following location will be deleted: /Workspace/Users/[USERNAME]/.bundle/job-runs-on-value-change/default + +Destroy: 1 deleted diff --git a/acceptance/bundle/resources/job_runs/on_value_change/script b/acceptance/bundle/resources/job_runs/on_value_change/script new file mode 100644 index 00000000000..2075827b559 --- /dev/null +++ b/acceptance/bundle/resources/job_runs/on_value_change/script @@ -0,0 +1,64 @@ +cleanup() { + trace $CLI bundle destroy --auto-approve + rm -f out.requests.txt +} +trap cleanup EXIT + +changes() { + $CLI bundle plan -o json | jq '.plan["resources.job_runs.my_run"].changes' +} + +title "first deploy triggers a run" +trace $CLI bundle deploy +trace read_id.py my_job +read_id.py my_run > /dev/null +trace print_requests.py //jobs/run-now + +title "resolved values are stored in state in trigger order" +trace print_state.py | jq '.state["resources.job_runs.my_run"].state.lifecycle.triggers_state' + +title "redeploy with unchanged values plans nothing" +trace $CLI bundle plan +trace $CLI bundle deploy +trace print_requests.py //jobs/run-now + +title "changing one value re-fires" +update_file.py databricks.yml "default: v1" "default: v2" +trace changes +trace $CLI bundle deploy +read_id.py my_run > /dev/null +trace print_requests.py --keep //jobs/runs/delete +trace print_requests.py //jobs/run-now + +title "removing the last on_value_change through a saved plan does not run" +update_file.py databricks.yml ', {on_value_change: "${workspace.root_path}"}' '' +trace $CLI bundle plan +$CLI bundle plan -o json > tmp.plan.json +trace $CLI bundle deploy --plan tmp.plan.json +trace print_requests.py //jobs/run-now + +title "re-adding it with an unchanged value does not re-fire" +update_file.py databricks.yml '{on_value_change: "${var.number}"}' '{on_value_change: "${var.number}"}, {on_value_change: "${workspace.root_path}"}' +trace $CLI bundle plan +trace $CLI bundle deploy +trace print_requests.py //jobs/run-now + +title "removing an on_value_change that is not the last shifts the rest and re-fires" +update_file.py databricks.yml ', {on_value_change: "${var.other}"}' '' +trace changes +trace $CLI bundle deploy +read_id.py my_run > /dev/null +trace print_requests.py //jobs/run-now + +title "removing the remaining on_value_change entries does not run" +update_file.py databricks.yml ', {on_value_change: "${var.stamp}"}, {on_value_change: "${var.number}"}, {on_value_change: "${workspace.root_path}"}' '' +trace $CLI bundle plan +trace $CLI bundle deploy +trace print_requests.py //jobs/run-now + +title "file changes still re-fire after value triggers are removed" +update_file.py watched.txt "v1" "v2" +trace changes +trace $CLI bundle deploy +read_id.py my_run > /dev/null +trace print_requests.py //jobs/run-now diff --git a/acceptance/bundle/resources/job_runs/on_value_change/test.toml b/acceptance/bundle/resources/job_runs/on_value_change/test.toml new file mode 100644 index 00000000000..608b2e13890 --- /dev/null +++ b/acceptance/bundle/resources/job_runs/on_value_change/test.toml @@ -0,0 +1,9 @@ +# The on_file_change content hash contains digit runs that the parent `\d{8,}` +# rule would mangle. Order=1 runs before it; Distinct keeps different hashes distinct. +Ignore = ["tmp.plan.json"] + +[[Repls]] +Old = '[0-9a-f]{64}' +New = '[FILE_HASH]' +Order = 1 +Distinct = true diff --git a/acceptance/bundle/resources/job_runs/on_value_change/watched.txt b/acceptance/bundle/resources/job_runs/on_value_change/watched.txt new file mode 100644 index 00000000000..626799f0f85 --- /dev/null +++ b/acceptance/bundle/resources/job_runs/on_value_change/watched.txt @@ -0,0 +1 @@ +v1 diff --git a/acceptance/bundle/resources/job_runs/on_value_change_empty/databricks.yml b/acceptance/bundle/resources/job_runs/on_value_change_empty/databricks.yml new file mode 100644 index 00000000000..25c1444bfca --- /dev/null +++ b/acceptance/bundle/resources/job_runs/on_value_change_empty/databricks.yml @@ -0,0 +1,21 @@ +bundle: + name: test-bundle + +variables: + optional: {default: ""} + +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} + lifecycle: + triggers: + - on_value_change: ${var.optional} diff --git a/acceptance/bundle/resources/job_runs/on_value_change_empty/out.test.toml b/acceptance/bundle/resources/job_runs/on_value_change_empty/out.test.toml new file mode 100644 index 00000000000..a927a5fbc06 --- /dev/null +++ b/acceptance/bundle/resources/job_runs/on_value_change_empty/out.test.toml @@ -0,0 +1,2 @@ +Cloud = false +EnvMatrix.DMS = ["", "true"] diff --git a/acceptance/bundle/resources/job_runs/on_value_change_empty/output.txt b/acceptance/bundle/resources/job_runs/on_value_change_empty/output.txt new file mode 100644 index 00000000000..48701895017 --- /dev/null +++ b/acceptance/bundle/resources/job_runs/on_value_change_empty/output.txt @@ -0,0 +1,104 @@ + +=== validate keeps the empty value +>>> [CLI] bundle validate -o json +{ + "triggers": [ + { + "on_value_change": "" + } + ], + "triggers_state": { + "on_value_change": [ + "" + ] + } +} + +=== a watch on an empty value deploys and records it +>>> [CLI] bundle deploy +Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/test-bundle/default/files... +Created jobs.my_job +Output from job_runs.my_run: id=[MY_RUN_ID]: Run URL: [DATABRICKS_URL]/jobs/[MY_JOB_ID]/runs/[MY_RUN_ID]?o=[WORKSPACE_ID] +Output from job_runs.my_run: id=[MY_RUN_ID]: SUCCESS +Created job_runs.my_run +Files: 1 uploaded, 0 deleted +Resources: 2 created, 0 changed, 0 deleted, 0 unchanged + +>>> print_requests.py //jobs/run-now +{ + "method": "POST", + "path": "/api/2.2/jobs/run-now", + "body": { + "idempotency_token": "[UUID]", + "job_id": [MY_JOB_ID] + } +} + +>>> print_state.py +{ + "triggers_state": { + "on_value_change": [ + "" + ] + } +} + +=== redeploy with the value still empty plans nothing +>>> [CLI] bundle plan +Plan: 0 to add, 0 to change, 0 to delete, 2 unchanged + +=== setting the value re-fires +>>> [CLI] bundle plan +recreate job_runs.my_run + +Plan: 1 to add, 0 to change, 1 to delete, 1 unchanged + +>>> [CLI] bundle deploy +Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/test-bundle/default/files... +Output from job_runs.my_run: id=[MY_RUN_ID_2]: Run URL: [DATABRICKS_URL]/jobs/[MY_JOB_ID]/runs/[MY_RUN_ID_2]?o=[WORKSPACE_ID] +Output from job_runs.my_run: id=[MY_RUN_ID_2]: SUCCESS +Recreated job_runs.my_run +Files: 1 uploaded, 0 deleted +Resources: 1 created, 0 changed, 1 deleted, 1 unchanged + +>>> print_requests.py //jobs/run-now +{ + "method": "POST", + "path": "/api/2.2/jobs/run-now", + "body": { + "idempotency_token": "[UUID]", + "job_id": [MY_JOB_ID] + } +} + +=== clearing the value again re-fires +>>> [CLI] bundle plan +recreate job_runs.my_run + +Plan: 1 to add, 0 to change, 1 to delete, 1 unchanged + +>>> [CLI] bundle deploy +Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/test-bundle/default/files... +Output from job_runs.my_run: id=[MY_RUN_ID_3]: Run URL: [DATABRICKS_URL]/jobs/[MY_JOB_ID]/runs/[MY_RUN_ID_3]?o=[WORKSPACE_ID] +Output from job_runs.my_run: id=[MY_RUN_ID_3]: SUCCESS +Recreated job_runs.my_run +Files: 1 uploaded, 0 deleted +Resources: 1 created, 0 changed, 1 deleted, 1 unchanged + +>>> print_requests.py //jobs/run-now +{ + "method": "POST", + "path": "/api/2.2/jobs/run-now", + "body": { + "idempotency_token": "[UUID]", + "job_id": [MY_JOB_ID] + } +} + +>>> [CLI] bundle destroy --auto-approve +The following resources will be deleted: + delete resources.jobs.my_job + +All files and directories at the following location will be deleted: /Workspace/Users/[USERNAME]/.bundle/test-bundle/default + +Destroy: 1 deleted diff --git a/acceptance/bundle/resources/job_runs/on_value_change_empty/script b/acceptance/bundle/resources/job_runs/on_value_change_empty/script new file mode 100644 index 00000000000..136dc0aaa21 --- /dev/null +++ b/acceptance/bundle/resources/job_runs/on_value_change_empty/script @@ -0,0 +1,32 @@ +cleanup() { + trace $CLI bundle destroy --auto-approve + rm -f out.requests.txt +} +trap cleanup EXIT + +title "validate keeps the empty value" +trace $CLI bundle validate -o json | jq .resources.job_runs.my_run.lifecycle + +title "a watch on an empty value deploys and records it" +trace $CLI bundle deploy +read_id.py my_job > /dev/null +read_id.py my_run > /dev/null +trace print_requests.py //jobs/run-now +trace print_state.py | jq '.state["resources.job_runs.my_run"].state.lifecycle' + +title "redeploy with the value still empty plans nothing" +trace $CLI bundle plan + +title "setting the value re-fires" +update_file.py databricks.yml '{default: ""}' '{default: set}' +trace $CLI bundle plan +trace $CLI bundle deploy +read_id.py my_run > /dev/null +trace print_requests.py //jobs/run-now + +title "clearing the value again re-fires" +update_file.py databricks.yml '{default: set}' '{default: ""}' +trace $CLI bundle plan +trace $CLI bundle deploy +read_id.py my_run > /dev/null +trace print_requests.py //jobs/run-now diff --git a/acceptance/bundle/resources/job_runs/on_value_change_errors/databricks.yml b/acceptance/bundle/resources/job_runs/on_value_change_errors/databricks.yml new file mode 100644 index 00000000000..43ab45ad51e --- /dev/null +++ b/acceptance/bundle/resources/job_runs/on_value_change_errors/databricks.yml @@ -0,0 +1,24 @@ +bundle: + name: job-runs-on-value-change-errors + +variables: + value: + default: resolved + complex: + type: complex + default: {key: value} + +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} + lifecycle: + triggers: [{on_file_change: watched.txt, on_value_change: "${var.value}"}] diff --git a/acceptance/bundle/resources/job_runs/on_value_change_errors/out.test.toml b/acceptance/bundle/resources/job_runs/on_value_change_errors/out.test.toml new file mode 100644 index 00000000000..a927a5fbc06 --- /dev/null +++ b/acceptance/bundle/resources/job_runs/on_value_change_errors/out.test.toml @@ -0,0 +1,2 @@ +Cloud = false +EnvMatrix.DMS = ["", "true"] diff --git a/acceptance/bundle/resources/job_runs/on_value_change_errors/output.txt b/acceptance/bundle/resources/job_runs/on_value_change_errors/output.txt new file mode 100644 index 00000000000..eaf3d16b41c --- /dev/null +++ b/acceptance/bundle/resources/job_runs/on_value_change_errors/output.txt @@ -0,0 +1,42 @@ + +=== multiple trigger fields in one entry are rejected +>>> musterr [CLI] bundle validate +Error: lifecycle.triggers entry must set only one of on_bundle_deploy, on_file_change, or on_value_change + in databricks.yml:24:20 + +Name: job-runs-on-value-change-errors +Target: default +Workspace: + User: [USERNAME] + Path: /Workspace/Users/[USERNAME]/.bundle/job-runs-on-value-change-errors/default + +Found 1 error + +=== an undefined variable is rejected +>>> musterr [CLI] bundle validate +Error: reference does not exist: ${var.missing} + +Name: job-runs-on-value-change-errors +Target: default +Workspace: + User: [USERNAME] + Path: /Workspace/Users/[USERNAME]/.bundle/job-runs-on-value-change-errors/default + +Found 1 error + +=== a complex value is rejected +>>> musterr [CLI] bundle validate +Warning: expected string, found map + at resources.job_runs.my_run.lifecycle.triggers[0].on_value_change + in databricks.yml:24:38 + +Error: lifecycle.triggers entry must set on_bundle_deploy, on_file_change, or on_value_change + in databricks.yml:24:20 + +Name: job-runs-on-value-change-errors +Target: default +Workspace: + User: [USERNAME] + Path: /Workspace/Users/[USERNAME]/.bundle/job-runs-on-value-change-errors/default + +Found 1 error and 1 warning diff --git a/acceptance/bundle/resources/job_runs/on_value_change_errors/script b/acceptance/bundle/resources/job_runs/on_value_change_errors/script new file mode 100644 index 00000000000..f9b543f8c1e --- /dev/null +++ b/acceptance/bundle/resources/job_runs/on_value_change_errors/script @@ -0,0 +1,10 @@ +title "multiple trigger fields in one entry are rejected" +trace musterr $CLI bundle validate + +title "an undefined variable is rejected" +update_file.py databricks.yml ' triggers: [{on_file_change: watched.txt, on_value_change: "${var.value}"}]' ' triggers: [{on_value_change: "${var.missing}"}]' +trace musterr $CLI bundle validate + +title "a complex value is rejected" +update_file.py databricks.yml ' triggers: [{on_value_change: "${var.missing}"}]' ' triggers: [{on_value_change: "${var.complex}"}]' +trace musterr $CLI bundle validate diff --git a/acceptance/bundle/resources/job_runs/on_value_change_errors/test.toml b/acceptance/bundle/resources/job_runs/on_value_change_errors/test.toml new file mode 100644 index 00000000000..a030353d571 --- /dev/null +++ b/acceptance/bundle/resources/job_runs/on_value_change_errors/test.toml @@ -0,0 +1 @@ +RecordRequests = false diff --git a/acceptance/bundle/resources/job_runs/on_value_change_errors/watched.txt b/acceptance/bundle/resources/job_runs/on_value_change_errors/watched.txt new file mode 100644 index 00000000000..0ed2fb018aa --- /dev/null +++ b/acceptance/bundle/resources/job_runs/on_value_change_errors/watched.txt @@ -0,0 +1 @@ +watched diff --git a/acceptance/bundle/resources/job_runs/on_value_change_ref/databricks.yml b/acceptance/bundle/resources/job_runs/on_value_change_ref/databricks.yml new file mode 100644 index 00000000000..f821d3e075c --- /dev/null +++ b/acceptance/bundle/resources/job_runs/on_value_change_ref/databricks.yml @@ -0,0 +1,37 @@ +bundle: + name: job-runs-on-value-change-ref + +variables: + suffix: + default: suffix + +resources: + jobs: + my_job: + name: my-job + tasks: + - task_key: main + notebook_task: + notebook_path: /Workspace/test + + other: + name: other-job + tasks: + - task_key: main + notebook_task: + notebook_path: /Workspace/test + + extra: + name: extra-job + tasks: + - task_key: main + notebook_task: + notebook_path: /Workspace/test + + job_runs: + my_run: + job_id: ${resources.jobs.my_job.id} + lifecycle: + triggers: + - on_value_change: ${resources.jobs.other.id} + - on_value_change: ${resources.jobs.other.id}-${var.suffix}-${resources.jobs.extra.id} diff --git a/acceptance/bundle/resources/job_runs/on_value_change_ref/out.test.toml b/acceptance/bundle/resources/job_runs/on_value_change_ref/out.test.toml new file mode 100644 index 00000000000..f7fafbcc731 --- /dev/null +++ b/acceptance/bundle/resources/job_runs/on_value_change_ref/out.test.toml @@ -0,0 +1,3 @@ +Cloud = false +EnvMatrix.DMS = ["", "true"] +EnvMatrix.READPLAN = ["", "1"] diff --git a/acceptance/bundle/resources/job_runs/on_value_change_ref/output.txt b/acceptance/bundle/resources/job_runs/on_value_change_ref/output.txt new file mode 100644 index 00000000000..727b5c36756 --- /dev/null +++ b/acceptance/bundle/resources/job_runs/on_value_change_ref/output.txt @@ -0,0 +1,121 @@ + +=== plan depends on both jobs in the expressions +>>> jq .plan["resources.job_runs.my_run"].depends_on tmp.plan.json +[ + { + "node": "resources.jobs.extra", + "label": "${resources.jobs.extra.id}" + }, + { + "node": "resources.jobs.my_job", + "label": "${resources.jobs.my_job.id}" + }, + { + "node": "resources.jobs.other", + "label": "${resources.jobs.other.id}" + } +] + +=== first deploy triggers a run +=== bundle deploy +Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/job-runs-on-value-change-ref/default/files... +Created jobs.extra +Created jobs.my_job +Created jobs.other +Output from job_runs.my_run: id=[MY_RUN_ID]: Run URL: [DATABRICKS_URL]/jobs/[MY_JOB_ID]/runs/[MY_RUN_ID]?o=[WORKSPACE_ID] +Output from job_runs.my_run: id=[MY_RUN_ID]: SUCCESS +Created job_runs.my_run +Files: 2 uploaded, 0 deleted +Resources: 4 created, 0 changed, 0 deleted, 0 unchanged + +>>> read_id.py my_job +[MY_JOB_ID] + +>>> read_id.py other +[OTHER_ID] + +>>> read_id.py extra +[EXTRA_ID] + +>>> print_requests.py //jobs/run-now +{ + "method": "POST", + "path": "/api/2.2/jobs/run-now", + "body": { + "idempotency_token": "[UUID]", + "job_id": [MY_JOB_ID] + } +} + +=== state records resolved values for pure and mixed references +>>> print_state.py +[ + "[OTHER_ID]", + "[OTHER_ID]-suffix-[EXTRA_ID]" +] + +=== redeploy with unchanged ids plans nothing +>>> [CLI] bundle plan +Plan: 0 to add, 0 to change, 0 to delete, 4 unchanged + +>>> [CLI] bundle deploy +Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/job-runs-on-value-change-ref/default/files... +Files: 0 uploaded, 0 deleted +Resources: 0 created, 0 changed, 0 deleted, 4 unchanged + +>>> print_requests.py //jobs/run-now + +=== updating a referenced job in place does not re-fire the run +>>> [CLI] bundle plan +update jobs.other + +Plan: 0 to add, 1 to change, 0 to delete, 3 unchanged + +>>> [CLI] bundle deploy +Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/job-runs-on-value-change-ref/default/files... +Updated jobs.other +Files: 1 uploaded, 0 deleted +Resources: 0 created, 1 changed, 0 deleted, 3 unchanged + +>>> print_requests.py //jobs/run-now + +=== replacing a referenced job re-fires the run +>>> [CLI] jobs delete [OTHER_ID] + +>>> [CLI] bundle plan +recreate job_runs.my_run +create jobs.other + +Plan: 2 to add, 0 to change, 1 to delete, 2 unchanged + +>>> [CLI] bundle deploy +Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/job-runs-on-value-change-ref/default/files... +Created jobs.other +Output from job_runs.my_run: id=[MY_RUN_ID_2]: Run URL: [DATABRICKS_URL]/jobs/[MY_JOB_ID]/runs/[MY_RUN_ID_2]?o=[WORKSPACE_ID] +Output from job_runs.my_run: id=[MY_RUN_ID_2]: SUCCESS +Recreated job_runs.my_run +Files: 0 uploaded, 0 deleted +Resources: 2 created, 0 changed, 1 deleted, 2 unchanged + +>>> read_id.py other +[OTHER_ID_3] + +>>> print_requests.py //jobs/run-now +{ + "method": "POST", + "path": "/api/2.2/jobs/run-now", + "body": { + "idempotency_token": "[UUID]", + "job_id": [MY_JOB_ID] + } +} + +>>> [CLI] bundle destroy --auto-approve +The following resources will be deleted: + delete resources.jobs.extra + delete resources.jobs.my_job + delete resources.jobs.other + +All files and directories at the following location will be deleted: /Workspace/Users/[USERNAME]/.bundle/job-runs-on-value-change-ref/default + +Destroy: 3 deleted diff --git a/acceptance/bundle/resources/job_runs/on_value_change_ref/script b/acceptance/bundle/resources/job_runs/on_value_change_ref/script new file mode 100644 index 00000000000..c3aedf9ef51 --- /dev/null +++ b/acceptance/bundle/resources/job_runs/on_value_change_ref/script @@ -0,0 +1,44 @@ +cleanup() { + trace $CLI bundle destroy --auto-approve + rm -f out.requests.txt +} +trap cleanup EXIT + +title "plan depends on both jobs in the expressions" +$CLI bundle plan -o json > tmp.plan.json +trace jq '.plan["resources.job_runs.my_run"].depends_on' tmp.plan.json | contains.py \ + 'resources.jobs.other' '${resources.jobs.other.id}' \ + 'resources.jobs.extra' '${resources.jobs.extra.id}' + +title "first deploy triggers a run" +# Not traced: readplanarg makes the command line differ per READPLAN variant. +title "bundle deploy\n" +$CLI bundle deploy $(readplanarg tmp.plan.json) +trace read_id.py my_job +trace read_id.py other +trace read_id.py extra +read_id.py my_run > /dev/null +trace print_requests.py //jobs/run-now + +title "state records resolved values for pure and mixed references" +trace print_state.py | jq '.state["resources.job_runs.my_run"].state.lifecycle.triggers_state.on_value_change' + +title "redeploy with unchanged ids plans nothing" +trace $CLI bundle plan +trace $CLI bundle deploy +trace print_requests.py //jobs/run-now + +title "updating a referenced job in place does not re-fire the run" +update_file.py databricks.yml "name: other-job" "name: other-job-renamed" +trace $CLI bundle plan +trace $CLI bundle deploy +trace print_requests.py //jobs/run-now + +title "replacing a referenced job re-fires the run" +other_id=$(read_id.py other) +trace $CLI jobs delete "$other_id" +trace $CLI bundle plan +trace $CLI bundle deploy +trace read_id.py other +read_id.py my_run > /dev/null +trace print_requests.py //jobs/run-now diff --git a/acceptance/bundle/resources/job_runs/on_value_change_ref/test.toml b/acceptance/bundle/resources/job_runs/on_value_change_ref/test.toml new file mode 100644 index 00000000000..a698f7a65d1 --- /dev/null +++ b/acceptance/bundle/resources/job_runs/on_value_change_ref/test.toml @@ -0,0 +1,5 @@ +# Deploy both by re-planning and from a plan saved on disk, so resource refs +# in on_value_change survive plan serialization. +EnvMatrix.READPLAN = ["", "1"] + +Ignore = ["tmp.plan.json"] diff --git a/bundle/config/mutator/drop_empty_strings.go b/bundle/config/mutator/drop_empty_strings.go index 028351ed943..83794505dd8 100644 --- a/bundle/config/mutator/drop_empty_strings.go +++ b/bundle/config/mutator/drop_empty_strings.go @@ -12,6 +12,9 @@ import ( type dropEmptyStrings struct{} +// lifecyclePattern matches the lifecycle block of every resource. +var lifecyclePattern = dyn.NewPattern(dyn.Key("resources"), dyn.AnyKey(), dyn.AnyKey(), dyn.Key("lifecycle")) + // DropEmptyStrings removes empty-string values on omitempty resource fields, so // they are not force-sent to the backend. An empty string reaches this point // either literally (policy_id: "") or via a variable that resolved to "", so @@ -21,6 +24,10 @@ type dropEmptyStrings struct{} // force-sends explicitly-set zero values, defeating the omitempty tag. Dropping // here fixes it and makes the result visible in `bundle validate -o json`, which // serializes the dynamic value. +// +// lifecycle is left as is: it configures the bundle and is never sent to the +// backend, and an empty string is meaningful there (job_runs +// lifecycle.triggers.on_value_change can watch a value that is ""). func DropEmptyStrings() bundle.Mutator { return &dropEmptyStrings{} } @@ -32,7 +39,16 @@ func (m *dropEmptyStrings) Name() string { func (m *dropEmptyStrings) Apply(ctx context.Context, b *bundle.Bundle) diag.Diagnostics { var diags diag.Diagnostics err := b.Config.Mutate(func(root dyn.Value) (dyn.Value, error) { - root, err := dyn.Map(root, "resources", func(_ dyn.Path, resources dyn.Value) (dyn.Value, error) { + lifecycles := make(map[string]dyn.Value) + _, err := dyn.MapByPattern(root, lifecyclePattern, func(p dyn.Path, lifecycle dyn.Value) (dyn.Value, error) { + lifecycles[p.String()] = lifecycle + return lifecycle, nil + }) + if err != nil { + return root, err + } + + root, err = dyn.Map(root, "resources", func(_ dyn.Path, resources dyn.Value) (dyn.Value, error) { // Normalize against the resources type so omitempty is known per field. // Only DropEmptyStrings is set: existing values are kept as-is otherwise. out, normDiags := convert.Normalize(config.Resources{}, resources, convert.DropEmptyStrings) @@ -43,6 +59,13 @@ func (m *dropEmptyStrings) Apply(ctx context.Context, b *bundle.Bundle) diag.Dia return root, err } + root, err = dyn.MapByPattern(root, lifecyclePattern, func(p dyn.Path, _ dyn.Value) (dyn.Value, error) { + return lifecycles[p.String()], nil + }) + if err != nil { + return root, err + } + // Keep an empty apps description instead of dropping it: an app update without // the field leaves the remote description unchanged, so removing description // from config would never converge. Force-sending "" clears it. diff --git a/bundle/config/mutator/resolve_job_run_file_triggers.go b/bundle/config/mutator/resolve_job_run_file_triggers.go index 35cffd0dcbe..d7db5499bc1 100644 --- a/bundle/config/mutator/resolve_job_run_file_triggers.go +++ b/bundle/config/mutator/resolve_job_run_file_triggers.go @@ -72,7 +72,10 @@ func (*resolveJobRunFileTriggers) Apply(ctx context.Context, b *bundle.Bundle) d out[pattern] = fingerprint } } - jr.Lifecycle.TriggersState = &resources.JobRunTriggersState{OnFileChange: out} + if jr.Lifecycle.TriggersState == nil { + jr.Lifecycle.TriggersState = &resources.JobRunTriggersState{} + } + jr.Lifecycle.TriggersState.OnFileChange = out } return diags } diff --git a/bundle/config/mutator/resolve_job_run_value_triggers.go b/bundle/config/mutator/resolve_job_run_value_triggers.go new file mode 100644 index 00000000000..b3539aacf76 --- /dev/null +++ b/bundle/config/mutator/resolve_job_run_value_triggers.go @@ -0,0 +1,46 @@ +package mutator + +import ( + "context" + + "github.com/databricks/cli/bundle" + "github.com/databricks/cli/bundle/config/resources" + "github.com/databricks/cli/libs/diag" +) + +type resolveJobRunValueTriggers struct{} + +// ResolveJobRunValueTriggers copies each job_run's interpolated on_value_change values, +// in trigger order, into lifecycle.triggers_state, where the planner compares them by +// position with the values recorded when the run last fired. +func ResolveJobRunValueTriggers() bundle.Mutator { + return &resolveJobRunValueTriggers{} +} + +func (*resolveJobRunValueTriggers) Name() string { + return "ResolveJobRunValueTriggers" +} + +func (*resolveJobRunValueTriggers) Apply(_ context.Context, b *bundle.Bundle) diag.Diagnostics { + for _, jr := range b.Config.Resources.JobRuns { + if jr == nil || jr.Lifecycle == nil { + continue + } + + var values []string + for _, trigger := range jr.Lifecycle.Triggers { + if trigger.OnValueChange != nil { + values = append(values, *trigger.OnValueChange) + } + } + if values == nil { + continue + } + + if jr.Lifecycle.TriggersState == nil { + jr.Lifecycle.TriggersState = &resources.JobRunTriggersState{} + } + jr.Lifecycle.TriggersState.OnValueChange = values + } + return nil +} diff --git a/bundle/config/mutator/resolve_job_run_value_triggers_test.go b/bundle/config/mutator/resolve_job_run_value_triggers_test.go new file mode 100644 index 00000000000..8968fb543fb --- /dev/null +++ b/bundle/config/mutator/resolve_job_run_value_triggers_test.go @@ -0,0 +1,27 @@ +package mutator_test + +import ( + "testing" + + "github.com/databricks/cli/bundle" + "github.com/databricks/cli/bundle/config" + "github.com/databricks/cli/bundle/config/mutator" + "github.com/databricks/cli/bundle/config/resources" + "github.com/stretchr/testify/assert" +) + +func TestResolveJobRunValueTriggers(t *testing.T) { + enabled := true + first := "first" + empty := "" + b := &bundle.Bundle{Config: config.Root{ + Resources: config.Resources{JobRuns: map[string]*resources.JobRun{"run": { + Lifecycle: &resources.JobRunLifecycle{ + Triggers: []resources.JobRunTrigger{{OnValueChange: &first}, {OnBundleDeploy: &enabled}, {OnValueChange: &empty}}, + }, + }}}, + }} + + assert.Empty(t, bundle.Apply(t.Context(), b, mutator.ResolveJobRunValueTriggers())) + assert.Equal(t, []string{"first", ""}, b.Config.Resources.JobRuns["run"].Lifecycle.TriggersState.OnValueChange) +} diff --git a/bundle/config/mutator/validate_job_run_triggers.go b/bundle/config/mutator/validate_job_run_triggers.go index d33a7263bde..cb12587a306 100644 --- a/bundle/config/mutator/validate_job_run_triggers.go +++ b/bundle/config/mutator/validate_job_run_triggers.go @@ -28,18 +28,19 @@ func (*validateJobRunTriggers) Apply(_ context.Context, b *bundle.Bundle) diag.D } for i, t := range jr.Lifecycle.Triggers { path := fmt.Sprintf("resources.job_runs.%s.lifecycle.triggers[%d]", name, i) - if t.OnBundleDeploy == nil && t.OnFileChange == nil { + armed := t.ArmedCount() + if armed == 0 { diags = diags.Append(diag.Diagnostic{ Severity: diag.Error, - Summary: "lifecycle.triggers entry must set on_bundle_deploy or on_file_change", + Summary: "lifecycle.triggers entry must set on_bundle_deploy, on_file_change, or on_value_change", Locations: b.Config.GetLocations(path), }) continue } - if t.OnBundleDeploy != nil && t.OnFileChange != nil { + if armed > 1 { diags = diags.Append(diag.Diagnostic{ Severity: diag.Error, - Summary: "lifecycle.triggers entry must set only one of on_bundle_deploy or on_file_change", + Summary: "lifecycle.triggers entry must set only one of on_bundle_deploy, on_file_change, or on_value_change", Locations: b.Config.GetLocations(path), }) continue diff --git a/bundle/config/mutator/validate_job_run_triggers_test.go b/bundle/config/mutator/validate_job_run_triggers_test.go index 3ec1e90b844..1ab7e9dda5f 100644 --- a/bundle/config/mutator/validate_job_run_triggers_test.go +++ b/bundle/config/mutator/validate_job_run_triggers_test.go @@ -48,14 +48,14 @@ func TestValidateJobRunTriggers(t *testing.T) { triggers: []resources.JobRunTrigger{ {}, }, - summary: "lifecycle.triggers entry must set on_bundle_deploy or on_file_change", + summary: "lifecycle.triggers entry must set on_bundle_deploy, on_file_change, or on_value_change", }, { name: "both keys on one entry", triggers: []resources.JobRunTrigger{ {OnBundleDeploy: &trueVal, OnFileChange: &fileChange}, }, - summary: "lifecycle.triggers entry must set only one of on_bundle_deploy or on_file_change", + summary: "lifecycle.triggers entry must set only one of on_bundle_deploy, on_file_change, or on_value_change", }, { name: "on_bundle_deploy false", diff --git a/bundle/config/resources/job_run.go b/bundle/config/resources/job_run.go index dff2e538cb0..147155994a8 100644 --- a/bundle/config/resources/job_run.go +++ b/bundle/config/resources/job_run.go @@ -44,20 +44,37 @@ type JobRunLifecycle struct { // JobRunTrigger is one lifecycle.triggers entry. type JobRunTrigger struct { OnBundleDeploy *bool `json:"on_bundle_deploy,omitempty"` - OnFileChange *string `json:"on_file_change,omitempty"` // path or glob relative to the defining YAML file; must resolve under the sync root + OnFileChange *string `json:"on_file_change,omitempty"` // path or glob relative to the defining YAML file; must resolve under the sync root + OnValueChange *string `json:"on_value_change,omitempty"` // expression whose resolved value re-fires the run when changed +} + +// ArmedCount returns how many trigger fields this entry sets. +func (t JobRunTrigger) ArmedCount() int { + n := 0 + if t.OnBundleDeploy != nil { + n++ + } + if t.OnFileChange != nil { + n++ + } + if t.OnValueChange != nil { + n++ + } + return n } // JobRunTriggersState is the resolved fingerprint of lifecycle.triggers. type JobRunTriggersState struct { OnBundleDeploy string `json:"on_bundle_deploy,omitempty"` OnFileChange map[string]string `json:"on_file_change,omitempty"` + OnValueChange []string `json:"on_value_change,omitempty"` // in lifecycle.triggers order } // IsEmpty reports whether no trigger is armed. An empty state is left off the // job run entirely, so this has to cover every field above: a new fingerprint // added without extending it would be dropped instead of persisted. func (s JobRunTriggersState) IsEmpty() bool { - return s.OnBundleDeploy == "" && len(s.OnFileChange) == 0 + return s.OnBundleDeploy == "" && len(s.OnFileChange) == 0 && len(s.OnValueChange) == 0 } // HasOnBundleDeploy reports whether any trigger re-fires on every deploy. diff --git a/bundle/direct/dresources/job_run.go b/bundle/direct/dresources/job_run.go index 90a628b0486..87fd1a68adf 100644 --- a/bundle/direct/dresources/job_run.go +++ b/bundle/direct/dresources/job_run.go @@ -368,7 +368,9 @@ func reportRunLine(ctx context.Context, runID int64, msg string) { // result_state either, so the lifecycle state is what tells the two apart. // Removing a trigger drops the change and leaves the last fingerprint in state, // so re-adding the trigger only re-fires when the watched files changed -// meanwhile. All other trigger changes recreate the run. +// meanwhile. on_value_change values are compared by position, so removing the +// last watches drops the change, and removing or reordering any other watch +// re-fires. All other trigger changes recreate the run. func (*ResourceJobRun) OverrideChangeDesc(_ context.Context, path *structpath.PathNode, change *ChangeDesc, remote *JobRunRemote) error { switch path.String() { case "lifecycle": @@ -388,6 +390,13 @@ func (*ResourceJobRun) OverrideChangeDesc(_ context.Context, path *structpath.Pa if isEmptyFileTriggerMap(change.New) || change.Old != nil { change.Reason = deployplan.ReasonDrop } + case "lifecycle.triggers_state.on_value_change": + // Lists of equal length are diffed per position, and any change there + // re-fires. A shorter list that keeps the leading values means watches + // were removed from the end. + if isLeadingValues(change.Old, change.New) { + change.Reason = deployplan.ReasonDrop + } case "result_state": // The planner passes no remote state when the run could not be read. if remote == nil || runIsTerminal(remote.State.LifeCycleState) { @@ -421,6 +430,14 @@ func isEmptyFileTriggerMap(v any) bool { return ok && len(m) == 0 } +// isLeadingValues reports whether newV holds the leading values of oldV and fewer +// of them (none, when newV is nil). +func isLeadingValues(oldV, newV any) bool { + oldValues, _ := oldV.([]string) + newValues, _ := newV.([]string) + return len(newValues) < len(oldValues) && slices.Equal(oldValues[:len(newValues)], newValues) +} + 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 index 5dc964c360f..b3d97f5e2d6 100644 --- a/bundle/direct/dresources/job_run_test.go +++ b/bundle/direct/dresources/job_run_test.go @@ -271,7 +271,8 @@ func TestJobRunStateOmitsEmptyLifecycle(t *testing.T) { func TestJobRunPrepareStateCopiesResolvedTriggers(t *testing.T) { enabled := true triggers := &resources.JobRunTriggersState{ - OnFileChange: map[string]string{"*.txt": "hash"}, + OnFileChange: map[string]string{"*.txt": "hash"}, + OnValueChange: []string{"resolved"}, } input := &resources.JobRun{ Lifecycle: &resources.JobRunLifecycle{ @@ -286,6 +287,7 @@ func TestJobRunPrepareStateCopiesResolvedTriggers(t *testing.T) { require.NotNil(t, state.Lifecycle.TriggersState) assert.NotSame(t, triggers, state.Lifecycle.TriggersState) assert.Equal(t, triggers.OnFileChange, state.Lifecycle.TriggersState.OnFileChange) + assert.Equal(t, triggers.OnValueChange, state.Lifecycle.TriggersState.OnValueChange) assert.NotEmpty(t, state.Lifecycle.TriggersState.OnBundleDeploy) assert.Empty(t, triggers.OnBundleDeploy) } diff --git a/bundle/internal/schema/annotations.yml b/bundle/internal/schema/annotations.yml index 08b65053e3b..e982a623a0a 100644 --- a/bundle/internal/schema/annotations.yml +++ b/bundle/internal/schema/annotations.yml @@ -1133,6 +1133,9 @@ resources: "on_file_change": "description": |- Path or glob relative to the defining YAML file. It must resolve under the sync root. Re-fire the run when a matched file's content hash changes, or when the set of matches appears or disappears. Only files the bundle syncs are hashed, so .gitignore and sync.exclude apply. Use * to match a single directory level; ** is not supported. + "on_value_change": + "description": |- + Interpolated expression. Re-fire the run when its resolved value changes. Values are compared by position among the on_value_change entries: adding one re-fires the run, removing the last ones does not, and removing or reordering any other one does. Resource references in the expression become deployment dependencies. "python_named_params": "description": |- PLACEHOLDER diff --git a/bundle/phases/initialize.go b/bundle/phases/initialize.go index 8c5a68d3130..823408061db 100644 --- a/bundle/phases/initialize.go +++ b/bundle/phases/initialize.go @@ -166,6 +166,11 @@ func Initialize(ctx context.Context, b *bundle.Bundle) { // Must run after PythonMutator to see development values set by Python mutators. validate.PipelineDevelopmentDeprecated(), + // Store job_runs on_value_change values in triggers_state. Must run after + // PythonMutator: triggers_state is not declared by the PyDABs JobRunLifecycle + // model, so exposing it to Python would fail resource loading. + mutator.ResolveJobRunValueTriggers(), + // Compute resources.volumes.*.volume_path and resolve references to it. Must run after // PythonMutator: volume_path is computed and read-only, not declared by the PyDABs Volume // model, so exposing it to Python would fail resource loading (like "deployment" below). diff --git a/bundle/schema/jsonschema.json b/bundle/schema/jsonschema.json index a060ec2e8a1..972247fea8a 100644 --- a/bundle/schema/jsonschema.json +++ b/bundle/schema/jsonschema.json @@ -1708,6 +1708,10 @@ "on_file_change": { "description": "Path or glob relative to the defining YAML file. It must resolve under the sync root. Re-fire the run when a matched file's content hash changes, or when the set of matches appears or disappears. Only files the bundle syncs are hashed, so .gitignore and sync.exclude apply. Use * to match a single directory level; ** is not supported.", "$ref": "#/$defs/string" + }, + "on_value_change": { + "description": "Interpolated expression. Re-fire the run when its resolved value changes. Values are compared by position among the on_value_change entries: adding one re-fires the run, removing the last ones does not, and removing or reordering any other one does. Resource references in the expression become deployment dependencies.", + "$ref": "#/$defs/string" } }, "additionalProperties": false diff --git a/python/databricks/bundles/job_runs/_models/job_run_trigger.py b/python/databricks/bundles/job_runs/_models/job_run_trigger.py index 75bdf970a4e..5180dd9947d 100644 --- a/python/databricks/bundles/job_runs/_models/job_run_trigger.py +++ b/python/databricks/bundles/job_runs/_models/job_run_trigger.py @@ -25,6 +25,11 @@ class JobRunTrigger: Path or glob relative to the defining YAML file. It must resolve under the sync root. Re-fire the run when a matched file's content hash changes, or when the set of matches appears or disappears. Only files the bundle syncs are hashed, so .gitignore and sync.exclude apply. Use * to match a single directory level; ** is not supported. """ + on_value_change: VariableOrOptional[str] = None + """ + Interpolated expression. Re-fire the run when its resolved value changes. Values are compared by position among the on_value_change entries: adding one re-fires the run, removing the last ones does not, and removing or reordering any other one does. Resource references in the expression become deployment dependencies. + """ + @classmethod def from_dict(cls, value: "JobRunTriggerDict") -> "Self": return _transform(cls, value) @@ -46,5 +51,10 @@ class JobRunTriggerDict(TypedDict, total=False): Path or glob relative to the defining YAML file. It must resolve under the sync root. Re-fire the run when a matched file's content hash changes, or when the set of matches appears or disappears. Only files the bundle syncs are hashed, so .gitignore and sync.exclude apply. Use * to match a single directory level; ** is not supported. """ + on_value_change: VariableOrOptional[str] + """ + Interpolated expression. Re-fire the run when its resolved value changes. Values are compared by position among the on_value_change entries: adding one re-fires the run, removing the last ones does not, and removing or reordering any other one does. Resource references in the expression become deployment dependencies. + """ + JobRunTriggerParam = JobRunTriggerDict | JobRunTrigger