Add global applier & destroyer timeouts

- Applier context existed, but was untested. Added cancellation test.
- Destroyer context added, with cancellation test.
- Added a new timeout flag to all kapply cmds that use the applier or
  destroyer.
- Added testutil AssertEqual and AssertNotEqual to enable event
  list comparisons in unit tests (which don't otherwise use
  Gomega-style Expect tests).
- Added a few logs to help with debugging test failures

cmd/apply/cmdapply.go

@@ -48,6 +48,8 @@ func GetApplyRunner(factory cmdutil.Factory, invFactory inventory.InventoryClien
 		fmt.Sprintf("Output format, must be one of %s", strings.Join(printers.SupportedPrinters(), ",")))
 	cmd.Flags().DurationVar(&r.period, "poll-period", 2*time.Second,
 		"Polling period for resource statuses.")
+	cmd.Flags().DurationVar(&r.timeout, "timeout", time.Duration(0),
+		"Timeout threshold for command execution")
 	cmd.Flags().DurationVar(&r.reconcileTimeout, "reconcile-timeout", time.Duration(0),
 		"Timeout threshold for waiting for all resources to reach the Current status.")
 	cmd.Flags().BoolVar(&r.noPrune, "no-prune", r.noPrune,
@@ -80,6 +82,7 @@ type ApplyRunner struct {
 	serverSideOptions      common.ServerSideOptions
 	output                 string
 	period                 time.Duration
+	timeout                time.Duration
 	reconcileTimeout       time.Duration
 	noPrune                bool
 	prunePropagationPolicy string
@@ -88,6 +91,17 @@ type ApplyRunner struct {
 }
 
 func (r *ApplyRunner) RunE(cmd *cobra.Command, args []string) error {
+	// If specified, cancel with timeout.
+	// Otherwise, cancel when completed to clean up.
+	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()
+
 	prunePropPolicy, err := flagutils.ConvertPropagationPolicy(r.prunePropagationPolicy)
 	if err != nil {
 		return err
@@ -147,7 +161,7 @@ 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,

cmd/destroy/cmddestroy.go

@@ -4,6 +4,7 @@
 package destroy
 
 import (
+	"context"
 	"fmt"
 	"strings"
 	"time"
@@ -42,6 +43,8 @@ func GetDestroyRunner(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, %q and %q.", flagutils.InventoryPolicyStrict, flagutils.InventoryPolicyAdopt, flagutils.InventoryPolicyForceAdopt))
+	cmd.Flags().DurationVar(&r.timeout, "timeout", time.Duration(0),
+		"Timeout threshold for command execution")
 	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.deletePropagationPolicy, "delete-propagation-policy",
@@ -67,12 +70,24 @@ type DestroyRunner struct {
 	loader     manifestreader.ManifestLoader
 
 	output                  string
+	timeout                 time.Duration
 	deleteTimeout           time.Duration
 	deletePropagationPolicy string
 	inventoryPolicy         string
 }
 
 func (r *DestroyRunner) RunE(cmd *cobra.Command, args []string) error {
+	// If specified, cancel with timeout.
+	// Otherwise, cancel when completed to clean up.
+	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,7 @@ 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, %q and %q.", flagutils.InventoryPolicyStrict, flagutils.InventoryPolicyAdopt, flagutils.InventoryPolicyForceAdopt))
+	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.
+	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

@@ -42,7 +42,7 @@ func GetStatusRunner(factory cmdutil.Factory, invFactory inventory.InventoryClie
 		"When to stop polling. Must be one of 'known', 'current', 'deleted', or 'forever'.")
 	c.Flags().StringVar(&r.output, "output", "events", "Output format.")
 	c.Flags().DurationVar(&r.timeout, "timeout", 0,
-		"How long to wait before exiting")
+		"Timeout threshold for command execution")
 
 	r.Command = c
 	return r
@@ -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.
+	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()
+
 	_, 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