Batch Jobs interdependencies

api/allocations.go

@@ -309,10 +309,11 @@ type AllocationMetric struct {
 	QuotaExhausted     []string
 	ResourcesExhausted map[string]*Resources
 	// Deprecated, replaced with ScoreMetaData
-	Scores            map[string]float64
-	AllocationTime    time.Duration
-	CoalescedFailures int
-	ScoreMetaData     []*NodeScoreMeta
+	Scores              map[string]float64
+	AllocationTime      time.Duration
+	CoalescedFailures   int
+	ScoreMetaData       []*NodeScoreMeta
+	BlockedDependencies []string
 }
 
 // NodeScoreMeta is used to serialize node scoring metadata

api/constraint.go

@@ -3,6 +3,11 @@
 
 package api
 
+import (
+	"errors"
+	"time"
+)
+
 const (
 	ConstraintDistinctProperty  = "distinct_property"
 	ConstraintDistinctHosts     = "distinct_hosts"
@@ -14,6 +19,13 @@ const (
 	ConstraintSetContainsAny    = "set_contains_any"
 	ConstraintAttributeIsSet    = "is_set"
 	ConstraintAttributeIsNotSet = "is_not_set"
+
+	JobDependencyComplete   = "jobComplete"   // All expected allocations are complete
+	JobDependencyRunning    = "jobRunning"    // All expected allocations are running
+	JobDependencyRecovering = "jobRecovering" // Some allocations are pending
+	JobDependencyLost       = "jobLost"       // All allocations are unknown
+	JobDependencyFailed     = "jobFailed"     // All allocations are failed, lost, or unplaced
+	JobDependencyStopped    = "jobStopped"    // The job has been manually stopped
 )
 
 // Constraint is used to serialize a job placement constraint.
@@ -31,3 +43,113 @@ func NewConstraint(left, operand, right string) *Constraint {
 		Operand: operand,
 	}
 }
+
+type JobDependency struct {
+	Name   string `hcl:"name"`
+	Status string `hcl:"status,optional"`
+}
+
+func NewJobDependency(name, status string) *JobDependency {
+	return &JobDependency{
+		Name:   name,
+		Status: status,
+	}
+}
+
+func (d *JobDependency) Canonicalize() {
+	if d.Status == "" {
+		d.Status = JobDependencyComplete
+	}
+}
+
+func (d *JobDependency) Copy() *JobDependency {
+	if d == nil {
+		return nil
+	}
+
+	copy := *d
+	return &copy
+}
+
+func (d *JobDependency) Validate() error {
+	if d.Name == "" {
+		return errors.New("dependency job name is required")
+	}
+
+	switch d.Status {
+	case JobDependencyComplete, JobDependencyRunning, JobDependencyRecovering,
+		JobDependencyLost, JobDependencyFailed, JobDependencyStopped:
+
+	default:
+		return errors.New("invalid state for dependency job")
+	}
+
+	return nil
+}
+
+// JobDependencies is used to serialize a job placement dependency.
+type JobDependencies struct {
+	Timeout *time.Duration   `hcl:"timeout,optional"`
+	Jobs    []*JobDependency `hcl:"job,block"`
+}
+
+func NewJobDependencies(timeout, actionOnTimeout string, jobs ...*JobDependency) *JobDependencies {
+	copyJobs := make([]*JobDependency, 0, len(jobs))
+	for _, job := range jobs {
+		copyJobs = append(copyJobs, job.Copy())
+	}
+
+	duration, _ := time.ParseDuration(timeout)
+	return &JobDependencies{
+		Timeout: &duration,
+		Jobs:    copyJobs,
+	}
+}
+
+func (d *JobDependencies) Canonicalize() {
+	if d.Timeout == nil {
+		d.Timeout = new(60 * time.Minute)
+	}
+	for _, job := range d.Jobs {
+		job.Canonicalize()
+	}
+}
+
+func (d *JobDependencies) Copy() *JobDependencies {
+	if d == nil {
+		return nil
+	}
+
+	jobs := make([]*JobDependency, 0, len(d.Jobs))
+	for _, job := range d.Jobs {
+		jobs = append(jobs, job.Copy())
+	}
+
+	return &JobDependencies{
+		Timeout: d.Timeout,
+		Jobs:    jobs,
+	}
+}
+
+func (d *JobDependencies) Validate() error {
+	if d == nil {
+		return nil
+	}
+
+	if d.Timeout == nil || *d.Timeout == 0 {
+		return errors.New("dependency timeout is required")
+	}
+
+	if len(d.Jobs) == 0 {
+		return errors.New("dependency requires at least one job block")
+	}
+
+	// Should we check that each dependency is unique??
+	for _, job := range d.Jobs {
+		if err := job.Validate(); err != nil {
+			return err
+		}
+	}
+
+	return nil
+}

api/constraint_test.go

