k3s/vendor/k8s.io/client-go/tools/pager/pager.go

248 lines
9.1 KiB
Go
Raw Normal View History

2019-01-12 04:58:27 +00:00
/*
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 pager
import (
"context"
"fmt"
"k8s.io/apimachinery/pkg/api/errors"
"k8s.io/apimachinery/pkg/api/meta"
metainternalversion "k8s.io/apimachinery/pkg/apis/meta/internalversion"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/runtime"
2019-08-30 18:33:25 +00:00
utilruntime "k8s.io/apimachinery/pkg/util/runtime"
2019-01-12 04:58:27 +00:00
)
const defaultPageSize = 500
2019-08-30 18:33:25 +00:00
const defaultPageBufferSize = 10
2019-01-12 04:58:27 +00:00
// ListPageFunc returns a list object for the given list options.
type ListPageFunc func(ctx context.Context, opts metav1.ListOptions) (runtime.Object, error)
// SimplePageFunc adapts a context-less list function into one that accepts a context.
func SimplePageFunc(fn func(opts metav1.ListOptions) (runtime.Object, error)) ListPageFunc {
return func(ctx context.Context, opts metav1.ListOptions) (runtime.Object, error) {
return fn(opts)
}
}
// ListPager assists client code in breaking large list queries into multiple
// smaller chunks of PageSize or smaller. PageFn is expected to accept a
// metav1.ListOptions that supports paging and return a list. The pager does
// not alter the field or label selectors on the initial options list.
type ListPager struct {
PageSize int64
PageFn ListPageFunc
FullListIfExpired bool
2019-08-30 18:33:25 +00:00
// Number of pages to buffer
PageBufferSize int32
2019-01-12 04:58:27 +00:00
}
// New creates a new pager from the provided pager function using the default
// options. It will fall back to a full list if an expiration error is encountered
// as a last resort.
func New(fn ListPageFunc) *ListPager {
return &ListPager{
PageSize: defaultPageSize,
PageFn: fn,
FullListIfExpired: true,
2019-08-30 18:33:25 +00:00
PageBufferSize: defaultPageBufferSize,
2019-01-12 04:58:27 +00:00
}
}
// TODO: introduce other types of paging functions - such as those that retrieve from a list
// of namespaces.
// List returns a single list object, but attempts to retrieve smaller chunks from the
// server to reduce the impact on the server. If the chunk attempt fails, it will load
// the full list instead. The Limit field on options, if unset, will default to the page size.
2020-03-26 21:07:15 +00:00
func (p *ListPager) List(ctx context.Context, options metav1.ListOptions) (runtime.Object, bool, error) {
2019-01-12 04:58:27 +00:00
if options.Limit == 0 {
options.Limit = p.PageSize
}
2019-12-12 01:27:03 +00:00
requestedResourceVersion := options.ResourceVersion
2019-01-12 04:58:27 +00:00
var list *metainternalversion.List
2020-03-26 21:07:15 +00:00
paginatedResult := false
2019-01-12 04:58:27 +00:00
for {
2019-08-30 18:33:25 +00:00
select {
case <-ctx.Done():
2020-03-26 21:07:15 +00:00
return nil, paginatedResult, ctx.Err()
2019-08-30 18:33:25 +00:00
default:
}
2019-01-12 04:58:27 +00:00
obj, err := p.PageFn(ctx, options)
if err != nil {
2019-12-12 01:27:03 +00:00
// Only fallback to full list if an "Expired" errors is returned, FullListIfExpired is true, and
// the "Expired" error occurred in page 2 or later (since full list is intended to prevent a pager.List from
// failing when the resource versions is established by the first page request falls out of the compaction
// during the subsequent list requests).
if !errors.IsResourceExpired(err) || !p.FullListIfExpired || options.Continue == "" {
2020-03-26 21:07:15 +00:00
return nil, paginatedResult, err
2019-01-12 04:58:27 +00:00
}
2019-12-12 01:27:03 +00:00
// the list expired while we were processing, fall back to a full list at
// the requested ResourceVersion.
2019-01-12 04:58:27 +00:00
options.Limit = 0
options.Continue = ""
2019-12-12 01:27:03 +00:00
options.ResourceVersion = requestedResourceVersion
2020-03-26 21:07:15 +00:00
result, err := p.PageFn(ctx, options)
return result, paginatedResult, err
2019-01-12 04:58:27 +00:00
}
m, err := meta.ListAccessor(obj)
if err != nil {
2020-03-26 21:07:15 +00:00
return nil, paginatedResult, fmt.Errorf("returned object must be a list: %v", err)
2019-01-12 04:58:27 +00:00
}
// exit early and return the object we got if we haven't processed any pages
if len(m.GetContinue()) == 0 && list == nil {
2020-03-26 21:07:15 +00:00
return obj, paginatedResult, nil
2019-01-12 04:58:27 +00:00
}
// initialize the list and fill its contents
if list == nil {
list = &metainternalversion.List{Items: make([]runtime.Object, 0, options.Limit+1)}
list.ResourceVersion = m.GetResourceVersion()
list.SelfLink = m.GetSelfLink()
}
if err := meta.EachListItem(obj, func(obj runtime.Object) error {
list.Items = append(list.Items, obj)
return nil
}); err != nil {
2020-03-26 21:07:15 +00:00
return nil, paginatedResult, err
2019-01-12 04:58:27 +00:00
}
// if we have no more items, return the list
if len(m.GetContinue()) == 0 {
2020-03-26 21:07:15 +00:00
return list, paginatedResult, nil
2019-01-12 04:58:27 +00:00
}
// set the next loop up
options.Continue = m.GetContinue()
2019-12-12 01:27:03 +00:00
// Clear the ResourceVersion on the subsequent List calls to avoid the
// `specifying resource version is not allowed when using continue` error.
// See https://github.com/kubernetes/kubernetes/issues/85221#issuecomment-553748143.
options.ResourceVersion = ""
2020-03-26 21:07:15 +00:00
// At this point, result is already paginated.
paginatedResult = true
2019-01-12 04:58:27 +00:00
}
}
2019-08-30 18:33:25 +00:00
// EachListItem fetches runtime.Object items using this ListPager and invokes fn on each item. If
// fn returns an error, processing stops and that error is returned. If fn does not return an error,
// any error encountered while retrieving the list from the server is returned. If the context
// cancels or times out, the context error is returned. Since the list is retrieved in paginated
// chunks, an "Expired" error (metav1.StatusReasonExpired) may be returned if the pagination list
// requests exceed the expiration limit of the apiserver being called.
//
// Items are retrieved in chunks from the server to reduce the impact on the server with up to
// ListPager.PageBufferSize chunks buffered concurrently in the background.
func (p *ListPager) EachListItem(ctx context.Context, options metav1.ListOptions, fn func(obj runtime.Object) error) error {
return p.eachListChunkBuffered(ctx, options, func(obj runtime.Object) error {
return meta.EachListItem(obj, fn)
})
}
// eachListChunkBuffered fetches runtimeObject list chunks using this ListPager and invokes fn on
// each list chunk. If fn returns an error, processing stops and that error is returned. If fn does
// not return an error, any error encountered while retrieving the list from the server is
// returned. If the context cancels or times out, the context error is returned. Since the list is
// retrieved in paginated chunks, an "Expired" error (metav1.StatusReasonExpired) may be returned if
// the pagination list requests exceed the expiration limit of the apiserver being called.
//
// Up to ListPager.PageBufferSize chunks are buffered concurrently in the background.
func (p *ListPager) eachListChunkBuffered(ctx context.Context, options metav1.ListOptions, fn func(obj runtime.Object) error) error {
if p.PageBufferSize < 0 {
return fmt.Errorf("ListPager.PageBufferSize must be >= 0, got %d", p.PageBufferSize)
}
// Ensure background goroutine is stopped if this call exits before all list items are
// processed. Cancelation error from this deferred cancel call is never returned to caller;
// either the list result has already been sent to bgResultC or the fn error is returned and
// the cancelation error is discarded.
ctx, cancel := context.WithCancel(ctx)
defer cancel()
chunkC := make(chan runtime.Object, p.PageBufferSize)
bgResultC := make(chan error, 1)
go func() {
defer utilruntime.HandleCrash()
var err error
defer func() {
close(chunkC)
bgResultC <- err
}()
err = p.eachListChunk(ctx, options, func(chunk runtime.Object) error {
select {
case chunkC <- chunk: // buffer the chunk, this can block
case <-ctx.Done():
return ctx.Err()
}
return nil
})
}()
for o := range chunkC {
err := fn(o)
if err != nil {
return err // any fn error should be returned immediately
}
}
// promote the results of our background goroutine to the foreground
return <-bgResultC
}
// eachListChunk fetches runtimeObject list chunks using this ListPager and invokes fn on each list
// chunk. If fn returns an error, processing stops and that error is returned. If fn does not return
// an error, any error encountered while retrieving the list from the server is returned. If the
// context cancels or times out, the context error is returned. Since the list is retrieved in
// paginated chunks, an "Expired" error (metav1.StatusReasonExpired) may be returned if the
// pagination list requests exceed the expiration limit of the apiserver being called.
func (p *ListPager) eachListChunk(ctx context.Context, options metav1.ListOptions, fn func(obj runtime.Object) error) error {
if options.Limit == 0 {
options.Limit = p.PageSize
}
for {
select {
case <-ctx.Done():
return ctx.Err()
default:
}
obj, err := p.PageFn(ctx, options)
if err != nil {
return err
}
m, err := meta.ListAccessor(obj)
if err != nil {
return fmt.Errorf("returned object must be a list: %v", err)
}
if err := fn(obj); err != nil {
return err
}
// if we have no more items, return.
if len(m.GetContinue()) == 0 {
return nil
}
// set the next loop up
options.Continue = m.GetContinue()
}
}