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..53ec0227df0 --- /dev/null +++ b/.nextchanges/bundles/job-runs-on-value-change.md @@ -0,0 +1 @@ +direct: `resources.job_runs` can set `lifecycle.triggers.on_value_change` to an interpolated expression (for example `${resources.jobs.foo.id}`) to re-fire the run when that value changes. Long resolved values are recorded as a SHA-256 hash. Removing the trigger does not recreate the existing run. diff --git a/acceptance/bundle/refschema/out.fields.txt b/acceptance/bundle/refschema/out.fields.txt index 63b82c1d595..dd5033ab682 100644 --- a/acceptance/bundle/refschema/out.fields.txt +++ b/acceptance/bundle/refschema/out.fields.txt @@ -914,9 +914,12 @@ resources.job_runs.*.lifecycle.triggers dresources.JobRunTriggersState STATE resources.job_runs.*.lifecycle.triggers.on_bundle_deploy string STATE resources.job_runs.*.lifecycle.triggers.on_file_change map[string]string STATE resources.job_runs.*.lifecycle.triggers.on_file_change.* string STATE +resources.job_runs.*.lifecycle.triggers.on_value_change []string STATE +resources.job_runs.*.lifecycle.triggers.on_value_change[*] string STATE 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.*.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..cbb68d9354a --- /dev/null +++ b/acceptance/bundle/resources/job_runs/on_value_change/databricks.yml @@ -0,0 +1,28 @@ +bundle: + name: job-runs-on-value-change + +variables: + stamp: + default: v1 + other: + default: a1 + long: + default: aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa + +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.stamp} + - on_value_change: ${var.other} + - on_value_change: ${var.long} 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..0938e678987 --- /dev/null +++ b/acceptance/bundle/resources/job_runs/on_value_change/out.test.toml @@ -0,0 +1,2 @@ +Cloud = false +EnvMatrix.DATABRICKS_BUNDLE_ENGINE = ["direct"] 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..fba1f6081f3 --- /dev/null +++ b/acceptance/bundle/resources/job_runs/on_value_change/output.txt @@ -0,0 +1,187 @@ + +=== first deploy triggers a run +>>> [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]: Run URL: [DATABRICKS_URL]/jobs/[MY_JOB_ID]/runs/[MY_RUN_ID]?o=[NUMID] +Output from job_runs.my_run: id=[MY_RUN_ID]: SUCCESS +Created job_runs.my_run +Created jobs.my_job +Files: 4 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] + } +} + +=== long values are stored as hashes +>>> print_state.py +[ + "v1", + "a1", + "sha256:d66304b6180365e47c858f6c84d3da065caf4b3350c9f45277a1af82e3dbb055" +] + +=== 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: 2 uploaded, 0 deleted +Resources: 0 created, 0 changed, 0 deleted, 2 unchanged + +>>> print_requests.py //jobs/run-now + +=== changing one value re-fires +>>> [CLI] bundle plan +recreate job_runs.my_run + +Plan: 1 to add, 0 to change, 1 to delete, 1 unchanged + +>>> changes +{ + "lifecycle.triggers.on_value_change": { + "action": "recreate", + "reason": "immutable", + "old": [ + "v1", + "a1", + "sha256:d66304b6180365e47c858f6c84d3da065caf4b3350c9f45277a1af82e3dbb055" + ], + "new": [ + "v2", + "a1", + "sha256:d66304b6180365e47c858f6c84d3da065caf4b3350c9f45277a1af82e3dbb055" + ] + }, + "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=[NUMID] +Output from job_runs.my_run: id=[MY_RUN_ID_2]: SUCCESS +Recreated job_runs.my_run +Files: 3 uploaded, 0 deleted +Resources: 1 created, 0 changed, 1 deleted, 1 unchanged + +>>> print_requests.py --keep //jobs/runs/delete +{ + "method": "POST", + "path": "/api/2.2/jobs/runs/delete", + "body": { + "run_id": [MY_RUN_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] + } +} + +=== converging two values re-fires +>>> [CLI] bundle plan +recreate job_runs.my_run + +Plan: 1 to add, 0 to change, 1 to delete, 1 unchanged + +>>> changes +{ + "lifecycle.triggers.on_value_change": { + "action": "recreate", + "reason": "immutable", + "old": [ + "v2", + "a1", + "sha256:d66304b6180365e47c858f6c84d3da065caf4b3350c9f45277a1af82e3dbb055" + ], + "new": [ + "v2", + "v2", + "sha256:d66304b6180365e47c858f6c84d3da065caf4b3350c9f45277a1af82e3dbb055" + ] + }, + "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=[NUMID] +Output from job_runs.my_run: id=[MY_RUN_ID_3]: SUCCESS +Recreated job_runs.my_run +Files: 3 uploaded, 0 deleted +Resources: 1 created, 0 changed, 1 deleted, 1 unchanged + +>>> print_requests.py --keep //jobs/runs/delete +{ + "method": "POST", + "path": "/api/2.2/jobs/runs/delete", + "body": { + "run_id": [MY_RUN_ID_2] + } +} + +>>> 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 one on_value_change is unchanged and 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: 3 uploaded, 0 deleted +Resources: 0 created, 0 changed, 0 deleted, 2 unchanged + +>>> print_requests.py //jobs/run-now + +=== removing the last on_value_change is unchanged and 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: 3 uploaded, 0 deleted +Resources: 0 created, 0 changed, 0 deleted, 2 unchanged + +>>> print_requests.py //jobs/run-now + +>>> [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-on-value-change/default + +Destroy: 2 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..f60be7c9f08 --- /dev/null +++ b/acceptance/bundle/resources/job_runs/on_value_change/script @@ -0,0 +1,58 @@ +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 "long values are stored as hashes" +trace print_state.py | jq '.state["resources.job_runs.my_run"].state.lifecycle.triggers.on_value_change' + +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 $CLI bundle plan +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 "converging two values re-fires" +update_file.py databricks.yml "default: a1" "default: v2" +trace $CLI bundle plan +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 one on_value_change is unchanged and does not run" +update_file.py databricks.yml ' - on_value_change: ${var.other} +' '' +trace $CLI bundle plan +trace $CLI bundle deploy +trace print_requests.py //jobs/run-now + +title "removing the last on_value_change is unchanged and does not run" +update_file.py databricks.yml ' lifecycle: + triggers: + - on_value_change: ${var.stamp} + - on_value_change: ${var.long} +' '' +trace $CLI bundle plan +trace $CLI bundle deploy +trace print_requests.py //jobs/run-now 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..84de0fdaf82 --- /dev/null +++ b/acceptance/bundle/resources/job_runs/on_value_change_ref/databricks.yml @@ -0,0 +1,32 @@ +bundle: + name: job-runs-on-value-change-ref + +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}-${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..57b0f616850 --- /dev/null +++ b/acceptance/bundle/resources/job_runs/on_value_change_ref/out.test.toml @@ -0,0 +1,3 @@ +Cloud = false +EnvMatrix.DATABRICKS_BUNDLE_ENGINE = ["direct"] +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..70349876476 --- /dev/null +++ b/acceptance/bundle/resources/job_runs/on_value_change_ref/output.txt @@ -0,0 +1,107 @@ + +=== plan depends on both jobs in the expression +>>> 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... +Output from job_runs.my_run: id=[MY_RUN_ID]: Run URL: [DATABRICKS_URL]/jobs/[MY_JOB_ID]/runs/[MY_RUN_ID]?o=[NUMID] +Output from job_runs.my_run: id=[MY_RUN_ID]: SUCCESS +Created job_runs.my_run +Created jobs.extra +Created jobs.my_job +Created jobs.other +Files: 6 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 the concatenated resolved ids +>>> print_state.py +[ + "[OTHER_ID]-[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: 2 uploaded, 0 deleted +Resources: 0 created, 0 changed, 0 deleted, 4 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... +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=[NUMID] +Output from job_runs.my_run: id=[MY_RUN_ID_2]: SUCCESS +Recreated job_runs.my_run +Created jobs.other +Files: 2 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.job_runs.my_run + 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: 4 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..707e8d8e839 --- /dev/null +++ b/acceptance/bundle/resources/job_runs/on_value_change_ref/script @@ -0,0 +1,38 @@ +cleanup() { + trace $CLI bundle destroy --auto-approve + rm -f out.requests.txt +} +trap cleanup EXIT + +title "plan depends on both jobs in the expression" +$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 the concatenated resolved ids" +trace print_state.py | jq '.state["resources.job_runs.my_run"].state.lifecycle.triggers.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 "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/validate_job_run_triggers.go b/bundle/config/mutator/validate_job_run_triggers.go index fa31032c689..2c397583127 100644 --- a/bundle/config/mutator/validate_job_run_triggers.go +++ b/bundle/config/mutator/validate_job_run_triggers.go @@ -27,7 +27,7 @@ func (*validateJobRunTriggers) Apply(_ context.Context, b *bundle.Bundle) diag.D continue } if jr.Lifecycle.PreventDestroy { - if summary := preventDestroyError(jr.HasOnBundleDeploy(), jr.HasOnFileChange()); summary != "" { + if summary := preventDestroyError(jr.ArmedTriggerNames()); summary != "" { diags = diags.Append(diag.Diagnostic{ Severity: diag.Error, Summary: summary, @@ -37,18 +37,20 @@ 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 { + switch t.ArmedCount() { + case 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 { + case 1: + // Exactly one key: valid. + default: 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 @@ -67,20 +69,28 @@ func (*validateJobRunTriggers) Apply(_ context.Context, b *bundle.Bundle) diag.D Locations: b.Config.GetLocations(path + ".on_file_change"), }) } + if t.OnValueChange != nil && strings.TrimSpace(*t.OnValueChange) == "" { + diags = diags.Append(diag.Diagnostic{ + Severity: diag.Error, + Summary: "lifecycle.triggers.on_value_change must be non-empty when set", + Locations: b.Config.GetLocations(path + ".on_value_change"), + }) + } } } return diags } -func preventDestroyError(onBundleDeploy, onFileChange bool) string { - switch { - case onBundleDeploy && onFileChange: - return "lifecycle.triggers.on_bundle_deploy and on_file_change are incompatible with lifecycle.prevent_destroy" - case onBundleDeploy: - return "lifecycle.triggers.on_bundle_deploy is incompatible with lifecycle.prevent_destroy" - case onFileChange: - return "lifecycle.triggers.on_file_change is incompatible with lifecycle.prevent_destroy" - default: +// preventDestroyError names the armed triggers that conflict with prevent_destroy, +// or returns an empty string when none are armed. +func preventDestroyError(names []string) string { + switch len(names) { + case 0: return "" + case 1: + return fmt.Sprintf("lifecycle.triggers.%s is incompatible with lifecycle.prevent_destroy", names[0]) + default: + last := len(names) - 1 + return fmt.Sprintf("lifecycle.triggers.%s and %s are incompatible with lifecycle.prevent_destroy", strings.Join(names[:last], ", "), names[last]) } } diff --git a/bundle/config/mutator/validate_job_run_triggers_test.go b/bundle/config/mutator/validate_job_run_triggers_test.go index 07a7276a41c..582dba68c1e 100644 --- a/bundle/config/mutator/validate_job_run_triggers_test.go +++ b/bundle/config/mutator/validate_job_run_triggers_test.go @@ -18,6 +18,10 @@ func TestValidateJobRunTriggers(t *testing.T) { emptyFile := "" whitespaceFile := " \t" + valueChange := "${resources.jobs.foo.id}" + emptyValue := "" + whitespaceValue := " \t" + tests := []struct { name string triggers []resources.JobRunTrigger @@ -41,6 +45,7 @@ func TestValidateJobRunTriggers(t *testing.T) { triggers: []resources.JobRunTrigger{ {OnFileChange: &fileChange}, {OnBundleDeploy: &trueVal}, + {OnValueChange: &valueChange}, }, }, { @@ -48,14 +53,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", @@ -103,10 +108,55 @@ func TestValidateJobRunTriggers(t *testing.T) { preventDestroy: true, summary: "lifecycle.triggers.on_bundle_deploy and on_file_change are incompatible with lifecycle.prevent_destroy", }, + { + name: "all triggers with prevent_destroy", + triggers: []resources.JobRunTrigger{ + {OnFileChange: &fileChange}, + {OnBundleDeploy: &trueVal}, + {OnValueChange: &valueChange}, + }, + preventDestroy: true, + summary: "lifecycle.triggers.on_bundle_deploy, on_file_change and on_value_change are incompatible with lifecycle.prevent_destroy", + }, { name: "prevent_destroy alone", preventDestroy: true, }, + { + name: "on_value_change set", + triggers: []resources.JobRunTrigger{ + {OnValueChange: &valueChange}, + }, + }, + { + name: "on_value_change empty", + triggers: []resources.JobRunTrigger{ + {OnValueChange: &emptyValue}, + }, + summary: "lifecycle.triggers.on_value_change must be non-empty when set", + }, + { + name: "on_value_change whitespace", + triggers: []resources.JobRunTrigger{ + {OnValueChange: &whitespaceValue}, + }, + summary: "lifecycle.triggers.on_value_change must be non-empty when set", + }, + { + name: "on_value_change with prevent_destroy", + triggers: []resources.JobRunTrigger{ + {OnValueChange: &valueChange}, + }, + preventDestroy: true, + summary: "lifecycle.triggers.on_value_change is incompatible with lifecycle.prevent_destroy", + }, + { + name: "on_value_change and on_file_change on one entry", + triggers: []resources.JobRunTrigger{ + {OnFileChange: &fileChange, OnValueChange: &valueChange}, + }, + summary: "lifecycle.triggers entry must set only one of on_bundle_deploy, on_file_change, or on_value_change", + }, } for _, tt := range tests { diff --git a/bundle/config/resources/job_run.go b/bundle/config/resources/job_run.go index 512077e2738..c36347146c6 100644 --- a/bundle/config/resources/job_run.go +++ b/bundle/config/resources/job_run.go @@ -59,6 +59,31 @@ func (r *JobRun) HasOnFileChange() bool { return false } +// ArmedTriggerNames returns the names of the trigger fields any entry arms, in +// schema order so that diagnostics naming them are stable. +func (r *JobRun) ArmedTriggerNames() []string { + if r.Lifecycle == nil { + return nil + } + var onBundleDeploy, onFileChange, onValueChange bool + for _, t := range r.Lifecycle.Triggers { + onBundleDeploy = onBundleDeploy || (t.OnBundleDeploy != nil && *t.OnBundleDeploy) + onFileChange = onFileChange || t.OnFileChange != nil + onValueChange = onValueChange || t.OnValueChange != nil + } + var names []string + if onBundleDeploy { + names = append(names, "on_bundle_deploy") + } + if onFileChange { + names = append(names, "on_file_change") + } + if onValueChange { + names = append(names, "on_value_change") + } + return names +} + func (r *JobRun) UnmarshalJSON(b []byte) error { return marshal.Unmarshal(b, r) } diff --git a/bundle/config/resources/lifecycle.go b/bundle/config/resources/lifecycle.go index 1f5d576d835..af0b98ab754 100644 --- a/bundle/config/resources/lifecycle.go +++ b/bundle/config/resources/lifecycle.go @@ -37,5 +37,21 @@ 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"` // interpolated expr; re-fire when the resolved value changes +} + +// ArmedCount returns the number of trigger fields set on this entry. +func (t JobRunTrigger) ArmedCount() int { + n := 0 + if t.OnBundleDeploy != nil { + n++ + } + if t.OnFileChange != nil { + n++ + } + if t.OnValueChange != nil { + n++ + } + return n } diff --git a/bundle/direct/bundle_plan.go b/bundle/direct/bundle_plan.go index 5b8829e3f59..2206ac180a0 100644 --- a/bundle/direct/bundle_plan.go +++ b/bundle/direct/bundle_plan.go @@ -1153,7 +1153,10 @@ func extractReferences(root dyn.Value, node string, stateType reflect.Type) (map // bundle:"readonly" field like volumes' volume_path — is dropped before deploy, so a // reference it carries cannot resolve into state and is not a dependency here. Such // references are still resolved earlier during initialize. - if structaccess.ValidatePath(stateType, fieldPath) == nil { + // Strict: config paths are native DABs paths, so tolerating Terraform's + // [0]-on-struct would let a config list path resolve onto a state field of + // a different shape (e.g. job_runs' lifecycle.triggers). + if structaccess.ValidatePathStrict(stateType, fieldPath) == nil { // Store the original string that contains references, not individual references. refs[fieldPath.String()] = ref.Str } diff --git a/bundle/direct/bundle_plan_test.go b/bundle/direct/bundle_plan_test.go index bf875680c2b..623a29376da 100644 --- a/bundle/direct/bundle_plan_test.go +++ b/bundle/direct/bundle_plan_test.go @@ -2,6 +2,7 @@ package direct import ( "bytes" + "reflect" "testing" "github.com/databricks/cli/bundle/config/resources" @@ -77,6 +78,33 @@ resources: }, refs) } +func TestExtractReferences_DoesNotTreatConfigSliceAsStateStruct(t *testing.T) { + type triggersState struct { + OnValueChange []string `json:"on_value_change,omitempty"` + } + type lifecycleState struct { + Triggers triggersState `json:"triggers"` + } + type state struct { + Lifecycle lifecycleState `json:"lifecycle"` + } + + const yml = ` +resources: + job_runs: + run: + lifecycle: + triggers: + - on_value_change: "${resources.jobs.watched.id}" +` + root, err := yamlloader.LoadYAML("test", bytes.NewBufferString(yml)) + require.NoError(t, err) + + refs, err := extractReferences(root, "resources.job_runs.run", reflect.TypeFor[*state]()) + require.NoError(t, err) + assert.Empty(t, refs) +} + func TestShouldSkipBackendDefault_ManagedPropertiesOnly(t *testing.T) { // Rules mirror the schemas backend_defaults in resources.yml, but the test is // deliberately self-contained so that edits to resources.yml don't break it. diff --git a/bundle/direct/dresources/job_run.go b/bundle/direct/dresources/job_run.go index 078ae1b1aa6..42e20f8967b 100644 --- a/bundle/direct/dresources/job_run.go +++ b/bundle/direct/dresources/job_run.go @@ -3,6 +3,8 @@ package dresources import ( "cmp" "context" + "crypto/sha256" + "encoding/hex" "errors" "fmt" "slices" @@ -13,8 +15,11 @@ import ( "github.com/databricks/cli/bundle/config/resources" "github.com/databricks/cli/bundle/deployplan" "github.com/databricks/cli/libs/cmdio" + "github.com/databricks/cli/libs/dyn" + "github.com/databricks/cli/libs/dyn/dynvar" "github.com/databricks/cli/libs/log" "github.com/databricks/cli/libs/structs/structpath" + "github.com/databricks/cli/libs/structs/structvar" "github.com/databricks/cli/libs/workspaceurls" "github.com/databricks/databricks-sdk-go" "github.com/databricks/databricks-sdk-go/marshal" @@ -26,12 +31,20 @@ import ( // jobRunTimeout matches the timeout `bundle run` allows a run (bundle/run/job.go). const jobRunTimeout = 24 * time.Hour +const ( + jobRunValueHashPrefix = "sha256:" + jobRunValueHashLength = len(jobRunValueHashPrefix) + sha256.Size*2 +) + // JobRunTriggersState is the persisted fingerprint of lifecycle.triggers. type JobRunTriggersState struct { // Fresh UUID each plan while armed so Old!=New forces recreate. OnBundleDeploy string `json:"on_bundle_deploy,omitempty"` // Content hashes from ResolveJobRunFileTriggers; any change recreates. OnFileChange map[string]string `json:"on_file_change,omitempty"` + // Fingerprints in trigger order. A list so two watches that resolve to the + // same value stay distinct (a map would treat converging them as a removal). + OnValueChange []string `json:"on_value_change,omitempty"` } // JobRunLifecycleState is the local-only trigger fingerprint. Nested by value, @@ -114,9 +127,61 @@ func (*ResourceJobRun) PrepareState(input *resources.JobRun) *JobRunState { if len(input.ResolvedFileTriggers) > 0 { state.Lifecycle.Triggers.OnFileChange = input.ResolvedFileTriggers } + if values := jobRunValueChangeState(input); len(values) > 0 { + state.Lifecycle.Triggers.OnValueChange = values + } + state.NormalizeAfterResolve() return state } +// PrepareInputConfig puts resource refs on the state path of each watched +// expression so the deploy graph depends on that value, not the config wrapper. +func (*ResourceJobRun) PrepareInputConfig(input *resources.JobRun, _ string) (*structvar.StructVar, error) { + refs := map[string]string{} + parent := structpath.MustParsePath("lifecycle.triggers.on_value_change") + for i, expr := range jobRunValueChangeState(input) { + if _, ok := dynvar.NewRef(dyn.V(expr)); !ok { + continue + } + refs[structpath.NewIndex(parent, i).String()] = expr + } + return &structvar.StructVar{Value: input, Refs: refs}, nil +} + +func jobRunValueChangeState(input *resources.JobRun) []string { + if input.Lifecycle == nil { + return nil + } + var out []string + for _, t := range input.Lifecycle.Triggers { + if t.OnValueChange == nil { + continue + } + out = append(out, strings.TrimSpace(*t.OnValueChange)) + } + return out +} + +// NormalizeAfterResolve hashes a watch once it no longer contains a reference. +func (s *JobRunState) NormalizeAfterResolve() { + values := s.Lifecycle.Triggers.OnValueChange + for i, value := range values { + if dynvar.ContainsVariableReference(value) { + continue + } + values[i] = compactJobRunValue(value) + } +} + +// compactJobRunValue hashes a value only when the digest is shorter than it. +func compactJobRunValue(value string) string { + if len(value) <= jobRunValueHashLength { + return value + } + sum := sha256.Sum256([]byte(value)) + return jobRunValueHashPrefix + hex.EncodeToString(sum[:]) +} + // makeJobRunRemote maps the GetRun response into the RunNow-shaped remote: GET // nests the params under overriding_parameters and returns job_parameters as a // list, so both are flattened back into RunNow. @@ -388,10 +453,17 @@ func (*ResourceJobRun) OverrideChangeDesc(_ context.Context, path *structpath.Pa change.Reason = "trigger removed" } case "lifecycle.triggers.on_file_change": + // Only a cleared trigger skips: a file dropping out of the map means the + // match disappeared, which is a real change. if change.New == nil { change.Action = deployplan.Skip change.Reason = "trigger removed" } + case "lifecycle.triggers.on_value_change": + if valueTriggersOnlyRemoved(change.Old, change.New) { + change.Action = deployplan.Skip + change.Reason = "trigger removed" + } case "result_state": // The planner passes no remote state when the run could not be read. if remote == nil || runIsTerminal(remote.State.LifeCycleState) { @@ -399,10 +471,45 @@ func (*ResourceJobRun) OverrideChangeDesc(_ context.Context, path *structpath.Pa } change.Action = deployplan.Skip change.Reason = "run in progress" + default: + parent := path.Parent() + if parent != nil && parent.String() == "lifecycle.triggers.on_value_change" { + // Classified on the list; a deletion that shifts later entries is not + // a change to those fingerprints. + change.Action = deployplan.Skip + change.Reason = deployplan.ReasonDrop + } } return nil } +// valueTriggersOnlyRemoved is true when new is old minus some watches (order kept). +// Converging two watches replaces a fingerprint, so it is not a removal. +func valueTriggersOnlyRemoved(oldValue, newValue any) bool { + if newValue == nil { + return true + } + oldList, okOld := oldValue.([]string) + newList, okNew := newValue.([]string) + if !okOld || !okNew { + return false + } + if len(newList) >= len(oldList) { + return false + } + return isStringSubsequence(oldList, newList) +} + +func isStringSubsequence(oldList, newList []string) bool { + i := 0 + for _, value := range oldList { + if i < len(newList) && value == newList[i] { + i++ + } + } + return i == len(newList) +} + // DoDelete deletes the run via jobs/runs/delete, on both destroy and the // recreate path. The API rejects a still-active run, which an interrupted wait // leaves behind, so cancel it first. diff --git a/bundle/direct/dresources/job_run_test.go b/bundle/direct/dresources/job_run_test.go index f9f21999316..f977f79d8b6 100644 --- a/bundle/direct/dresources/job_run_test.go +++ b/bundle/direct/dresources/job_run_test.go @@ -10,8 +10,10 @@ import ( "testing" "time" + "github.com/databricks/cli/bundle/config/resources" "github.com/databricks/cli/bundle/deployplan" "github.com/databricks/cli/libs/structs/structpath" + "github.com/databricks/cli/libs/structs/structvar" "github.com/databricks/cli/libs/testserver" "github.com/databricks/databricks-sdk-go" "github.com/databricks/databricks-sdk-go/service/jobs" @@ -346,6 +348,33 @@ func TestJobRunCreateSendsAFreshIdempotencyToken(t *testing.T) { assert.Empty(t, config.IdempotencyToken) } +func TestCompactJobRunValue(t *testing.T) { + assert.Equal(t, strings.Repeat("a", jobRunValueHashLength), compactJobRunValue(strings.Repeat("a", jobRunValueHashLength))) + assert.Equal( + t, + "sha256:d66304b6180365e47c858f6c84d3da065caf4b3350c9f45277a1af82e3dbb055", + compactJobRunValue(strings.Repeat("a", jobRunValueHashLength+1)), + ) +} + +func TestJobRunValueChangeStateNormalizesAfterAllReferencesResolve(t *testing.T) { + expr := "${resources.jobs.other.id}-${resources.jobs.extra.id}" + var trigger resources.JobRunTrigger + trigger.OnValueChange = &expr + var input resources.JobRun + input.Lifecycle = &resources.JobRunLifecycle{} + input.Lifecycle.Triggers = []resources.JobRunTrigger{trigger} + state := (&ResourceJobRun{}).PrepareState(&input) + path := structpath.NewIndex(structpath.MustParsePath("lifecycle.triggers.on_value_change"), 0) + sv := structvar.NewStructVar(state, map[string]string{path.String(): expr}) + + require.NoError(t, sv.ResolveRef("${resources.jobs.other.id}", int64(123))) + assert.Equal(t, []string{"123-${resources.jobs.extra.id}"}, state.Lifecycle.Triggers.OnValueChange) + + require.NoError(t, sv.ResolveRef("${resources.jobs.extra.id}", int64(456))) + assert.Equal(t, []string{"123-456"}, state.Lifecycle.Triggers.OnValueChange) +} + func TestJobRunDeleteLeavesFinishedRunAlone(t *testing.T) { var cancelled atomic.Bool server := testserver.New(t) @@ -374,21 +403,30 @@ func TestJobRunOverrideChangeDescTriggerRemoved(t *testing.T) { for _, tt := range []struct { name string path string + old any new any action deployplan.ActionType }{ - {"cleared on_bundle_deploy string", "lifecycle.triggers.on_bundle_deploy", "", deployplan.Skip}, - {"nil on_bundle_deploy", "lifecycle.triggers.on_bundle_deploy", nil, deployplan.Skip}, - {"rotated on_bundle_deploy", "lifecycle.triggers.on_bundle_deploy", "uuid", deployplan.Recreate}, - {"cleared on_file_change", "lifecycle.triggers.on_file_change", nil, deployplan.Skip}, - {"changed on_file_change map", "lifecycle.triggers.on_file_change", map[string]string{"a.txt": "h"}, deployplan.Recreate}, + {"cleared on_bundle_deploy string", "lifecycle.triggers.on_bundle_deploy", "old", "", deployplan.Skip}, + {"nil on_bundle_deploy", "lifecycle.triggers.on_bundle_deploy", "old", nil, deployplan.Skip}, + {"rotated on_bundle_deploy", "lifecycle.triggers.on_bundle_deploy", "old", "uuid", deployplan.Recreate}, + {"cleared on_file_change", "lifecycle.triggers.on_file_change", map[string]string{"a.txt": "h"}, nil, deployplan.Skip}, + {"changed on_file_change map", "lifecycle.triggers.on_file_change", map[string]string{"a.txt": "h"}, map[string]string{"a.txt": "new"}, deployplan.Recreate}, // A file dropping out of the map is a real change, so the skip must not - // extend to paths below on_file_change. - {"cleared on_file_change child", "lifecycle.triggers.on_file_change['a.txt']", nil, deployplan.Recreate}, - {"result_state with unreadable remote", "result_state", nil, deployplan.Recreate}, + // extend to on_file_change entries. + {"removed one on_file_change", "lifecycle.triggers.on_file_change", map[string]string{"a.txt": "h", "b.txt": "h"}, map[string]string{"a.txt": "h"}, deployplan.Recreate}, + {"cleared on_file_change child", "lifecycle.triggers.on_file_change['a.txt']", "h", nil, deployplan.Recreate}, + {"cleared on_value_change", "lifecycle.triggers.on_value_change", []string{"a"}, nil, deployplan.Skip}, + {"removed last on_value_change", "lifecycle.triggers.on_value_change", []string{"a", "b"}, []string{"a"}, deployplan.Skip}, + {"removed first on_value_change", "lifecycle.triggers.on_value_change", []string{"a", "b"}, []string{"b"}, deployplan.Skip}, + {"changed on_value_change", "lifecycle.triggers.on_value_change", []string{"a"}, []string{"b"}, deployplan.Recreate}, + {"converged on_value_change", "lifecycle.triggers.on_value_change", []string{"a", "b"}, []string{"a", "a"}, deployplan.Recreate}, + {"cleared on_value_change child", "lifecycle.triggers.on_value_change[1]", "b", nil, deployplan.Skip}, + {"shifted on_value_change child", "lifecycle.triggers.on_value_change[0]", "a", "b", deployplan.Skip}, + {"result_state with unreadable remote", "result_state", "", nil, deployplan.Recreate}, } { t.Run(tt.name, func(t *testing.T) { - change := &ChangeDesc{Action: deployplan.Recreate, New: tt.new} + change := &ChangeDesc{Action: deployplan.Recreate, Old: tt.old, New: tt.new} require.NoError(t, r.OverrideChangeDesc(t.Context(), structpath.MustParsePath(tt.path), change, nil)) assert.Equal(t, tt.action, change.Action) }) diff --git a/bundle/internal/schema/annotations.yml b/bundle/internal/schema/annotations.yml index 8b17164066d..8f788ac35ad 100644 --- a/bundle/internal/schema/annotations.yml +++ b/bundle/internal/schema/annotations.yml @@ -1020,6 +1020,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. Incompatible with lifecycle.prevent_destroy. + "on_value_change": + "description": |- + Interpolated expression. Re-fire the run when the resolved value changes. Resource references in the expression become deployment dependencies. Long resolved values are recorded as a SHA-256 hash instead of the value itself. Incompatible with lifecycle.prevent_destroy. "python_named_params": "description": |- PLACEHOLDER diff --git a/bundle/schema/jsonschema.json b/bundle/schema/jsonschema.json index 593447629ba..2feec7b2142 100644 --- a/bundle/schema/jsonschema.json +++ b/bundle/schema/jsonschema.json @@ -1398,6 +1398,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. Incompatible with lifecycle.prevent_destroy.", "$ref": "#/$defs/string" + }, + "on_value_change": { + "description": "Interpolated expression. Re-fire the run when the resolved value changes. Resource references in the expression become deployment dependencies. Long resolved values are recorded as a SHA-256 hash instead of the value itself. Incompatible with lifecycle.prevent_destroy.", + "$ref": "#/$defs/string" } }, "additionalProperties": false diff --git a/libs/structs/structaccess/typecheck.go b/libs/structs/structaccess/typecheck.go index 7147fa0f435..d4bf129b71e 100644 --- a/libs/structs/structaccess/typecheck.go +++ b/libs/structs/structaccess/typecheck.go @@ -34,6 +34,16 @@ func ValidatePath(t reflect.Type, path *structpath.PathNode) error { return ValidatePattern(t, (*structpath.PatternNode)(path)) } +// ValidatePathStrict is ValidatePath without the Terraform compatibility that +// treats [0] on a struct as a no-op. Use it for paths that are native to DABs, +// where indexing a struct is always a mismatch. +func ValidatePathStrict(t reflect.Type, path *structpath.PathNode) error { + if path.IsRoot() { + return nil + } + return validateNodeSlice(t, (*structpath.PatternNode)(path).AsSlice(), false) +} + // ValidatePattern reports whether the given pattern path is valid for the provided type. // It returns nil if the path resolves fully, or an error indicating where resolution failed. // Patterns may include wildcards ([*] and .*). @@ -41,11 +51,11 @@ func ValidatePattern(t reflect.Type, path *structpath.PatternNode) error { if path.IsRoot() { return nil } - return validateNodeSlice(t, path.AsSlice()) + return validateNodeSlice(t, path.AsSlice(), true) } -// validateNodeSlice is the implementation for ValidatePattern. -func validateNodeSlice(t reflect.Type, nodes []*structpath.PatternNode) error { +// validateNodeSlice is the implementation for ValidatePattern and ValidatePathStrict. +func validateNodeSlice(t reflect.Type, nodes []*structpath.PatternNode, allowStructIndex bool) error { cur := t for _, node := range nodes { // Always dereference pointers at the type level. @@ -64,7 +74,7 @@ func validateNodeSlice(t reflect.Type, nodes []*structpath.PatternNode) error { if kind != reflect.Slice && kind != reflect.Array { // Terraform represents single-block fields as lists and uses [0] to access them. // Treat [0] on a struct as a no-op so TF-style paths work against DABs structs. - if idx == 0 && kind == reflect.Struct { + if allowStructIndex && idx == 0 && kind == reflect.Struct { continue } return fmt.Errorf("%s: cannot index %s", node.String(), kind) diff --git a/libs/structs/structvar/structvar.go b/libs/structs/structvar/structvar.go index b60f8c15ee7..d6c42d4e740 100644 --- a/libs/structs/structvar/structvar.go +++ b/libs/structs/structvar/structvar.go @@ -22,6 +22,12 @@ type StructVar struct { Refs map[string]string `json:"vars,omitempty"` } +// ReferenceNormalizer lets a value re-derive fields that depend on resolved +// references. ResolveRef calls it after each substitution. +type ReferenceNormalizer interface { + NormalizeAfterResolve() +} + // StructVarJSON is the serialized form of StructVar for persisting in plan files. type StructVarJSON struct { Value json.RawMessage `json:"value"` @@ -123,6 +129,10 @@ func (sv *StructVar) ResolveRef(reference string, value any) error { return ErrNotFound } + if normalizer, ok := sv.Value.(ReferenceNormalizer); ok { + normalizer.NormalizeAfterResolve() + } + return nil }