make oc adm inspect run in parallel and remove rate limits

pkg/cli/admin/inspect/inspect.go

@@ -205,14 +205,8 @@ func (o *InspectOptions) Run() error {
 	}
 
 	// finally, gather polymorphic resources specified by the user
-	allErrs := []error{}
 	ctx := NewResourceContext()
-	for _, info := range infos {
-		err := InspectResource(info, ctx, o)
-		if err != nil {
-			allErrs = append(allErrs, err)
-		}
-	}
+	allErrs := ParallelInspectResource(infos, ctx, o)
 
 	// now gather all the events into a single file and produce a unified file
 	if err := CreateEventFilterPage(o.destDir); err != nil {
@@ -230,11 +224,10 @@ func (o *InspectOptions) Run() error {
 // gatherConfigResourceData gathers all config.openshift.io resources
 func (o *InspectOptions) gatherConfigResourceData(destDir string, ctx *resourceContext) error {
 	// determine if we've already collected configResourceData
-	if ctx.visited.Has(configResourceDataKey) {
+	if ctx.visited(configResourceDataKey) {
 		klog.V(1).Infof("Skipping previously-collected config.openshift.io resource data")
 		return nil
 	}
-	ctx.visited.Insert(configResourceDataKey)
 
 	klog.V(1).Infof("Gathering config.openshift.io resource data...\n")
 
@@ -273,11 +266,10 @@ func (o *InspectOptions) gatherConfigResourceData(destDir string, ctx *resourceC
 // gatherOperatorResourceData gathers all kubeapiserver.operator.openshift.io resources
 func (o *InspectOptions) gatherOperatorResourceData(destDir string, ctx *resourceContext) error {
 	// determine if we've already collected operatorResourceData
-	if ctx.visited.Has(operatorResourceDataKey) {
+	if ctx.visited(operatorResourceDataKey) {
 		klog.V(1).Infof("Skipping previously-collected operator.openshift.io resource data")
 		return nil
 	}
-	ctx.visited.Insert(operatorResourceDataKey)
 
 	// ensure destination path exists
 	if err := os.MkdirAll(destDir, os.ModePerm); err != nil {

pkg/cli/admin/inspect/resource.go

@@ -4,6 +4,7 @@ import (
 	"fmt"
 	"os"
 	"path"
+	"sync"
 
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
@@ -24,14 +25,42 @@ const (
 	operatorResourceDataKey = "/cluster-scoped-resources/operator.openshift.io"
 )
 
+func ParallelInspectResource(infos []*resource.Info, context *resourceContext, o *InspectOptions) []error {
+	if len(infos) == 0 {
+		return []error{}
+	}
+
+	errCh := make(chan error, len(infos))
+	wg := sync.WaitGroup{}
+	for i := range infos {
+		info := infos[i]
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+
+			err := InspectResource(info, context, o)
+			if err != nil {
+				errCh <- err
+			}
+		}()
+	}
+	wg.Wait()
+	allErrs := []error{}
+	close(errCh)
+	for err := range errCh {
+		allErrs = append(allErrs, err)
+	}
+
+	return allErrs
+}
+
 // InspectResource receives an object to gather debugging data for, and a context to keep track of
 // already-seen objects when following related-object reference chains.
 func InspectResource(info *resource.Info, context *resourceContext, o *InspectOptions) error {
-	if context.visited.Has(infoToContextKey(info)) {
+	if context.visited(infoToContextKey(info)) {
 		klog.V(1).Infof("Skipping previously-inspected resource: %q ...", infoToContextKey(info))
 		return nil
 	}
-	context.visited.Insert(infoToContextKey(info))
 
 	switch info.ResourceMapping().Resource.GroupResource() {
 	case configv1.GroupVersion.WithResource("clusteroperators").GroupResource():
@@ -69,23 +98,21 @@ func InspectResource(info *resource.Info, context *resourceContext, o *InspectOp
 			errs = append(errs, err)
 		}
 		resourcesToCollect := namespaceResourcesToCollect()
+		allResourceInfosToCollect := []*resource.Info{}
 		for _, resource := range resourcesToCollect {
-			if context.visited.Has(resourceToContextKey(resource, info.Name)) {
+			if context.visited(resourceToContextKey(resource, info.Name)) {
 				continue
 			}
 			resourceInfos, err := groupResourceToInfos(o.configFlags, resource, info.Name)
 			if err != nil {
 				errs = append(errs, err)
 				continue
 			}
-			for _, resourceInfo := range resourceInfos {
-				if err := InspectResource(resourceInfo, context, o); err != nil {
-					errs = append(errs, err)
-					continue
-				}
-			}
+			allResourceInfosToCollect = append(allResourceInfosToCollect, resourceInfos...)
 		}
 
+		gatherErrs := ParallelInspectResource(allResourceInfosToCollect, context, o)
+		errs = append(errs, gatherErrs...)
 		return errors.NewAggregate(errs)
 
 	case corev1.SchemeGroupVersion.WithResource("secrets").GroupResource():
@@ -127,8 +154,9 @@ func gatherRelatedObjects(context *resourceContext, unstr *unstructured.Unstruct
 	}
 
 	errs := []error{}
+	allRelatedInfos := []*resource.Info{}
 	for _, relatedRef := range relatedObjReferences {
-		if context.visited.Has(objectRefToContextKey(relatedRef)) {
+		if context.peekVisited(objectRefToContextKey(relatedRef)) {
 			continue
 		}
 
@@ -137,15 +165,11 @@ func gatherRelatedObjects(context *resourceContext, unstr *unstructured.Unstruct
 			errs = append(errs, fmt.Errorf("skipping gathering %s due to error: %v", objectReferenceToString(relatedRef), err))
 			continue
 		}
-
-		for _, relatedInfo := range relatedInfos {
-			if err := InspectResource(relatedInfo, context, o); err != nil {
-				errs = append(errs, fmt.Errorf("skipping gathering %s due to error: %v", objectReferenceToString(relatedRef), err))
-				continue
-			}
-		}
+		allRelatedInfos = append(allRelatedInfos, relatedInfos...)
 	}
 
+	gatherErrs := ParallelInspectResource(allRelatedInfos, context, o)
+	errs = append(errs, gatherErrs...)
 	return errors.NewAggregate(errs)
 }
 

pkg/cli/admin/inspect/util.go

@@ -7,6 +7,7 @@ import (
 	"os"
 	"path"
 	"path/filepath"
+	"sync"
 
 	configv1 "github.com/openshift/api/config/v1"
 	corev1 "k8s.io/api/core/v1"
@@ -23,15 +24,35 @@ import (
 
 // resourceContext is used to keep track of previously seen objects
 type resourceContext struct {
-	visited sets.String
+	lock sync.Mutex
+
+	alreadyVisited sets.String
 }
 
 func NewResourceContext() *resourceContext {
 	return &resourceContext{
-		visited: sets.NewString(),
+		alreadyVisited: sets.NewString(),
 	}
 }
 
+// visited returns whether or not an item already has already been visited and adds it to the list
+func (r *resourceContext) visited(resource string) bool {
+	r.lock.Lock()
+	defer r.lock.Unlock()
+
+	ret := r.alreadyVisited.Has(resource)
+	r.alreadyVisited.Insert(resource)
+	return ret
+}
+
+// visited returns whether or not an item already has already been visited and does NOT add it to the list
+func (r *resourceContext) peekVisited(resource string) bool {
+	r.lock.Lock()
+	defer r.lock.Unlock()
+
+	return r.alreadyVisited.Has(resource)
+}
+
 func objectReferenceToString(ref *configv1.ObjectReference) string {
 	resource := ref.Resource
 	group := ref.Group