cacher: apply key for initial events

For case of SendInitialEvents, a buffer of objects is created. That
process takes a significant amount of memory and CPU when the resource
is of a large volume. Many objects may be not relevant when key is provided.
This commit applies key when composing the buffer for SendInitialEvents.

Signed-off-by: Eric Lin <[email protected]>

Author: linxiulei

staging/src/k8s.io/apiserver/pkg/storage/cacher/cache_watcher_test.go

@@ -300,7 +300,7 @@ func TestResourceVersionAfterInitEvents(t *testing.T) {
 		store.Add(elem)
 	}
 
-	wci, err := newCacheIntervalFromStore(numObjects, store, getAttrsFunc)
+	wci, err := newCacheIntervalFromStore(numObjects, store, getAttrsFunc, "", false)
 	if err != nil {
 		t.Fatal(err)
 	}

staging/src/k8s.io/apiserver/pkg/storage/cacher/cacher.go

@@ -622,7 +622,7 @@ func (c *Cacher) Watch(ctx context.Context, key string, opts storage.ListOptions
 	defer c.watchCache.RUnlock()
 
 	var cacheInterval *watchCacheInterval
-	cacheInterval, err = c.watchCache.getAllEventsSinceLocked(requiredResourceVersion, opts)
+	cacheInterval, err = c.watchCache.getAllEventsSinceLocked(requiredResourceVersion, key, opts)
 	if err != nil {
 		// To match the uncached watch implementation, once we have passed authn/authz/admission,
 		// and successfully parsed a resource version, other errors must fail with a watch event of type ERROR,

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

@@ -712,9 +712,10 @@ func (w *watchCache) isIndexValidLocked(index int) bool {
 // getAllEventsSinceLocked returns a watchCacheInterval that can be used to
 // retrieve events since a certain resourceVersion. This function assumes to
 // be called under the watchCache lock.
-func (w *watchCache) getAllEventsSinceLocked(resourceVersion uint64, opts storage.ListOptions) (*watchCacheInterval, error) {
+func (w *watchCache) getAllEventsSinceLocked(resourceVersion uint64, key string, opts storage.ListOptions) (*watchCacheInterval, error) {
+	_, matchesSingle := opts.Predicate.MatchesSingle()
 	if opts.SendInitialEvents != nil && *opts.SendInitialEvents {
-		return w.getIntervalFromStoreLocked()
+		return w.getIntervalFromStoreLocked(key, matchesSingle)
 	}
 
 	size := w.endIndex - w.startIndex
@@ -743,7 +744,7 @@ func (w *watchCache) getAllEventsSinceLocked(resourceVersion uint64, opts storag
 			// current state and only then start watching from that point.
 			//
 			// TODO: In v2 api, we should stop returning the current state - #13969.
-			return w.getIntervalFromStoreLocked()
+			return w.getIntervalFromStoreLocked(key, matchesSingle)
 		}
 		// SendInitialEvents = false and resourceVersion = 0
 		// means that the request would like to start watching
@@ -769,8 +770,8 @@ func (w *watchCache) getAllEventsSinceLocked(resourceVersion uint64, opts storag
 // getIntervalFromStoreLocked returns a watchCacheInterval
 // that covers the entire storage state.
 // This function assumes to be called under the watchCache lock.
-func (w *watchCache) getIntervalFromStoreLocked() (*watchCacheInterval, error) {
-	ci, err := newCacheIntervalFromStore(w.resourceVersion, w.store, w.getAttrsFunc)
+func (w *watchCache) getIntervalFromStoreLocked(key string, matchesSingle bool) (*watchCacheInterval, error) {
+	ci, err := newCacheIntervalFromStore(w.resourceVersion, w.store, w.getAttrsFunc, key, matchesSingle)
 	if err != nil {
 		return nil, err
 	}

staging/src/k8s.io/apiserver/pkg/storage/cacher/watch_cache_interval.go

@@ -133,9 +133,22 @@ func (s sortableWatchCacheEvents) Swap(i, j int) {
 // returned by Next() need to be events from a List() done on the underlying store of
 // the watch cache.
 // The items returned in the interval will be sorted by Key.
-func newCacheIntervalFromStore(resourceVersion uint64, store cache.Indexer, getAttrsFunc attrFunc) (*watchCacheInterval, error) {
+func newCacheIntervalFromStore(resourceVersion uint64, store cache.Indexer, getAttrsFunc attrFunc, key string, matchesSingle bool) (*watchCacheInterval, error) {
 	buffer := &watchCacheIntervalBuffer{}
-	allItems := store.List()
+	var allItems []interface{}
+
+	if matchesSingle {
+		item, exists, err := store.GetByKey(key)
+		if err != nil {
+			return nil, err
+		}
+
+		if exists {
+			allItems = append(allItems, item)
+		}
+	} else {
+		allItems = store.List()
+	}
 	buffer.buffer = make([]*watchCacheEvent, len(allItems))
 	for i, item := range allItems {
 		elem, ok := item.(*storeElement)

staging/src/k8s.io/apiserver/pkg/storage/cacher/watch_cache_interval_test.go

@@ -391,7 +391,7 @@ func TestCacheIntervalNextFromStore(t *testing.T) {
 		store.Add(elem)
 	}
 
-	wci, err := newCacheIntervalFromStore(rv, store, getAttrsFunc)
+	wci, err := newCacheIntervalFromStore(rv, store, getAttrsFunc, "", false)
 	if err != nil {
 		t.Fatal(err)
 	}

staging/src/k8s.io/apiserver/pkg/storage/cacher/watch_cache_test.go

@@ -105,7 +105,7 @@ func (w *testWatchCache) getCacheIntervalForEvents(resourceVersion uint64, opts
 	w.RLock()
 	defer w.RUnlock()
 
-	return w.getAllEventsSinceLocked(resourceVersion, opts)
+	return w.getAllEventsSinceLocked(resourceVersion, "", opts)
 }
 
 // newTestWatchCache just adds a fake clock.

staging/src/k8s.io/apiserver/pkg/storage/selection_predicate.go

@@ -118,7 +118,7 @@ func (s *SelectionPredicate) MatchesObjectAttributes(l labels.Set, f fields.Set)
 // MatchesSingleNamespace will return (namespace, true) if and only if s.Field matches on the object's
 // namespace.
 func (s *SelectionPredicate) MatchesSingleNamespace() (string, bool) {
-	if len(s.Continue) > 0 {
+	if len(s.Continue) > 0 || s.Field == nil {
 		return "", false
 	}
 	if namespace, ok := s.Field.RequiresExactMatch("metadata.namespace"); ok {
@@ -130,7 +130,7 @@ func (s *SelectionPredicate) MatchesSingleNamespace() (string, bool) {
 // MatchesSingle will return (name, true) if and only if s.Field matches on the object's
 // name.
 func (s *SelectionPredicate) MatchesSingle() (string, bool) {
-	if len(s.Continue) > 0 {
+	if len(s.Continue) > 0 || s.Field == nil {
 		return "", false
 	}
 	// TODO: should be namespace.name