provide -no-shutdown-delay flag for job/alloc stop

Some operators use very long group/task `shutdown_delay` settings to
safely drain network connections to their workloads after service
deregistration. But during incident response, they may want to cause
that drain to be skipped so they can quickly shed load.

Provide a `-no-shutdown-delay` flag on the `nomad alloc stop` and
`nomad job stop` commands that bypasses the delay. This sets a new
desired transition state on the affected allocations that the
allocation/task runner will identify during pre-kill on the client.

Note (as documented here) that using this flag will almost always result in failed
inbound network connections for workloads as the tasks will exit
before clients receive updated service discovery information and
won't be gracefully drained.

Also note that an allocation that is already waiting for the
`shutdown_delay` will not be interrupted by running `job stop` or
`alloc stop` again with `-no-shutdown-delay` set; once we've set an
allocation to stop the allocation runner ignores further updates.

api/jobs.go

@@ -299,6 +299,11 @@ type DeregisterOptions struct {
 	// is useful when an operator wishes to push through a job deregistration
 	// in busy clusters with a large evaluation backlog.
 	EvalPriority int
+
+	// NoShutdownDelay, if set to true, will override the group and
+	// task shutdown_delay configuration and set it to "0s" for any
+	// allocations stopped as a result of this Deregister call.
+	NoShutdownDelay bool
 }
 
 // DeregisterOpts is used to remove an existing job. See DeregisterOptions