@@ -5,11 +5,14 @@ package api
 
 import (
 	"testing"
+	"time"
 
 	"github.com/hashicorp/nomad/api/internal/testutil"
 	"github.com/shoenig/test/must"
 )
 
+var timeoutDuration = 10 * time.Minute
+
 func TestCompose_Constraints(t *testing.T) {
 	testutil.Parallel(t)
 
@@ -21,3 +24,41 @@ func TestCompose_Constraints(t *testing.T) {
 	}
 	must.Eq(t, expect, c)
 }
+
+func TestCompose_Dependencies(t *testing.T) {
+	testutil.Parallel(t)
+
+	d := NewJobDependencies("10m", "reject", &JobDependency{Name: "service-123", Status: JobDependencyComplete})
+	d.Canonicalize()
+
+	must.Eq(t, &timeoutDuration, d.Timeout)
+	must.Len(t, 1, d.Jobs)
+	must.Eq(t, "service-123", d.Jobs[0].Name)
+	must.Eq(t, JobDependencyComplete, d.Jobs[0].Status)
+	must.NoError(t, d.Validate())
+
+	copy := d.Copy()
+	must.Eq(t, d, copy)
+	must.True(t, d.Jobs[0] != copy.Jobs[0])
+}
+
+func TestCompose_Dependencies_DefaultsAndValidation(t *testing.T) {
+	testutil.Parallel(t)
+
+	d := &JobDependencies{
+		Timeout: &timeoutDuration,
+		Jobs: []*JobDependency{{
+			Name: "service-123",
+		}},
+	}
+	d.Canonicalize()
+
+	must.Eq(t, JobDependencyComplete, d.Jobs[0].Status)
+	must.NoError(t, d.Validate())
+
+	bad := &JobDependencies{
+		Timeout: &timeoutDuration,
+		Jobs:    []*JobDependency{{Name: "service-123", Status: "unexpectedState"}},
+	}
+	must.Error(t, bad.Validate())
+}

api/jobs.go

@@ -46,15 +46,11 @@ const (
 	// For Client configuration, if no region information is given,
 	// the client node will default to be part of the GlobalRegion.
 	GlobalRegion = "global"
-)
 
-const (
 	// RegisterEnforceIndexErrPrefix is the prefix to use in errors caused by
 	// enforcing the job modify index during registers.
 	RegisterEnforceIndexErrPrefix = "Enforcing job modify index"
-)
 
