Merge pull request #53768 from smarterclayton/chunking_cli

Automatic merge from submit-queue. If you want to cherry-pick this change to another branch, please follow the instructions <a href="https://github.com/kubernetes/community/blob/master/contributors/devel/cherry-picks.md">here</a>.

Support api chunking in kubectl get

This enables chunking in the resource builder to make it easy to
retrieve resources in pages and visit partial result sets. This adds
`--chunk-size` to `kubectl get` only so that users can get comfortable
with the use of chunking in beta. Future changes will enable chunking
for all CLI commands so that bulk actions can be performed more
efficiently.

```
$ kubectl get pods --all-namespaces
... print batch of 500 pods ...
... print second batch of 500 pods ...
...
```

@kubernetes/sig-cli-pr-reviews @kubernetes/sig-api-machinery-pr-reviews

```release-note
`kubectl get` will by default fetch large lists of resources in chunks of up to 500 items rather than requesting all resources up front from the server. This reduces the perceived latency of managing large clusters since the server returns the first set of results to the client much more quickly.  A new flag `--chunk-size=SIZE` may be used to alter the number of items or disable this feature when `0` is passed.  This is a beta feature.
```

hack/make-rules/test-cmd-util.sh

@@ -1374,7 +1374,7 @@ run_kubectl_get_tests() {
   fi
 
   ### Test kubectl get all
-  output_message=$(kubectl --v=6 --namespace default get all 2>&1 "${kube_flags[@]}")
+  output_message=$(kubectl --v=6 --namespace default get all --chunk-size=0 2>&1 "${kube_flags[@]}")
   # Post-condition: Check if we get 200 OK from all the url(s)
   kube::test::if_has_string "${output_message}" "/api/v1/namespaces/default/pods 200 OK"
   kube::test::if_has_string "${output_message}" "/api/v1/namespaces/default/replicationcontrollers 200 OK"
@@ -1385,6 +1385,17 @@ run_kubectl_get_tests() {
   kube::test::if_has_string "${output_message}" "/apis/extensions/v1beta1/namespaces/default/deployments 200 OK"
   kube::test::if_has_string "${output_message}" "/apis/extensions/v1beta1/namespaces/default/replicasets 200 OK"
 
+  ### Test kubectl get chunk size
+  output_message=$(kubectl --v=6 get clusterrole --chunk-size=10 2>&1 "${kube_flags[@]}")
+  # Post-condition: Check if we get a limit and continue
+  kube::test::if_has_string "${output_message}" "/clusterroles?limit=10 200 OK"
+  kube::test::if_has_string "${output_message}" "/v1/clusterroles?continue="
+
+  ### Test kubectl get chunk size defaults to 500
+  output_message=$(kubectl --v=6 get clusterrole 2>&1 "${kube_flags[@]}")
+  # Post-condition: Check if we get a limit and continue
+  kube::test::if_has_string "${output_message}" "/clusterroles?limit=500 200 OK"
+
   ### Test --allow-missing-template-keys
   # Pre-condition: no POD exists
   create_and_use_new_namespace

pkg/kubectl/cmd/apply.go

@@ -463,7 +463,15 @@ func (p *pruner) prune(namespace string, mapping *meta.RESTMapping, shortOutput,
 		return err
 	}
 
-	objList, err := resource.NewHelper(c, mapping).List(namespace, mapping.GroupVersionKind.Version, p.selector, false, includeUninitialized)
+	objList, err := resource.NewHelper(c, mapping).List(
+		namespace,
+		mapping.GroupVersionKind.Version,
+		false,
+		&metav1.ListOptions{
+			LabelSelector:        p.selector,
+			IncludeUninitialized: includeUninitialized,
+		},
+	)
 	if err != nil {
 		return err
 	}

pkg/kubectl/cmd/get.go

@@ -51,6 +51,7 @@ type GetOptions struct {
 
 	IgnoreNotFound bool
 	Raw            string
+	ChunkSize      int64
 }
 
 var (
@@ -137,6 +138,7 @@ func NewCmdGet(f cmdutil.Factory, out io.Writer, errOut io.Writer) *cobra.Comman
 	cmd.Flags().Bool("show-kind", false, "If present, list the resource type for the requested object(s).")
 	cmd.Flags().Bool("all-namespaces", false, "If present, list the requested object(s) across all namespaces. Namespace in current context is ignored even if specified with --namespace.")
 	cmd.Flags().BoolVar(&options.IgnoreNotFound, "ignore-not-found", false, "Treat \"resource not found\" as a successful retrieval.")
+	cmd.Flags().Int64Var(&options.ChunkSize, "chunk-size", 500, "Return large lists in chunks rather than all at once. Pass 0 to disable. This flag is beta and may change in the future.")
 	cmd.Flags().StringSliceP("label-columns", "L", []string{}, "Accepts a comma separated list of labels that are going to be presented as columns. Names are case-sensitive. You can also use multiple flag options like -L label1 -L label2...")
 	cmd.Flags().Bool("export", false, "If true, use 'export' for the resources.  Exported resources are stripped of cluster-specific information.")
 	addOpenAPIPrintColumnFlags(cmd)
@@ -223,6 +225,7 @@ func RunGet(f cmdutil.Factory, out, errOut io.Writer, cmd *cobra.Command, args [
 			FilenameParam(enforceNamespace, &options.FilenameOptions).
 			SelectorParam(selector).
 			ExportParam(export).
+			RequestChunksOf(options.ChunkSize).
 			IncludeUninitialized(includeUninitialized).
 			ResourceTypeOrNameArgs(true, args...).
 			SingleResourceType().
@@ -325,6 +328,7 @@ func RunGet(f cmdutil.Factory, out, errOut io.Writer, cmd *cobra.Command, args [
 		FilenameParam(enforceNamespace, &options.FilenameOptions).
 		SelectorParam(selector).
 		ExportParam(export).
+		RequestChunksOf(options.ChunkSize).
 		IncludeUninitialized(includeUninitialized).
 		ResourceTypeOrNameArgs(true, args...).
 		ContinueOnError().

pkg/kubectl/resource/builder.go

@@ -56,6 +56,7 @@ type Builder struct {
 	selector             *string
 	selectAll            bool
 	includeUninitialized bool
+	limitChunks          int64
 
 	resources []string
 
@@ -338,6 +339,14 @@ func (b *Builder) RequireNamespace() *Builder {
 	return b
 }
 
+// RequestChunksOf attempts to load responses from the server in batches of size limit
+// to avoid long delays loading and transferring very large lists. If unset defaults to
+// no chunking.
+func (b *Builder) RequestChunksOf(chunkSize int64) *Builder {
+	b.limitChunks = chunkSize
+	return b
+}
+
 // SelectEverythingParam
 func (b *Builder) SelectAllParam(selectAll bool) *Builder {
 	if selectAll && b.selector != nil {
@@ -636,7 +645,7 @@ func (b *Builder) visitBySelector() *Result {
 		if mapping.Scope.Name() != meta.RESTScopeNameNamespace {
 			selectorNamespace = ""
 		}
-		visitors = append(visitors, NewSelector(client, mapping, selectorNamespace, *b.selector, b.export, b.includeUninitialized))
+		visitors = append(visitors, NewSelector(client, mapping, selectorNamespace, *b.selector, b.export, b.includeUninitialized, b.limitChunks))
 	}
 	if b.continueOnError {
 		result.visitor = EagerVisitorList(visitors)

pkg/kubectl/resource/helper.go

@@ -63,21 +63,15 @@ func (m *Helper) Get(namespace, name string, export bool) (runtime.Object, error
 	return req.Do().Get()
 }
 
-// TODO: add field selector
-func (m *Helper) List(namespace, apiVersion string, selector string, export, includeUninitialized bool) (runtime.Object, error) {
+func (m *Helper) List(namespace, apiVersion string, export bool, options *metav1.ListOptions) (runtime.Object, error) {
 	req := m.RESTClient.Get().
 		NamespaceIfScoped(namespace, m.NamespaceScoped).
 		Resource(m.Resource).
-		VersionedParams(&metav1.ListOptions{
-			LabelSelector: selector,
-		}, metav1.ParameterCodec)
+		VersionedParams(options, metav1.ParameterCodec)
 	if export {
 		// TODO: I should be part of ListOptions
 		req.Param("export", strconv.FormatBool(export))
 	}
-	if includeUninitialized {
-		req.Param("includeUninitialized", strconv.FormatBool(includeUninitialized))
-	}
 	return req.Do().Get()
 }
 

pkg/kubectl/resource/helper_test.go

@@ -359,7 +359,7 @@ func TestHelperList(t *testing.T) {
 			RESTClient:      client,
 			NamespaceScoped: true,
 		}
-		obj, err := modifier.List("bar", legacyscheme.Registry.GroupOrDie(api.GroupName).GroupVersion.String(), "foo=baz", false, false)
+		obj, err := modifier.List("bar", legacyscheme.Registry.GroupOrDie(api.GroupName).GroupVersion.String(), false, &metav1.ListOptions{LabelSelector: "foo=baz"})
 		if (err != nil) != test.Err {
 			t.Errorf("unexpected error: %t %v", test.Err, err)
 		}

pkg/kubectl/resource/selector.go

@@ -21,6 +21,7 @@ import (
 
 	"k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/apimachinery/pkg/api/meta"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/watch"
 )
 
@@ -32,10 +33,11 @@ type Selector struct {
 	Selector             string
 	Export               bool
 	IncludeUninitialized bool
+	LimitChunks          int64
 }
 
 // NewSelector creates a resource selector which hides details of getting items by their label selector.
-func NewSelector(client RESTClient, mapping *meta.RESTMapping, namespace string, selector string, export, includeUninitialized bool) *Selector {
+func NewSelector(client RESTClient, mapping *meta.RESTMapping, namespace string, selector string, export, includeUninitialized bool, limitChunks int64) *Selector {
 	return &Selector{
 		Client:               client,
 		Mapping:              mapping,
@@ -43,42 +45,65 @@ func NewSelector(client RESTClient, mapping *meta.RESTMapping, namespace string,
 		Selector:             selector,
 		Export:               export,
 		IncludeUninitialized: includeUninitialized,
+		LimitChunks:          limitChunks,
 	}
 }
 
-// Visit implements Visitor
+// Visit implements Visitor and uses request chunking by default.
 func (r *Selector) Visit(fn VisitorFunc) error {
-	list, err := NewHelper(r.Client, r.Mapping).List(r.Namespace, r.ResourceMapping().GroupVersionKind.GroupVersion().String(), r.Selector, r.Export, r.IncludeUninitialized)
-	if err != nil {
-		if errors.IsBadRequest(err) || errors.IsNotFound(err) {
-			if se, ok := err.(*errors.StatusError); ok {
-				// modify the message without hiding this is an API error
-				if len(r.Selector) == 0 {
-					se.ErrStatus.Message = fmt.Sprintf("Unable to list %q: %v", r.Mapping.Resource, se.ErrStatus.Message)
-				} else {
-					se.ErrStatus.Message = fmt.Sprintf("Unable to find %q that match the selector %q: %v", r.Mapping.Resource, r.Selector, se.ErrStatus.Message)
-				}
-				return se
+	var continueToken string
+	for {
+		list, err := NewHelper(r.Client, r.Mapping).List(
+			r.Namespace,
+			r.ResourceMapping().GroupVersionKind.GroupVersion().String(),
+			r.Export,
+			&metav1.ListOptions{
+				LabelSelector:        r.Selector,
+				IncludeUninitialized: r.IncludeUninitialized,
+				Limit:                r.LimitChunks,
+				Continue:             continueToken,
+			},
+		)
+		if err != nil {
+			if errors.IsResourceExpired(err) {
+				return err
 			}
-			if len(r.Selector) == 0 {
-				return fmt.Errorf("Unable to list %q: %v", r.Mapping.Resource, err)
-			} else {
+			if errors.IsBadRequest(err) || errors.IsNotFound(err) {
+				if se, ok := err.(*errors.StatusError); ok {
+					// modify the message without hiding this is an API error
+					if len(r.Selector) == 0 {
+						se.ErrStatus.Message = fmt.Sprintf("Unable to list %q: %v", r.Mapping.Resource, se.ErrStatus.Message)
+					} else {
+						se.ErrStatus.Message = fmt.Sprintf("Unable to find %q that match the selector %q: %v", r.Mapping.Resource, r.Selector, se.ErrStatus.Message)
+					}
+					return se
+				}
+				if len(r.Selector) == 0 {
+					return fmt.Errorf("Unable to list %q: %v", r.Mapping.Resource, err)
+				}
 				return fmt.Errorf("Unable to find %q that match the selector %q: %v", r.Mapping.Resource, r.Selector, err)
 			}
+			return err
 		}
-		return err
-	}
-	accessor := r.Mapping.MetadataAccessor
-	resourceVersion, _ := accessor.ResourceVersion(list)
-	info := &Info{
-		Client:    r.Client,
-		Mapping:   r.Mapping,
-		Namespace: r.Namespace,
+		accessor := r.Mapping.MetadataAccessor
+		resourceVersion, _ := accessor.ResourceVersion(list)
+		nextContinueToken, _ := accessor.Continue(list)
+		info := &Info{
+			Client:    r.Client,
+			Mapping:   r.Mapping,
+			Namespace: r.Namespace,
 
-		Object:          list,
-		ResourceVersion: resourceVersion,
+			Object:          list,
+			ResourceVersion: resourceVersion,
+		}
+		if err := fn(info, nil); err != nil {
+			return err
+		}
+		if len(nextContinueToken) == 0 {
+			return nil
+		}
+		continueToken = nextContinueToken
 	}
-	return fn(info, nil)
 }
 
 func (r *Selector) Watch(resourceVersion string) (watch.Interface, error) {

staging/src/k8s.io/apimachinery/pkg/api/meta/interfaces.go

@@ -75,6 +75,9 @@ type MetadataAccessor interface {
 	Annotations(obj runtime.Object) (map[string]string, error)
 	SetAnnotations(obj runtime.Object, annotations map[string]string) error
 
+	Continue(obj runtime.Object) (string, error)
+	SetContinue(obj runtime.Object, c string) error
+
 	runtime.ResourceVersioner
 }
 

staging/src/k8s.io/apimachinery/pkg/api/meta/meta.go

@@ -367,6 +367,23 @@ func (resourceAccessor) SetResourceVersion(obj runtime.Object, version string) e
 	return nil
 }
 
+func (resourceAccessor) Continue(obj runtime.Object) (string, error) {
+	accessor, err := ListAccessor(obj)
+	if err != nil {
+		return "", err
+	}
+	return accessor.GetContinue(), nil
+}
+
+func (resourceAccessor) SetContinue(obj runtime.Object, version string) error {
+	accessor, err := ListAccessor(obj)
+	if err != nil {
+		return err
+	}
+	accessor.SetContinue(version)
+	return nil
+}
+
 // extractFromOwnerReference extracts v to o. v is the OwnerReferences field of an object.
 func extractFromOwnerReference(v reflect.Value, o *metav1.OwnerReference) error {
 	if err := runtime.Field(v, "APIVersion", &o.APIVersion); err != nil {