github/kubernetes
1
0
Fork 0
mirror of https://github.com/k3s-io/kubernetes.git synced 2026-07-24 21:13:16 +00:00
Code Issues Projects Releases Wiki Activity

Override getKeys when cacher is enabled

Browse Source
This commit is contained in:
Marek Siarkowicz
2025-06-23 16:35:06 +02:00
parent 7cb2417999
commit e2c6b7fdf9
8 changed files with 86 additions and 25 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
staging/src/k8s.io/apiserver/pkg/storage/cacher/cacher.go
View File

@@ -459,7 +459,7 @@ func NewCacherFromConfig(config Config) (*Cacher, error) {
}, time.Second, stopCh,
)
}()
config.Storage.SetKeysFunc(cacher.getKeys)
return cacher, nil
}
@@ -1289,6 +1289,14 @@ func (c *Cacher) setInitialEventsEndBookmarkIfRequested(cacheInterval *watchCach
}
}
func (c *Cacher) getKeys(ctx context.Context) ([]string, error) {
rev, err := c.storage.GetCurrentResourceVersion(ctx)
if err != nil {
return nil, err
}
return c.watchCache.WaitUntilFreshAndGetKeys(ctx, rev)
}
func (c *Cacher) Ready() bool {
_, err := c.ready.check()
return err == nil

2
staging/src/k8s.io/apiserver/pkg/storage/cacher/cacher_whitebox_test.go
View File

@@ -192,6 +192,8 @@ func (d *dummyStorage) GuaranteedUpdate(_ context.Context, _ string, _ runtime.O
func (d *dummyStorage) Stats(_ context.Context) (storage.Stats, error) {
return storage.Stats{}, fmt.Errorf("unimplemented")
}
func (d *dummyStorage) SetKeysFunc(storage.KeysFunc) {
}
func (d *dummyStorage) ReadinessCheck() error {
return nil
}

4
staging/src/k8s.io/apiserver/pkg/storage/cacher/delegator.go
View File

@@ -99,6 +99,10 @@ func (c *CacheDelegator) GetCurrentResourceVersion(ctx context.Context) (uint64,
return c.storage.GetCurrentResourceVersion(ctx)
}
func (c *CacheDelegator) SetKeysFunc(keys storage.KeysFunc) {
c.storage.SetKeysFunc(keys)
}
func (c *CacheDelegator) Delete(ctx context.Context, key string, out runtime.Object, preconditions *storage.Preconditions, validateDeletion storage.ValidateObjectFunc, cachedExistingObject runtime.Object, opts storage.DeleteOptions) error {
// Ignore the suggestion and try to pass down the current version of the object
// read from cache.

18
staging/src/k8s.io/apiserver/pkg/storage/cacher/watch_cache.go
View File

@@ -512,6 +512,24 @@ func (w *watchCache) WaitUntilFreshAndList(ctx context.Context, resourceVersion
return w.list(ctx, resourceVersion, key, opts)
}
// 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) WaitUntilFreshAndGetKeys(ctx context.Context, resourceVersion uint64) (keys []string, err error) {
if delegator.ConsistentReadSupported() && w.notFresh(resourceVersion) {
w.waitingUntilFresh.Add()
err = w.waitUntilFreshAndBlock(ctx, resourceVersion)
w.waitingUntilFresh.Remove()
} else {
err = w.waitUntilFreshAndBlock(ctx, resourceVersion)
}
defer w.RUnlock()
if err != nil {
return nil, err
}
return w.store.ListKeys(), nil
}
// NOTICE: Structure follows the shouldDelegateList function in
// staging/src/k8s.io/apiserver/pkg/storage/cacher/delegator.go
func (w *watchCache) list(ctx context.Context, resourceVersion uint64, key string, opts storage.ListOptions) (resp listResp, index string, err error) {

24
staging/src/k8s.io/apiserver/pkg/storage/etcd3/stats.go
View File

@@ -18,6 +18,7 @@ package etcd3
import (
"context"
"strings"
"sync"
"sync/atomic"
@@ -32,10 +33,12 @@ import (
const sizerRefreshInterval = time.Minute
type keysFunc func(context.Context) ([]string, error)
func newStatsCache(getKeys keysFunc) *statsCache {
func newStatsCache(prefix string, getKeys storage.KeysFunc) *statsCache {
if prefix[len(prefix)-1] != '/' {
prefix += "/"
}
sc := &statsCache{
prefix: prefix,
getKeys: getKeys,
stop: make(chan struct{}),
keys: make(map[string]sizeRevision),
@@ -58,7 +61,8 @@ func newStatsCache(getKeys keysFunc) *statsCache {
// This approach may leak keys if delete events are not observed,
// thus we run a background goroutine to periodically cleanup keys if needed.
type statsCache struct {
getKeys keysFunc
prefix string
getKeys storage.KeysFunc
stop chan struct{}
wg sync.WaitGroup
lastKeyCleanup atomic.Pointer[time.Time]
@@ -89,6 +93,10 @@ func (sc *statsCache) Stats(ctx context.Context) (storage.Stats, error) {
return stats, nil
}
func (sc *statsCache) SetKeysFunc(keys storage.KeysFunc) {
sc.getKeys = keys
}
func (sc *statsCache) Close() {
close(sc.stop)
sc.wg.Wait()
@@ -122,6 +130,14 @@ func (sc *statsCache) cleanKeysIfNeeded(ctx context.Context) {
func (sc *statsCache) cleanKeys(keepKeys []string) {
newKeys := make(map[string]sizeRevision, len(keepKeys))
for _, key := range keepKeys {
// Handle cacher keys not having prefix.
if !strings.HasPrefix(key, sc.prefix) {
startIndex := 0
if key[0] == '/' {
startIndex = 1
}
key = sc.prefix + key[startIndex:]
}
keySizeRevision, ok := sc.keys[key]
if !ok {
continue

38
staging/src/k8s.io/apiserver/pkg/storage/etcd3/stats_test.go
View File

@@ -27,77 +27,77 @@ import (
func TestStatsCache(t *testing.T) {
ctx := t.Context()
store := newStatsCache(func(ctx context.Context) ([]string, error) { return []string{}, nil })
store := newStatsCache("/prefix", func(ctx context.Context) ([]string, error) { return []string{}, nil })
defer store.Close()
stats, err := store.Stats(ctx)
require.NoError(t, err)
assert.Equal(t, int64(0), stats.EstimatedAverageObjectSizeBytes)
store.UpdateKey(&mvccpb.KeyValue{Key: []byte("foo1"), Value: []byte("0123456789"), ModRevision: 2})
store.getKeys = func(ctx context.Context) ([]string, error) { return []string{"foo1"}, nil }
store.UpdateKey(&mvccpb.KeyValue{Key: []byte("/prefix/foo1"), Value: []byte("0123456789"), ModRevision: 2})
store.getKeys = func(ctx context.Context) ([]string, error) { return []string{"/prefix/foo1"}, nil }
stats, err = store.Stats(ctx)
require.NoError(t, err)
assert.Equal(t, int64(10), stats.EstimatedAverageObjectSizeBytes)
store.UpdateKey(&mvccpb.KeyValue{Key: []byte("foo2"), Value: []byte("01234567890123456789"), ModRevision: 3})
store.getKeys = func(ctx context.Context) ([]string, error) { return []string{"foo1", "foo2"}, nil }
store.UpdateKey(&mvccpb.KeyValue{Key: []byte("/prefix/foo2"), Value: []byte("01234567890123456789"), ModRevision: 3})
store.getKeys = func(ctx context.Context) ([]string, error) { return []string{"/prefix/foo1", "/prefix/foo2"}, nil }
stats, err = store.Stats(ctx)
require.NoError(t, err)
assert.Equal(t, int64(15), stats.EstimatedAverageObjectSizeBytes)
store.UpdateKey(&mvccpb.KeyValue{Key: []byte("foo1"), Value: []byte("012345678901234567890123456789"), ModRevision: 4})
store.UpdateKey(&mvccpb.KeyValue{Key: []byte("/prefix/foo1"), Value: []byte("012345678901234567890123456789"), ModRevision: 4})
stats, err = store.Stats(ctx)
require.NoError(t, err)
assert.Equal(t, int64(25), stats.EstimatedAverageObjectSizeBytes)
store.UpdateKey(&mvccpb.KeyValue{Key: []byte("foo2"), Value: []byte("0123456789"), ModRevision: 5})
store.UpdateKey(&mvccpb.KeyValue{Key: []byte("/prefix/foo2"), Value: []byte("0123456789"), ModRevision: 5})
stats, err = store.Stats(ctx)
require.NoError(t, err)
assert.Equal(t, int64(20), stats.EstimatedAverageObjectSizeBytes)
store.DeleteKey(&mvccpb.KeyValue{Key: []byte("foo1"), ModRevision: 6})
store.getKeys = func(ctx context.Context) ([]string, error) { return []string{"foo2"}, nil }
store.DeleteKey(&mvccpb.KeyValue{Key: []byte("/prefix/foo1"), ModRevision: 6})
store.getKeys = func(ctx context.Context) ([]string, error) { return []string{"/prefix/foo2"}, nil }
stats, err = store.Stats(ctx)
require.NoError(t, err)
assert.Equal(t, int64(10), stats.EstimatedAverageObjectSizeBytes)
// Snapshot revision from revision 3
store.Update([]*mvccpb.KeyValue{
{Key: []byte("foo1"), Value: []byte("0123456789"), ModRevision: 2},
{Key: []byte("foo2"), Value: []byte("01234567890123456789"), ModRevision: 3},
{Key: []byte("/prefix/foo1"), Value: []byte("0123456789"), ModRevision: 2},
{Key: []byte("/prefix/foo2"), Value: []byte("01234567890123456789"), ModRevision: 3},
})
stats, err = store.Stats(ctx)
require.NoError(t, err)
assert.Equal(t, int64(10), stats.EstimatedAverageObjectSizeBytes)
// Replay from revision 2
store.UpdateKey(&mvccpb.KeyValue{Key: []byte("foo1"), Value: []byte("0123456789"), ModRevision: 2})
store.UpdateKey(&mvccpb.KeyValue{Key: []byte("/prefix/foo1"), Value: []byte("0123456789"), ModRevision: 2})
stats, err = store.Stats(ctx)
require.NoError(t, err)
assert.Equal(t, int64(10), stats.EstimatedAverageObjectSizeBytes)
store.UpdateKey(&mvccpb.KeyValue{Key: []byte("foo2"), Value: []byte("01234567890123456789"), ModRevision: 3})
store.UpdateKey(&mvccpb.KeyValue{Key: []byte("/prefix/foo2"), Value: []byte("01234567890123456789"), ModRevision: 3})
stats, err = store.Stats(ctx)
require.NoError(t, err)
assert.Equal(t, int64(10), stats.EstimatedAverageObjectSizeBytes)
store.UpdateKey(&mvccpb.KeyValue{Key: []byte("foo1"), Value: []byte("012345678901234567890123456789"), ModRevision: 4})
store.UpdateKey(&mvccpb.KeyValue{Key: []byte("/prefix/foo1"), Value: []byte("012345678901234567890123456789"), ModRevision: 4})
stats, err = store.Stats(ctx)
require.NoError(t, err)
assert.Equal(t, int64(10), stats.EstimatedAverageObjectSizeBytes)
store.UpdateKey(&mvccpb.KeyValue{Key: []byte("foo2"), Value: []byte("0123456789"), ModRevision: 5})
store.UpdateKey(&mvccpb.KeyValue{Key: []byte("/prefix/foo2"), Value: []byte("0123456789"), ModRevision: 5})
stats, err = store.Stats(ctx)
require.NoError(t, err)
assert.Equal(t, int64(10), stats.EstimatedAverageObjectSizeBytes)
store.DeleteKey(&mvccpb.KeyValue{Key: []byte("foo1"), ModRevision: 6})
store.DeleteKey(&mvccpb.KeyValue{Key: []byte("/prefix/foo1"), ModRevision: 6})
stats, err = store.Stats(ctx)
require.NoError(t, err)
assert.Equal(t, int64(10), stats.EstimatedAverageObjectSizeBytes)
store.DeleteKey(&mvccpb.KeyValue{Key: []byte("foo1"), ModRevision: 7})
store.DeleteKey(&mvccpb.KeyValue{Key: []byte("/prefix/foo1"), ModRevision: 7})
store.getKeys = func(ctx context.Context) ([]string, error) { return []string{}, nil }
stats, err = store.Stats(ctx)
require.NoError(t, err)
@@ -106,8 +106,8 @@ func TestStatsCache(t *testing.T) {
// Old snapshot might restore old revision if keys were recreated
store.getKeys = func(ctx context.Context) ([]string, error) { return []string{"foo1", "foo2"}, nil }
store.Update([]*mvccpb.KeyValue{
{Key: []byte("foo1"), Value: []byte("0123456789"), ModRevision: 2},
{Key: []byte("foo2"), Value: []byte("01234567890123456789"), ModRevision: 3},
{Key: []byte("/prefix/foo1"), Value: []byte("0123456789"), ModRevision: 2},
{Key: []byte("/prefix/foo2"), Value: []byte("01234567890123456789"), ModRevision: 3},
})
stats, err = store.Stats(ctx)
require.NoError(t, err)

8
staging/src/k8s.io/apiserver/pkg/storage/etcd3/store.go
View File

@@ -183,7 +183,7 @@ func New(c *kubernetes.Client, codec runtime.Codec, newFunc, newListFunc func()
newListFunc: newListFunc,
}
if utilfeature.DefaultFeatureGate.Enabled(features.SizeBasedListCostEstimate) {
stats := newStatsCache(s.getKeys)
stats := newStatsCache(pathPrefix, s.getKeys)
s.stats = stats
w.stats = stats
}
@@ -633,6 +633,12 @@ func (s *store) Stats(ctx context.Context) (stats storage.Stats, err error) {
}, nil
}
func (s *store) SetKeysFunc(keys storage.KeysFunc) {
if s.stats != nil {
s.stats.SetKeysFunc(keys)
}
}
func (s *store) getKeys(ctx context.Context) ([]string, error) {
startTime := time.Now()
resp, err := s.client.KV.Get(ctx, s.pathPrefix, clientv3.WithPrefix(), clientv3.WithKeysOnly())

7
staging/src/k8s.io/apiserver/pkg/storage/interfaces.go
View File

@@ -267,8 +267,15 @@ type Interface interface {
// GetCurrentResourceVersion gets the current resource version from etcd.
// This method issues an empty list request and reads only the ResourceVersion from the object metadata
GetCurrentResourceVersion(ctx context.Context) (uint64, error)
// SetKeysFunc allows to override the function used to get keys from storage.
// This allows to replace default function that fetches keys from storage with one using cache.
SetKeysFunc(KeysFunc)
}
// KeysFunc is a function prototype to fetch keys from storage.
type KeysFunc func(context.Context) ([]string, error)
// GetOptions provides the options that may be provided for storage get operations.
type GetOptions struct {
// IgnoreNotFound determines what is returned if the requested object is not found. If