-const (
 	// JobPeriodicLaunchSuffix is the string appended to the periodic jobs ID
 	// when launching derived instances of it.
 	JobPeriodicLaunchSuffix = "/periodic-"
@@ -1113,6 +1109,7 @@ type Job struct {
 	Datacenters      []string                `hcl:"datacenters,optional"`
 	NodePool         *string                 `mapstructure:"node_pool" hcl:"node_pool,optional"`
 	Constraints      []*Constraint           `hcl:"constraint,block"`
+	Dependencies     *JobDependencies        `hcl:"dependency,block"`
 	Affinities       []*Affinity             `hcl:"affinity,block"`
 	TaskGroups       []*TaskGroup            `hcl:"group,block"`
 	Update           *UpdateStrategy         `hcl:"update,block"`
@@ -1234,6 +1231,10 @@ func (j *Job) Canonicalize() {
 		j.Multiregion.Canonicalize()
 	}
 
+	if j.Dependencies != nil {
+		j.Dependencies.Canonicalize()
+	}
+
 	for _, tg := range j.TaskGroups {
 		tg.Canonicalize(j)
 	}

ci/test-core.json

@@ -42,6 +42,7 @@
     "nomad/stream/...",
     "nomad/structs/...",
     "nomad/volumewatcher/...",
+    "nomad/dependency/...",
     "plugins/...",
     "scheduler/...",
     "scheduler/feasible/...",

command/agent/job_endpoint.go

@@ -12,6 +12,7 @@ import (
 	"slices"
 	"strconv"
 	"strings"
+	"time"
 
 	"github.com/golang/snappy"
 	"github.com/hashicorp/nomad/acl"
@@ -1145,6 +1146,7 @@ func ApiJobToStructJob(job *api.Job) *structs.Job {
 		Version:        *job.Version,
 		Constraints:    ApiConstraintsToStructs(job.Constraints),
 		Affinities:     ApiAffinitiesToStructs(job.Affinities),
+		Dependencies:   ApiDependencyToStructs(job.Dependencies),
 		UI:             ApiJobUIConfigToStructs(job.UI),
 		VersionTag:     ApiJobVersionTagToStructs(job.VersionTag),
 	}
@@ -2262,6 +2264,34 @@ func ApiAffinitiesToStructs(in []*api.Affinity) []*structs.Affinity {
 	return out
 }
 
+func ApiDependencyToStructs(in *api.JobDependencies) *structs.JobDependencies {
+	if in == nil {
+		return nil
+	}
+
+	jobs := make([]*structs.JobDependency, 0, len(in.Jobs))
+	for _, j := range in.Jobs {
+		if j == nil {
+			continue
+		}
+
+		jobs = append(jobs, &structs.JobDependency{
+			Name:   j.Name,
+			Status: j.Status,
+		})
+	}
+
+	var timeout time.Duration
+	if in.Timeout != nil {
+		timeout = *in.Timeout
+	}
+
+	return &structs.JobDependencies{
+		Timeout: timeout,
+		Jobs:    jobs,
+	}
+}
+
 func ApiJobUIConfigToStructs(jobUI *api.JobUIConfig) *structs.JobUIConfig {
 	if jobUI == nil {
 		return nil

command/agent/job_endpoint_test.go

@@ -4676,3 +4676,22 @@ func TestConversion_ApiJobVersionTagToStructs(t *testing.T) {
 		must.Eq(t, expected, result)
 	})
 }
+
+func TestConversion_ApiDependencyToStructs(t *testing.T) {
+	t.Run("nil dependency", func(t *testing.T) {
+		must.Nil(t, ApiDependencyToStructs(nil))
+	})
+	dur := time.Duration(10 * time.Minute)
+	t.Run("maps timeout, action and nested jobs", func(t *testing.T) {
+		in := &api.JobDependencies{
+			Timeout: &dur,
+			Jobs: []*api.JobDependency{
+				{Name: "service-123", Status: "completed"},
+			},
+		}
+
+		out := ApiDependencyToStructs(in)
+		must.Eq(t, 10*time.Minute, out.Timeout)
+		must.Eq(t, []*structs.JobDependency{{Name: "service-123", Status: "completed"}}, out.Jobs)
+	})
+}

command/monitor.go

@@ -340,7 +340,11 @@ func formatAllocMetrics(
 	var out string
 
 	if metrics.NodesEvaluated == 0 {
-		out += fmt.Sprintf("%s* No nodes were eligible for evaluation\n", prefix)
+		if len(metrics.BlockedDependencies) > 0 {
+			out += fmt.Sprintf("%s* Job blocked by: %s\n", prefix, strings.Join(metrics.BlockedDependencies, ", "))
+		} else {
+			out += fmt.Sprintf("%s* No nodes were eligible for evaluation\n", prefix)
+		}
 	}
 
 	// Print a helpful message if the user has asked for a DC that has no

jobspec2/go.mod

@@ -11,6 +11,7 @@ require (
 	github.com/hashicorp/nomad/api v0.0.0-20260814142628-f3fe893c53d2
 	github.com/mitchellh/reflectwalk v1.0.2
 	github.com/shoenig/test v1.13.2
+	github.com/stretchr/testify v1.11.1
 	github.com/zclconf/go-cty v1.19.0
 	github.com/zclconf/go-cty-yaml v1.2.0
 )
@@ -21,6 +22,7 @@ require (
 	github.com/apparentlymart/go-textseg/v15 v15.0.0 // indirect
 	github.com/apparentlymart/go-textseg/v17 v17.0.1 // indirect
 	github.com/bmatcuk/doublestar v1.3.4 // indirect
+	github.com/davecgh/go-spew v1.1.1 // indirect
 	github.com/go-viper/mapstructure/v2 v2.5.0 // indirect
 	github.com/google/go-cmp v0.7.0 // indirect
 	github.com/google/uuid v1.6.0 // indirect
@@ -30,10 +32,11 @@ require (
 	github.com/hashicorp/go-rootcerts v1.0.2 // indirect
 	github.com/mitchellh/go-homedir v1.1.0 // indirect
 	github.com/mitchellh/go-wordwrap v1.0.1 // indirect
-	github.com/stretchr/testify v1.11.1 // indirect
+	github.com/pmezard/go-difflib v1.0.0 // indirect
 	golang.org/x/crypto v0.57.0 // indirect
 	golang.org/x/mod v0.41.0 // indirect
 	golang.org/x/sync v0.23.0 // indirect
 	golang.org/x/text v0.42.0 // indirect
 	golang.org/x/tools v0.49.0 // indirect
+	gopkg.in/yaml.v3 v3.0.1 // indirect
 )

jobspec2/go.sum

@@ -58,5 +58,7 @@ golang.org/x/text v0.42.0 h1:JbOZXgfeCPU9gacVtYliJqOhD+zhrEqK4LfdpmlUZqI=
 golang.org/x/text v0.42.0/go.mod h1:ojzP1Z+2QtioaF8DTtO8K5q7JWVVYwZKenzujK0Zd0E=
 golang.org/x/tools v0.49.0 h1:3NI7VXzL9+1WZD52Dx2ttoPwD5DWrFGpl9mFZDlmisI=
 golang.org/x/tools v0.49.0/go.mod h1:SJNXV9DBKT0UbdttsQjbfJlAE/q+y36++zo3uL3N0Oo=
+gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405 h1:yhCVgyC4o1eVCa2tZl7eS0r+SDo693bJlVdllGtEeKM=
+gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
 gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
 gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=