Add support for implicit paging in un/structured clients

Author: Nimrod Shneor

pkg/client/client_test.go

@@ -19,6 +19,7 @@ package client_test
 import (
 	"context"
 	"fmt"
+	"strconv"
 	"sync/atomic"
 	"time"
 
@@ -47,6 +48,14 @@ func deleteDeployment(ctx context.Context, dep *appsv1.Deployment, ns string) {
 	}
 }
 
+func deletePod(ctx context.Context, pod *corev1.Pod, ns string) {
+	_, err := clientset.CoreV1().Pods(ns).Get(ctx, pod.Name, metav1.GetOptions{})
+	if err == nil {
+		err = clientset.CoreV1().Pods(ns).Delete(ctx, pod.Name, metav1.DeleteOptions{})
+		Expect(err).NotTo(HaveOccurred())
+	}
+}
+
 func deleteNamespace(ctx context.Context, ns *corev1.Namespace) {
 	ns, err := clientset.CoreV1().Namespaces().Get(ctx, ns.Name, metav1.GetOptions{})
 	if err != nil {
@@ -2085,6 +2094,88 @@ var _ = Describe("Client", func() {
 				Expect(deps.Items[1].Name).To(Equal(dep4.Name))
 			}, serverSideTimeoutSeconds)
 
+			It("should filter results using page and limit options", func() {
+				buildPod := func(suffix string) *corev1.Pod {
+					return &corev1.Pod{
+						ObjectMeta: metav1.ObjectMeta{
+							Name: fmt.Sprintf("pod-%s", suffix),
+						},
+						Spec: corev1.PodSpec{
+							Containers: []corev1.Container{{Name: "nginx", Image: "nginx"}},
+						},
+					}
+				}
+
+				By("creating 150 pods")
+				workLoad := 150
+				for workLoad > 0 {
+					pod := buildPod(strconv.Itoa(workLoad))
+					defer deletePod(ctx, pod, ns)
+					pod, err := clientset.
+						CoreV1().
+						Pods(ns).
+						Create(ctx, pod, metav1.CreateOptions{})
+					Expect(err).NotTo(HaveOccurred())
+					workLoad--
+					defer deletePod(ctx, pod, ns)
+				}
+
+				cl, err := client.New(cfg, client.Options{})
+				Expect(err).NotTo(HaveOccurred())
+
+				var offset int64
+				By("listing first one hundred pods when page=0 is used")
+				pods := &corev1.PodList{}
+				err = cl.List(context.Background(), pods,
+					client.Page{
+						Offset: &offset,
+					},
+				)
+				Expect(err).NotTo(HaveOccurred())
+				Expect(pods.Items).To(HaveLen(client.DefaultPageLimit))
+				Expect(pods.Continue).NotTo(BeEmpty())
+				Expect(pods.Items[0].Name).To(Equal("pod-1"))
+
+				By("listing the second page of results with page=1")
+				pods = &corev1.PodList{}
+				offset = 1
+				err = cl.List(context.Background(), pods,
+					client.Page{
+						Offset: &offset,
+					},
+				)
+				Expect(err).NotTo(HaveOccurred())
+				Expect(pods.Continue).To(BeEmpty())
+				Expect(pods.Items).To(HaveLen(50))
+
+				By("listing first page of fifty pods when page=0 and limit=50 are specified")
+				pods = &corev1.PodList{}
+				offset = 0
+				err = cl.List(context.Background(), pods,
+					client.Page{
+						Offset: &offset,
+					},
+					client.Limit(50),
+				)
+				Expect(err).NotTo(HaveOccurred())
+				Expect(pods.Continue).NotTo(BeEmpty())
+				Expect(pods.Items).To(HaveLen(50))
+
+				By("listing second page of fifty pods when page=1 and limit=50 are specified")
+				pods = &corev1.PodList{}
+				offset = 1
+				err = cl.List(context.Background(), pods,
+					client.Page{
+						Offset: &offset,
+					},
+					client.Limit(50),
+				)
+				Expect(err).NotTo(HaveOccurred())
+				Expect(pods.Continue).NotTo(BeEmpty())
+				Expect(pods.Items).To(HaveLen(50))
+
+			}, serverSideTimeoutSeconds)
+
 			PIt("should fail if the object doesn't have meta", func() {
 
 			})

pkg/client/options.go

@@ -23,6 +23,11 @@ import (
 	"k8s.io/apimachinery/pkg/selection"
 )
 
+const (
+	// DefaultPageLimit represents the default limit used when specifyinh the 'Page' ListOption.
+	DefaultPageLimit = 100
+)
+
 // {{{ "Functional" Option Interfaces
 
 // CreateOption is some configuration that modifies options for a create request.
@@ -341,6 +346,14 @@ type ListOptions struct {
 	// it has expired. This field is not supported if watch is true in the Raw ListOptions.
 	Continue string
 
+	// Page is used to retrieve results at a certain offset from the server.
+	// To avoid confusion paging start at index one.
+	// The limit used as a single page size is one hundred items by default.
+	// The server may reject requests for continuation tokens
+	// it does not recognize and will return a 410 error if the token can no longer be used because
+	// it has expired. This field is not supported if watch is true in the Raw ListOptions.
+	Page Page
+
 	// Raw represents raw ListOptions, as passed to the API server.  Note
 	// that these may not be respected by all implementations of interface,
 	// and the LabelSelector, FieldSelector, Limit and Continue fields are ignored.
@@ -521,6 +534,19 @@ func (c Continue) ApplyToList(opts *ListOptions) {
 	opts.Continue = string(c)
 }
 
+// Page sets a page offset for the result set from the server.
+// The limit used as a single page size is one hundred items by default.
+// Page does not implement DeleteAllOfOption interface because the server
+// does not support setting it for deletecollection operations.
+type Page struct {
+	Offset *int64
+}
+
+// ApplyToList applies this configuration to the given an List options.
+func (p Page) ApplyToList(opts *ListOptions) {
+	opts.Page.Offset = p.Offset
+}
+
 // }}}
 
 // {{{ Update Options

pkg/client/typed_client.go

@@ -151,6 +151,30 @@ func (c *typedClient) List(ctx context.Context, obj ObjectList, opts ...ListOpti
 	}
 	listOpts := ListOptions{}
 	listOpts.ApplyOptions(opts)
+
+	// This is an implicit implementation of paging as kube only provides limit and
+	// continuation token as a mechanism for retrieving large sets of results.
+	// For more details, see the following:
+	// https://kubernetes.io/docs/reference/using-api/api-concepts/#retrieving-large-results-sets-in-chunks
+	if listOpts.Page.Offset != nil {
+		offset := *listOpts.Page.Offset
+		if listOpts.Limit == 0 {
+			Limit(DefaultPageLimit).ApplyToList(&listOpts)
+		}
+		for offset > 0 {
+			err := r.Get().
+				NamespaceIfScoped(listOpts.Namespace, r.isNamespaced()).
+				Resource(r.resource()).
+				VersionedParams(listOpts.AsListOptions(), c.paramCodec).
+				Do(ctx).Into(obj)
+			if err != nil {
+				return err
+			}
+			listOpts.Continue = obj.GetContinue()
+			offset--
+		}
+	}
+
 	return r.Get().
 		NamespaceIfScoped(listOpts.Namespace, r.isNamespaced()).
 		Resource(r.resource()).

pkg/client/unstructured_client.go

@@ -213,6 +213,29 @@ func (uc *unstructuredClient) List(ctx context.Context, obj ObjectList, opts ...
 		return err
 	}
 
+	// This is an implicit implementation of paging as kube only provides a limit and
+	// continuation token params as a mechanism for retrieving large sets of results.
+	// For more details, see the following:
+	// https://kubernetes.io/docs/reference/using-api/api-concepts/#retrieving-large-results-sets-in-chunks
+	if listOpts.Page.Offset != nil {
+		offset := *listOpts.Page.Offset
+		if listOpts.Limit == 0 {
+			Limit(DefaultPageLimit).ApplyToList(&listOpts)
+		}
+		for offset > 0 {
+			err := r.Get().
+				NamespaceIfScoped(listOpts.Namespace, r.isNamespaced()).
+				Resource(r.resource()).
+				VersionedParams(listOpts.AsListOptions(), uc.paramCodec).
+				Do(ctx).Into(obj)
+			if err != nil {
+				return err
+			}
+			listOpts.Continue = obj.GetContinue()
+			offset--
+		}
+	}
+
 	return r.Get().
 		NamespaceIfScoped(listOpts.Namespace, r.isNamespaced()).
 		Resource(r.resource()).