github/kubernetes
1
0
Fork 0
mirror of https://github.com/k3s-io/kubernetes.git synced 2026-01-06 07:57:35 +00:00
Code Issues Projects Releases Wiki Activity

cache: reflector should never stop watching

Browse Source 
A recent change tries to separate resync and relist. The motivation
was to avoid triggering relist when a resync is required.

However, the change is not effective since it stops the watcher. As hongchao
mentioned in the original comment, today's storage interface will not deliever
any progress notification to the watch chan. So any watcher that does not receive
events for the last few seconds will not be able to catch up from the previous
index after a hard close since the index of the last received event is out of
the cache window inside etcd2.

This pull request tries to fix this issue by not stoping watcher when a resync is
required.
This commit is contained in:
Xiang Li
2016-05-19 21:42:10 -07:00
parent 0257f54a8c
commit abbbd7c2f2
2 changed files with 38 additions and 55 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
pkg/client/cache/reflector_test.go vendored
View File

@@ -30,6 +30,8 @@ import (
"k8s.io/kubernetes/pkg/watch"
)
var nevererrc chan error
type testLW struct {
ListFunc func() (runtime.Object, error)
WatchFunc func(options api.ListOptions) (watch.Interface, error)
@@ -84,7 +86,7 @@ func TestRunUntil(t *testing.T) {
// Synchronously add a dummy pod into the watch channel so we
// know the RunUntil go routine is in the watch handler.
fw.Add(&api.Pod{ObjectMeta: api.ObjectMeta{Name: "bar"}})
stopCh <- struct{}{}
close(stopCh)
select {
case _, ok := <-fw.ResultChan():
if ok {
@@ -129,7 +131,7 @@ func TestReflectorWatchHandlerError(t *testing.T) {
fw.Stop()
}()
var resumeRV string
err := g.watchHandler(fw, &resumeRV, neverExitWatch, wait.NeverStop)
err := g.watchHandler(fw, &resumeRV, nevererrc, wait.NeverStop)
if err == nil {
t.Errorf("unexpected non-error")
}
@@ -149,7 +151,7 @@ func TestReflectorWatchHandler(t *testing.T) {
fw.Stop()
}()
var resumeRV string
err := g.watchHandler(fw, &resumeRV, neverExitWatch, wait.NeverStop)
err := g.watchHandler(fw, &resumeRV, nevererrc, wait.NeverStop)
if err != nil {
t.Errorf("unexpected error %v", err)
}
@@ -198,7 +200,7 @@ func TestReflectorStopWatch(t *testing.T) {
var resumeRV string
stopWatch := make(chan struct{}, 1)
stopWatch <- struct{}{}
err := g.watchHandler(fw, &resumeRV, neverExitWatch, stopWatch)
err := g.watchHandler(fw, &resumeRV, nevererrc, stopWatch)
if err != errorStopRequested {
t.Errorf("expected stop error, got %q", err)
}