Add a new parameter to avoid starting a replacement for lost allocs

.changelog/16867.txt

@@ -0,0 +1,3 @@
+```release-note:feature
+**Reschedule on Lost**: Adds the ability to prevent tasks on down nodes from being rescheduled
+```

api/jobs_test.go

@@ -312,8 +312,9 @@ func TestJobs_Canonicalize(t *testing.T) {
 				},
 				TaskGroups: []*TaskGroup{
 					{
-						Name:  pointerOf(""),
-						Count: pointerOf(1),
+						Name:                    pointerOf(""),
+						Count:                   pointerOf(1),
+						PreventRescheduleOnLost: pointerOf(false),
 						EphemeralDisk: &EphemeralDisk{
 							Sticky:  pointerOf(false),
 							Migrate: pointerOf(false),
@@ -399,8 +400,9 @@ func TestJobs_Canonicalize(t *testing.T) {
 				JobModifyIndex:    pointerOf(uint64(0)),
 				TaskGroups: []*TaskGroup{
 					{
-						Name:  pointerOf(""),
-						Count: pointerOf(1),
+						Name:                    pointerOf(""),
+						Count:                   pointerOf(1),
+						PreventRescheduleOnLost: pointerOf(false),
 						EphemeralDisk: &EphemeralDisk{
 							Sticky:  pointerOf(false),
 							Migrate: pointerOf(false),
@@ -491,8 +493,9 @@ func TestJobs_Canonicalize(t *testing.T) {
 				},
 				TaskGroups: []*TaskGroup{
 					{
-						Name:  pointerOf("bar"),
-						Count: pointerOf(1),
+						Name:                    pointerOf("bar"),
+						PreventRescheduleOnLost: pointerOf(false),
+						Count:                   pointerOf(1),
 						EphemeralDisk: &EphemeralDisk{
 							Sticky:  pointerOf(false),
 							Migrate: pointerOf(false),
@@ -555,8 +558,9 @@ func TestJobs_Canonicalize(t *testing.T) {
 				},
 				TaskGroups: []*TaskGroup{
 					{
-						Name:  pointerOf("cache"),
-						Count: pointerOf(1),
+						Name:                    pointerOf("cache"),
+						Count:                   pointerOf(1),
+						PreventRescheduleOnLost: pointerOf(true),
 						RestartPolicy: &RestartPolicy{
 							Interval: pointerOf(5 * time.Minute),
 							Attempts: pointerOf(10),
@@ -666,8 +670,9 @@ func TestJobs_Canonicalize(t *testing.T) {
 				},
 				TaskGroups: []*TaskGroup{
 					{
-						Name:  pointerOf("cache"),
-						Count: pointerOf(1),
+						Name:                    pointerOf("cache"),
+						Count:                   pointerOf(1),
+						PreventRescheduleOnLost: pointerOf(true),
 						RestartPolicy: &RestartPolicy{
 							Interval:        pointerOf(5 * time.Minute),
 							Attempts:        pointerOf(10),
@@ -795,7 +800,6 @@ func TestJobs_Canonicalize(t *testing.T) {
 				},
 			},
 		},
-
 		{
 			name: "periodic",
 			input: &Job{
@@ -865,7 +869,8 @@ func TestJobs_Canonicalize(t *testing.T) {
 				},
 				TaskGroups: []*TaskGroup{
 					{
-						Name: pointerOf("bar"),
+						Name:                    pointerOf("bar"),
+						PreventRescheduleOnLost: pointerOf(true),
 						Consul: &Consul{
 							Namespace: "",
 						},
@@ -885,7 +890,8 @@ func TestJobs_Canonicalize(t *testing.T) {
 						},
 					},
 					{
-						Name: pointerOf("baz"),
+						Name:                    pointerOf("baz"),
+						PreventRescheduleOnLost: pointerOf(false),
 						Tasks: []*Task{
 							{
 								Name: "task1",
@@ -930,8 +936,9 @@ func TestJobs_Canonicalize(t *testing.T) {
 				},
 				TaskGroups: []*TaskGroup{
 					{
-						Name:  pointerOf("bar"),
-						Count: pointerOf(1),
+						Name:                    pointerOf("bar"),
+						Count:                   pointerOf(1),
+						PreventRescheduleOnLost: pointerOf(true),
 						EphemeralDisk: &EphemeralDisk{
 							Sticky:  pointerOf(false),
 							Migrate: pointerOf(false),
@@ -979,8 +986,9 @@ func TestJobs_Canonicalize(t *testing.T) {
 						},
 					},
 					{
-						Name:  pointerOf("baz"),
-						Count: pointerOf(1),
+						Name:                    pointerOf("baz"),
+						PreventRescheduleOnLost: pointerOf(false),
+						Count:                   pointerOf(1),
 						EphemeralDisk: &EphemeralDisk{
 							Sticky:  pointerOf(false),
 							Migrate: pointerOf(false),
@@ -1038,7 +1046,8 @@ func TestJobs_Canonicalize(t *testing.T) {
 				ParentID: pointerOf("lol"),
 				TaskGroups: []*TaskGroup{
 					{
-						Name: pointerOf("bar"),
+						Name:                    pointerOf("bar"),
+						PreventRescheduleOnLost: pointerOf(true),
 						RestartPolicy: &RestartPolicy{
 							Delay:    pointerOf(15 * time.Second),
 							Attempts: pointerOf(2),
@@ -1111,8 +1120,9 @@ func TestJobs_Canonicalize(t *testing.T) {
 				},
 				TaskGroups: []*TaskGroup{
 					{
-						Name:  pointerOf("bar"),
-						Count: pointerOf(1),
+						Name:                    pointerOf("bar"),
+						PreventRescheduleOnLost: pointerOf(true),
+						Count:                   pointerOf(1),
 						EphemeralDisk: &EphemeralDisk{
 							Sticky:  pointerOf(false),
 							Migrate: pointerOf(false),
@@ -1166,8 +1176,9 @@ func TestJobs_Canonicalize(t *testing.T) {
 						},
 					},
 					{
-						Name:  pointerOf("baz"),
-						Count: pointerOf(1),
+						Name:                    pointerOf("baz"),
+						PreventRescheduleOnLost: pointerOf(false),
+						Count:                   pointerOf(1),
 						EphemeralDisk: &EphemeralDisk{
 							Sticky:  pointerOf(false),
 							Migrate: pointerOf(false),
@@ -1223,7 +1234,6 @@ func TestJobs_Canonicalize(t *testing.T) {
 				},
 			},
 		},
-
 		{
 			name: "multiregion",
 			input: &Job{

api/tasks.go

@@ -459,6 +459,7 @@ type TaskGroup struct {
 	MaxClientDisconnect       *time.Duration            `mapstructure:"max_client_disconnect" hcl:"max_client_disconnect,optional"`
 	Scaling                   *ScalingPolicy            `hcl:"scaling,block"`
 	Consul                    *Consul                   `hcl:"consul,block"`
+	PreventRescheduleOnLost   *bool                     `hcl:"prevent_reschedule_on_lost,optional"`
 }
 
 // NewTaskGroup creates a new TaskGroup.
@@ -577,7 +578,9 @@ func (g *TaskGroup) Canonicalize(job *Job) {
 	for _, s := range g.Services {
 		s.Canonicalize(nil, g, job)
 	}
-
+	if g.PreventRescheduleOnLost == nil {
+		g.PreventRescheduleOnLost = pointerOf(false)
+	}
 }
 
 // These needs to be in sync with DefaultServiceJobRestartPolicy in

client/client.go

@@ -2666,10 +2666,11 @@ func (c *Client) updateAlloc(update *structs.Allocation) {
 		return
 	}
 
+	alloc := ar.Alloc()
 	// Reconnect unknown allocations if they were updated and are not terminal.
 	reconnect := update.ClientStatus == structs.AllocClientStatusUnknown &&
-		update.AllocModifyIndex > ar.Alloc().AllocModifyIndex &&
-		!update.ServerTerminalStatus()
+		update.AllocModifyIndex > alloc.AllocModifyIndex &&
+		(!update.ServerTerminalStatus() || !alloc.PreventRescheduleOnLost())
 	if reconnect {
 		err = ar.Reconnect(update)
 		if err != nil {

command/agent/job_endpoint.go

@@ -1135,6 +1135,12 @@ func ApiTgToStructsTG(job *structs.Job, taskGroup *api.TaskGroup, tg *structs.Ta
 		RenderTemplates: *taskGroup.RestartPolicy.RenderTemplates,
 	}
 
+	if taskGroup.PreventRescheduleOnLost == nil {
+		tg.PreventRescheduleOnLost = false
+	} else {
+		tg.PreventRescheduleOnLost = *taskGroup.PreventRescheduleOnLost
+	}
+
 	if taskGroup.ShutdownDelay != nil {
 		tg.ShutdownDelay = taskGroup.ShutdownDelay
 	}

command/agent/job_endpoint_test.go

@@ -3058,6 +3058,7 @@ func TestJobs_ApiJobToStructsJob(t *testing.T) {
 						Operand: "z",
 					},
 				},
+				PreventRescheduleOnLost: false,
 				Affinities: []*structs.Affinity{
 					{
 						LTarget: "x",
@@ -3552,8 +3553,9 @@ func TestJobs_ApiJobToStructsJob(t *testing.T) {
 		},
 		TaskGroups: []*structs.TaskGroup{
 			{
-				Name:  "group1",
-				Count: 5,
+				Name:                    "group1",
+				Count:                   5,
+				PreventRescheduleOnLost: false,
 				Constraints: []*structs.Constraint{
 					{
 						LTarget: "x",

nomad/core_sched_test.go

@@ -1940,20 +1940,21 @@ func TestCoreScheduler_PartitionJobReap(t *testing.T) {
 // Tests various scenarios when allocations are eligible to be GCed
 func TestAllocation_GCEligible(t *testing.T) {
 	type testCase struct {
-		Desc                string
-		GCTime              time.Time
-		ClientStatus        string
-		DesiredStatus       string
-		JobStatus           string
-		JobStop             bool
-		AllocJobModifyIndex uint64
-		JobModifyIndex      uint64
-		ModifyIndex         uint64
-		NextAllocID         string
-		ReschedulePolicy    *structs.ReschedulePolicy
-		RescheduleTrackers  []*structs.RescheduleEvent
-		ThresholdIndex      uint64
-		ShouldGC            bool
+		Desc                    string
+		GCTime                  time.Time
+		ClientStatus            string
+		DesiredStatus           string
+		JobStatus               string
+		JobStop                 bool
+		PreventRescheduleOnLost *bool
+		AllocJobModifyIndex     uint64
+		JobModifyIndex          uint64
+		ModifyIndex             uint64
+		NextAllocID             string
+		ReschedulePolicy        *structs.ReschedulePolicy
+		RescheduleTrackers      []*structs.RescheduleEvent
+		ThresholdIndex          uint64
+		ShouldGC                bool
 	}
 
 	fail := time.Now()
@@ -2120,6 +2121,14 @@ func TestAllocation_GCEligible(t *testing.T) {
 			JobStop:  true,
 			ShouldGC: true,
 		},
+		{
+			Desc:          "GC when alloc is lost and eligible for reschedule",
+			ClientStatus:  structs.AllocClientStatusLost,
+			DesiredStatus: structs.AllocDesiredStatusStop,
+			GCTime:        fail,
+			JobStatus:     structs.JobStatusDead,
+			ShouldGC:      true,
+		},
 		{
 			Desc:             "GC when job status is dead",
 			ClientStatus:     structs.AllocClientStatusFailed,
@@ -2155,6 +2164,14 @@ func TestAllocation_GCEligible(t *testing.T) {
 			},
 			ShouldGC: true,
 		},
+		{
+			Desc:          "GC when alloc is unknown and but desired state is running",
+			ClientStatus:  structs.AllocClientStatusUnknown,
+			DesiredStatus: structs.AllocDesiredStatusRun,
+			GCTime:        fail,
+			JobStatus:     structs.JobStatusRunning,
+			ShouldGC:      false,
+		},
 	}
 
 	for _, tc := range harness {
@@ -2166,6 +2183,9 @@ func TestAllocation_GCEligible(t *testing.T) {
 		alloc.NextAllocation = tc.NextAllocID
 		job := mock.Job()
 		alloc.TaskGroup = job.TaskGroups[0].Name
+		if tc.PreventRescheduleOnLost != nil {
+			job.TaskGroups[0].PreventRescheduleOnLost = *tc.PreventRescheduleOnLost
+		}
 		job.TaskGroups[0].ReschedulePolicy = tc.ReschedulePolicy
 		if tc.JobStatus != "" {
 			job.Status = tc.JobStatus

nomad/mock/job.go

@@ -31,8 +31,9 @@ func Job() *structs.Job {
 		},
 		TaskGroups: []*structs.TaskGroup{
 			{
-				Name:  "web",
-				Count: 10,
+				Name:                    "web",
+				Count:                   10,
+				PreventRescheduleOnLost: false,
 				Constraints: []*structs.Constraint{
 					{
 						LTarget: "${attr.consul.version}",

nomad/plan_apply.go

@@ -737,6 +737,11 @@ func evaluateNodePlan(snap *state.StateSnapshot, plan *structs.Plan, nodeID stri
 			return true, "", nil
 		}
 		return false, "node is disconnected and contains invalid updates", nil
+	} else if node.Status == structs.NodeStatusDown {
+		if isValidForDownNode(plan, node.ID) {
+			return true, "", nil
+		}
+		return false, "node is down and contains invalid updates", nil
 	} else if node.Status != structs.NodeStatusReady {
 		return false, "node is not ready for placements", nil
 	}
@@ -790,3 +795,17 @@ func isValidForDisconnectedNode(plan *structs.Plan, nodeID string) bool {
 
 	return true
 }
+
+// The plan is only valid for a node down if it only contains
+// updates to mark allocations as unknown and those allocations are configured
+// as non reschedulables when lost or if the allocs are being updated to lost.
+func isValidForDownNode(plan *structs.Plan, nodeID string) bool {
+	for _, alloc := range plan.NodeAllocation[nodeID] {
+		if !(alloc.ClientStatus == structs.AllocClientStatusUnknown && alloc.PreventRescheduleOnLost()) &&
+			(alloc.ClientStatus != structs.AllocClientStatusLost) {
+			return false
+		}
+	}
+
+	return true
+}