Merge pull request kubernetes#119207 from serathius/progress-notify

Implement conditionalProgressRequester that allows requesting watch progress notification if watch cache is not fresh

Author: k8s-ci-robot

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

@@ -104,7 +104,7 @@ type Config struct {
 
 	Codec runtime.Codec
 
-	Clock clock.Clock
+	Clock clock.WithTicker
 }
 
 type watchersMap map[int]*cacheWatcher
@@ -329,6 +329,10 @@ type Cacher struct {
 	expiredBookmarkWatchers []*cacheWatcher
 }
 
+func (c *Cacher) RequestWatchProgress(ctx context.Context) error {
+	return c.storage.RequestWatchProgress(ctx)
+}
+
 // NewCacherFromConfig creates a new Cacher responsible for servicing WATCH and LIST requests from
 // its internal cache and updating its cache in the background based on the
 // given configuration.
@@ -397,9 +401,9 @@ func NewCacherFromConfig(config Config) (*Cacher, error) {
 		// so that future reuse does not get a spurious timeout.
 		<-cacher.timer.C
 	}
-
+	progressRequester := newConditionalProgressRequester(config.Storage.RequestWatchProgress, config.Clock)
 	watchCache := newWatchCache(
-		config.KeyFunc, cacher.processEvent, config.GetAttrsFunc, config.Versioner, config.Indexers, config.Clock, config.GroupResource)
+		config.KeyFunc, cacher.processEvent, config.GetAttrsFunc, config.Versioner, config.Indexers, config.Clock, config.GroupResource, progressRequester)
 	listerWatcher := NewListerWatcher(config.Storage, config.ResourcePrefix, config.NewListFunc)
 	reflectorName := "storage/cacher.go:" + config.ResourcePrefix
 
