expose apply concurrency as a flag

pkg/kapp/clusterapply/applying_changes.go

@@ -8,15 +8,20 @@ import (
 	"github.com/k14s/kapp/pkg/kapp/util"
 )
 
+type ApplyingChangesOpts struct {
+	Concurrency int
+}
+
 type ApplyingChanges struct {
 	numTotal             int // for ui
+	opts                 ApplyingChangesOpts
 	applied              map[*ctldgraph.Change]struct{}
 	clusterChangeFactory ClusterChangeFactory
 	ui                   UI
 }
 
-func NewApplyingChanges(numTotal int, clusterChangeFactory ClusterChangeFactory, ui UI) *ApplyingChanges {
-	return &ApplyingChanges{numTotal, map[*ctldgraph.Change]struct{}{}, clusterChangeFactory, ui}
+func NewApplyingChanges(numTotal int, opts ApplyingChangesOpts, clusterChangeFactory ClusterChangeFactory, ui UI) *ApplyingChanges {
+	return &ApplyingChanges{numTotal, opts, map[*ctldgraph.Change]struct{}{}, clusterChangeFactory, ui}
 }
 
 func (c *ApplyingChanges) Apply(allChanges []*ctldgraph.Change) ([]WaitingChange, error) {
@@ -45,7 +50,7 @@ func (c *ApplyingChanges) Apply(allChanges []*ctldgraph.Change) ([]WaitingChange
 	// Example errors w/o throttling:
 	// - "...: grpc: the client connection is closing (reason: )"
 	// - "...: context canceled (reason: )"
-	applyThrottle := util.NewThrottle(5)
+	applyThrottle := util.NewThrottle(c.opts.Concurrency)
 
 	for _, change := range nonAppliedChanges {
 		c.markApplied(change)

pkg/kapp/clusterapply/cluster_change_set.go

@@ -2,15 +2,14 @@ package clusterapply
 
 import (
 	"fmt"
-	"time"
 
 	ctldiff "github.com/k14s/kapp/pkg/kapp/diff"
 	ctldgraph "github.com/k14s/kapp/pkg/kapp/diffgraph"
 )
 
 type ClusterChangeSetOpts struct {
-	WaitTimeout       time.Duration
-	WaitCheckInterval time.Duration
+	ApplyingChangesOpts
+	WaitingChangesOpts
 }
 
 type ClusterChangeSet struct {
@@ -81,8 +80,9 @@ func (c ClusterChangeSet) Apply(changesGraph *ctldgraph.ChangeGraph) error {
 	expectedNumChanges := len(changesGraph.All())
 
 	blockedChanges := ctldgraph.NewBlockedChanges(changesGraph)
-	applyingChanges := NewApplyingChanges(expectedNumChanges, c.clusterChangeFactory, c.ui)
-	waitingChanges := NewWaitingChanges(expectedNumChanges, c.opts, c.ui)
+	applyingChanges := NewApplyingChanges(
+		expectedNumChanges, c.opts.ApplyingChangesOpts, c.clusterChangeFactory, c.ui)
+	waitingChanges := NewWaitingChanges(expectedNumChanges, c.opts.WaitingChangesOpts, c.ui)
 
 	for {
 		appliedChanges, err := applyingChanges.Apply(blockedChanges.Unblocked())

pkg/kapp/clusterapply/waiting_changes.go

@@ -7,11 +7,16 @@ import (
 	ctldgraph "github.com/k14s/kapp/pkg/kapp/diffgraph"
 )
 
+type WaitingChangesOpts struct {
+	Timeout       time.Duration
+	CheckInterval time.Duration
+}
+
 type WaitingChanges struct {
 	numTotal       int // for ui
 	numWaited      int // for ui
 	trackedChanges []WaitingChange
-	opts           ClusterChangeSetOpts
+	opts           WaitingChangesOpts
 	ui             UI
 }
 
@@ -20,7 +25,7 @@ type WaitingChange struct {
 	Cluster *ClusterChange
 }
 
-func NewWaitingChanges(numTotal int, opts ClusterChangeSetOpts, ui UI) *WaitingChanges {
+func NewWaitingChanges(numTotal int, opts WaitingChangesOpts, ui UI) *WaitingChanges {
 	return &WaitingChanges{numTotal, 0, nil, opts, ui}
 }
 
@@ -76,11 +81,11 @@ func (c *WaitingChanges) WaitForAny() ([]WaitingChange, error) {
 			return doneChanges, nil
 		}
 
-		if time.Now().Sub(startTime) > c.opts.WaitTimeout {
-			return nil, fmt.Errorf("timed out waiting after %s", c.opts.WaitTimeout)
+		if time.Now().Sub(startTime) > c.opts.Timeout {
+			return nil, fmt.Errorf("timed out waiting after %s", c.opts.Timeout)
 		}
 
-		time.Sleep(c.opts.WaitCheckInterval)
+		time.Sleep(c.opts.CheckInterval)
 	}
 }
 

pkg/kapp/cmd/app/apply_flags.go

@@ -36,16 +36,17 @@ func (s *ApplyFlags) SetWithDefaults(prefix string, defaults ApplyFlags, cmd *co
 	}
 
 	cmd.Flags().BoolVar(&s.ApplyIgnored, prefix+"apply-ignored", defaults.ApplyIgnored, "Set to apply ignored changes")
+	cmd.Flags().IntVar(&s.ApplyingChangesOpts.Concurrency, prefix+"apply-concurrency", 5, "Maximum number of concurrent apply operations")
 
 	cmd.Flags().StringVar(&s.AddOrUpdateChangeOpts.DefaultUpdateStrategy, prefix+"apply-default-update-strategy",
 		defaults.AddOrUpdateChangeOpts.DefaultUpdateStrategy, "Change default update strategy")
 
 	cmd.Flags().BoolVar(&s.Wait, prefix+"wait", defaults.Wait, "Set to wait for changes to be applied")
 	cmd.Flags().BoolVar(&s.WaitIgnored, prefix+"wait-ignored", defaults.WaitIgnored, "Set to wait for ignored changes to be applied")
 
-	cmd.Flags().DurationVar(&s.WaitTimeout, prefix+"wait-timeout",
+	cmd.Flags().DurationVar(&s.WaitingChangesOpts.Timeout, prefix+"wait-timeout",
 		mustParseDuration("15m"), "Maximum amount of time to wait")
-	cmd.Flags().DurationVar(&s.WaitCheckInterval, prefix+"wait-check-interval",
+	cmd.Flags().DurationVar(&s.WaitingChangesOpts.CheckInterval, prefix+"wait-check-interval",
 		mustParseDuration("1s"), "Amount of time to sleep between checks while waiting")
 }
 

pkg/kapp/util/throttle.go

@@ -1,10 +1,17 @@
 package util
 
+import (
+	"fmt"
+)
+
 type Throttle struct {
 	ch chan struct{}
 }
 
 func NewThrottle(max int) Throttle {
+	if max < 1 {
+		panic(fmt.Sprintf("Expected maximum throttle to be >= 1, but was %d", max))
+	}
 	ch := make(chan struct{}, max)
 	for i := 0; i < max; i++ {
 		ch <- struct{}{}