diff --git a/acceptance/bundle/refschema/out.fields.txt b/acceptance/bundle/refschema/out.fields.txt index 5e774e1b580..28ccd1dcd02 100644 --- a/acceptance/bundle/refschema/out.fields.txt +++ b/acceptance/bundle/refschema/out.fields.txt @@ -914,6 +914,8 @@ resources.internal_immutable_snapshots.*.remote_root string INPUT resources.internal_immutable_snapshots.*.zip_path string INPUT STATE resources.job_runs.*.dbt_commands []string ALL resources.job_runs.*.dbt_commands[*] string ALL +resources.job_runs.*.depends_on []string INPUT STATE +resources.job_runs.*.depends_on[*] string INPUT STATE resources.job_runs.*.id string INPUT resources.job_runs.*.idempotency_token string ALL resources.job_runs.*.jar_params []string ALL diff --git a/acceptance/bundle/resources/job_runs/interrupted_run/databricks.yml b/acceptance/bundle/resources/job_runs/interrupted_run/databricks.yml index 506daeccca8..318570c07ce 100644 --- a/acceptance/bundle/resources/job_runs/interrupted_run/databricks.yml +++ b/acceptance/bundle/resources/job_runs/interrupted_run/databricks.yml @@ -10,6 +10,18 @@ resources: notebook_task: notebook_path: /Workspace/test + downstream_job: + name: downstream-job + tasks: + - task_key: main + notebook_task: + notebook_path: /Workspace/test + job_runs: my_run: job_id: ${resources.jobs.my_job.id} + + downstream_run: + job_id: ${resources.jobs.downstream_job.id} + depends_on: + - ${resources.job_runs.my_run.id} diff --git a/acceptance/bundle/resources/job_runs/interrupted_run/output.txt b/acceptance/bundle/resources/job_runs/interrupted_run/output.txt index 07c2bfffa54..ccbf0f5ad60 100644 --- a/acceptance/bundle/resources/job_runs/interrupted_run/output.txt +++ b/acceptance/bundle/resources/job_runs/interrupted_run/output.txt @@ -2,6 +2,7 @@ === the deploy stops waiting before the run finishes >>> errcode [CLI] bundle deploy Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/job-runs-interrupted-run/default/files... +Created jobs.downstream_job Created jobs.my_job Error: cannot create resources.job_runs.my_run: waiting after creating id=[MY_RUN_ID]: Fault injected by test. (403 INJECTED) @@ -10,6 +11,8 @@ HTTP Status: 403 Forbidden API error_code: INJECTED API message: Fault injected by test. +Error: cannot create resources.job_runs.downstream_run: dependency failed: resources.job_runs.my_run + Files: 3 uploaded, 0 deleted Exit code: 1 @@ -31,51 +34,92 @@ Exit code: 1 } } -=== a run that is still going is skipped ->>> jq .plan["resources.job_runs.my_run"] tmp.plan.json +=== a run that is still going is skipped and blocks its dependent +>>> jq {upstream: .plan["resources.job_runs.my_run"], downstream: .plan["resources.job_runs.downstream_run"]} tmp.plan.json { - "depends_on": [ - { - "node": "resources.jobs.my_job", - "label": "${resources.jobs.my_job.id}" - } - ], - "action": "skip", - "remote_state": { - "job_id": [MY_JOB_ID], - "run_id": [MY_RUN_ID], - "run_name": "my-job", - "run_page_url": "[DATABRICKS_URL]/jobs/[MY_JOB_ID]/runs/[MY_RUN_ID]?o=[NUMID]", - "run_type": "JOB_RUN", - "state": { - "life_cycle_state": "RUNNING" + "upstream": { + "depends_on": [ + { + "node": "resources.jobs.my_job", + "label": "${resources.jobs.my_job.id}" + } + ], + "action": "skip", + "remote_state": { + "job_id": [MY_JOB_ID], + "run_id": [MY_RUN_ID], + "run_name": "my-job", + "run_page_url": "[DATABRICKS_URL]/jobs/[MY_JOB_ID]/runs/[MY_RUN_ID]?o=[NUMID]", + "run_type": "JOB_RUN", + "state": { + "life_cycle_state": "RUNNING" + } + }, + "changes": { + "result_state": { + "action": "skip", + "reason": "run in progress", + "new": "SUCCESS" + } } }, - "changes": { - "result_state": { - "action": "skip", - "reason": "run in progress", - "new": "SUCCESS" + "downstream": { + "depends_on": [ + { + "node": "resources.job_runs.my_run", + "label": "${resources.job_runs.my_run.id}" + }, + { + "node": "resources.jobs.downstream_job", + "label": "${resources.jobs.downstream_job.id}" + } + ], + "action": "create", + "new_state": { + "value": { + "depends_on": [ + "[MY_RUN_ID]" + ], + "job_id": [NUMID], + "result_state": "SUCCESS" + } } } } -=== Badness: the next deploy does not wait for that run +=== the next deploy resumes waiting before starting the dependent run >>> [CLI] bundle deploy --plan tmp.plan.json Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/job-runs-interrupted-run/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 +Output from job_runs.downstream_run: id=[NUMID]: Run URL: [DATABRICKS_URL]/jobs/[NUMID]/runs/[NUMID]?o=[NUMID] +Output from job_runs.downstream_run: id=[NUMID]: SUCCESS +Created job_runs.downstream_run Files: 1 uploaded, 0 deleted -Resources: 0 created, 0 changed, 0 deleted, 2 unchanged +Resources: 1 created, 0 changed, 0 deleted, 3 unchanged -=== the same run is still tracked, and nothing was cancelled, deleted or triggered +=== both successful runs are tracked >>> read_state.py job_runs my_run id result_state job_runs my_run id='[MY_RUN_ID]' result_state=None ->>> print_requests.py //jobs +>>> read_state.py job_runs downstream_run id depends_on result_state +job_runs downstream_run id='[NUMID]' depends_on=['[MY_RUN_ID]'] result_state=None + +>>> print_requests.py //jobs/run-now +{ + "method": "POST", + "path": "/api/2.2/jobs/run-now", + "body": { + "idempotency_token": "[UUID]", + "job_id": [NUMID] + } +} >>> [CLI] bundle destroy --auto-approve The following resources will be deleted: + delete resources.jobs.downstream_job delete resources.jobs.my_job All files and directories at the following location will be deleted: /Workspace/Users/[USERNAME]/.bundle/job-runs-interrupted-run/default -Destroy: 1 deleted +Destroy: 2 deleted diff --git a/acceptance/bundle/resources/job_runs/interrupted_run/script b/acceptance/bundle/resources/job_runs/interrupted_run/script index 5d82c41ff49..d427c8d8d6d 100644 --- a/acceptance/bundle/resources/job_runs/interrupted_run/script +++ b/acceptance/bundle/resources/job_runs/interrupted_run/script @@ -17,15 +17,16 @@ trace read_id.py my_job trace read_id.py my_run trace print_requests.py //jobs/run-now -title "a run that is still going is skipped" +title "a run that is still going is skipped and blocks its dependent" $CLI bundle plan -o json > tmp.plan.json -trace jq '.plan["resources.job_runs.my_run"]' tmp.plan.json +trace jq '{upstream: .plan["resources.job_runs.my_run"], downstream: .plan["resources.job_runs.downstream_run"]}' tmp.plan.json # Replay the plan captured while the run was still going: a fresh plan would # re-read it, and the testserver would settle it on that GET. -title "Badness: the next deploy does not wait for that run" +title "the next deploy resumes waiting before starting the dependent run" trace $CLI bundle deploy --plan tmp.plan.json -title "the same run is still tracked, and nothing was cancelled, deleted or triggered" +title "both successful runs are tracked" trace read_state.py job_runs my_run id result_state -trace print_requests.py //jobs +trace read_state.py job_runs downstream_run id depends_on result_state +trace print_requests.py //jobs/run-now diff --git a/acceptance/bundle/resources/job_runs/interrupted_run/test.toml b/acceptance/bundle/resources/job_runs/interrupted_run/test.toml index 46de6c88c3c..dc6f4cafee7 100644 --- a/acceptance/bundle/resources/job_runs/interrupted_run/test.toml +++ b/acceptance/bundle/resources/job_runs/interrupted_run/test.toml @@ -1,4 +1,3 @@ # The interruption is staged by injecting a fault into the testserver, so this # stays off cloud. -Badness = "An in-progress run is planned as skip, so a deploy after an interrupted wait does not resume waiting for it to finish." Ignore = ["tmp.plan.json"] diff --git a/bundle/config/resources/job_run.go b/bundle/config/resources/job_run.go index ff9a91a8899..1c14e329c7c 100644 --- a/bundle/config/resources/job_run.go +++ b/bundle/config/resources/job_run.go @@ -21,6 +21,9 @@ type JobRun struct { ID string `json:"id,omitempty" bundle:"readonly"` jobs.RunNow + // DependsOn lists job runs that must succeed before this run starts. + DependsOn []string `json:"depends_on,omitempty"` + // Lifecycle shadows BaseResource.Lifecycle so job_runs can set triggers. Lifecycle *JobRunLifecycle `json:"lifecycle,omitempty"` diff --git a/bundle/config/validate/validate_job_run_dependencies.go b/bundle/config/validate/validate_job_run_dependencies.go new file mode 100644 index 00000000000..16fd5bd60f5 --- /dev/null +++ b/bundle/config/validate/validate_job_run_dependencies.go @@ -0,0 +1,78 @@ +package validate + +import ( + "context" + "fmt" + "maps" + "slices" + + "github.com/databricks/cli/bundle" + "github.com/databricks/cli/libs/diag" + "github.com/databricks/cli/libs/dyn" + "github.com/databricks/cli/libs/dyn/dynvar" +) + +func ValidateJobRunDependencies() bundle.ReadOnlyMutator { + return &validateJobRunDependencies{} +} + +type validateJobRunDependencies struct{ bundle.RO } + +func (v *validateJobRunDependencies) Name() string { + return "validate:job_run_dependencies" +} + +func (v *validateJobRunDependencies) Apply(_ context.Context, b *bundle.Bundle) diag.Diagnostics { + var diags diag.Diagnostics + + for _, name := range slices.Sorted(maps.Keys(b.Config.Resources.JobRuns)) { + jobRun := b.Config.Resources.JobRuns[name] + if jobRun == nil { + continue + } + + for i, dependency := range jobRun.DependsOn { + path := dyn.NewPath( + dyn.Key("resources"), + dyn.Key("job_runs"), + dyn.Key(name), + dyn.Key("depends_on"), + dyn.Index(i), + ) + refPath, ok := dynvar.PureReferenceToPath(dependency) + if !ok || !isJobRunIDReference(refPath) { + diags = append(diags, invalidJobRunDependency(b, path)) + continue + } + + target := refPath[2].Key() + if _, ok := b.Config.Resources.JobRuns[target]; !ok { + diags = append(diags, diag.Diagnostic{ + Severity: diag.Error, + Summary: fmt.Sprintf("depends_on references undefined job run %q", target), + Paths: []dyn.Path{path}, + Locations: b.Config.GetLocations(path.String()), + }) + } + } + } + + return diags +} + +func isJobRunIDReference(path dyn.Path) bool { + return len(path) == 4 && + path[0].Key() == "resources" && + path[1].Key() == "job_runs" && + path[2].Key() != "" && + path[3].Key() == "id" +} + +func invalidJobRunDependency(b *bundle.Bundle, path dyn.Path) diag.Diagnostic { + return diag.Diagnostic{ + Severity: diag.Error, + Summary: "depends_on entries must be job run ID references, for example ${resources.job_runs.prepare.id}", + Paths: []dyn.Path{path}, + Locations: b.Config.GetLocations(path.String()), + } +} diff --git a/bundle/config/validate/validate_job_run_dependencies_test.go b/bundle/config/validate/validate_job_run_dependencies_test.go new file mode 100644 index 00000000000..cebca25e042 --- /dev/null +++ b/bundle/config/validate/validate_job_run_dependencies_test.go @@ -0,0 +1,75 @@ +package validate + +import ( + "testing" + + "github.com/databricks/cli/bundle" + "github.com/databricks/cli/bundle/config" + "github.com/databricks/cli/bundle/config/resources" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestValidateJobRunDependencies(t *testing.T) { + tests := []struct { + name string + dependency string + summary string + }{ + { + name: "job run id reference", + dependency: "${resources.job_runs.prepare.id}", + }, + { + name: "literal name", + dependency: "prepare", + summary: "depends_on entries must be job run ID references, for example ${resources.job_runs.prepare.id}", + }, + { + name: "job definition reference", + dependency: "${resources.jobs.prepare.id}", + summary: "depends_on entries must be job run ID references, for example ${resources.job_runs.prepare.id}", + }, + { + name: "job run outcome reference", + dependency: "${resources.job_runs.prepare.state.result_state}", + summary: "depends_on entries must be job run ID references, for example ${resources.job_runs.prepare.id}", + }, + { + name: "reference with extra text", + dependency: "run ${resources.job_runs.prepare.id}", + summary: "depends_on entries must be job run ID references, for example ${resources.job_runs.prepare.id}", + }, + { + name: "undefined job run", + dependency: "${resources.job_runs.missing.id}", + summary: `depends_on references undefined job run "missing"`, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + b := &bundle.Bundle{ + Config: config.Root{ + Resources: config.Resources{ + JobRuns: map[string]*resources.JobRun{ + "prepare": {}, + "publish": {DependsOn: []string{tt.dependency}}, + }, + }, + }, + } + + diags := ValidateJobRunDependencies().Apply(t.Context(), b) + if tt.summary == "" { + assert.Empty(t, diags) + return + } + + require.Len(t, diags, 1) + assert.Equal(t, tt.summary, diags[0].Summary) + require.Len(t, diags[0].Paths, 1) + assert.Equal(t, "resources.job_runs.publish.depends_on[0]", diags[0].Paths[0].String()) + }) + } +} diff --git a/bundle/direct/bundle_apply.go b/bundle/direct/bundle_apply.go index a673a3239c8..d41e2d6b661 100644 --- a/bundle/direct/bundle_apply.go +++ b/bundle/direct/bundle_apply.go @@ -113,6 +113,8 @@ func (b *DeploymentBundle) Apply(ctx context.Context, client *databricks.Workspa MaxWait: unitWait, } + var waitedAfterSkip bool + if action == deployplan.Delete { if entry.IsStateOnlyDelete() { // The resource is already deleted remotely (Gone) or has no delete @@ -135,7 +137,29 @@ func (b *DeploymentBundle) Apply(ctx context.Context, client *databricks.Workspa return true } - // We don't keep NewState around for 'skip' nodes + // We don't keep NewState around for 'skip' nodes. A skipped resource can + // still be in progress after an earlier deploy abandoned its wait, so resume + // that wait before releasing any blocking dependents. + if action == deployplan.Skip && hasBlockingDependents(g, resourceKey) { + id := b.StateDB.GetResourceID(resourceKey) + if id == "" { + logdiag.LogError(ctx, fmt.Errorf("%s: internal error: missing entry in state", errorPrefix)) + return false + } + waitCtx := d.withResourceKey(ctx) + remoteState, err := retryOnTransient(waitCtx, func() (any, error) { + return d.Adapter.WaitAfterSkip(waitCtx, id, entry.RemoteState) + }) + if err != nil { + logdiag.LogError(ctx, fmt.Errorf("%s: waiting for skipped resource id=%s: %w", errorPrefix, id, err)) + return false + } + if err := d.SetRemoteState(remoteState); err != nil { + logdiag.LogError(ctx, fmt.Errorf("%s: %w", errorPrefix, err)) + return false + } + waitedAfterSkip = d.RemoteState != nil + } if action != deployplan.Skip { if !b.resolveReferences(ctx, resourceKey, entry, errorPrefix, false) { @@ -186,10 +210,12 @@ func (b *DeploymentBundle) Apply(ctx context.Context, client *databricks.Workspa return false } - err = d.refreshRemoteState(ctx, id) - if err != nil { - logdiag.LogError(ctx, fmt.Errorf("%s: failed to read remote state: %w", errorPrefix, err)) - return false + if !waitedAfterSkip { + err = d.refreshRemoteState(ctx, id) + if err != nil { + logdiag.LogError(ctx, fmt.Errorf("%s: failed to read remote state: %w", errorPrefix, err)) + return false + } } b.RemoteStateCache.Store(resourceKey, d.RemoteState) } diff --git a/bundle/direct/bundle_plan_test.go b/bundle/direct/bundle_plan_test.go index ccd7f827ad5..cb269c5a320 100644 --- a/bundle/direct/bundle_plan_test.go +++ b/bundle/direct/bundle_plan_test.go @@ -80,6 +80,31 @@ resources: }, refs) } +func TestExtractReferences_JobRunDependsOn(t *testing.T) { + adapters, err := dresources.InitAll(nil) + require.NoError(t, err) + + const yml = ` +resources: + job_runs: + prepare: + job_id: 1 + publish: + job_id: 2 + depends_on: + - ${resources.job_runs.prepare.id} +` + root, err := yamlloader.LoadYAML("test", bytes.NewBufferString(yml)) + require.NoError(t, err) + + refs, err := extractReferences(root, "resources.job_runs.publish", adapters["job_runs"].StateType()) + require.NoError(t, err) + + assert.Equal(t, map[string]string{ + "depends_on[0]": "${resources.job_runs.prepare.id}", + }, refs) +} + func TestShouldSkipBackendDefault_ManagedPropertiesOnly(t *testing.T) { // Rules mirror the schemas backend_defaults in schemas.yml, but the test is // deliberately self-contained so that edits to schemas.yml don't break it. @@ -357,7 +382,8 @@ func TestJobRunFinishedWithoutSuccessIsRecreate(t *testing.T) { } // A run that has not stopped yet may still succeed, so the plan leaves it -// alone rather than recreating it. Skip does not resume an abandoned wait. +// alone rather than recreating it. Apply resumes the wait when it has a +// blocking dependent. func TestJobRunInProgressIsSkip(t *testing.T) { for _, lifeCycleState := range []jobs.RunLifeCycleState{ jobs.RunLifeCycleStatePending, diff --git a/bundle/direct/dresources/README.md b/bundle/direct/dresources/README.md index 5487042c965..6f1f3e45d7d 100644 --- a/bundle/direct/dresources/README.md +++ b/bundle/direct/dresources/README.md @@ -39,10 +39,12 @@ Do **not** derive update mask field names from `entry.Changes`. The paths in `en If a resource has fields that must not be sent in updates (deploy-only, lifecycle-only, etc.), document them explicitly with a `var` block and a comment explaining each exclusion. -## Async APIs: WaitAfterCreate / WaitAfterUpdate +## Async APIs: WaitAfterCreate / WaitAfterSkip / WaitAfterUpdate For resources whose create or update is asynchronous (the resource is not immediately ready after the call returns), implement `WaitAfterCreate` and/or `WaitAfterUpdate` instead of polling inline inside DoCreate/DoUpdate. These are the correct extension points in the framework, and polling inline bypasses state persistence timing. +Implement `WaitAfterSkip` when an interrupted wait must be resumed before blocking dependents can proceed. It receives the remote state captured by the plan, allowing an already-ready resource to return without another read. + ## Slice ordering: KeyedSlices If the API may return a slice's elements in a different order between calls (e.g., `depends_on` in job tasks, `privileges` in grants), implement `KeyedSlices` to compare elements by a natural key rather than by index. Without this, every deploy after any reordering shows phantom diffs. diff --git a/bundle/direct/dresources/adapter.go b/bundle/direct/dresources/adapter.go index 6210a952112..c882d459865 100644 --- a/bundle/direct/dresources/adapter.go +++ b/bundle/direct/dresources/adapter.go @@ -99,6 +99,10 @@ type IResource interface { // TODO: wait status should be persisted in the state. WaitAfterCreate(ctx context.Context, id string, newState any) (remoteState any, e error) + // [Optional] WaitAfterSkip waits for an unchanged resource to become ready when it has blocking dependents. + // This resumes waits abandoned by an earlier interrupted deployment. + WaitAfterSkip(ctx context.Context, id string, remoteState any) (updatedRemoteState any, e error) + // [Optional] WaitAfterUpdate waits for the resource to become ready after update. Returns optionally updated remote state. WaitAfterUpdate(ctx context.Context, id string, newState any) (remoteState any, e error) @@ -136,6 +140,7 @@ type Adapter struct { doUpdate *calladapt.BoundCaller doUpdateWithID *calladapt.BoundCaller waitAfterCreate *calladapt.BoundCaller + waitAfterSkip *calladapt.BoundCaller waitAfterUpdate *calladapt.BoundCaller waitAfterDelete *calladapt.BoundCaller overrideChangeDesc *calladapt.BoundCaller @@ -178,6 +183,7 @@ func NewAdapter(typedNil any, resourceType string, client *databricks.WorkspaceC doUpdateWithID: nil, doResize: nil, waitAfterCreate: nil, + waitAfterSkip: nil, waitAfterUpdate: nil, waitAfterDelete: nil, overrideChangeDesc: nil, @@ -283,6 +289,11 @@ func (a *Adapter) initMethods(resource any) error { return err } + a.waitAfterSkip, err = calladapt.PrepareCall(resource, reflect.TypeFor[IResource](), "WaitAfterSkip") + if err != nil { + return err + } + a.waitAfterUpdate, err = calladapt.PrepareCall(resource, reflect.TypeFor[IResource](), "WaitAfterUpdate") if err != nil { return err @@ -420,6 +431,14 @@ func (a *Adapter) validate() error { validations = append(validations, "WaitAfterCreate remoteState return", a.waitAfterCreate.OutTypes[0], remoteType) } + if a.waitAfterSkip != nil { + if len(a.waitAfterSkip.OutTypes) != 2 { + return fmt.Errorf("WaitAfterSkip must return (remoteType, error), got %d return values", len(a.waitAfterSkip.OutTypes)) + } + validations = append(validations, "WaitAfterSkip remoteState", a.waitAfterSkip.InTypes[2], remoteType) + validations = append(validations, "WaitAfterSkip remoteState return", a.waitAfterSkip.OutTypes[0], remoteType) + } + if a.waitAfterUpdate != nil { validations = append(validations, "WaitAfterUpdate newState", a.waitAfterUpdate.InTypes[2], stateType) // WaitAfterUpdate must return (remoteType, error) @@ -656,6 +675,22 @@ func (a *Adapter) WaitAfterCreate(ctx context.Context, id string, newState any) return remoteState, nil } +// WaitAfterSkip waits for an unchanged resource to become ready when another +// resource depends on it. If the resource doesn't implement this method, this is a no-op. +func (a *Adapter) WaitAfterSkip(ctx context.Context, id string, remoteState any) (any, error) { + if a.waitAfterSkip == nil { + return nil, nil + } + + outs, err := a.waitAfterSkip.Call(ctx, id, remoteState) + if err != nil { + return nil, err + } + + updatedRemoteState := normalizeNilPointer(outs[0]) + return updatedRemoteState, nil +} + // WaitAfterUpdate waits for the resource to become ready after update. // If the resource doesn't implement this method, this is a no-op. // Returns the updated remoteState if available, otherwise returns nil. diff --git a/bundle/direct/dresources/job_run.go b/bundle/direct/dresources/job_run.go index 36031ec612e..17028cc4586 100644 --- a/bundle/direct/dresources/job_run.go +++ b/bundle/direct/dresources/job_run.go @@ -35,6 +35,10 @@ type JobRunLifecycleState struct { type JobRunState struct { jobs.RunNow + // Local-only resolved run IDs used to order job runs. DoCreate sends only + // RunNow, so these are never included in the Jobs API request. + DependsOn []string `json:"depends_on,omitempty"` + // Always SUCCESS during planning and cleared before persistence. ResultState jobs.RunResultState `json:"result_state,omitempty"` @@ -90,6 +94,7 @@ func (*ResourceJobRun) New(client *databricks.WorkspaceClient) *ResourceJobRun { func (*ResourceJobRun) PrepareState(input *resources.JobRun) *JobRunState { state := &JobRunState{ RunNow: input.RunNow, + DependsOn: slices.Clone(input.DependsOn), ResultState: jobs.RunResultStateSuccess, Lifecycle: nil, } @@ -181,6 +186,7 @@ func (r *ResourceJobRun) DoRead(ctx context.Context, id string) (*JobRunRemote, func (*ResourceJobRun) RemapState(remote *JobRunRemote) *JobRunState { return &JobRunState{ RunNow: remote.RunNow, + DependsOn: nil, ResultState: remote.ResultState, Lifecycle: nil, } @@ -207,6 +213,15 @@ func (r *ResourceJobRun) WaitAfterCreate(ctx context.Context, id string, _ *JobR return r.waitForRun(ctx, id) } +// WaitAfterSkip resumes waiting for a previously triggered run when another +// resource depends on it. This covers deploys interrupted during WaitAfterCreate. +func (r *ResourceJobRun) WaitAfterSkip(ctx context.Context, id string, remote *JobRunRemote) (*JobRunRemote, error) { + if runIsTerminal(remote.State.LifeCycleState) && remote.ResultState == jobs.RunResultStateSuccess { + return remote, nil + } + return r.waitForRun(ctx, id) +} + // waitForRun polls the run until it stops, and fails unless it succeeded. func (r *ResourceJobRun) waitForRun(ctx context.Context, id string) (*JobRunRemote, error) { runID, err := parseRunID(id) diff --git a/bundle/direct/dresources/job_run_test.go b/bundle/direct/dresources/job_run_test.go index 5da8feb39e0..d8ba03e92cb 100644 --- a/bundle/direct/dresources/job_run_test.go +++ b/bundle/direct/dresources/job_run_test.go @@ -290,6 +290,16 @@ func TestJobRunPrepareStateCopiesResolvedTriggers(t *testing.T) { assert.Empty(t, triggers.OnBundleDeploy) } +func TestJobRunPrepareStateCopiesDependencies(t *testing.T) { + input := &resources.JobRun{DependsOn: []string{"${resources.job_runs.prepare.id}"}} + + state := (&ResourceJobRun{}).PrepareState(input) + + assert.Equal(t, input.DependsOn, state.DependsOn) + state.DependsOn[0] = "123" + assert.Equal(t, "${resources.job_runs.prepare.id}", input.DependsOn[0]) +} + // The planner diffs RemapState(remote) against PrepareState(config), so a run // that did not end in SUCCESS has to surface as a difference on result_state. func TestJobRunRemapStateCarriesTheOutcome(t *testing.T) { @@ -358,17 +368,39 @@ func TestJobRunWaitPollsUntilTerminal(t *testing.T) { assert.Equal(t, int32(3), gets.Load(), "expected the wait to poll past both RUNNING reads") } +func TestJobRunWaitAfterSkipKeepsSettledRun(t *testing.T) { + remote := &JobRunRemote{ + ResultState: jobs.RunResultStateSuccess, + State: &jobs.RunState{ + LifeCycleState: jobs.RunLifeCycleStateTerminated, + ResultState: jobs.RunResultStateSuccess, + }, + } + + got, err := (&ResourceJobRun{}).WaitAfterSkip(t.Context(), "123", remote) + + require.NoError(t, err) + assert.Same(t, remote, got) +} + func TestJobRunCreateSendsAFreshIdempotencyToken(t *testing.T) { var tokens []string server := testserver.New(t) server.Handle("POST", "/api/2.2/jobs/run-now", func(req testserver.Request) any { + var raw map[string]any + require.NoError(t, json.Unmarshal(req.Body, &raw)) + assert.NotContains(t, raw, "depends_on") + var body jobs.RunNow require.NoError(t, json.Unmarshal(req.Body, &body)) tokens = append(tokens, body.IdempotencyToken) return jobs.RunNowResponse{RunId: int64(123 + len(tokens))} }) r := (&ResourceJobRun{}).New(jobRunClientFor(t, server)) - config := &JobRunState{RunNow: jobs.RunNow{JobId: 456}} + config := &JobRunState{ + RunNow: jobs.RunNow{JobId: 456}, + DependsOn: []string{"123"}, + } for range 2 { _, _, err := r.DoCreate(t.Context(), config) diff --git a/bundle/direct/dresources/type_test.go b/bundle/direct/dresources/type_test.go index 9d7f6770180..e3736779776 100644 --- a/bundle/direct/dresources/type_test.go +++ b/bundle/direct/dresources/type_test.go @@ -62,7 +62,8 @@ var knownMissingInRemoteType = map[string][]string{ "zip_path", }, "job_runs": { - // Local-only trigger fingerprints under lifecycle. + // Local-only dependency IDs and trigger fingerprints. + "depends_on", "lifecycle", }, } diff --git a/bundle/internal/schema/annotations.yml b/bundle/internal/schema/annotations.yml index e275e928478..13228f9131c 100644 --- a/bundle/internal/schema/annotations.yml +++ b/bundle/internal/schema/annotations.yml @@ -1022,6 +1022,9 @@ resources: The deployment waits for the run to finish and fails if it does not succeed, so other resources can reference the run's outcome, for example `${resources.job_runs..state.result_state}`. A run that did not succeed is run again on the next deployment. "$fields": + "depends_on": + "description": |- + Job runs that must finish successfully before this run starts. Each entry must reference a job run ID, for example `${resources.job_runs.prepare.id}`. When a dependency re-runs, this run re-runs after it succeeds. "lifecycle": "description": |- Settings that control the deployment lifecycle of the resource, such as preventing it from being destroyed and when the run re-fires. diff --git a/bundle/phases/initialize.go b/bundle/phases/initialize.go index 9d9c682ef62..a2a55237a0f 100644 --- a/bundle/phases/initialize.go +++ b/bundle/phases/initialize.go @@ -191,6 +191,9 @@ func Initialize(ctx context.Context, b *bundle.Bundle) { // Reject configured job_runs.idempotency_token; the CLI sets it on run-now. validate.ValidateJobRunIdempotencyToken(), + // Require job_runs.depends_on entries to reference existing job run IDs. + validate.ValidateJobRunDependencies(), + // Reject invalid job_runs.lifecycle.triggers (empty, false, prevent_destroy). mutator.ValidateJobRunTriggers(), diff --git a/bundle/schema/jsonschema.json b/bundle/schema/jsonschema.json index cfa257d0ed8..07b63fe53e7 100644 --- a/bundle/schema/jsonschema.json +++ b/bundle/schema/jsonschema.json @@ -1411,6 +1411,10 @@ "doNotSuggest": true, "deprecated": true }, + "depends_on": { + "description": "Job runs that must finish successfully before this run starts. Each entry must reference a job run ID, for example `${resources.job_runs.prepare.id}`. When a dependency re-runs, this run re-runs after it succeeds.", + "$ref": "#/$defs/slice/string" + }, "jar_params": { "description": "[Private Preview] A list of parameters for jobs with Spark JAR tasks, for example `\"jar_params\": [\"john doe\", \"35\"]`.\nThe parameters are used to invoke the main function of the main class specified in the Spark JAR task.\nIf not specified upon `run-now`, it defaults to an empty list.\njar_params cannot be specified in conjunction with notebook_params.\nThe JSON representation of this field (for example `{\"jar_params\":[\"john doe\",\"35\"]}`) cannot exceed 10,000 bytes.\n\n⚠ **Deprecation note** Use [job parameters](https://docs.databricks.com/jobs/job-parameters.html#job-parameter-pushdown) to pass information down to tasks.", "$ref": "#/$defs/slice/string", diff --git a/python/databricks/bundles/job_runs/_models/job_run.py b/python/databricks/bundles/job_runs/_models/job_run.py index 82397df768c..9950222f6b5 100644 --- a/python/databricks/bundles/job_runs/_models/job_run.py +++ b/python/databricks/bundles/job_runs/_models/job_run.py @@ -42,6 +42,11 @@ class JobRun(Resource): The ID of the job to be executed """ + depends_on: VariableOrList[str] = field(default_factory=list) + """ + Job runs that must finish successfully before this run starts. Each entry must reference a job run ID, for example `${resources.job_runs.prepare.id}`. When a dependency re-runs, this run re-runs after it succeeds. + """ + job_parameters: VariableOrDict[str] = field(default_factory=dict) """ Job-level parameters used in the run. for example `"param": "overriding_val"` @@ -95,6 +100,11 @@ class JobRunDict(TypedDict, total=False): The ID of the job to be executed """ + depends_on: VariableOrList[str] + """ + Job runs that must finish successfully before this run starts. Each entry must reference a job run ID, for example `${resources.job_runs.prepare.id}`. When a dependency re-runs, this run re-runs after it succeeds. + """ + job_parameters: VariableOrDict[str] """ Job-level parameters used in the run. for example `"param": "overriding_val"` diff --git a/python/databricks_tests/core/_generated/job_runs.py b/python/databricks_tests/core/_generated/job_runs.py index 09b1c10f6eb..c6e6250f570 100644 --- a/python/databricks_tests/core/_generated/job_runs.py +++ b/python/databricks_tests/core/_generated/job_runs.py @@ -15,6 +15,7 @@ def _test_case(): ResourceTestCase( add_resource=Resources.add_job_run, dict_example={ + "depends_on": ["depends_on"], "job_id": 0, "job_parameters": {"key": "value"}, "lifecycle": {}, @@ -24,6 +25,7 @@ def _test_case(): "queue": {"enabled": True}, }, dataclass_example=JobRun( + depends_on=["depends_on"], job_id=0, job_parameters={"key": "value"}, lifecycle=JobRunLifecycle(),