@@ -419,6 +423,7 @@ func NewCacherFromConfig(config Config) (*Cacher, error) {
 	cacher.reflector = reflector
 
 	go cacher.dispatchEvents()
+	go progressRequester.Run(stopCh)
 
 	cacher.stopWg.Add(1)
 	go func() {

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

@@ -328,7 +328,7 @@ type setupOptions struct {
 	keyFunc        func(runtime.Object) (string, error)
 	indexerFuncs   map[string]storage.IndexerFunc
 	pagingEnabled  bool
-	clock          clock.Clock
+	clock          clock.WithTicker
 }
 
 type setupOption func(*setupOptions)

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

@@ -90,6 +90,10 @@ type dummyStorage struct {
 	watchFn   func(_ context.Context, _ string, _ storage.ListOptions) (watch.Interface, error)
 }
 
+func (d *dummyStorage) RequestWatchProgress(ctx context.Context) error {
+	return nil
+}
+
 type dummyWatch struct {
 	ch chan watch.Event
 }

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

@@ -196,6 +196,10 @@ type watchCache struct {
 
 	// For testing cache interval invalidation.
 	indexValidator indexValidator
+
+	// Requests progress notification if there are requests waiting for watch
+	// to be fresh
+	waitingUntilFresh *conditionalProgressRequester
 }
 
 func newWatchCache(
@@ -204,8 +208,9 @@ func newWatchCache(
 	getAttrsFunc func(runtime.Object) (labels.Set, fields.Set, error),
 	versioner storage.Versioner,
 	indexers *cache.Indexers,
-	clock clock.Clock,
-	groupResource schema.GroupResource) *watchCache {
+	clock clock.WithTicker,
+	groupResource schema.GroupResource,
+	progressRequester *conditionalProgressRequester) *watchCache {
 	wc := &watchCache{
 		capacity:            defaultLowerBoundCapacity,
 		keyFunc:             keyFunc,
@@ -222,6 +227,7 @@ func newWatchCache(
 		clock:               clock,
 		versioner:           versioner,
 		groupResource:       groupResource,
+		waitingUntilFresh:   progressRequester,
 	}
 	metrics.WatchCacheCapacity.WithLabelValues(groupResource.String()).Set(float64(wc.capacity))
 	wc.cond = sync.NewCond(wc.RLocker())

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

@@ -287,6 +287,7 @@ func TestCacheIntervalNextFromWatchCache(t *testing.T) {
 	for _, c := range cases {
 		t.Run(c.name, func(t *testing.T) {
 			wc := newTestWatchCache(capacity, &cache.Indexers{})
+			defer wc.Stop()
 			for i := 0; i < c.eventsAddedToWatchcache; i++ {
 				wc.Add(makeTestPod(fmt.Sprintf("pod%d", i), uint64(i)))
 			}

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

@@ -68,6 +68,9 @@ func makeTestStoreElement(pod *v1.Pod) *storeElement {
 
 type testWatchCache struct {
 	*watchCache
+
+	bookmarkRevision chan int64
+	stopCh           chan struct{}
 }
 
 func (w *testWatchCache) getAllEventsSince(resourceVersion uint64) ([]*watchCacheEvent, error) {
@@ -112,19 +115,42 @@ func newTestWatchCache(capacity int, indexers *cache.Indexers) *testWatchCache {
 	}
 	versioner := storage.APIObjectVersioner{}
 	mockHandler := func(*watchCacheEvent) {}
-	wc := newWatchCache(keyFunc, mockHandler, getAttrsFunc, versioner, indexers, testingclock.NewFakeClock(time.Now()), schema.GroupResource{Resource: "pods"})
+	wc := &testWatchCache{}
+	wc.bookmarkRevision = make(chan int64, 1)
+	wc.stopCh = make(chan struct{})
+	clock := testingclock.NewFakeClock(time.Now())
+	pr := newConditionalProgressRequester(wc.RequestWatchProgress, clock)
+	go pr.Run(wc.stopCh)
+	wc.watchCache = newWatchCache(keyFunc, mockHandler, getAttrsFunc, versioner, indexers, clock, schema.GroupResource{Resource: "pods"}, pr)
 	// To preserve behavior of tests that assume a given capacity,
 	// resize it to th expected size.
 	wc.capacity = capacity
 	wc.cache = make([]*watchCacheEvent, capacity)
 	wc.lowerBoundCapacity = min(capacity, defaultLowerBoundCapacity)
 	wc.upperBoundCapacity = max(capacity, defaultUpperBoundCapacity)
 
-	return &testWatchCache{watchCache: wc}
+	return wc
+}
+
+func (w *testWatchCache) RequestWatchProgress(ctx context.Context) error {
+	go func() {
+		select {
+		case rev := <-w.bookmarkRevision:
+			w.UpdateResourceVersion(fmt.Sprintf("%d", rev))
+		case <-ctx.Done():
+			return
+		}
+	}()
+	return nil
+}
+
+func (w *testWatchCache) Stop() {
+	close(w.stopCh)
 }
 
 func TestWatchCacheBasic(t *testing.T) {
 	store := newTestWatchCache(2, &cache.Indexers{})
+	defer store.Stop()
 
 	// Test Add/Update/Delete.
 	pod1 := makeTestPod("pod", 1)
@@ -202,6 +228,7 @@ func TestWatchCacheBasic(t *testing.T) {
 
 func TestEvents(t *testing.T) {
 	store := newTestWatchCache(5, &cache.Indexers{})
+	defer store.Stop()
 
 	// no dynamic-size cache to fit old tests.
 	store.lowerBoundCapacity = 5
@@ -326,6 +353,7 @@ func TestEvents(t *testing.T) {
 
 func TestMarker(t *testing.T) {
 	store := newTestWatchCache(3, &cache.Indexers{})
+	defer store.Stop()
 
 	// First thing that is called when propagated from storage is Replace.
 	store.Replace([]interface{}{
@@ -380,7 +408,7 @@ func TestWaitUntilFreshAndList(t *testing.T) {
 			return []string{pod.Spec.NodeName}, nil
 		},
 	})
-
+	defer store.Stop()
 	// In background, update the store.
 	go func() {
 		store.Add(makeTestPodDetails("pod1", 2, "node1", map[string]string{"label": "value1"}))
@@ -463,6 +491,7 @@ func TestWaitUntilFreshAndList(t *testing.T) {
 func TestWaitUntilFreshAndGet(t *testing.T) {
 	ctx := context.Background()
 	store := newTestWatchCache(3, &cache.Indexers{})
+	defer store.Stop()
 
 	// In background, update the store.
 	go func() {
@@ -489,6 +518,7 @@ func TestWaitUntilFreshAndGet(t *testing.T) {
 func TestWaitUntilFreshAndListTimeout(t *testing.T) {
 	ctx := context.Background()
 	store := newTestWatchCache(3, &cache.Indexers{})
+	defer store.Stop()
 	fc := store.clock.(*testingclock.FakeClock)
 
 	// In background, step clock after the below call starts the timer.
@@ -529,6 +559,7 @@ func (t *testLW) Watch(options metav1.ListOptions) (watch.Interface, error) {
 func TestReflectorForWatchCache(t *testing.T) {
 	ctx := context.Background()
 	store := newTestWatchCache(5, &cache.Indexers{})
+	defer store.Stop()
 
 	{
 		_, version, _, err := store.WaitUntilFreshAndList(ctx, 0, nil)
@@ -792,6 +823,7 @@ func TestDynamicCache(t *testing.T) {
 	for _, test := range tests {
 		t.Run(test.name, func(t *testing.T) {
 			store := newTestWatchCache(test.cacheCapacity, &cache.Indexers{})
+			defer store.Stop()
 			store.cache = make([]*watchCacheEvent, test.cacheCapacity)
 			store.startIndex = test.startIndex
 			store.lowerBoundCapacity = test.lowerBoundCapacity
@@ -840,6 +872,7 @@ func checkCacheElements(cache *testWatchCache) bool {
 
 func TestCacheIncreaseDoesNotBreakWatch(t *testing.T) {
 	store := newTestWatchCache(2, &cache.Indexers{})
+	defer store.Stop()
 
 	now := store.clock.Now()
 	addEvent := func(key string, rv uint64, t time.Time) {
@@ -988,6 +1021,7 @@ func TestSuggestedWatchChannelSize(t *testing.T) {
 	for _, test := range testCases {
 		t.Run(test.name, func(t *testing.T) {
 			store := newTestWatchCache(test.capacity, &cache.Indexers{})
+			defer store.Stop()
 			got := store.suggestedWatchChannelSize(test.indexExists, test.triggerUsed)
 			if got != test.expected {
 				t.Errorf("unexpected channel size got: %v, expected: %v", got, test.expected)
@@ -998,6 +1032,7 @@ func TestSuggestedWatchChannelSize(t *testing.T) {
 
 func BenchmarkWatchCache_updateCache(b *testing.B) {
 	store := newTestWatchCache(defaultUpperBoundCapacity, &cache.Indexers{})
+	defer store.Stop()
 	store.cache = store.cache[:0]
 	store.upperBoundCapacity = defaultUpperBoundCapacity
 	loadEventWithDuration(store, defaultUpperBoundCapacity, 0)

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

@@ -0,0 +1,117 @@
+/*
+Copyright 2023 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package cacher
+
+import (
+	"context"
+	"sync"
+	"time"
+
+	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
+	"k8s.io/apimachinery/pkg/util/wait"
+
+	"k8s.io/klog/v2"
+	"k8s.io/utils/clock"
+)
+
+const (
+	// progressRequestPeriod determines period of requesting progress
+	// from etcd when there is a request waiting for watch cache to be fresh.
+	progressRequestPeriod = 100 * time.Millisecond
+)
+
+func newConditionalProgressRequester(requestWatchProgress WatchProgressRequester, clock clock.WithTicker) *conditionalProgressRequester {
+	pr := &conditionalProgressRequester{
+		clock:                clock,
+		requestWatchProgress: requestWatchProgress,
+	}
+	pr.cond = sync.NewCond(pr.mux.RLocker())
+	return pr
+}
+
+type WatchProgressRequester func(ctx context.Context) error
+
+// conditionalProgressRequester will request progress notification if there
+// is a request waiting for watch cache to be fresh.
+type conditionalProgressRequester struct {
+	clock                clock.WithTicker
+	requestWatchProgress WatchProgressRequester
+
+	mux     sync.RWMutex
+	cond    *sync.Cond
+	waiting int
+	stopped bool
+}
+
+func (pr *conditionalProgressRequester) Run(stopCh <-chan struct{}) {
+	ctx := wait.ContextForChannel(stopCh)
+	go func() {
+		defer utilruntime.HandleCrash()
+		<-stopCh
+		pr.mux.Lock()
+		defer pr.mux.Unlock()
+		pr.stopped = true
+		pr.cond.Signal()
+	}()
+	ticker := pr.clock.NewTicker(progressRequestPeriod)
+	defer ticker.Stop()
+	for {
+		stopped := func() bool {
+			pr.mux.RLock()
+			defer pr.mux.RUnlock()
+			for pr.waiting == 0 && !pr.stopped {
+				pr.cond.Wait()
+			}
+			return pr.stopped
+		}()
+		if stopped {
+			return
+		}
+
+		select {
+		case <-ticker.C():
+			shouldRequest := func() bool {
+				pr.mux.RLock()
+				defer pr.mux.RUnlock()
+				return pr.waiting > 0 && !pr.stopped
+			}()
+			if !shouldRequest {
+				continue
+			}
+			err := pr.requestWatchProgress(ctx)
+			if err != nil {
+				klog.V(4).InfoS("Error requesting bookmark", "err", err)
+			}
+		case <-stopCh:
+			return
+		}
+	}
+}
+
+func (pr *conditionalProgressRequester) Add() {
+	pr.mux.Lock()
+	defer pr.mux.Unlock()
+	pr.waiting += 1
+	pr.cond.Signal()
+}
+
+func (pr *conditionalProgressRequester) Remove() {
+	pr.mux.Lock()
+	defer pr.mux.Unlock()
+	pr.waiting -= 1
+	pr.cond.Signal()
+}