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 index 05f07adf1b8..dbe3ff59929 100644 --- 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 index d3547940024..e2bb511e423 100644 --- 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 index 75ac693fe48..a90849176b3 100644 --- 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 index bf3fd023f4d..f0c204899ad 100644 --- 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 index 157360a191a..a8582c51bba 100644 --- 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 index 4bc8d323915..51c3d36ca7b 100644 --- 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 index 9f9016dd899..a737b9ccb7a 100644 --- 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 index 2fed9f486fe..dc2b35da091 100644 --- 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 index 34e91361768..1216b1f1e27 100644 --- 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 index 8f4ae504a91..4d4c1e158ca 100644 --- 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 index 108f2acb479..f2a1f105373 100644 --- 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 index 05c0256f50d..47ac30f0a78 100644 --- 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))