mirrors/kubernetes
0
0 Fork 0
mirror of https://github.com/kubernetes/kubernetes.git synced 2026-10-28 13:45:38 +08:00
Code Activity

Merge pull request #143410 from serathius/watchcache-selector-on-watcher

Browse Source 
Watchcache selector on watcher
This commit is contained in:
kubernetes-prow[bot] authored and GitHub committed 2026-10-26 12:55:59 +00:00
2 parents fb59ba2a2a c2252902d2
commit 5ea567004b
4 files changed
+40 -34
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

staging/src/k8s.io/apiserver/pkg/storage/cacher/cache_watcher.go
+4
View File
@@ -98,6 +98,8 @@ type cacheWatcher struct {
// state holds a numeric value indicating the current state of the watcher
state int
selector watchSelector
}
func newCacheWatcher(
@@ -111,6 +113,7 @@ func newCacheWatcher(
watcherMetrics *metrics.WatcherMetricsObservers,
clk clock.Clock,
identifier string,
selector watchSelector,
) *cacheWatcher {
if clk == nil {
clk = clock.RealClock{}
@@ -129,6 +132,7 @@ func newCacheWatcher(
watcherMetrics: watcherMetrics,
clock: clk,
identifier: identifier,
selector: selector,
}
}
staging/src/k8s.io/apiserver/pkg/storage/cacher/cache_watcher_test.go
+11 -11
View File
@@ -67,7 +67,7 @@ func TestCacheWatcherCleanupNotBlockedByResult(t *testing.T) {
}
// set the size of the buffer of w.result to 0, so that the writes to
// w.result is blocked.
w = newCacheWatcher(0, filterAny, forget, storage.APIObjectVersioner{}, time.Now(), false, schema.GroupResource{Resource: "pods"}, metrics.NewNoopWatcherMetricsObservers(), nil, "")
w = newCacheWatcher(0, filterAny, forget, storage.APIObjectVersioner{}, time.Now(), false, schema.GroupResource{Resource: "pods"}, metrics.NewNoopWatcherMetricsObservers(), nil, "", watchSelector{})
w.input <- watcherInput{interval: intervalFromEvents(initEvents)}
go w.process(context.Background())
w.Stop()
@@ -192,7 +192,7 @@ TestCase:
testCase.events[j].ResourceVersion = uint64(j) + 1
}
w := newCacheWatcher(0, filter, forget, storage.APIObjectVersioner{}, time.Now(), false, schema.GroupResource{Resource: "pods"}, metrics.NewNoopWatcherMetricsObservers(), nil, "")
w := newCacheWatcher(0, filter, forget, storage.APIObjectVersioner{}, time.Now(), false, schema.GroupResource{Resource: "pods"}, metrics.NewNoopWatcherMetricsObservers(), nil, "", watchSelector{})
w.input <- watcherInput{interval: intervalFromEvents(testCase.events)}
go w.process(context.Background())
@@ -229,7 +229,7 @@ func TestCacheWatcherStoppedInAnotherGoroutine(t *testing.T) {
// timeout to zero and run the Stop goroutine concurrently.
// May sure that the watch will not be blocked on Stop.
for i := 0; i < maxRetriesToProduceTheRaceCondition; i++ {
w = newCacheWatcher(0, filterAny, forget, storage.APIObjectVersioner{}, time.Now(), false, schema.GroupResource{Resource: "pods"}, metrics.NewNoopWatcherMetricsObservers(), nil, "")
w = newCacheWatcher(0, filterAny, forget, storage.APIObjectVersioner{}, time.Now(), false, schema.GroupResource{Resource: "pods"}, metrics.NewNoopWatcherMetricsObservers(), nil, "", watchSelector{})
go w.Stop()
select {
case <-done:
@@ -241,7 +241,7 @@ func TestCacheWatcherStoppedInAnotherGoroutine(t *testing.T) {
deadline := time.Now().Add(time.Hour)
// After that, verifies the cacheWatcher.process goroutine works correctly.
for i := 0; i < maxRetriesToProduceTheRaceCondition; i++ {
w = newCacheWatcher(2, filterAny, emptyFunc, storage.APIObjectVersioner{}, deadline, false, schema.GroupResource{Resource: "pods"}, metrics.NewNoopWatcherMetricsObservers(), nil, "")
w = newCacheWatcher(2, filterAny, emptyFunc, storage.APIObjectVersioner{}, deadline, false, schema.GroupResource{Resource: "pods"}, metrics.NewNoopWatcherMetricsObservers(), nil, "", watchSelector{})
w.input <- watcherInput{interval: intervalFromEvents(nil)}
w.input <- watcherInput{event: &watchCacheEvent{Object: &v1.Pod{}, ResourceVersion: uint64(i + 1)}}
ctx, cancel := context.WithDeadline(context.Background(), deadline)
@@ -316,7 +316,7 @@ func TestResourceVersionAfterInitEvents(t *testing.T) {
forget := func(_ bool) {}
deadline := time.Now().Add(time.Minute)
w := newCacheWatcher(numObjects+1, filterAny, forget, storage.APIObjectVersioner{}, deadline, true, schema.GroupResource{Resource: "pods"}, metrics.NewNoopWatcherMetricsObservers(), nil, "")
w := newCacheWatcher(numObjects+1, filterAny, forget, storage.APIObjectVersioner{}, deadline, true, schema.GroupResource{Resource: "pods"}, metrics.NewNoopWatcherMetricsObservers(), nil, "", watchSelector{})
w.input <- watcherInput{interval: wci}
// Simulate a situation when the last event will that was already in
@@ -357,7 +357,7 @@ func TestTimeBucketWatchersBasic(t *testing.T) {
forget := func(bool) {}
newWatcher := func(deadline time.Time) *cacheWatcher {
w := newCacheWatcher(0, filterAny, forget, storage.APIObjectVersioner{}, deadline, true, schema.GroupResource{Resource: "pods"}, metrics.NewNoopWatcherMetricsObservers(), nil, "")
w := newCacheWatcher(0, filterAny, forget, storage.APIObjectVersioner{}, deadline, true, schema.GroupResource{Resource: "pods"}, metrics.NewNoopWatcherMetricsObservers(), nil, "", watchSelector{})
w.setBookmarkAfterResourceVersion(0)
return w
}
@@ -423,7 +423,7 @@ func TestCacheWatcherDraining(t *testing.T) {
makeWatchCacheEvent(5),
makeWatchCacheEvent(6),
}
w = newCacheWatcher(1, filterAny, forget, storage.APIObjectVersioner{}, time.Now(), true, schema.GroupResource{Resource: "pods"}, metrics.NewNoopWatcherMetricsObservers(), nil, "")
w = newCacheWatcher(1, filterAny, forget, storage.APIObjectVersioner{}, time.Now(), true, schema.GroupResource{Resource: "pods"}, metrics.NewNoopWatcherMetricsObservers(), nil, "", watchSelector{})
w.input <- watcherInput{interval: intervalFromEvents(initEvents)}
go w.process(context.Background())
if !w.add(makeWatchCacheEvent(7), time.NewTimer(1*time.Second)) {
@@ -464,7 +464,7 @@ func TestCacheWatcherDrainingRequestedButNotDrained(t *testing.T) {
makeWatchCacheEvent(5),
makeWatchCacheEvent(6),
}
w = newCacheWatcher(1, filterAny, forget, storage.APIObjectVersioner{}, time.Now(), true, schema.GroupResource{Resource: "pods"}, metrics.NewNoopWatcherMetricsObservers(), nil, "")
w = newCacheWatcher(1, filterAny, forget, storage.APIObjectVersioner{}, time.Now(), true, schema.GroupResource{Resource: "pods"}, metrics.NewNoopWatcherMetricsObservers(), nil, "", watchSelector{})
w.input <- watcherInput{interval: intervalFromEvents(initEvents)}
go w.process(context.Background())
if !w.add(makeWatchCacheEvent(7), time.NewTimer(1*time.Second)) {
@@ -501,7 +501,7 @@ func TestCacheWatcherDrainingNoBookmarkAfterResourceVersionReceived(t *testing.T
{Object: &v1.Pod{}, ResourceVersion: 1},
{Object: &v1.Pod{}, ResourceVersion: 2},
}
w = newCacheWatcher(0, filterAny, forget, storage.APIObjectVersioner{}, time.Now(), true, schema.GroupResource{Resource: "pods"}, metrics.NewNoopWatcherMetricsObservers(), nil, "")
w = newCacheWatcher(0, filterAny, forget, storage.APIObjectVersioner{}, time.Now(), true, schema.GroupResource{Resource: "pods"}, metrics.NewNoopWatcherMetricsObservers(), nil, "", watchSelector{})
w.setBookmarkAfterResourceVersion(10)
w.input <- watcherInput{interval: intervalFromEvents(initEvents)}
go w.process(context.Background())
@@ -574,7 +574,7 @@ func TestCacheWatcherDrainingNoBookmarkAfterResourceVersionSent(t *testing.T) {
})
fakeClock := testingclock.NewFakeClock(time.Now())
w = newCacheWatcher(2, filterAny, forget, storage.APIObjectVersioner{}, time.Now(), true, schema.GroupResource{Resource: "pods"}, metrics.NewWatcherMetricsObservers(schema.GroupResource{Resource: "pods"}), fakeClock, "")
w = newCacheWatcher(2, filterAny, forget, storage.APIObjectVersioner{}, time.Now(), true, schema.GroupResource{Resource: "pods"}, metrics.NewWatcherMetricsObservers(schema.GroupResource{Resource: "pods"}), fakeClock, "", watchSelector{})
w.setBookmarkAfterResourceVersion(10)
w.input <- watcherInput{interval: intervalFromEvents(initEvents)}
go w.process(ctx)
@@ -672,7 +672,7 @@ apiserver_watch_events_dispatch_duration_seconds_count{group="",resource="pods",
func TestBookmarkAfterResourceVersionWatchers(t *testing.T) {
newWatcher := func(id string, deadline time.Time) *cacheWatcher {
w := newCacheWatcher(0, filterAny, func(bool) {}, storage.APIObjectVersioner{}, deadline, true, schema.GroupResource{Resource: "pods"}, metrics.NewNoopWatcherMetricsObservers(), nil, id)
w := newCacheWatcher(0, filterAny, func(bool) {}, storage.APIObjectVersioner{}, deadline, true, schema.GroupResource{Resource: "pods"}, metrics.NewNoopWatcherMetricsObservers(), nil, id, watchSelector{})
w.setBookmarkAfterResourceVersion(10)
return w
}
staging/src/k8s.io/apiserver/pkg/storage/cacher/cacher.go
+20 -19
View File
@@ -147,32 +147,32 @@ type indexedWatchers struct {
indexWatchers map[string]watchersMap
}
func (i *indexedWatchers) addWatcher(w *cacheWatcher, number int, selector watchSelector) {
if selector.indexValue != nil {
if _, ok := i.indexWatchers[*selector.indexValue]; !ok {
i.indexWatchers[*selector.indexValue] = watchersMap{}
func (i *indexedWatchers) addWatcher(w *cacheWatcher, number int) {
if w.selector.indexValue != nil {
if _, ok := i.indexWatchers[*w.selector.indexValue]; !ok {
i.indexWatchers[*w.selector.indexValue] = watchersMap{}
}
i.indexWatchers[*selector.indexValue].addWatcher(w, number)
i.indexWatchers[*w.selector.indexValue].addWatcher(w, number)
} else {
scopedWatchers, ok := i.scopeWatchers[selector.scope]
scopedWatchers, ok := i.scopeWatchers[w.selector.scope]
if !ok {
scopedWatchers = watchersMap{}
i.scopeWatchers[selector.scope] = scopedWatchers
i.scopeWatchers[w.selector.scope] = scopedWatchers
}
scopedWatchers.addWatcher(w, number)
}
}
func (i *indexedWatchers) deleteWatcher(number int, selector watchSelector) {
if selector.indexValue != nil {
i.indexWatchers[*selector.indexValue].deleteWatcher(number)
if len(i.indexWatchers[*selector.indexValue]) == 0 {
delete(i.indexWatchers, *selector.indexValue)
func (i *indexedWatchers) deleteWatcher(w *cacheWatcher, number int) {
if w.selector.indexValue != nil {
i.indexWatchers[*w.selector.indexValue].deleteWatcher(number)
if len(i.indexWatchers[*w.selector.indexValue]) == 0 {
delete(i.indexWatchers, *w.selector.indexValue)
}
} else {
i.scopeWatchers[selector.scope].deleteWatcher(number)
if len(i.scopeWatchers[selector.scope]) == 0 {
delete(i.scopeWatchers, selector.scope)
i.scopeWatchers[w.selector.scope].deleteWatcher(number)
if len(i.scopeWatchers[w.selector.scope]) == 0 {
delete(i.scopeWatchers, w.selector.scope)
}
}
}
@@ -578,6 +578,7 @@ func (c *Cacher) Watch(ctx context.Context, key string, opts storage.ListOptions
c.watcherMetrics,
c.clock,
identifier,
selector,
)
// We explicitly use thread unsafe version and do locking ourself to ensure that
@@ -631,10 +632,10 @@ func (c *Cacher) Watch(ctx context.Context, key string, opts storage.ListOptions
}
// Update watcher.forget function once we can compute it.
watcher.forget = forgetWatcher(c, watcher, c.watcherIdx, selector)
watcher.forget = forgetWatcher(c, watcher, c.watcherIdx)
// Update the bookMarkAfterResourceVersion
watcher.setBookmarkAfterResourceVersion(bookmarkAfterResourceVersion)
c.watchers.addWatcher(watcher, c.watcherIdx, selector)
c.watchers.addWatcher(watcher, c.watcherIdx)
addedWatcher = true
// Add it to the queue only when the client support watch bookmarks.
@@ -1267,7 +1268,7 @@ func (c *Cacher) prepareKey(key string, recursive bool) (string, error) {
return storage.PrepareKey(c.resourcePrefix, key, recursive)
}
func forgetWatcher(c *Cacher, w *cacheWatcher, index int, selector watchSelector) func(bool) {
func forgetWatcher(c *Cacher, w *cacheWatcher, index int) func(bool) {
return func(drainWatcher bool) {
c.Lock()
defer c.Unlock()
@@ -1277,7 +1278,7 @@ func forgetWatcher(c *Cacher, w *cacheWatcher, index int, selector watchSelector
// It's possible that the watcher is already not in the structure (e.g. in case of
// simultaneous Stop() and terminateAllWatchers(), but it is safe to call stopLocked()
// on a watcher multiple times.
c.watchers.deleteWatcher(index, selector)
c.watchers.deleteWatcher(w, index)
c.stopWatcherLocked(w)
}
}
staging/src/k8s.io/apiserver/pkg/storage/cacher/cacher_whitebox_test.go
+5 -4
View File
@@ -2582,13 +2582,13 @@ func TestStartDispatchingSelectsWatchers(t *testing.T) {
{namespace: "other-ns", name: "pod"}, {namespace: "ns", name: "other-pod"}, {namespace: "other-ns"}, {name: "other-pod"},
} {
for range 2 {
cacher.watchers.addWatcher(&cacheWatcher{identifier: scope.namespace + "/" + scope.name}, number, watchSelector{scope: scope})
cacher.watchers.addWatcher(&cacheWatcher{identifier: scope.namespace + "/" + scope.name, selector: watchSelector{scope: scope}}, number)
number++
}
}
for _, value := range []string{"node-a", "node-b", "node-c"} {
for range 2 {
cacher.watchers.addWatcher(&cacheWatcher{identifier: value}, number, watchSelector{indexValue: &value})
cacher.watchers.addWatcher(&cacheWatcher{identifier: value, selector: watchSelector{indexValue: &value}}, number)
number++
}
}
@@ -2702,13 +2702,14 @@ func TestForgetWatcher(t *testing.T) {
metrics.NewNoopWatcherMetricsObservers(),
testingclock.NewFakeClock(time.Now()),
"1",
watchSelector{},
)
forgetWatcherFn = forgetWatcher(cacher, w, 0, watchSelector{})
forgetWatcherFn = forgetWatcher(cacher, w, 0)
addWatcher := func(w *cacheWatcher) {
cacher.Lock()
defer cacher.Unlock()
cacher.watchers.addWatcher(w, 0, watchSelector{})
cacher.watchers.addWatcher(w, 0)
}
addWatcher(w)