Fill partial pages on the server rather than forcing client to

staging/src/k8s.io/apiserver/pkg/storage/etcd3/store.go

@@ -396,11 +396,11 @@ func (s *store) GetToList(ctx context.Context, key string, resourceVersion strin
 	if err != nil {
 		return storage.NewInternalError(err.Error())
 	}
-	elems := []*elemForDecode{{
-		data: data,
-		rev:  uint64(getResp.Kvs[0].ModRevision),
-	}}
-	if err := decodeList(elems, storage.SimpleFilter(pred), listPtr, s.codec, s.versioner); err != nil {
+	v, err := conversion.EnforcePtr(listPtr)
+	if err != nil || v.Kind() != reflect.Slice {
+		panic("need ptr to slice")
+	}
+	if err := appendListItem(v, data, uint64(getResp.Kvs[0].ModRevision), storage.SimpleFilter(pred), s.codec, s.versioner); err != nil {
 		return err
 	}
 	// update version with cluster level revision
@@ -472,7 +472,14 @@ func (s *store) List(ctx context.Context, key, resourceVersion string, pred stor
 	if err != nil {
 		return err
 	}
-	key = path.Join(s.pathPrefix, key)
+	v, err := conversion.EnforcePtr(listPtr)
+	if err != nil || v.Kind() != reflect.Slice {
+		panic("need ptr to slice")
+	}
+
+	if s.pathPrefix != "" {
+		key = path.Join(s.pathPrefix, key)
+	}
 	// We need to make sure the key ended with "/" so that we only get children "directories".
 	// e.g. if we have key "/a", "/a/b", "/ab", getting keys with prefix "/a" will return all three,
 	// while with prefix "/a/" will return only "/a/b" which is the correct answer.
@@ -481,81 +488,127 @@ func (s *store) List(ctx context.Context, key, resourceVersion string, pred stor
 	}
 	keyPrefix := key
 
+	filter := storage.SimpleFilter(pred)
+
 	// set the appropriate clientv3 options to filter the returned data set
+	var paging bool
 	options := make([]clientv3.OpOption, 0, 4)
 	if s.pagingEnabled && pred.Limit > 0 {
+		paging = true
 		options = append(options, clientv3.WithLimit(pred.Limit))
 	}
+
 	var returnedRV int64
 	switch {
 	case s.pagingEnabled && len(pred.Continue) > 0:
 		continueKey, continueRV, err := decodeContinue(pred.Continue, keyPrefix)
 		if err != nil {
-			return err
+			return apierrors.NewBadRequest(fmt.Sprintf("invalid continue token: %v", err))
+		}
+
+		if len(resourceVersion) > 0 && resourceVersion != "0" {
+			return apierrors.NewBadRequest("specifying resource version is not allowed when using continue")
 		}
 
-		options = append(options, clientv3.WithRange(clientv3.GetPrefixRangeEnd(key)))
+		rangeEnd := clientv3.GetPrefixRangeEnd(keyPrefix)
+		options = append(options, clientv3.WithRange(rangeEnd))
 		key = continueKey
 
 		options = append(options, clientv3.WithRev(continueRV))
 		returnedRV = continueRV
 
-	case len(resourceVersion) > 0:
-		fromRV, err := strconv.ParseInt(resourceVersion, 10, 64)
-		if err != nil {
-			return fmt.Errorf("invalid resource version: %v", err)
+	case s.pagingEnabled && pred.Limit > 0:
+		if len(resourceVersion) > 0 {
+			fromRV, err := strconv.ParseInt(resourceVersion, 10, 64)
+			if err != nil {
+				return apierrors.NewBadRequest(fmt.Sprintf("invalid resource version: %v", err))
+			}
+			if fromRV > 0 {
+				options = append(options, clientv3.WithRev(fromRV))
+			}
+			returnedRV = fromRV
 		}
 
-		options = append(options, clientv3.WithPrefix(), clientv3.WithRev(fromRV))
-		returnedRV = fromRV
+		rangeEnd := clientv3.GetPrefixRangeEnd(keyPrefix)
+		options = append(options, clientv3.WithRange(rangeEnd))
 
 	default:
-		options = append(options, clientv3.WithPrefix())
-	}
+		if len(resourceVersion) > 0 {
+			fromRV, err := strconv.ParseInt(resourceVersion, 10, 64)
+			if err != nil {
+				return apierrors.NewBadRequest(fmt.Sprintf("invalid resource version: %v", err))
+			}
+			if fromRV > 0 {
+				options = append(options, clientv3.WithRev(fromRV))
+			}
+			returnedRV = fromRV
+		}
 
-	getResp, err := s.client.KV.Get(ctx, key, options...)
-	if err != nil {
-		return interpretListError(err, len(pred.Continue) > 0)
+		options = append(options, clientv3.WithPrefix())
 	}
 
-	elems := make([]*elemForDecode, 0, len(getResp.Kvs))
-	for _, kv := range getResp.Kvs {
-		data, _, err := s.transformer.TransformFromStorage(kv.Value, authenticatedDataString(kv.Key))
+	// loop until we have filled the requested limit from etcd or there are no more results
+	var lastKey []byte
+	var hasMore bool
+	for {
+		getResp, err := s.client.KV.Get(ctx, key, options...)
 		if err != nil {
-			utilruntime.HandleError(fmt.Errorf("unable to transform key %q: %v", kv.Key, err))
-			continue
+			return interpretListError(err, len(pred.Continue) > 0)
 		}
+		hasMore = getResp.More
 
-		elems = append(elems, &elemForDecode{
-			data: data,
-			rev:  uint64(kv.ModRevision),
-		})
-	}
+		if len(getResp.Kvs) == 0 && getResp.More {
+			return fmt.Errorf("no results were found, but etcd indicated there were more values remaining")
+		}
 
-	if err := decodeList(elems, storage.SimpleFilter(pred), listPtr, s.codec, s.versioner); err != nil {
-		return err
-	}
+		// take items from the response until the bucket is full, filtering as we go
+		for _, kv := range getResp.Kvs {
+			if paging && int64(v.Len()) >= pred.Limit {
+				hasMore = true
+				break
+			}
+			lastKey = kv.Key
+
+			data, _, err := s.transformer.TransformFromStorage(kv.Value, authenticatedDataString(kv.Key))
+			if err != nil {
+				utilruntime.HandleError(fmt.Errorf("unable to transform key %q: %v", kv.Key, err))
+				continue
+			}
 
-	// indicate to the client which resource version was returned
-	if returnedRV == 0 {
-		returnedRV = getResp.Header.Revision
+			if err := appendListItem(v, data, uint64(kv.ModRevision), filter, s.codec, s.versioner); err != nil {
+				return err
+			}
+		}
+
+		// indicate to the client which resource version was returned
+		if returnedRV == 0 {
+			returnedRV = getResp.Header.Revision
+		}
+
+		// no more results remain or we didn't request paging
+		if !hasMore || !paging {
+			break
+		}
+		// we're paging but we have filled our bucket
+		if int64(v.Len()) >= pred.Limit {
+			break
+		}
+		key = string(lastKey) + "\x00"
 	}
-	switch {
-	case !getResp.More:
-		// no continuation
-		return s.versioner.UpdateList(listObj, uint64(returnedRV), "")
-	case len(getResp.Kvs) == 0:
-		return fmt.Errorf("no results were found, but etcd indicated there were more values")
-	default:
+
+	// instruct the client to begin querying from immediately after the last key we returned
+	// we never return a key that the client wouldn't be allowed to see
+	if hasMore {
 		// we want to start immediately after the last key
-		// TODO: this reveals info about certain keys
-		key := string(getResp.Kvs[len(getResp.Kvs)-1].Key)
-		next, err := encodeContinue(key+"\x00", keyPrefix, returnedRV)
+		next, err := encodeContinue(string(lastKey)+"\x00", keyPrefix, returnedRV)
 		if err != nil {
 			return err
 		}
 		return s.versioner.UpdateList(listObj, uint64(returnedRV), next)
 	}
+
+	// no continuation
+	return s.versioner.UpdateList(listObj, uint64(returnedRV), "")
 }
 
 // Watch implements storage.Interface.Watch.
@@ -677,23 +730,16 @@ func decode(codec runtime.Codec, versioner storage.Versioner, value []byte, objP
 	return nil
 }
 
-// decodeList decodes a list of values into a list of objects, with resource version set to corresponding rev.
-// On success, ListPtr would be set to the list of objects.
-func decodeList(elems []*elemForDecode, filter storage.FilterFunc, listPtr interface{}, codec runtime.Codec, versioner storage.Versioner) error {
-	v, err := conversion.EnforcePtr(listPtr)
-	if err != nil || v.Kind() != reflect.Slice {
-		panic("need ptr to slice")
+// appendListItem decodes and appends the object (if it passes filter) to v, which must be a slice.
+func appendListItem(v reflect.Value, data []byte, rev uint64, filter storage.FilterFunc, codec runtime.Codec, versioner storage.Versioner) error {
+	obj, _, err := codec.Decode(data, nil, reflect.New(v.Type().Elem()).Interface().(runtime.Object))
+	if err != nil {
+		return err
 	}
-	for _, elem := range elems {
-		obj, _, err := codec.Decode(elem.data, nil, reflect.New(v.Type().Elem()).Interface().(runtime.Object))
-		if err != nil {
-			return err
-		}
-		// being unable to set the version does not prevent the object from being extracted
-		versioner.UpdateObject(obj, elem.rev)
-		if filter(obj) {
-			v.Set(reflect.Append(v, reflect.ValueOf(obj).Elem()))
-		}
+	// being unable to set the version does not prevent the object from being extracted
+	versioner.UpdateObject(obj, rev)
+	if filter(obj) {
+		v.Set(reflect.Append(v, reflect.ValueOf(obj).Elem()))
 	}
 	return nil
 }

test/e2e/apimachinery/BUILD

@@ -9,6 +9,7 @@ go_library(
     name = "go_default_library",
     srcs = [
         "aggregator.go",
+        "chunking.go",
         "custom_resource_definition.go",
         "etcd_failure.go",
         "framework.go",
@@ -58,6 +59,7 @@ go_library(
         "//vendor/k8s.io/client-go/kubernetes:go_default_library",
         "//vendor/k8s.io/client-go/util/cert:go_default_library",
         "//vendor/k8s.io/client-go/util/retry:go_default_library",
+        "//vendor/k8s.io/client-go/util/workqueue:go_default_library",
         "//vendor/k8s.io/kube-aggregator/pkg/apis/apiregistration/v1beta1:go_default_library",
         "//vendor/k8s.io/sample-apiserver/pkg/apis/wardle/v1alpha1:go_default_library",
     ],

test/e2e/apimachinery/chunking.go

+/*
+Copyright 2017 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package apimachinery
+
+import (
+	"fmt"
+	"math/rand"
+
+	. "github.com/onsi/ginkgo"
+	. "github.com/onsi/gomega"
+
+	"k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/client-go/util/workqueue"
+	"k8s.io/kubernetes/test/e2e/framework"
+)
+
+const numberOfTotalResources = 400
+
+var _ = SIGDescribe("Servers with support for API chunking", func() {
+	f := framework.NewDefaultFramework("chunking")
+
+	It("should return chunks of results for list calls", func() {
+		ns := f.Namespace.Name
+		c := f.ClientSet
+		client := c.Core().PodTemplates(ns)
+
+		By("creating a large number of resources")
+		workqueue.Parallelize(20, numberOfTotalResources, func(i int) {
+			for tries := 3; tries >= 0; tries-- {
+				_, err := client.Create(&v1.PodTemplate{
+					ObjectMeta: metav1.ObjectMeta{
+						Name: fmt.Sprintf("template-%04d", i),
+					},
+					Template: v1.PodTemplateSpec{
+						Spec: v1.PodSpec{
+							Containers: []v1.Container{
+								{Name: "test", Image: "test2"},
+							},
+						},
+					},
+				})
+				if err == nil {
+					return
+				}
+				framework.Logf("Got an error creating template %d: %v", i, err)
+			}
+			Fail("Unable to create template %d, exiting", i)
+		})
+
+		By("retrieving those results in paged fashion several times")
+		for i := 0; i < 3; i++ {
+			opts := metav1.ListOptions{}
+			found := 0
+			var lastRV string
+			for {
+				opts.Limit = int64(rand.Int31n(numberOfTotalResources/10) + 1)
+				list, err := client.List(opts)
+				Expect(err).ToNot(HaveOccurred())
+				framework.Logf("Retrieved %d/%d results with rv %s and continue %s", len(list.Items), opts.Limit, list.ResourceVersion, list.Continue)
+
+				if len(lastRV) == 0 {
+					lastRV = list.ResourceVersion
+				}
+				if lastRV != list.ResourceVersion {
+					Expect(list.ResourceVersion).To(Equal(lastRV))
+				}
+				for _, item := range list.Items {
+					Expect(item.Name).To(Equal(fmt.Sprintf("template-%04d", found)))
+					found++
+				}
+				if len(list.Continue) == 0 {
+					break
+				}
+				opts.Continue = list.Continue
+			}
+			Expect(found).To(BeNumerically("==", numberOfTotalResources))
+		}
+
+		By("retrieving those results all at once")
+		list, err := client.List(metav1.ListOptions{Limit: numberOfTotalResources + 1})
+		Expect(err).ToNot(HaveOccurred())
+		Expect(list.Items).To(HaveLen(numberOfTotalResources))
+	})
+})