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 74f9d776b4e..075cfdf0cfc 100644 --- a/staging/src/k8s.io/apiserver/pkg/storage/cacher/cacher.go +++ b/staging/src/k8s.io/apiserver/pkg/storage/cacher/cacher.go @@ -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 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 c12b21bfb59..5c5cbb7d33a 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 @@ -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 } diff --git a/staging/src/k8s.io/apiserver/pkg/storage/cacher/delegator.go b/staging/src/k8s.io/apiserver/pkg/storage/cacher/delegator.go index b65dffa8796..fb93f3e4309 100644 --- a/staging/src/k8s.io/apiserver/pkg/storage/cacher/delegator.go +++ b/staging/src/k8s.io/apiserver/pkg/storage/cacher/delegator.go @@ -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. diff --git a/staging/src/k8s.io/apiserver/pkg/storage/cacher/watch_cache.go b/staging/src/k8s.io/apiserver/pkg/storage/cacher/watch_cache.go index 192dc057250..6926b3aa6fa 100644 --- a/staging/src/k8s.io/apiserver/pkg/storage/cacher/watch_cache.go +++ b/staging/src/k8s.io/apiserver/pkg/storage/cacher/watch_cache.go @@ -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) { diff --git a/staging/src/k8s.io/apiserver/pkg/storage/etcd3/stats.go b/staging/src/k8s.io/apiserver/pkg/storage/etcd3/stats.go index e2dc7f22f47..fa1b1caa8d9 100644 --- a/staging/src/k8s.io/apiserver/pkg/storage/etcd3/stats.go +++ b/staging/src/k8s.io/apiserver/pkg/storage/etcd3/stats.go @@ -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 diff --git a/staging/src/k8s.io/apiserver/pkg/storage/etcd3/stats_test.go b/staging/src/k8s.io/apiserver/pkg/storage/etcd3/stats_test.go index 77637fe3336..e67436f8326 100644 --- a/staging/src/k8s.io/apiserver/pkg/storage/etcd3/stats_test.go +++ b/staging/src/k8s.io/apiserver/pkg/storage/etcd3/stats_test.go @@ -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) 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 808a87f2c6d..54cbfa182a5 100644 --- a/staging/src/k8s.io/apiserver/pkg/storage/etcd3/store.go +++ b/staging/src/k8s.io/apiserver/pkg/storage/etcd3/store.go @@ -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()) diff --git a/staging/src/k8s.io/apiserver/pkg/storage/interfaces.go b/staging/src/k8s.io/apiserver/pkg/storage/interfaces.go index 582e369de9b..036de7a4f15 100644 --- a/staging/src/k8s.io/apiserver/pkg/storage/interfaces.go +++ b/staging/src/k8s.io/apiserver/pkg/storage/interfaces.go @@ -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