github/kubernetes
1
0
Fork 0
mirror of https://github.com/k3s-io/kubernetes.git synced 2025-07-29 06:27:05 +00:00
Code Issues Projects Releases Wiki Activity

Fix draining cacher tests

Browse Source
This commit is contained in:
Wojciech Tyczyński 2022-07-28 10:20:23 +02:00
parent debace151c
commit a530a6898a
1 changed files with 29 additions and 11 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -1586,6 +1586,19 @@ func TestCacheIntervalInvalidationStopsWatch(t *testing.T) {
} }
} }
func makeWatchCacheEvent(rv uint64) *watchCacheEvent {
return &watchCacheEvent{
Type: watch.Added,
Object: &v1.Pod{
ObjectMeta: metav1.ObjectMeta{
Name: fmt.Sprintf("pod-%d", rv),
ResourceVersion: fmt.Sprintf("%d", rv),
},
},
ResourceVersion: rv,
}
}
// TestCacheWatcherDraining verifies the cacheWatcher.process goroutine is properly cleaned up when draining was requested // TestCacheWatcherDraining verifies the cacheWatcher.process goroutine is properly cleaned up when draining was requested
func TestCacheWatcherDraining(t *testing.T) { func TestCacheWatcherDraining(t *testing.T) {
var lock sync.RWMutex var lock sync.RWMutex
@ -1600,18 +1613,23 @@ func TestCacheWatcherDraining(t *testing.T) {
w.stopLocked() w.stopLocked()
} }
initEvents := []*watchCacheEvent{ initEvents := []*watchCacheEvent{
{Object: &v1.Pod{}}, makeWatchCacheEvent(5),
{Object: &v1.Pod{}}, makeWatchCacheEvent(6),
} }
w = newCacheWatcher(1, filter, forget, testVersioner{}, time.Now(), true, objectType, "") w = newCacheWatcher(1, filter, forget, testVersioner{}, time.Now(), true, objectType, "")
go w.processInterval(context.Background(), intervalFromEvents(initEvents), 0) go w.processInterval(context.Background(), intervalFromEvents(initEvents), 1)
if !w.add(&watchCacheEvent{Object: &v1.Pod{}}, time.NewTimer(1*time.Second)) { if !w.add(makeWatchCacheEvent(7), time.NewTimer(1*time.Second)) {
t.Fatal("failed adding an even to the watcher") t.Fatal("failed adding an even to the watcher")
} }
forget(true) // drain the watcher forget(true) // drain the watcher
<-w.ResultChan()
<-w.ResultChan() eventCount := 0
<-w.ResultChan() for range w.ResultChan() {
eventCount++
}
if eventCount != 3 {
t.Errorf("Unexpected number of objects received: %d, expected: 3", eventCount)
}
if err := wait.PollImmediate(1*time.Second, 5*time.Second, func() (bool, error) { if err := wait.PollImmediate(1*time.Second, 5*time.Second, func() (bool, error) {
lock.RLock() lock.RLock()
defer lock.RUnlock() defer lock.RUnlock()
@ -1636,12 +1654,12 @@ func TestCacheWatcherDrainingRequestedButNotDrained(t *testing.T) {
w.stopLocked() w.stopLocked()
} }
initEvents := []*watchCacheEvent{ initEvents := []*watchCacheEvent{
{Object: &v1.Pod{}}, makeWatchCacheEvent(5),
{Object: &v1.Pod{}}, makeWatchCacheEvent(6),
} }
w = newCacheWatcher(1, filter, forget, testVersioner{}, time.Now(), true, objectType, "") w = newCacheWatcher(1, filter, forget, testVersioner{}, time.Now(), true, objectType, "")
go w.processInterval(context.Background(), intervalFromEvents(initEvents), 0) go w.processInterval(context.Background(), intervalFromEvents(initEvents), 1)
if !w.add(&watchCacheEvent{Object: &v1.Pod{}}, time.NewTimer(1*time.Second)) { if !w.add(makeWatchCacheEvent(7), time.NewTimer(1*time.Second)) {
t.Fatal("failed adding an even to the watcher") t.Fatal("failed adding an even to the watcher")
} }
forget(true) // drain the watcher forget(true) // drain the watcher