Add context and timeouts

- Add timeout flag to apply, destory, and preview cmds
- Default to no timeout
- Add context to TaskContext
- Pass context through to all functions that use it
- Rename prune-timeout flag to delete-timeout and add actual impl
- Rename PruneOptions to Pruner (there's already prune.Options)
- Rename PropagationPolicy to DeletionPropagationPolicy consistently
- Wrap func calls and defs for readability
- Remove Prune bool from solver.Options to simplify
  AppendPruneWaitTasks to always prune.
  Change Applier to only call AppendPruneWaitTasks if NoPrune.
  Change Destroyer to always call AppendPruneWaitTasks.
- Replace task.ClearTimeout with context.Cancel
- Add task.OnStatusEvent to fix WaitTask leaky abstraction.
  baseRunner was calling private funcs from WaitTask.
- Move ResourceStatusCollector into TaskContext. It doesn't need to
  persist between runs. Moving it simplifies baseRunner.
  This required making some of the collector funcs public.
- Remove Identifier from collector.ResourceStatus. Redundant with id.
- Add collector.Get/Put funcs for encapsulation and thread-safety
- Add locking to ResourceStatusCollector because a test was complaining
  about a race condition when updating status from another goroutine,
  like the poller does.
- Add logging to taskrunner tests for easier debugging concurrency
- Update task tests for cancel and timeout
- Rename options vars to "opts" for consistency
- Rewrite WaitTask context/timeout handling.
  TaskChannel is now only written to from one place, gated by an
  internal error channel and sync.Once.
  WaitTask now distinguishes between wait task timeout and parent
  context timeout.
- Update task tests to reflect new behavior with context cancel/timeout

cmd/apply/cmdapply.go

@@ -52,13 +52,15 @@ func GetApplyRunner(factory cmdutil.Factory, invFactory inventory.InventoryClien
 		"Timeout threshold for waiting for all resources to reach the Current status.")
 	cmd.Flags().BoolVar(&r.noPrune, "no-prune", r.noPrune,
 		"If true, do not prune previously applied objects.")
-	cmd.Flags().StringVar(&r.prunePropagationPolicy, "prune-propagation-policy",
-		"Background", "Propagation policy for pruning")
-	cmd.Flags().DurationVar(&r.pruneTimeout, "prune-timeout", time.Duration(0),
-		"Timeout threshold for waiting for all pruned resources to be deleted")
+	cmd.Flags().StringVar(&r.deletePolicy, "delete-propagation-policy",
+		"Background", "Propagation policy for deletion")
+	cmd.Flags().DurationVar(&r.deleteTimeout, "delete-timeout", time.Duration(0),
+		"Timeout threshold for waiting for all deleted resources to complete deletion")
 	cmd.Flags().StringVar(&r.inventoryPolicy, flagutils.InventoryPolicyFlag, flagutils.InventoryPolicyStrict,
 		"It determines the behavior when the resources don't belong to current inventory. Available options "+
 			fmt.Sprintf("%q and %q.", flagutils.InventoryPolicyStrict, flagutils.InventoryPolicyAdopt))
+	cmd.Flags().DurationVar(&r.timeout, "timeout", time.Duration(0),
+		"Timeout threshold for command execution")
 
 	r.Command = cmd
 	return r
@@ -77,18 +79,30 @@ type ApplyRunner struct {
 	invFactory inventory.InventoryClientFactory
 	loader     manifestreader.ManifestLoader
 
-	serverSideOptions      common.ServerSideOptions
-	output                 string
-	period                 time.Duration
-	reconcileTimeout       time.Duration
-	noPrune                bool
-	prunePropagationPolicy string
-	pruneTimeout           time.Duration
-	inventoryPolicy        string
+	serverSideOptions common.ServerSideOptions
+	output            string
+	period            time.Duration
+	reconcileTimeout  time.Duration
+	noPrune           bool
+	deletePolicy      string
+	deleteTimeout     time.Duration
+	inventoryPolicy   string
+	timeout           time.Duration
 }
 
 func (r *ApplyRunner) RunE(cmd *cobra.Command, args []string) error {
-	prunePropPolicy, err := flagutils.ConvertPropagationPolicy(r.prunePropagationPolicy)
+	// If specified, cancel with timeout.
+	// Otherwise, cancel when completed to clean up timer.
+	ctx := cmd.Context()
+	var cancel func()
+	if r.timeout != 0 {
+		ctx, cancel = context.WithTimeout(ctx, r.timeout)
+	} else {
+		ctx, cancel = context.WithCancel(ctx)
+	}
+	defer cancel()
+
+	deletePolicy, err := flagutils.ConvertPropagationPolicy(r.deletePolicy)
 	if err != nil {
 		return err
 	}
@@ -102,7 +116,7 @@ func (r *ApplyRunner) RunE(cmd *cobra.Command, args []string) error {
 	// we do need status events event if we are not waiting for status. The
 	// printers should be updated to handle this.
 	var printStatusEvents bool
-	if r.reconcileTimeout != time.Duration(0) || r.pruneTimeout != time.Duration(0) {
+	if r.reconcileTimeout != time.Duration(0) || r.deleteTimeout != time.Duration(0) {
 		printStatusEvents = true
 	}
 
@@ -147,18 +161,19 @@ func (r *ApplyRunner) RunE(cmd *cobra.Command, args []string) error {
 	if err != nil {
 		return err
 	}
-	ch := a.Run(context.Background(), inv, objs, apply.Options{
+
+	ch := a.Run(ctx, inv, objs, apply.Options{
 		ServerSideOptions: r.serverSideOptions,
 		PollInterval:      r.period,
 		ReconcileTimeout:  r.reconcileTimeout,
 		// If we are not waiting for status, tell the applier to not
 		// emit the events.
-		EmitStatusEvents:       printStatusEvents,
-		NoPrune:                r.noPrune,
-		DryRunStrategy:         common.DryRunNone,
-		PrunePropagationPolicy: prunePropPolicy,
-		PruneTimeout:           r.pruneTimeout,
-		InventoryPolicy:        inventoryPolicy,
+		EmitStatusEvents:          printStatusEvents,
+		NoPrune:                   r.noPrune,
+		DryRunStrategy:            common.DryRunNone,
+		DeletionPropagationPolicy: deletePolicy,
+		DeleteTimeout:             r.deleteTimeout,
+		InventoryPolicy:           inventoryPolicy,
 	})
 
 	// The printer will print updates from the channel. It will block

cmd/destroy/cmddestroy.go

@@ -4,6 +4,7 @@
 package destroy
 
 import (
+	"context"
 	"fmt"
 	"strings"
 	"time"
@@ -46,6 +47,8 @@ func GetDestroyRunner(factory cmdutil.Factory, invFactory inventory.InventoryCli
 		"Timeout threshold for waiting for all deleted resources to complete deletion")
 	cmd.Flags().StringVar(&r.deletePropagationPolicy, "delete-propagation-policy",
 		"Background", "Propagation policy for deletion")
+	cmd.Flags().DurationVar(&r.timeout, "timeout", time.Duration(0),
+		"Timeout threshold for command execution")
 
 	r.Command = cmd
 	return r
@@ -70,9 +73,21 @@ type DestroyRunner struct {
 	deleteTimeout           time.Duration
 	deletePropagationPolicy string
 	inventoryPolicy         string
+	timeout                 time.Duration
 }
 
 func (r *DestroyRunner) RunE(cmd *cobra.Command, args []string) error {
+	// If specified, cancel with timeout.
+	// Otherwise, cancel when completed to clean up timer.
+	ctx := cmd.Context()
+	var cancel func()
+	if r.timeout != 0 {
+		ctx, cancel = context.WithTimeout(ctx, r.timeout)
+	} else {
+		ctx, cancel = context.WithCancel(ctx)
+	}
+	defer cancel()
+
 	deletePropPolicy, err := flagutils.ConvertPropagationPolicy(r.deletePropagationPolicy)
 	if err != nil {
 		return err
@@ -117,7 +132,8 @@ func (r *DestroyRunner) RunE(cmd *cobra.Command, args []string) error {
 	// Run the destroyer. It will return a channel where we can receive updates
 	// to keep track of progress and any issues.
 	printStatusEvents := r.deleteTimeout != time.Duration(0)
-	ch := d.Run(inv, apply.DestroyerOptions{
+
+	ch := d.Run(ctx, inv, apply.DestroyerOptions{
 		DeleteTimeout:           r.deleteTimeout,
 		DeletePropagationPolicy: deletePropPolicy,
 		InventoryPolicy:         inventoryPolicy,

cmd/preview/cmdpreview.go

@@ -7,6 +7,7 @@ import (
 	"context"
 	"fmt"
 	"strings"
+	"time"
 
 	"github.com/spf13/cobra"
 	"k8s.io/cli-runtime/pkg/genericclioptions"
@@ -57,6 +58,8 @@ func GetPreviewRunner(factory cmdutil.Factory, invFactory inventory.InventoryCli
 	cmd.Flags().StringVar(&r.inventoryPolicy, flagutils.InventoryPolicyFlag, flagutils.InventoryPolicyStrict,
 		"It determines the behavior when the resources don't belong to current inventory. Available options "+
 			fmt.Sprintf("%q and %q.", flagutils.InventoryPolicyStrict, flagutils.InventoryPolicyAdopt))
+	cmd.Flags().DurationVar(&r.timeout, "timeout", time.Duration(0),
+		"Timeout threshold for command execution")
 
 	r.Command = cmd
 	return r
@@ -80,10 +83,22 @@ type PreviewRunner struct {
 	serverSideOptions common.ServerSideOptions
 	output            string
 	inventoryPolicy   string
+	timeout           time.Duration
 }
 
 // RunE is the function run from the cobra command.
 func (r *PreviewRunner) RunE(cmd *cobra.Command, args []string) error {
+	// If specified, cancel with timeout.
+	// Otherwise, cancel when completed to clean up timer.
+	ctx := cmd.Context()
+	var cancel func()
+	if r.timeout != 0 {
+		ctx, cancel = context.WithTimeout(ctx, r.timeout)
+	} else {
+		ctx, cancel = context.WithCancel(ctx)
+	}
+	defer cancel()
+
 	var ch <-chan event.Event
 
 	drs := common.DryRunClient
@@ -139,9 +154,6 @@ func (r *PreviewRunner) RunE(cmd *cobra.Command, args []string) error {
 			return err
 		}
 
-		// Create a context
-		ctx := context.Background()
-
 		// Run the applier. It will return a channel where we can receive updates
 		// to keep track of progress and any issues.
 		ch = a.Run(ctx, inv, objs, apply.Options{
@@ -156,7 +168,7 @@ func (r *PreviewRunner) RunE(cmd *cobra.Command, args []string) error {
 		if err != nil {
 			return err
 		}
-		ch = d.Run(inv, apply.DestroyerOptions{
+		ch = d.Run(ctx, inv, apply.DestroyerOptions{
 			InventoryPolicy: inventoryPolicy,
 			DryRunStrategy:  drs,
 		})

cmd/status/cmdstatus.go

@@ -72,6 +72,17 @@ type StatusRunner struct {
 // poller to compute status for each of the resources. One of the printer
 // implementations takes care of printing the output.
 func (r *StatusRunner) runE(cmd *cobra.Command, args []string) error {
+	// If specified, cancel with timeout.
+	// Otherwise, cancel when completed to clean up timer.
+	ctx := cmd.Context()
+	var cancel func()
+	if r.timeout != 0 {
+		ctx, cancel = context.WithTimeout(ctx, r.timeout)
+	} else {
+		ctx, cancel = context.WithCancel(ctx)
+	}
+	defer cancel()
+
 	_, err := common.DemandOneDirectory(args)
 	if err != nil {
 		return err
@@ -125,17 +136,6 @@ func (r *StatusRunner) runE(cmd *cobra.Command, args []string) error {
 		return fmt.Errorf("error creating printer: %w", err)
 	}
 
-	// If the user has specified a timeout, we create a context with timeout,
-	// otherwise we create a context with cancel.
-	ctx := context.Background()
-	var cancel func()
-	if r.timeout != 0 {
-		ctx, cancel = context.WithTimeout(ctx, r.timeout)
-	} else {
-		ctx, cancel = context.WithCancel(ctx)
-	}
-	defer cancel()
-
 	// Choose the appropriate ObserverFunc based on the criteria for when
 	// the command should exit.
 	var cancelFunc collector.ObserverFunc

cmd/status/cmdstatus_test.go

@@ -236,12 +236,15 @@ deployment.apps/foo is InProgress: inProgress
 				timeout:   tc.timeout,
 			}
 
-			cmd := &cobra.Command{}
+			cmd := &cobra.Command{
+				RunE: runner.runE,
+			}
 			cmd.SetIn(strings.NewReader(tc.input))
 			var buf bytes.Buffer
 			cmd.SetOut(&buf)
 
-			err := runner.runE(cmd, []string{})
+			// execute with cobra to handle setting the default context
+			err := cmd.Execute()
 
 			if tc.expectedErrMsg != "" {
 				if !assert.Error(t, err) {