Merge pull request #116584 from justinsb/parallel_discovery

kubectl prunev2: issue discovery requests in parallel

Kubernetes-commit: 228722578aa6b4c11694d504db29ffd1a07455f1

go.mod

@@ -30,10 +30,10 @@ require (
 	github.com/stretchr/testify v1.8.1
 	golang.org/x/sys v0.6.0
 	gopkg.in/yaml.v2 v2.4.0
-	k8s.io/api v0.0.0-20230315055832-c79498c4e63b
+	k8s.io/api v0.0.0-20230315055835-286fd7ee0419
 	k8s.io/apimachinery v0.0.0-20230315054728-8d1258da8f38
 	k8s.io/cli-runtime v0.0.0-20230315090346-9185a9dd0ead
-	k8s.io/client-go v0.0.0-20230315061852-9ff2627505a4
+	k8s.io/client-go v0.0.0-20230315061906-445660b561c3
 	k8s.io/component-base v0.0.0-20230315065615-6b9bb8ecc3d0
 	k8s.io/component-helpers v0.0.0-20230315070331-853a1039365a
 	k8s.io/klog/v2 v2.90.1
@@ -94,10 +94,10 @@ require (
 )
 
 replace (
-	k8s.io/api => k8s.io/api v0.0.0-20230315055832-c79498c4e63b
+	k8s.io/api => k8s.io/api v0.0.0-20230315055835-286fd7ee0419
 	k8s.io/apimachinery => k8s.io/apimachinery v0.0.0-20230315054728-8d1258da8f38
 	k8s.io/cli-runtime => k8s.io/cli-runtime v0.0.0-20230315090346-9185a9dd0ead
-	k8s.io/client-go => k8s.io/client-go v0.0.0-20230315061852-9ff2627505a4
+	k8s.io/client-go => k8s.io/client-go v0.0.0-20230315061906-445660b561c3
 	k8s.io/code-generator => k8s.io/code-generator v0.0.0-20230315053024-8fead9f64de8
 	k8s.io/component-base => k8s.io/component-base v0.0.0-20230315065615-6b9bb8ecc3d0
 	k8s.io/component-helpers => k8s.io/component-helpers v0.0.0-20230315070331-853a1039365a

go.sum

@@ -539,14 +539,14 @@ honnef.co/go/tools v0.0.0-20190523083050-ea95bdfd59fc/go.mod h1:rf3lG4BRIbNafJWh
 honnef.co/go/tools v0.0.1-2019.2.3/go.mod h1:a3bituU0lyd329TUQxRnasdCoJDkEUEAqEt0JzvZhAg=
 honnef.co/go/tools v0.0.1-2020.1.3/go.mod h1:X/FiERA/W4tHapMX5mGpAtMSVEeEUOyHaw9vFzvIQ3k=
 honnef.co/go/tools v0.0.1-2020.1.4/go.mod h1:X/FiERA/W4tHapMX5mGpAtMSVEeEUOyHaw9vFzvIQ3k=
-k8s.io/api v0.0.0-20230315055832-c79498c4e63b h1:+XtheLjvEgurlWvtf13tutNwmW6WUHGezbiapc4s6EM=
-k8s.io/api v0.0.0-20230315055832-c79498c4e63b/go.mod h1:aZ6MBt4NMLXSxkSKFkoDaP4hTutnZIvH5dCSpOis9g4=
+k8s.io/api v0.0.0-20230315055835-286fd7ee0419 h1:4aCwVAD4ugAc04mLcDwgZicreRARKKTaukyfvfeV7xM=
+k8s.io/api v0.0.0-20230315055835-286fd7ee0419/go.mod h1:aZ6MBt4NMLXSxkSKFkoDaP4hTutnZIvH5dCSpOis9g4=
 k8s.io/apimachinery v0.0.0-20230315054728-8d1258da8f38 h1:n1qDRCTPAXwyXYg7eSpWDO9FdW79lwAQ9dAr1vETpn4=
 k8s.io/apimachinery v0.0.0-20230315054728-8d1258da8f38/go.mod h1:5ikh59fK3AJ287GUvpUsryoMFtH9zj/ARfWCo3AyXTM=
 k8s.io/cli-runtime v0.0.0-20230315090346-9185a9dd0ead h1:HYtR8a33cxjM2XGQvAZaDXXjhY3nz9uWvbeYJSz6yl0=
 k8s.io/cli-runtime v0.0.0-20230315090346-9185a9dd0ead/go.mod h1:wg6/UJ9wwfEEAXnyuuajHSr+YnMXL1K0yUrRHjXz3Do=
-k8s.io/client-go v0.0.0-20230315061852-9ff2627505a4 h1:mjPYL7mzIzB9PHI4ttnYkGY5zat6ObsndQNh7Gx12OQ=
-k8s.io/client-go v0.0.0-20230315061852-9ff2627505a4/go.mod h1:daDXfDBtiJdqKrqqFL0WtQ6hHzsDxP1lCEdTt3+kijU=
+k8s.io/client-go v0.0.0-20230315061906-445660b561c3 h1:urBW+EemzKj3PArMtVVuRerDGvBiXcnS+OAoOjqQoaM=
+k8s.io/client-go v0.0.0-20230315061906-445660b561c3/go.mod h1:D6IqTVZ0quZZGxz2yiqMrndPUjMSPJ6QNq5SiSz+Mv0=
 k8s.io/component-base v0.0.0-20230315065615-6b9bb8ecc3d0 h1:IjneP02MOB07PIP9+PQjKrOIZEZ5T7umR+GIZkU4h0U=
 k8s.io/component-base v0.0.0-20230315065615-6b9bb8ecc3d0/go.mod h1:kTuptveA6tUMLMKnaq4AbIAAk7IcdhwkbljAV3JZRpM=
 k8s.io/component-helpers v0.0.0-20230315070331-853a1039365a h1:PIKao3YG7ZMiE+CtdM09YDcowRhGTdvCeEgWj3J9DKI=

pkg/cmd/apply/applyset_pruner.go

@@ -19,6 +19,7 @@ package apply
 import (
 	"context"
 	"fmt"
+	"sync"
 
 	"k8s.io/apimachinery/pkg/api/meta"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -65,8 +66,17 @@ func (p *PruneObject) String() string {
 // FindAllObjectsToPrune returns the list of objects that will be pruned.
 // Calling this instead of Prune can be useful for dry-run / diff behaviour.
 func (a *ApplySet) FindAllObjectsToPrune(ctx context.Context, dynamicClient dynamic.Interface, visitedUids sets.Set[types.UID]) ([]PruneObject, error) {
-	var allObjects []PruneObject
-	// TODO: Run discovery in parallel (and maybe in consistent order?)
+	type task struct {
+		namespace   string
+		restMapping *meta.RESTMapping
+
+		err     error
+		results []PruneObject
+	}
+	var tasks []*task
+
+	// We run discovery in parallel, in as many goroutines as priority and fairness will allow
+	// (We don't expect many requests in real-world scenarios - maybe tens, unlikely to be hundreds)
 	for _, restMapping := range a.AllPrunableResources() {
 		switch restMapping.Scope.Name() {
 		case meta.RESTScopeNameNamespace:
@@ -75,25 +85,48 @@ func (a *ApplySet) FindAllObjectsToPrune(ctx context.Context, dynamicClient dyna
 					// Just double-check because otherwise we get cryptic error messages
 					return nil, fmt.Errorf("unexpectedly encountered empty namespace during prune of namespace-scoped resource %v", restMapping.GroupVersionKind)
 				}
-				pruneObjects, err := a.findObjectsToPrune(ctx, dynamicClient, visitedUids, namespace, restMapping)
-				if err != nil {
-					return nil, fmt.Errorf("listing %v objects for prune: %w", restMapping.GroupVersionKind.String(), err)
-				}
-				allObjects = append(allObjects, pruneObjects...)
+				tasks = append(tasks, &task{
+					namespace:   namespace,
+					restMapping: restMapping,
+				})
 			}
 
 		case meta.RESTScopeNameRoot:
-			pruneObjects, err := a.findObjectsToPrune(ctx, dynamicClient, visitedUids, metav1.NamespaceNone, restMapping)
-			if err != nil {
-				return nil, fmt.Errorf("listing %v objects for prune: %w", restMapping.GroupVersionKind.String(), err)
-			}
-			allObjects = append(allObjects, pruneObjects...)
+			tasks = append(tasks, &task{
+				restMapping: restMapping,
+			})
 
 		default:
 			return nil, fmt.Errorf("unhandled scope %q", restMapping.Scope.Name())
 		}
 	}
 
+	var wg sync.WaitGroup
+
+	for i := range tasks {
+		task := tasks[i]
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+
+			results, err := a.findObjectsToPrune(ctx, dynamicClient, visitedUids, task.namespace, task.restMapping)
+			if err != nil {
+				task.err = fmt.Errorf("listing %v objects for pruning: %w", task.restMapping.GroupVersionKind.String(), err)
+			} else {
+				task.results = results
+			}
+		}()
+	}
+	// Wait for all the goroutines to finish
+	wg.Wait()
+
+	var allObjects []PruneObject
+	for _, task := range tasks {
+		if task.err != nil {
+			return nil, task.err
+		}
+		allObjects = append(allObjects, task.results...)
+	}
 	return allObjects, nil
 }