Add RemainingItemCount to ListMeta

Chao Xu · Chao Xu · commit 58c18309a84f · 2019-05-06T12:58:51.000-07:00
diff --git a/staging/src/k8s.io/apimachinery/pkg/apis/meta/v1/meta.go b/staging/src/k8s.io/apimachinery/pkg/apis/meta/v1/meta.go
@@ -94,6 +94,8 @@ type ListInterface interface {
 	SetSelfLink(selfLink string)
 	GetContinue() string
 	SetContinue(c string)
+	GetRemainingItemCount() int64
+	SetRemainingItemCount(c int64)
 }
 
 // Type exposes the type and APIVersion of versioned or internal API objects.
@@ -111,6 +113,8 @@ func (meta *ListMeta) GetSelfLink() string               { return meta.SelfLink
 func (meta *ListMeta) SetSelfLink(selfLink string)       { meta.SelfLink = selfLink }
 func (meta *ListMeta) GetContinue() string               { return meta.Continue }
 func (meta *ListMeta) SetContinue(c string)              { meta.Continue = c }
+func (meta *ListMeta) GetRemainingItemCount() int64      { return meta.RemainingItemCount }
+func (meta *ListMeta) SetRemainingItemCount(c int64)     { meta.RemainingItemCount = c }
 
 func (obj *TypeMeta) GetObjectKind() schema.ObjectKind { return obj }
 
diff --git a/staging/src/k8s.io/apimachinery/pkg/apis/meta/v1/types.go b/staging/src/k8s.io/apimachinery/pkg/apis/meta/v1/types.go
@@ -81,6 +81,14 @@ type ListMeta struct {
 	// identical to the value in the first response, unless you have received this token from an error
 	// message.
 	Continue string `json:"continue,omitempty" protobuf:"bytes,3,opt,name=continue"`
+
+	// RemainingItemCount is the number of subsequent items in the list which are not included in this
+	// list response. If the list request contained label or field selectors, then the number of
+	// remaining items is unknown and this field will be unset. If the list is complete (either
+	// because it is unpaginated or because this is the last page), then there are no more remaining
+	// items and this field will also be unset.  Servers older than v1.15 do not set this field.
+	// +optional
+	RemainingItemCount int64 `json:"remainingItemCount,omitempty" protobuf:"bytes,4,opt,name=remainingItemCount"`
 }
 
 // These are internal finalizer values for Kubernetes-like APIs, must be qualified name unless defined here
diff --git a/staging/src/k8s.io/apimachinery/pkg/apis/meta/v1/unstructured/helpers.go b/staging/src/k8s.io/apimachinery/pkg/apis/meta/v1/unstructured/helpers.go
@@ -275,6 +275,14 @@ func getNestedString(obj map[string]interface{}, fields ...string) string {
 	return val
 }
 
+func getNestedInt64(obj map[string]interface{}, fields ...string) int64 {
+	val, found, err := NestedInt64(obj, fields...)
+	if !found || err != nil {
+		return 0
+	}
+	return val
+}
+
 func jsonPath(fields []string) string {
 	return "." + strings.Join(fields, ".")
 }
diff --git a/staging/src/k8s.io/apimachinery/pkg/apis/meta/v1/unstructured/unstructured_list.go b/staging/src/k8s.io/apimachinery/pkg/apis/meta/v1/unstructured/unstructured_list.go
@@ -166,6 +166,14 @@ func (u *UnstructuredList) SetContinue(c string) {
 	u.setNestedField(c, "metadata", "continue")
 }
 
+func (u *UnstructuredList) GetRemainingItemCount() int64 {
+	return getNestedInt64(u.Object, "metadata", "remainingItemCount")
+}
+
+func (u *UnstructuredList) SetRemainingItemCount(c int64) {
+	u.setNestedField(c, "metadata", "remainingItemCount")
+}
+
 func (u *UnstructuredList) SetGroupVersionKind(gvk schema.GroupVersionKind) {
 	u.SetAPIVersion(gvk.GroupVersion().String())
 	u.SetKind(gvk.Kind)
diff --git a/staging/src/k8s.io/apimachinery/pkg/test/api_meta_meta_test.go b/staging/src/k8s.io/apimachinery/pkg/test/api_meta_meta_test.go
@@ -193,28 +193,31 @@ func TestGenericTypeMeta(t *testing.T) {
 }
 
 type InternalTypeMeta struct {
-	Kind              string                  `json:"kind,omitempty"`
-	Namespace         string                  `json:"namespace,omitempty"`
-	Name              string                  `json:"name,omitempty"`
-	GenerateName      string                  `json:"generateName,omitempty"`
-	UID               string                  `json:"uid,omitempty"`
-	CreationTimestamp metav1.Time             `json:"creationTimestamp,omitempty"`
-	SelfLink          string                  `json:"selfLink,omitempty"`
-	ResourceVersion   string                  `json:"resourceVersion,omitempty"`
-	Continue          string                  `json:"next,omitempty"`
-	APIVersion        string                  `json:"apiVersion,omitempty"`
-	Labels            map[string]string       `json:"labels,omitempty"`
-	Annotations       map[string]string       `json:"annotations,omitempty"`
-	Finalizers        []string                `json:"finalizers,omitempty"`
-	OwnerReferences   []metav1.OwnerReference `json:"ownerReferences,omitempty"`
+	Kind               string                  `json:"kind,omitempty"`
+	Namespace          string                  `json:"namespace,omitempty"`
+	Name               string                  `json:"name,omitempty"`
+	GenerateName       string                  `json:"generateName,omitempty"`
+	UID                string                  `json:"uid,omitempty"`
+	CreationTimestamp  metav1.Time             `json:"creationTimestamp,omitempty"`
+	SelfLink           string                  `json:"selfLink,omitempty"`
+	ResourceVersion    string                  `json:"resourceVersion,omitempty"`
+	Continue           string                  `json:"next,omitempty"`
+	RemainingItemCount int64                   `json:"remainingItemCount,omitempty"`
+	APIVersion         string                  `json:"apiVersion,omitempty"`
+	Labels             map[string]string       `json:"labels,omitempty"`
+	Annotations        map[string]string       `json:"annotations,omitempty"`
+	Finalizers         []string                `json:"finalizers,omitempty"`
+	OwnerReferences    []metav1.OwnerReference `json:"ownerReferences,omitempty"`
 }
 
-func (m *InternalTypeMeta) GetResourceVersion() string   { return m.ResourceVersion }
-func (m *InternalTypeMeta) SetResourceVersion(rv string) { m.ResourceVersion = rv }
-func (m *InternalTypeMeta) GetSelfLink() string          { return m.SelfLink }
-func (m *InternalTypeMeta) SetSelfLink(link string)      { m.SelfLink = link }
-func (m *InternalTypeMeta) GetContinue() string          { return m.Continue }
-func (m *InternalTypeMeta) SetContinue(c string)         { m.Continue = c }
+func (m *InternalTypeMeta) GetResourceVersion() string    { return m.ResourceVersion }
+func (m *InternalTypeMeta) SetResourceVersion(rv string)  { m.ResourceVersion = rv }
+func (m *InternalTypeMeta) GetSelfLink() string           { return m.SelfLink }
+func (m *InternalTypeMeta) SetSelfLink(link string)       { m.SelfLink = link }
+func (m *InternalTypeMeta) GetContinue() string           { return m.Continue }
+func (m *InternalTypeMeta) SetContinue(c string)          { m.Continue = c }
+func (m *InternalTypeMeta) GetRemainingItemCount() int64  { return m.RemainingItemCount }
+func (m *InternalTypeMeta) SetRemainingItemCount(c int64) { m.RemainingItemCount = c }
 
 type MyAPIObject struct {
 	TypeMeta InternalTypeMeta `json:",inline"`
diff --git a/staging/src/k8s.io/apiserver/pkg/storage/cacher/cacher.go b/staging/src/k8s.io/apiserver/pkg/storage/cacher/cacher.go
@@ -604,7 +604,7 @@ func (c *Cacher) GetToList(ctx context.Context, key string, resourceVersion stri
 		}
 	}
 	if c.versioner != nil {
-		if err := c.versioner.UpdateList(listObj, readResourceVersion, ""); err != nil {
+		if err := c.versioner.UpdateList(listObj, readResourceVersion, "", 0); err != nil {
 			return err
 		}
 	}
@@ -679,7 +679,7 @@ func (c *Cacher) List(ctx context.Context, key string, resourceVersion string, p
 	}
 	trace.Step(fmt.Sprintf("Filtered %d items", listVal.Len()))
 	if c.versioner != nil {
-		if err := c.versioner.UpdateList(listObj, readResourceVersion, ""); err != nil {
+		if err := c.versioner.UpdateList(listObj, readResourceVersion, "", 0); err != nil {
 			return err
 		}
 	}
diff --git a/staging/src/k8s.io/apiserver/pkg/storage/cacher/cacher_whitebox_test.go b/staging/src/k8s.io/apiserver/pkg/storage/cacher/cacher_whitebox_test.go
@@ -217,13 +217,14 @@ type testVersioner struct{}
 func (testVersioner) UpdateObject(obj runtime.Object, resourceVersion uint64) error {
 	return meta.NewAccessor().SetResourceVersion(obj, strconv.FormatUint(resourceVersion, 10))
 }
-func (testVersioner) UpdateList(obj runtime.Object, resourceVersion uint64, continueValue string) error {
+func (testVersioner) UpdateList(obj runtime.Object, resourceVersion uint64, continueValue string, count int64) error {
 	listAccessor, err := meta.ListAccessor(obj)
 	if err != nil || listAccessor == nil {
 		return err
 	}
 	listAccessor.SetResourceVersion(strconv.FormatUint(resourceVersion, 10))
 	listAccessor.SetContinue(continueValue)
+	listAccessor.SetRemainingItemCount(count)
 	return nil
 }
 func (testVersioner) PrepareObjectForStorage(obj runtime.Object) error {
diff --git a/staging/src/k8s.io/apiserver/pkg/storage/etcd/api_object_versioner.go b/staging/src/k8s.io/apiserver/pkg/storage/etcd/api_object_versioner.go
@@ -44,7 +44,7 @@ func (a APIObjectVersioner) UpdateObject(obj runtime.Object, resourceVersion uin
 }
 
 // UpdateList implements Versioner
-func (a APIObjectVersioner) UpdateList(obj runtime.Object, resourceVersion uint64, nextKey string) error {
+func (a APIObjectVersioner) UpdateList(obj runtime.Object, resourceVersion uint64, nextKey string, count int64) error {
 	listAccessor, err := meta.ListAccessor(obj)
 	if err != nil || listAccessor == nil {
 		return err
@@ -55,6 +55,7 @@ func (a APIObjectVersioner) UpdateList(obj runtime.Object, resourceVersion uint6
 	}
 	listAccessor.SetResourceVersion(versionString)
 	listAccessor.SetContinue(nextKey)
+	listAccessor.SetRemainingItemCount(count)
 	return nil
 }
 
diff --git a/staging/src/k8s.io/apiserver/pkg/storage/etcd3/store.go b/staging/src/k8s.io/apiserver/pkg/storage/etcd3/store.go
@@ -413,7 +413,7 @@ func (s *store) GetToList(ctx context.Context, key string, resourceVersion strin
 		}
 	}
 	// update version with cluster level revision
-	return s.versioner.UpdateList(listObj, uint64(getResp.Header.Revision), "")
+	return s.versioner.UpdateList(listObj, uint64(getResp.Header.Revision), "", 0)
 }
 
 func (s *store) Count(key string) (int64, error) {
@@ -576,9 +576,10 @@ func (s *store) List(ctx context.Context, key, resourceVersion string, pred stor
 	// loop until we have filled the requested limit from etcd or there are no more results
 	var lastKey []byte
 	var hasMore bool
+	var getResp *clientv3.GetResponse
 	for {
 		startTime := time.Now()
-		getResp, err := s.client.KV.Get(ctx, key, options...)
+		getResp, err = s.client.KV.Get(ctx, key, options...)
 		metrics.RecordEtcdRequestLatency("list", getTypeName(listPtr), startTime)
 		if err != nil {
 			return interpretListError(err, len(pred.Continue) > 0, continueKey, keyPrefix)
@@ -639,11 +640,17 @@ func (s *store) List(ctx context.Context, key, resourceVersion string, pred stor
 		if err != nil {
 			return err
 		}
-		return s.versioner.UpdateList(listObj, uint64(returnedRV), next)
+		remainingItemCount := getResp.Count - pred.Limit
+		// getResp.Count counts in objects that do not match the pred.
+		// Instead of returning inaccurate count, return 0.
+		if !pred.Empty() {
+			remainingItemCount = 0
+		}
+		return s.versioner.UpdateList(listObj, uint64(returnedRV), next, remainingItemCount)
 	}
 
 	// no continuation
-	return s.versioner.UpdateList(listObj, uint64(returnedRV), "")
+	return s.versioner.UpdateList(listObj, uint64(returnedRV), "", 0)
 }
 
 // growSlice takes a slice value and grows its capacity up
diff --git a/staging/src/k8s.io/apiserver/pkg/storage/etcd3/store_test.go b/staging/src/k8s.io/apiserver/pkg/storage/etcd3/store_test.go
@@ -825,14 +825,15 @@ func TestList(t *testing.T) {
 	}
 
 	tests := []struct {
-		name           string
-		disablePaging  bool
-		rv             string
-		prefix         string
-		pred           storage.SelectionPredicate
-		expectedOut    []*example.Pod
-		expectContinue bool
-		expectError    bool
+		name                       string
+		disablePaging              bool
+		rv                         string
+		prefix                     string
+		pred                       storage.SelectionPredicate
+		expectedOut                []*example.Pod
+		expectContinue             bool
+		expectedRemainingItemCount int64
+		expectError                bool
 	}{
 		{
 			name:        "rejects invalid resource version",
@@ -882,8 +883,9 @@ func TestList(t *testing.T) {
 				Field: fields.Everything(),
 				Limit: 1,
 			},
-			expectedOut:    []*example.Pod{preset[1].storedObj},
-			expectContinue: true,
+			expectedOut:                []*example.Pod{preset[1].storedObj},
+			expectContinue:             true,
+			expectedRemainingItemCount: 1,
 		},
 		{
 			name:          "test List with limit when paging disabled",
@@ -1061,6 +1063,9 @@ func TestList(t *testing.T) {
 			t.Errorf("(%s): length of list want=%d, got=%d", tt.name, len(tt.expectedOut), len(out.Items))
 			continue
 		}
+		if e, a := tt.expectedRemainingItemCount, out.ListMeta.RemainingItemCount; e != a {
+			t.Errorf("(%s): remainingItemCount want=%d, got=%d", tt.name, e, a)
+		}
 		for j, wantPod := range tt.expectedOut {
 			getPod := &out.Items[j]
 			if !reflect.DeepEqual(wantPod, getPod) {
diff --git a/staging/src/k8s.io/apiserver/pkg/storage/interfaces.go b/staging/src/k8s.io/apiserver/pkg/storage/interfaces.go
@@ -40,10 +40,12 @@ type Versioner interface {
 	// from database.
 	UpdateObject(obj runtime.Object, resourceVersion uint64) error
 	// UpdateList sets the resource version into an API list object. Returns an error if the object
-	// cannot be updated correctly. May return nil if the requested object does not need metadata
-	// from database. continueValue is optional and indicates that more results are available if
-	// the client passes that value to the server in a subsequent call.
-	UpdateList(obj runtime.Object, resourceVersion uint64, continueValue string) error
+	// cannot be updated correctly. May return nil if the requested object does not need metadata from
+	// database. continueValue is optional and indicates that more results are available if the client
+	// passes that value to the server in a subsequent call. remainingItemCount indicates the number
+	// of remaining objects if the list is partial. The remainingItemCount field is omitted during
+	// serialization if it is set to 0.
+	UpdateList(obj runtime.Object, resourceVersion uint64, continueValue string, remainingItemCount int64) error
 	// PrepareObjectForStorage should set SelfLink and ResourceVersion to the empty value. Should
 	// return an error if the specified object cannot be updated.
 	PrepareObjectForStorage(obj runtime.Object) error
diff --git a/test/e2e/apimachinery/chunking.go b/test/e2e/apimachinery/chunking.go
@@ -26,7 +26,7 @@ import (
 	"github.com/onsi/ginkgo"
 	"github.com/onsi/gomega"
 
-	"k8s.io/api/core/v1"
+	v1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/util/wait"
@@ -88,6 +88,7 @@ var _ = SIGDescribe("Servers with support for API chunking", func() {
 					lastRV = list.ResourceVersion
 				}
 				gomega.Expect(list.ResourceVersion).To(gomega.Equal(lastRV))
+				gomega.Expect(int(list.RemainingItemCount) + len(list.Items) + found).To(gomega.BeNumerically("==", numberOfTotalResources))
 				for _, item := range list.Items {
 					gomega.Expect(item.Name).To(gomega.Equal(fmt.Sprintf("template-%04d", found)))
 					found++
@@ -120,6 +121,7 @@ var _ = SIGDescribe("Servers with support for API chunking", func() {
 		gomega.Expect(err).ToNot(gomega.HaveOccurred(), "failed to list pod templates in namespace: %s, given limit: %d", ns, opts.Limit)
 		firstToken := list.Continue
 		firstRV := list.ResourceVersion
+		gomega.Expect(int(list.RemainingItemCount) + len(list.Items)).To(gomega.BeNumerically("==", numberOfTotalResources))
 		framework.Logf("Retrieved %d/%d results with rv %s and continue %s", len(list.Items), opts.Limit, list.ResourceVersion, firstToken)
 
 		ginkgo.By("retrieving the second page until the token expires")
@@ -153,7 +155,8 @@ var _ = SIGDescribe("Servers with support for API chunking", func() {
 		gomega.Expect(err).ToNot(gomega.HaveOccurred(), "failed to list pod templates in namespace: %s, given inconsistent continue token %s and limit: %d", ns, opts.Continue, opts.Limit)
 		gomega.Expect(list.ResourceVersion).ToNot(gomega.Equal(firstRV))
 		gomega.Expect(len(list.Items)).To(gomega.BeNumerically("==", opts.Limit))
-		found := oneTenth
+		found := int(oneTenth)
+		gomega.Expect(int(list.RemainingItemCount) + len(list.Items) + found).To(gomega.BeNumerically("==", numberOfTotalResources))
 		for _, item := range list.Items {
 			gomega.Expect(item.Name).To(gomega.Equal(fmt.Sprintf("template-%04d", found)))
 			found++
@@ -165,6 +168,7 @@ var _ = SIGDescribe("Servers with support for API chunking", func() {
 		for {
 			list, err := client.List(opts)
 			gomega.Expect(err).ToNot(gomega.HaveOccurred(), "failed to list pod templates in namespace: %s, given limit: %d", ns, opts.Limit)
+			gomega.Expect(int(list.RemainingItemCount) + len(list.Items) + found).To(gomega.BeNumerically("==", numberOfTotalResources))
 			framework.Logf("Retrieved %d/%d results with rv %s and continue %s", len(list.Items), opts.Limit, list.ResourceVersion, list.Continue)
 			gomega.Expect(len(list.Items)).To(gomega.BeNumerically("<=", opts.Limit))
 			gomega.Expect(list.ResourceVersion).To(gomega.Equal(lastRV))

Original file line number	Diff line number	Diff line change
`@@ -275,6 +275,14 @@ func getNestedString(obj map[string]interface{}, fields ...string) string {`
`275`	`275`	`return val`
`276`	`276`	`}`
`277`	`277`
	`278`	`+func getNestedInt64(obj map[string]interface{}, fields ...string) int64 {`
	`279`	`+ val, found, err := NestedInt64(obj, fields...)`
	`280`	`+ if !found \|\| err != nil {`
	`281`	`+ return 0`
	`282`	`+ }`
	`283`	`+ return val`
	`284`	`+}`
	`285`	`+`
`278`	`286`	`func jsonPath(fields []string) string {`
`279`	`287`	`return "." + strings.Join(fields, ".")`
`280`	`288`	`}`
Original file line number	Diff line number	Diff line change
`@@ -604,7 +604,7 @@ func (c *Cacher) GetToList(ctx context.Context, key string, resourceVersion stri`
`604`	`604`	`}`
`605`	`605`	`}`
`606`	`606`	`if c.versioner != nil {`
`607`		`- if err := c.versioner.UpdateList(listObj, readResourceVersion, ""); err != nil {`
	`607`	`+ if err := c.versioner.UpdateList(listObj, readResourceVersion, "", 0); err != nil {`
`608`	`608`	`return err`
`609`	`609`	`}`
`610`	`610`	`}`
`@@ -679,7 +679,7 @@ func (c *Cacher) List(ctx context.Context, key string, resourceVersion string, p`
`679`	`679`	`}`
`680`	`680`	`trace.Step(fmt.Sprintf("Filtered %d items", listVal.Len()))`
`681`	`681`	`if c.versioner != nil {`
`682`		`- if err := c.versioner.UpdateList(listObj, readResourceVersion, ""); err != nil {`
	`682`	`+ if err := c.versioner.UpdateList(listObj, readResourceVersion, "", 0); err != nil {`
`683`	`683`	`return err`
`684`	`684`	`}`
`685`	`685`	`}`
Original file line number	Diff line number	Diff line change
`@@ -217,13 +217,14 @@ type testVersioner struct{}`
`217`	`217`	`func (testVersioner) UpdateObject(obj runtime.Object, resourceVersion uint64) error {`
`218`	`218`	`return meta.NewAccessor().SetResourceVersion(obj, strconv.FormatUint(resourceVersion, 10))`
`219`	`219`	`}`
`220`		`-func (testVersioner) UpdateList(obj runtime.Object, resourceVersion uint64, continueValue string) error {`
	`220`	`+func (testVersioner) UpdateList(obj runtime.Object, resourceVersion uint64, continueValue string, count int64) error {`
`221`	`221`	`listAccessor, err := meta.ListAccessor(obj)`
`222`	`222`	`if err != nil \|\| listAccessor == nil {`
`223`	`223`	`return err`
`224`	`224`	`}`
`225`	`225`	`listAccessor.SetResourceVersion(strconv.FormatUint(resourceVersion, 10))`
`226`	`226`	`listAccessor.SetContinue(continueValue)`
	`227`	`+ listAccessor.SetRemainingItemCount(count)`
`227`	`228`	`return nil`
`228`	`229`	`}`
`229`	`230`	`func (testVersioner) PrepareObjectForStorage(obj runtime.Object) error {`
Original file line number	Diff line number	Diff line change
`@@ -44,7 +44,7 @@ func (a APIObjectVersioner) UpdateObject(obj runtime.Object, resourceVersion uin`
`44`	`44`	`}`
`45`	`45`
`46`	`46`	`// UpdateList implements Versioner`
`47`		`-func (a APIObjectVersioner) UpdateList(obj runtime.Object, resourceVersion uint64, nextKey string) error {`
	`47`	`+func (a APIObjectVersioner) UpdateList(obj runtime.Object, resourceVersion uint64, nextKey string, count int64) error {`
`48`	`48`	`listAccessor, err := meta.ListAccessor(obj)`
`49`	`49`	`if err != nil \|\| listAccessor == nil {`
`50`	`50`	`return err`
`@@ -55,6 +55,7 @@ func (a APIObjectVersioner) UpdateList(obj runtime.Object, resourceVersion uint6`
`55`	`55`	`}`
`56`	`56`	`listAccessor.SetResourceVersion(versionString)`
`57`	`57`	`listAccessor.SetContinue(nextKey)`
	`58`	`+ listAccessor.SetRemainingItemCount(count)`
`58`	`59`	`return nil`
`59`	`60`	`}`
`60`	`61`