kubernetes
/
apiserver
mirror of https://github.com/kubernetes/apiserver.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #129440 from serathius/watchcache-extract-list-response 

Extract list response struct to manage all the response fields

Kubernetes-commit: 0e78cf18d19592e63d069e9a8bd72c2e87dbee0c
This commit is contained in:
Kubernetes Publisher 2025-01-07 14:08:41 -08:00
parent 2cad252f0c 62c90b1dff
commit 893803dc85
6 changed files with 78 additions and 71 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
go.mod
View File

@ -52,7 +52,7 @@ require (
gopkg.in/evanphx/json-patch.v4 v4.12.0
gopkg.in/natefinch/lumberjack.v2 v2.2.1
gopkg.in/square/go-jose.v2 v2.6.0
k8s.io/api v0.0.0-20250106201850-018f9a59fc40
k8s.io/api v0.0.0-20250107083829-9e7d345b161c
k8s.io/apimachinery v0.0.0-20250106201545-3e8e52d6a125
k8s.io/client-go v0.0.0-20250106202309-ca1e525bb47b
k8s.io/component-base v0.0.0-20250106203308-7356cc73422d

4
go.sum
View File

@ -365,8 +365,8 @@ gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
honnef.co/go/tools v0.0.0-20190102054323-c2f93a96b099/go.mod h1:rf3lG4BRIbNafJWhAfAdb/ePZxsR/4RtNHQocxwk9r4=
honnef.co/go/tools v0.0.0-20190523083050-ea95bdfd59fc/go.mod h1:rf3lG4BRIbNafJWhAfAdb/ePZxsR/4RtNHQocxwk9r4=
k8s.io/api v0.0.0-20250106201850-018f9a59fc40 h1:Iwg5OLnTsPB6uTOBsPevZgVRZVQRIC/JRtQ1kGBQm2k=
k8s.io/api v0.0.0-20250106201850-018f9a59fc40/go.mod h1:bcTx9U6z8NtpUbnc3va/3EXi9J7dYqxrv93ABhJNcQ0=
k8s.io/api v0.0.0-20250107083829-9e7d345b161c h1:3OdBzgr0PF+mKK9xH/ycqxxZacdV2JJnaxgj32yUc8o=
k8s.io/api v0.0.0-20250107083829-9e7d345b161c/go.mod h1:bcTx9U6z8NtpUbnc3va/3EXi9J7dYqxrv93ABhJNcQ0=
k8s.io/apimachinery v0.0.0-20250106201545-3e8e52d6a125 h1:DKE2wIHpBRYleqyn7WO7dmFTMvZWXwaKvB3qyZaKV/o=
k8s.io/apimachinery v0.0.0-20250106201545-3e8e52d6a125/go.mod h1:iKrKOPQq2bzYghWivf0Pyy2ZJ8A6WtFW16/tSLqk0Sg=
k8s.io/client-go v0.0.0-20250106202309-ca1e525bb47b h1:VYZC/349kyKmT4SJmtItkoQuTuOdCu3UHEfI1wBMhls=

27
pkg/storage/cacher/cacher.go
View File

@ -825,20 +825,25 @@ func shouldDelegateListOnNotReadyCache(opts storage.ListOptions) bool {
return noLabelSelector && noFieldSelector && hasLimit
}
func (c *Cacher) listItems(ctx context.Context, listRV uint64, key string, pred storage.SelectionPredicate, recursive bool) ([]interface{}, uint64, string, error) {
func (c *Cacher) listItems(ctx context.Context, listRV uint64, key string, pred storage.SelectionPredicate, recursive bool) (listResp, string, error) {
if !recursive {
obj, exists, readResourceVersion, err := c.watchCache.WaitUntilFreshAndGet(ctx, listRV, key)
if err != nil {
return nil, 0, "", err
return listResp{}, "", err
}
if exists {
return []interface{}{obj}, readResourceVersion, "", nil
return listResp{Items: []interface{}{obj}, ResourceVersion: readResourceVersion}, "", nil
}
return nil, readResourceVersion, "", nil
return listResp{ResourceVersion: readResourceVersion}, "", nil
}
return c.watchCache.WaitUntilFreshAndList(ctx, listRV, key, pred.MatcherIndex(ctx))
}
type listResp struct {
Items []interface{}
ResourceVersion uint64
}
// GetList implements storage.Interface
func (c *Cacher) GetList(ctx context.Context, key string, opts storage.ListOptions, listObj runtime.Object) error {
recursive := opts.Recursive
@ -914,7 +919,7 @@ func (c *Cacher) GetList(ctx context.Context, key string, opts storage.ListOptio
return fmt.Errorf("need a pointer to slice, got %v", listVal.Kind())
}
objs, readResourceVersion, indexUsed, err := c.listItems(ctx, listRV, preparedKey, pred, recursive)
resp, indexUsed, err := c.listItems(ctx, listRV, preparedKey, pred, recursive)
success := "true"
fallback := "false"
if err != nil {
@ -933,7 +938,7 @@ func (c *Cacher) GetList(ctx context.Context, key string, opts storage.ListOptio
if consistentRead {
metrics.ConsistentReadTotal.WithLabelValues(c.resourcePrefix, success, fallback).Add(1)
}
span.AddEvent("Listed items from cache", attribute.Int("count", len(objs)))
span.AddEvent("Listed items from cache", attribute.Int("count", len(resp.Items)))
// store pointer of eligible objects,
// Why not directly put object in the items of listObj?
// the elements in ListObject are Struct type, making slice will bring excessive memory consumption.
@ -942,7 +947,7 @@ func (c *Cacher) GetList(ctx context.Context, key string, opts storage.ListOptio
var lastSelectedObjectKey string
var hasMoreListItems bool
limit := computeListLimit(opts)
for i, obj := range objs {
for i, obj := range resp.Items {
elem, ok := obj.(*storeElement)
if !ok {
return fmt.Errorf("non *storeElement returned from storage: %v", obj)
@ -952,7 +957,7 @@ func (c *Cacher) GetList(ctx context.Context, key string, opts storage.ListOptio
lastSelectedObjectKey = elem.Key
}
if limit > 0 && int64(len(selectedObjects)) >= limit {
hasMoreListItems = i < len(objs)-1
hasMoreListItems = i < len(resp.Items)-1
break
}
}
@ -969,16 +974,16 @@ func (c *Cacher) GetList(ctx context.Context, key string, opts storage.ListOptio
}
span.AddEvent("Filtered items", attribute.Int("count", listVal.Len()))
if c.versioner != nil {
continueValue, remainingItemCount, err := storage.PrepareContinueToken(lastSelectedObjectKey, key, int64(readResourceVersion), int64(len(objs)), hasMoreListItems, opts)
continueValue, remainingItemCount, err := storage.PrepareContinueToken(lastSelectedObjectKey, key, int64(resp.ResourceVersion), int64(len(resp.Items)), hasMoreListItems, opts)
if err != nil {
return err
}
if err = c.versioner.UpdateList(listObj, readResourceVersion, continueValue, remainingItemCount); err != nil {
if err = c.versioner.UpdateList(listObj, resp.ResourceVersion, continueValue, remainingItemCount); err != nil {
return err
}
}
metrics.RecordListCacheMetrics(c.resourcePrefix, indexUsed, len(objs), listVal.Len())
metrics.RecordListCacheMetrics(c.resourcePrefix, indexUsed, len(resp.Items), listVal.Len())
return nil
}

2
pkg/storage/cacher/cacher_whitebox_test.go
View File

@ -3079,7 +3079,7 @@ func TestListIndexer(t *testing.T) {
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
pred := storagetesting.CreatePodPredicate(tt.fieldSelector, true, tt.indexFields)
_, _, usedIndex, err := cacher.listItems(ctx, 0, "/pods/"+tt.requestedNamespace, pred, tt.recursive)
_, usedIndex, err := cacher.listItems(ctx, 0, "/pods/"+tt.requestedNamespace, pred, tt.recursive)
if err != nil {
t.Errorf("Unexpected error: %v", err)
}

30
pkg/storage/cacher/watch_cache.go
View File

@ -451,7 +451,7 @@ func (s sortableStoreElements) Swap(i, j int) {
// WaitUntilFreshAndList returns list of pointers to `storeElement` objects along
// with their ResourceVersion and the name of the index, if any, that was used.
func (w *watchCache) WaitUntilFreshAndList(ctx context.Context, resourceVersion uint64, key string, matchValues []storage.MatchValue) (result []interface{}, rv uint64, index string, err error) {
func (w *watchCache) WaitUntilFreshAndList(ctx context.Context, resourceVersion uint64, key string, matchValues []storage.MatchValue) (resp listResp, index string, err error) {
requestWatchProgressSupported := etcdfeature.DefaultFeatureSupportChecker.Supports(storage.RequestWatchProgress)
if utilfeature.DefaultFeatureGate.Enabled(features.ConsistentListFromCache) && requestWatchProgressSupported && w.notFresh(resourceVersion) {
w.waitingUntilFresh.Add()
@ -463,32 +463,34 @@ func (w *watchCache) WaitUntilFreshAndList(ctx context.Context, resourceVersion
defer w.RUnlock()
if err != nil {
return result, rv, index, err
return listResp{}, "", err
}
var prefixFilteredAndOrdered bool
result, rv, index, prefixFilteredAndOrdered, err = func() ([]interface{}, uint64, string, bool, error) {
// This isn't the place where we do "final filtering" - only some "prefiltering" is happening here. So the only
// requirement here is to NOT miss anything that should be returned. We can return as many non-matching items as we
// want - they will be filtered out later. The fact that we return less things is only further performance improvement.
// TODO: if multiple indexes match, return the one with the fewest items, so as to do as much filtering as possible.
for _, matchValue := range matchValues {
if result, err := w.store.ByIndex(matchValue.IndexName, matchValue.Value); err == nil {
return result, w.resourceVersion, matchValue.IndexName, false, nil
result, err = filterPrefixAndOrder(key, result)
return listResp{
Items: result,
ResourceVersion: w.resourceVersion,
}, matchValue.IndexName, err
}
}
if store, ok := w.store.(orderedLister); ok {
result, _ := store.ListPrefix(key, "", 0)
return result, w.resourceVersion, "", true, nil
return listResp{
Items: result,
ResourceVersion: w.resourceVersion,
}, "", nil
}
return w.store.List(), w.resourceVersion, "", false, nil
}()
if !prefixFilteredAndOrdered {
result := w.store.List()
result, err = filterPrefixAndOrder(key, result)
if err != nil {
return nil, 0, "", err
}
}
return result, w.resourceVersion, index, nil
return listResp{
Items: result,
ResourceVersion: w.resourceVersion,
}, "", err
}
func filterPrefixAndOrder(prefix string, items []interface{}) ([]interface{}, error) {

64
pkg/storage/cacher/watch_cache_test.go
View File

@ -462,15 +462,15 @@ func TestWaitUntilFreshAndList(t *testing.T) {
}()
// list by empty MatchValues.
list, resourceVersion, indexUsed, err := store.WaitUntilFreshAndList(ctx, 5, "prefix/", nil)
resp, indexUsed, err := store.WaitUntilFreshAndList(ctx, 5, "prefix/", nil)
if err != nil {
t.Fatalf("unexpected error: %v", err)
}
if resourceVersion != 5 {
t.Errorf("unexpected resourceVersion: %v, expected: 5", resourceVersion)
if resp.ResourceVersion != 5 {
t.Errorf("unexpected resourceVersion: %v, expected: 5", resp.ResourceVersion)
}
if len(list) != 3 {
t.Errorf("unexpected list returned: %#v", list)
if len(resp.Items) != 3 {
t.Errorf("unexpected list returned: %#v", resp)
}
if indexUsed != "" {
t.Errorf("Used index %q but expected none to be used", indexUsed)
@ -481,15 +481,15 @@ func TestWaitUntilFreshAndList(t *testing.T) {
{IndexName: "l:label", Value: "value1"},
{IndexName: "f:spec.nodeName", Value: "node2"},
}
list, resourceVersion, indexUsed, err = store.WaitUntilFreshAndList(ctx, 5, "prefix/", matchValues)
resp, indexUsed, err = store.WaitUntilFreshAndList(ctx, 5, "prefix/", matchValues)
if err != nil {
t.Fatalf("unexpected error: %v", err)
}
if resourceVersion != 5 {
t.Errorf("unexpected resourceVersion: %v, expected: 5", resourceVersion)
if resp.ResourceVersion != 5 {
t.Errorf("unexpected resourceVersion: %v, expected: 5", resp.ResourceVersion)
}
if len(list) != 2 {
t.Errorf("unexpected list returned: %#v", list)
if len(resp.Items) != 2 {
t.Errorf("unexpected list returned: %#v", resp)
}
if indexUsed != "l:label" {
t.Errorf("Used index %q but expected %q", indexUsed, "l:label")
@ -500,15 +500,15 @@ func TestWaitUntilFreshAndList(t *testing.T) {
{IndexName: "l:not-exist-label", Value: "whatever"},
{IndexName: "f:spec.nodeName", Value: "node2"},
}
list, resourceVersion, indexUsed, err = store.WaitUntilFreshAndList(ctx, 5, "prefix/", matchValues)
resp, indexUsed, err = store.WaitUntilFreshAndList(ctx, 5, "prefix/", matchValues)
if err != nil {
t.Fatalf("unexpected error: %v", err)
}
if resourceVersion != 5 {
t.Errorf("unexpected resourceVersion: %v, expected: 5", resourceVersion)
if resp.ResourceVersion != 5 {
t.Errorf("unexpected resourceVersion: %v, expected: 5", resp.ResourceVersion)
}
if len(list) != 1 {
t.Errorf("unexpected list returned: %#v", list)
if len(resp.Items) != 1 {
t.Errorf("unexpected list returned: %#v", resp)
}
if indexUsed != "f:spec.nodeName" {
t.Errorf("Used index %q but expected %q", indexUsed, "f:spec.nodeName")
@ -518,15 +518,15 @@ func TestWaitUntilFreshAndList(t *testing.T) {
matchValues = []storage.MatchValue{
{IndexName: "l:not-exist-label", Value: "whatever"},
}
list, resourceVersion, indexUsed, err = store.WaitUntilFreshAndList(ctx, 5, "prefix/", matchValues)
resp, indexUsed, err = store.WaitUntilFreshAndList(ctx, 5, "prefix/", matchValues)
if err != nil {
t.Fatalf("unexpected error: %v", err)
}
if resourceVersion != 5 {
t.Errorf("unexpected resourceVersion: %v, expected: 5", resourceVersion)
if resp.ResourceVersion != 5 {
t.Errorf("unexpected resourceVersion: %v, expected: 5", resp.ResourceVersion)
}
if len(list) != 3 {
t.Errorf("unexpected list returned: %#v", list)
if len(resp.Items) != 3 {
t.Errorf("unexpected list returned: %#v", resp)
}
if indexUsed != "" {
t.Errorf("Used index %q but expected none to be used", indexUsed)
@ -546,15 +546,15 @@ func TestWaitUntilFreshAndListFromCache(t *testing.T) {
}()
// list from future revision. Requires watch cache to request bookmark to get it.
list, resourceVersion, indexUsed, err := store.WaitUntilFreshAndList(ctx, 3, "prefix/", nil)
resp, indexUsed, err := store.WaitUntilFreshAndList(ctx, 3, "prefix/", nil)
if err != nil {
t.Fatalf("unexpected error: %v", err)
}
if resourceVersion != 3 {
t.Errorf("unexpected resourceVersion: %v, expected: 6", resourceVersion)
if resp.ResourceVersion != 3 {
t.Errorf("unexpected resourceVersion: %v, expected: 6", resp.ResourceVersion)
}
if len(list) != 1 {
t.Errorf("unexpected list returned: %#v", list)
if len(resp.Items) != 1 {
t.Errorf("unexpected list returned: %#v", resp)
}
if indexUsed != "" {
t.Errorf("Used index %q but expected none to be used", indexUsed)
@ -626,7 +626,7 @@ func TestWaitUntilFreshAndListTimeout(t *testing.T) {
store.Add(makeTestPod("bar", 4))
}()
_, _, _, err := store.WaitUntilFreshAndList(ctx, 4, "", nil)
_, _, err := store.WaitUntilFreshAndList(ctx, 4, "", nil)
if !errors.IsTimeout(err) {
t.Errorf("expected timeout error but got: %v", err)
}
@ -655,12 +655,12 @@ func TestReflectorForWatchCache(t *testing.T) {
defer store.Stop()
{
_, version, _, err := store.WaitUntilFreshAndList(ctx, 0, "", nil)
resp, _, err := store.WaitUntilFreshAndList(ctx, 0, "", nil)
if err != nil {
t.Fatalf("unexpected error: %v", err)
}
if version != 0 {
t.Errorf("unexpected resource version: %d", version)
if resp.ResourceVersion != 0 {
t.Errorf("unexpected resource version: %d", resp.ResourceVersion)
}
}
@ -678,12 +678,12 @@ func TestReflectorForWatchCache(t *testing.T) {
r.ListAndWatch(wait.NeverStop)
{
_, version, _, err := store.WaitUntilFreshAndList(ctx, 10, "", nil)
resp, _, err := store.WaitUntilFreshAndList(ctx, 10, "", nil)
if err != nil {
t.Fatalf("unexpected error: %v", err)
}
if version != 10 {
t.Errorf("unexpected resource version: %d", version)
if resp.ResourceVersion != 10 {
t.Errorf("unexpected resource version: %d", resp.ResourceVersion)
}
}
}