Use more effiicent ListPager API.

Use ListPager.EachListItem instead of ListPager.List.  Allows for
incremental conversion page-by-page instead of loading up all pages
into an upstream-format List object and then converting en-masse.

Author: fasaxc

libcalico-go/lib/backend/k8s/resources/list_pager.go

@@ -16,6 +16,7 @@ package resources
 
 import (
 	"context"
+	"sync/atomic"
 
 	"github.com/sirupsen/logrus"
 	"k8s.io/apimachinery/pkg/api/meta"
@@ -39,28 +40,47 @@ func pagedList(
 	*model.KVPairList,
 	error,
 ) {
-	lp := pager.New(listFunc)
+	// Wrap our incoming listFunc with one that stashes the revision and number
+	// of items we've seen so far.  This allows us to use the more efficient
+	// EachListItem() method, while also capturing the list metadata that we
+	// need.
+	listResourceVersion := ""
+	var numItemsLoaded atomic.Int64 // listFunc is called from background goroutine.
+	shimmedListFunc := func(ctx context.Context, opts metav1.ListOptions) (runtime.Object, error) {
+		obj, err := listFunc(ctx, opts)
+		m, err := meta.ListAccessor(obj)
+		if err != nil {
+			return nil, err
+		}
+		if m.GetResourceVersion() != "" {
+			listResourceVersion = m.GetResourceVersion()
+		}
+		numItemsLoaded.Add(int64(meta.LenList(obj)))
+		return obj, err
+	}
+	lp := pager.New(shimmedListFunc)
+
 	opts := metav1.ListOptions{ResourceVersion: revision}
 	if revision != "" {
 		opts.ResourceVersionMatch = metav1.ResourceVersionMatchNotOlderThan
 	}
-	result, isPaged, err := lp.List(ctx, opts)
-	if err != nil {
-		return nil, K8sErrorToCalico(err, list)
-	}
-	logCtx := log.WithField("pagedList", isPaged)
-	logCtx.Debug("List() call completed, convert results")
 
-	// For each item in the response, convert it to a KVPair and add it to the list.
-	kvps := []*model.KVPair{}
-	err = meta.EachListItem(result, func(obj runtime.Object) error {
+	var kvps []*model.KVPair
+	err := lp.EachListItem(ctx, opts, func(obj runtime.Object) error {
 		res := obj.(Resource)
 		result, err := toKVPs(res)
 		if err != nil {
-			logCtx.WithError(err).WithField("Item", res).Warning("unable to process resource, skipping")
+			log.WithError(err).WithField("Item", res).Warning("Unable to process resource, skipping")
 			return nil
 		}
 		if result != nil {
+			if kvps == nil {
+				// Try to guess a suitable result slice capacity.  In practice,
+				// this will be the size of the first page (but that's usually
+				// the only page.)
+				ratio := len(result)
+				kvps = make([]*model.KVPair, 0, int(numItemsLoaded.Load())*ratio)
+			}
 			kvps = append(kvps, result...)
 		}
 		return nil
@@ -69,13 +89,12 @@ func pagedList(
 		return nil, K8sErrorToCalico(err, list)
 	}
 
-	// Extract list revision information.
-	m, err := meta.ListAccessor(result)
-	if err != nil {
-		return nil, err
+	if listResourceVersion == "" {
+		log.WithField("list", list).Panic("Failed to extract resource version from list.")
 	}
+
 	return &model.KVPairList{
 		KVPairs:  kvps,
-		Revision: m.GetResourceVersion(),
+		Revision: listResourceVersion,
 	}, nil
 }

libcalico-go/lib/backend/k8s/resources/profile.go

@@ -186,54 +186,60 @@ func (c *profileClient) List(ctx context.Context, list model.ListInterface, revi
 		return nil, err
 	}
 
-	// Enumerate matching namespaces, paginated.
-	listFunc := func(ctx context.Context, opts metav1.ListOptions) (runtime.Object, error) {
-		if saName != "" {
-			// We've been asked to list a particular service account, skip
-			// listing namespaces.
-			return &v1.NamespaceList{}, nil
+	var nsKVPs *model.KVPairList
+	if saName == "" {
+		// We haven't been asked for a specific service account, so we
+		// must query the namespaces.
+		listFunc := func(ctx context.Context, opts metav1.ListOptions) (runtime.Object, error) {
+			if nsName != "" {
+				opts.FieldSelector = fields.OneTermEqualSelector("metadata.name", nsName).String()
+			}
+			return c.clientSet.CoreV1().Namespaces().List(ctx, opts)
 		}
-		if nsName != "" {
-			opts.FieldSelector = fields.OneTermEqualSelector("metadata.name", nsName).String()
+		convertFunc := func(r Resource) ([]*model.KVPair, error) {
+			ns := r.(*v1.Namespace)
+			kvp, err := c.getNsKv(ns)
+			if err != nil {
+				return nil, err
+			}
+			return []*model.KVPair{kvp}, nil
 		}
-		return c.clientSet.CoreV1().Namespaces().List(ctx, opts)
-	}
-	convertFunc := func(r Resource) ([]*model.KVPair, error) {
-		ns := r.(*v1.Namespace)
-		kvp, err := c.getNsKv(ns)
+		nsKVPs, err = pagedList(ctx, logContext.WithField("from", "namespaces"), nsRev, list, convertFunc, listFunc)
 		if err != nil {
 			return nil, err
 		}
-		return []*model.KVPair{kvp}, nil
-	}
-	nsKVPs, err := pagedList(ctx, logContext.WithField("from", "namespaces"), nsRev, list, convertFunc, listFunc)
-	if err != nil {
-		return nil, err
-	}
-
-	// Enumerate matching service accounts, paginated.
-	listFunc = func(ctx context.Context, opts metav1.ListOptions) (runtime.Object, error) {
-		if nsName != "" && saName == "" {
-			// We've been asked to list a particular namespace, skip
-			// listing service accounts.
-			return &v1.ServiceAccountList{}, nil
+	} else {
+		// We have been asked for a specific service account, so we don't
+		// need to query namespaces.
+		nsKVPs = &model.KVPairList{}
+	}
+
+	var saKVPs *model.KVPairList
+	if nsName == "" || saName != "" {
+		// We haven't been asked for a specific namespace, so we need to
+		// query the service accounts.
+		listFunc := func(ctx context.Context, opts metav1.ListOptions) (runtime.Object, error) {
+			if saName != "" {
+				opts.FieldSelector = fields.OneTermEqualSelector("metadata.name", saName).String()
+			}
+			return c.clientSet.CoreV1().ServiceAccounts(nsName).List(ctx, opts)
 		}
-		if saName != "" {
-			opts.FieldSelector = fields.OneTermEqualSelector("metadata.name", saName).String()
+		convertFunc := func(r Resource) ([]*model.KVPair, error) {
+			sa := r.(*v1.ServiceAccount)
+			kvp, err := c.getSaKv(sa)
+			if err != nil {
+				return nil, err
+			}
+			return []*model.KVPair{kvp}, nil
 		}
-		return c.clientSet.CoreV1().ServiceAccounts(nsName).List(ctx, opts)
-	}
-	convertFunc = func(r Resource) ([]*model.KVPair, error) {
-		sa := r.(*v1.ServiceAccount)
-		kvp, err := c.getSaKv(sa)
+		saKVPs, err = pagedList(ctx, logContext.WithField("from", "serviceaccounts"), saRev, list, convertFunc, listFunc)
 		if err != nil {
 			return nil, err
 		}
-		return []*model.KVPair{kvp}, nil
-	}
-	saKVPs, err := pagedList(ctx, logContext.WithField("from", "serviceaccounts"), saRev, list, convertFunc, listFunc)
-	if err != nil {
-		return nil, err
+	} else {
+		// We have been asked for a specific namespace, so we don't need to
+		// query service accounts.
+		saKVPs = &model.KVPairList{}
 	}
 
 	// Return a merged KVPairList including both results and the default-allow profile.