k8s大集群list问题记录

简单总结

当我们用进行list请求时,如果是以下情况,则list请求会直接访问到etcd而不是经过apiserver的cache,所以会对etcd的压力会比较大
1 list请求没有设置resourceVersion
2 apiserver开启APIListChunking特性(默认已开启)并且list请求设置了Continue
3 apiserver开启APIListChunking特性(默认已开启)并且list请求设置了Limit且resourceVersion不等于'0'
4 list请求设置了match并且不等于NotOlderThan

就算是我们经过apiserver的cache,但是当预期list的资源在k8s内存在大量时候,需要进行大量的过滤,对apiserver的压力会比较大

相关代码

staging/src/k8s.io/apiserver/pkg/registry/generic/registry/store.go中

func (e *Store) List(ctx context.Context, options *metainternalversion.ListOptions) (runtime.Object, error) {
    label := labels.Everything()
    ...
    out, err := e.ListPredicate(ctx, e.PredicateFunc(label, field), options)
    ...
    return out, nil
}

func (e *Store) ListPredicate(ctx context.Context, p storage.SelectionPredicate, options *metainternalversion.ListOptions) (runtime.Object, error) {
    ...
    err := e.Storage.GetList(ctx, e.KeyRootFunc(ctx), storageOpts, list)
    ...
}

staging/src/k8s.io/apiserver/pkg/storage/cacher/cacher.go中(try get from etcd)

func (c *Cacher) GetList(ctx context.Context, key string, opts storage.ListOptions, listObj runtime.Object) error {
    ...
    if shouldDelegateList(opts) {
        return c.storage.GetList(ctx, key, opts, listObj)
    }
    ...
    objs, readResourceVersion, indexUsed, err := c.listItems(ctx, listRV, key, pred, recursive)
    if err != nil {
        return err
    }
    span.AddEvent("Listed items from cache", attribute.Int("count", len(objs)))
    if len(objs) > listVal.Cap() && pred.Label.Empty() && pred.Field.Empty() {
        // Resize the slice appropriately, since we already know that none
        // of the elements will be filtered out.
        listVal.Set(reflect.MakeSlice(reflect.SliceOf(c.objectType.Elem()), 0, len(objs)))
        span.AddEvent("Resized result")
    }
    for _, obj := range objs {
        elem, ok := obj.(*storeElement)
        if !ok {
            return fmt.Errorf("non *storeElement returned from storage: %v", obj)
        }
        if filter(elem.Key, elem.Labels, elem.Fields) {
            listVal.Set(reflect.Append(listVal, reflect.ValueOf(elem.Object).Elem()))
        }
    }
    ...
}

func shouldDelegateList(opts storage.ListOptions) bool {
    resourceVersion := opts.ResourceVersion
    pred := opts.Predicate
    match := opts.ResourceVersionMatch
    pagingEnabled := utilfeature.DefaultFeatureGate.Enabled(features.APIListChunking)
    hasContinuation := pagingEnabled && len(pred.Continue) > 0
    hasLimit := pagingEnabled && pred.Limit > 0 && resourceVersion != "0"
    unsupportedMatch := match != "" && match != metav1.ResourceVersionMatchNotOlderThan

    // If resourceVersion is not specified, serve it from underlying
    // storage (for backward compatibility). If a continuation is
    // requested, serve it from the underlying storage as well.
    // Limits are only sent to storage when resourceVersion is non-zero
    // since the watch cache isn't able to perform continuations, and
    // limits are ignored when resource version is zero
    return resourceVersion == "" || hasContinuation || hasLimit || unsupportedMatch
}

staging/src/k8s.io/apiserver/pkg/storage/etcd3/store.go(get from etcd)

func (s *store) GetList(ctx context.Context, key string, opts storage.ListOptions, listObj runtime.Object) error {
    ...
}