@@ -312,8 +317,8 @@ func (j *Jobs) DeregisterOpts(jobID string, opts *DeregisterOptions, q *WriteOpt
 	// Protect against nil opts. url.Values expects a string, and so using
 	// fmt.Sprintf is the best way to do this.
 	if opts != nil {
-		endpoint += fmt.Sprintf("?purge=%t&global=%t&eval_priority=%v",
-			opts.Purge, opts.Global, opts.EvalPriority)
+		endpoint += fmt.Sprintf("?purge=%t&global=%t&eval_priority=%v&no_shutdown_delay=%t",
+			opts.Purge, opts.Global, opts.EvalPriority, opts.NoShutdownDelay)
 	}
 
 	wm, err := j.client.delete(endpoint, &resp, q)

client/allocrunner/alloc_runner_hooks.go

@@ -338,7 +338,7 @@ func (ar *allocRunner) preKillHooks() {
 			ar.logger.Trace("running alloc pre shutdown hook", "name", name, "start", start)
 		}
 
-		pre.PreKill()
+		pre.PreKill(ar.alloc)
 
 		if ar.logger.IsTrace() {
 			end := time.Now()

client/allocrunner/groupservice_hook.go

@@ -29,7 +29,6 @@ type groupServiceHook struct {
 	consulClient        consul.ConsulServiceAPI
 	consulNamespace     string
 	prerun              bool
-	delay               time.Duration
 	deregistered        bool
 	networkStatusGetter networkStatusGetter
 
@@ -41,6 +40,7 @@ type groupServiceHook struct {
 	networks       structs.Networks
 	ports          structs.AllocatedPorts
 	taskEnvBuilder *taskenv.Builder
+	delay          time.Duration
 
 	// Since Update() may be called concurrently with any other hook all
 	// hook methods must be fully serialized
@@ -164,24 +164,24 @@ func (h *groupServiceHook) PreTaskRestart() error {
 		h.mu.Unlock()
 	}()
 
-	h.preKillLocked()
+	h.preKillLocked(nil)
 	return h.prerunLocked()
 }
 
-func (h *groupServiceHook) PreKill() {
+func (h *groupServiceHook) PreKill(alloc *structs.Allocation) {
 	h.mu.Lock()
 	defer h.mu.Unlock()
-	h.preKillLocked()
+	h.preKillLocked(alloc)
 }
 
 // implements the PreKill hook but requires the caller hold the lock
-func (h *groupServiceHook) preKillLocked() {
+func (h *groupServiceHook) preKillLocked(alloc *structs.Allocation) {
 	// If we have a shutdown delay deregister group services and then wait
 	// before continuing to kill tasks.
 	h.deregister()
 	h.deregistered = true
 
-	if h.delay == 0 {
+	if h.delay == 0 || (alloc != nil && alloc.DesiredTransition.ShouldIgnoreShutdownDelay()) {
 		return
 	}
 

client/allocrunner/interfaces/runner_lifecycle.go

@@ -22,7 +22,7 @@ type RunnerPrerunHook interface {
 type RunnerPreKillHook interface {
 	RunnerHook
 
-	PreKill()
+	PreKill(*structs.Allocation)
 }
 
 // RunnerPostrunHooks are executed after calling TaskRunner.Run, even for

client/allocrunner/taskrunner/task_runner.go

@@ -889,7 +889,7 @@ func (tr *TaskRunner) handleKill(resultCh <-chan *drivers.ExitResult) *drivers.E
 	// Wait for task ShutdownDelay after running prekill hooks
 	// This allows for things like service de-registration to run
 	// before waiting to kill task
-	if delay := tr.Task().ShutdownDelay; delay != 0 {
+	if delay := tr.Task().ShutdownDelay; delay != 0 && !tr.alloc.DesiredTransition.ShouldIgnoreShutdownDelay() {
 		tr.logger.Debug("waiting before killing task", "shutdown_delay", delay)
 
 		select {

client/allocrunner/taskrunner/task_runner_test.go

@@ -14,6 +14,10 @@ import (
 	"time"
 
 	"github.com/golang/snappy"
+	"github.com/kr/pretty"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
 	"github.com/hashicorp/nomad/client/allocdir"
 	"github.com/hashicorp/nomad/client/allocrunner/interfaces"
 	"github.com/hashicorp/nomad/client/config"
@@ -26,16 +30,14 @@ import (
 	agentconsul "github.com/hashicorp/nomad/command/agent/consul"
 	mockdriver "github.com/hashicorp/nomad/drivers/mock"
 	"github.com/hashicorp/nomad/drivers/rawexec"
+	"github.com/hashicorp/nomad/helper"
 	"github.com/hashicorp/nomad/helper/testlog"
 	"github.com/hashicorp/nomad/helper/uuid"
 	"github.com/hashicorp/nomad/nomad/mock"
 	"github.com/hashicorp/nomad/nomad/structs"
 	"github.com/hashicorp/nomad/plugins/device"
 	"github.com/hashicorp/nomad/plugins/drivers"
 	"github.com/hashicorp/nomad/testutil"
-	"github.com/kr/pretty"
-	"github.com/stretchr/testify/assert"
-	"github.com/stretchr/testify/require"
 )
 
 type MockTaskStateUpdater struct {
@@ -996,6 +998,74 @@ WAIT:
 	}
 }
 
+// TestTaskRunner_NoShutdownDelay asserts services are removed from
+// Consul and tasks are killed without waiting for ${shutdown_delay}
+// when the alloc has the NoShutdownDelay transition flag set.
+func TestTaskRunner_NoShutdownDelay(t *testing.T) {
+	t.Parallel()
+
+	alloc := mock.Alloc()
+	alloc.DesiredTransition = structs.DesiredTransition{NoShutdownDelay: helper.BoolToPtr(true)}
+	task := alloc.Job.TaskGroups[0].Tasks[0]
+	task.Services[0].Tags = []string{"tag1"}
+	task.Services = task.Services[:1] // only need 1 for this test
+	task.Driver = "mock_driver"
+	task.Config = map[string]interface{}{
+		"run_for": "1000s",
+	}
+
+	// set the shutdown delay high; we'll timeout well before this
+	task.ShutdownDelay = 1 * time.Minute
+
+	tr, conf, cleanup := runTestTaskRunner(t, alloc, task.Name)
+	defer cleanup()
+
+	mockConsul := conf.Consul.(*consulapi.MockConsulServiceClient)
+
+	testWaitForTaskToStart(t, tr)
+
+	testutil.WaitForResult(func() (bool, error) {
+		ops := mockConsul.GetOps()
+		if n := len(ops); n != 1 {
+			return false, fmt.Errorf("expected 1 consul operation. Found %d", n)
+		}
+		return ops[0].Op == "add", fmt.Errorf("consul operation was not a registration: %#v", ops[0])
+	}, func(err error) {
+		t.Fatalf("err: %v", err)
+	})
+
+	killed := make(chan error)
+	go func() {
+		err := tr.Kill(context.Background(), structs.NewTaskEvent("test"))
+		killed <- err
+	}()
+
+	// Wait for first de-registration call. Note that unlike
+	// TestTaskRunner_ShutdownDelay, we're racing with task exit
+	// and can't assert that we only get the first deregistration op
+	// (from serviceHook.PreKill).
+	testutil.WaitForResult(func() (bool, error) {
+		ops := mockConsul.GetOps()
+		if n := len(ops); n < 2 {
+			return false, fmt.Errorf("expected at least 2 consul operations.")
+		}
+		return ops[1].Op == "remove", fmt.Errorf(
+			"consul operation was not a deregistration: %#v", ops[1])
+	}, func(err error) {
+		t.Fatalf("err: %v", err)
+	})
+
+	// Wait for the task to exit
+	select {
+	case <-tr.WaitCh():
+	case <-time.After(time.Duration(testutil.TestMultiplier()) * time.Second):
+		t.Fatalf("task kill did not ignore shutdown delay")
+	}
+
+	err := <-killed
+	require.NoError(t, err, "killing task returned unexpected error")
+}
+
 // TestTaskRunner_Dispatch_Payload asserts that a dispatch job runs and the
 // payload was written to disk.
 func TestTaskRunner_Dispatch_Payload(t *testing.T) {

command/agent/alloc_endpoint.go

@@ -138,8 +138,18 @@ func (s *HTTPServer) allocStop(allocID string, resp http.ResponseWriter, req *ht
 		return nil, CodedError(405, ErrInvalidMethod)
 	}
 
+	noShutdownDelay := false
+	if noShutdownDelayQS := req.URL.Query().Get("no_shutdown_delay"); noShutdownDelayQS != "" {
+		var err error
+		noShutdownDelay, err = strconv.ParseBool(noShutdownDelayQS)
+		if err != nil {
+			return nil, fmt.Errorf("no_shutdown_delay
 value is not a boolean: %v", err)
+		}
+	}
+
 	sr := &structs.AllocStopRequest{
-		AllocID: allocID,
+		AllocID:         allocID,
+		NoShutdownDelay: noShutdownDelay,
 	}
 	s.parseWriteRequest(req, &sr.WriteRequest)
 

command/agent/job_endpoint.go

@@ -453,6 +453,18 @@ func (s *HTTPServer) jobDelete(resp http.ResponseWriter, req *http.Request,
 		return nil, err
 	}
 
+	// Identify the no_shutdown_delay query param and parse.
+	noShutdownDelayStr := req.URL.Query().Get("no_shutdown_delay")
+	var noShutdownDelay bool
+	if noShutdownDelayStr != "" {
+		var err error
+		noShutdownDelay, err = strconv.ParseBool(noShutdownDelayStr)
+		if err != nil {
+			return nil, fmt.Errorf("Failed to parse value of %qq (%v) as a bool: %v", "no_shutdown_delay", noShutdownDelayStr, err)
+		}
+	}
+	args.NoShutdownDelay = noShutdownDelay
+
 	// Validate the evaluation priority if the user supplied a non-default
 	// value. It's more efficient to do it here, within the agent rather than
 	// sending a bad request for the server to reject.

command/alloc_stop.go

@@ -38,6 +38,14 @@ Stop Specific Options:
     screen, which can be used to examine the rescheduling evaluation using the
     eval-status command.
 
+  -no-shutdown-delay
+	Ignore the the group and task shutdown_delay configuration so there is no
+    delay between service deregistration and task shutdown. Note that using
+    this flag will result in failed network connections to the allocation
+    being stopped. Also note that an allocation that is already waiting for
+    the shutdown_delay will not be interrupted by running alloc stop again with
+    this flag.
+
   -verbose
     Show full information.
 `
@@ -47,12 +55,13 @@ Stop Specific Options:
 func (c *AllocStopCommand) Name() string { return "alloc stop" }
 
 func (c *AllocStopCommand) Run(args []string) int {
-	var detach, verbose bool
+	var detach, verbose, noShutdownDelay bool
 
 	flags := c.Meta.FlagSet(c.Name(), FlagSetClient)
 	flags.Usage = func() { c.Ui.Output(c.Help()) }
 	flags.BoolVar(&detach, "detach", false, "")
 	flags.BoolVar(&verbose, "verbose", false, "")
+	flags.BoolVar(&noShutdownDelay, "no-shutdown-delay", false, "")
 
 	if err := flags.Parse(args); err != nil {
 		return 1
@@ -115,7 +124,12 @@ func (c *AllocStopCommand) Run(args []string) int {
 		return 1
 	}
 
-	resp, err := client.Allocations().Stop(alloc, nil)
+	var opts *api.QueryOptions
+	if noShutdownDelay {
+		opts = &api.QueryOptions{Params: map[string]string{"no_shutdown_delay": "true"}}
+	}
+
+	resp, err := client.Allocations().Stop(alloc, opts)
 	if err != nil {
 		c.Ui.Error(fmt.Sprintf("Error stopping allocation: %s", err))
 		return 1

command/job_stop.go

@@ -43,14 +43,22 @@ Stop Options:
     Override the priority of the evaluations produced as a result of this job
     deregistration. By default, this is set to the priority of the job.
 
-  -purge
-    Purge is used to stop the job and purge it from the system. If not set, the
-    job will still be queryable and will be purged by the garbage collector.
-
   -global
     Stop a multi-region job in all its regions. By default job stop will stop
     only a single region at a time. Ignored for single-region jobs.
 
+  -no-shutdown-delay
+	Ignore the the group and task shutdown_delay configuration so there is no
+    delay between service deregistration and task shutdown. Note that using
+    this flag will result in failed network connections to the allocation
+    being stopped. Also note that an allocation that is already waiting for
+    the shutdown_delay will not be interrupted by running job stop again with
+    this flag.
+
+  -purge
+    Purge is used to stop the job and purge it from the system. If not set, the
+    job will still be queryable and will be purged by the garbage collector.
+
   -yes
     Automatic yes to prompts.
 
@@ -67,12 +75,13 @@ func (c *JobStopCommand) Synopsis() string {
 func (c *JobStopCommand) AutocompleteFlags() complete.Flags {
 	return mergeAutocompleteFlags(c.Meta.AutocompleteFlags(FlagSetClient),
 		complete.Flags{
-			"-detach":        complete.PredictNothing,
-			"-eval-priority": complete.PredictNothing,
-			"-purge":         complete.PredictNothing,
-			"-global":        complete.PredictNothing,
-			"-yes":           complete.PredictNothing,
-			"-verbose":       complete.PredictNothing,
+			"-detach":            complete.PredictNothing,
+			"-eval-priority":     complete.PredictNothing,
+			"-purge":             complete.PredictNothing,
+			"-global":            complete.PredictNothing,
+			"-no-shutdown-delay": complete.PredictNothing,
+			"-yes":               complete.PredictNothing,
+			"-verbose":           complete.PredictNothing,
 		})
 }
 
@@ -94,14 +103,15 @@ func (c *JobStopCommand) AutocompleteArgs() complete.Predictor {
 func (c *JobStopCommand) Name() string { return "job stop" }
 
 func (c *JobStopCommand) Run(args []string) int {
-	var detach, purge, verbose, global, autoYes bool
+	var detach, purge, verbose, global, autoYes, noShutdownDelay bool
 	var evalPriority int
 
 	flags := c.Meta.FlagSet(c.Name(), FlagSetClient)
 	flags.Usage = func() { c.Ui.Output(c.Help()) }
 	flags.BoolVar(&detach, "detach", false, "")
 	flags.BoolVar(&verbose, "verbose", false, "")
 	flags.BoolVar(&global, "global", false, "")
+	flags.BoolVar(&noShutdownDelay, "no-shutdown-delay", false, "")
 	flags.BoolVar(&autoYes, "yes", false, "")
 	flags.BoolVar(&purge, "purge", false, "")
 	flags.IntVar(&evalPriority, "eval-priority", 0, "")
@@ -199,7 +209,7 @@ func (c *JobStopCommand) Run(args []string) int {
 	}
 
 	// Invoke the stop
-	opts := &api.DeregisterOptions{Purge: purge, Global: global, EvalPriority: evalPriority}
+	opts := &api.DeregisterOptions{Purge: purge, Global: global, EvalPriority: evalPriority, NoShutdownDelay: noShutdownDelay}
 	wq := &api.WriteOptions{Namespace: jobs[0].JobSummary.Namespace}
 	evalID, _, err := client.Jobs().DeregisterOpts(*job.ID, opts, wq)
 	if err != nil {

nomad/alloc_endpoint.go

@@ -320,7 +320,8 @@ func (a *Alloc) Stop(args *structs.AllocStopRequest, reply *structs.AllocStopRes
 		Evals: []*structs.Evaluation{eval},
 		Allocs: map[string]*structs.DesiredTransition{
 			args.AllocID: {
-				Migrate: helper.BoolToPtr(true),
+				Migrate:         helper.BoolToPtr(true),
+				NoShutdownDelay: helper.BoolToPtr(args.NoShutdownDelay),
 			},
 		},
 	}