From b1169ed91e78fec5a62518c77b7713804a9e04fa Mon Sep 17 00:00:00 2001 From: Daniel Smith Date: Mon, 15 Sep 2014 18:10:10 -0700 Subject: [PATCH] Fix cache to use the "List then Watch" pattern. --- pkg/client/cache/fifo.go | 17 ++++ pkg/client/cache/fifo_test.go | 26 +++++ pkg/client/cache/reflector.go | 95 ++++++++++++++----- pkg/client/cache/reflector_test.go | 147 ++++++++++++++++++++++++----- pkg/client/cache/store.go | 14 +++ pkg/client/cache/store_test.go | 64 +++++++++---- 6 files changed, 302 insertions(+), 61 deletions(-) diff --git a/pkg/client/cache/fifo.go b/pkg/client/cache/fifo.go index 07463e72c35..40261651974 100644 --- a/pkg/client/cache/fifo.go +++ b/pkg/client/cache/fifo.go @@ -116,6 +116,23 @@ func (f *FIFO) Pop() interface{} { } } +// Replace will delete the contents of 'f', using instead the given map. +// 'f' takes ownersip of the map, you should not reference the map again +// after calling this function. f's queue is reset, too; upon return, it +// will contain the items in the map, in no particular order. +func (f *FIFO) Replace(idToObj map[string]interface{}) { + f.lock.Lock() + defer f.lock.Unlock() + f.items = idToObj + f.queue = f.queue[:0] + for id := range idToObj { + f.queue = append(f.queue, id) + } + if len(f.queue) > 0 { + f.cond.Broadcast() + } +} + // NewFIFO returns a Store which can be used to queue up items to // process. func NewFIFO() *FIFO { diff --git a/pkg/client/cache/fifo_test.go b/pkg/client/cache/fifo_test.go index f86360aa327..e4a4a4f88d7 100644 --- a/pkg/client/cache/fifo_test.go +++ b/pkg/client/cache/fifo_test.go @@ -81,3 +81,29 @@ func TestFIFO_addUpdate(t *testing.T) { t.Errorf("item did not get removed") } } + +func TestFIFO_addReplace(t *testing.T) { + f := NewFIFO() + f.Add("foo", 10) + f.Replace(map[string]interface{}{"foo": 15}) + got := make(chan int, 2) + go func() { + for { + got <- f.Pop().(int) + } + }() + + first := <-got + if e, a := 15, first; e != a { + t.Errorf("Didn't get updated value (%v), got %v", e, a) + } + select { + case unexpected := <-got: + t.Errorf("Got second value %v", unexpected) + case <-time.After(50 * time.Millisecond): + } + _, exists := f.Get("foo") + if exists { + t.Errorf("item did not get removed") + } +} diff --git a/pkg/client/cache/reflector.go b/pkg/client/cache/reflector.go index 31a774410f9..1e309975048 100644 --- a/pkg/client/cache/reflector.go +++ b/pkg/client/cache/reflector.go @@ -17,6 +17,7 @@ limitations under the License. package cache import ( + "fmt" "reflect" "time" @@ -26,58 +27,106 @@ import ( "github.com/golang/glog" ) +// ListerWatcher is any object that knows how to perform an initial list and start a watch on a resource. +type ListerWatcher interface { + // List should return a list type object; the Items field will be extracted, and the + // ResourceVersion field will be used to start the watch in the right place. + List() (runtime.Object, error) + // Watch should begin a watch at the specified version. + Watch(resourceVersion uint64) (watch.Interface, error) +} + // Reflector watches a specified resource and causes all changes to be reflected in the given store. type Reflector struct { // The type of object we expect to place in the store. expectedType reflect.Type // The destination to sync up with the watch source store Store - // watchFactory is called to initiate watches. - watchFactory WatchFactory + // listerWatcher is used to perform lists and watches. + listerWatcher ListerWatcher // period controls timing between one watch ending and // the beginning of the next one. period time.Duration } -// WatchFactory should begin a watch at the specified version. -type WatchFactory func(resourceVersion uint64) (watch.Interface, error) - // NewReflector creates a new Reflector object which will keep the given store up to // date with the server's contents for the given resource. Reflector promises to // only put things in the store that have the type of expectedType. -func NewReflector(watchFactory WatchFactory, expectedType interface{}, store Store) *Reflector { - gc := &Reflector{ - watchFactory: watchFactory, - store: store, - expectedType: reflect.TypeOf(expectedType), - period: time.Second, +func NewReflector(lw ListerWatcher, expectedType interface{}, store Store) *Reflector { + r := &Reflector{ + listerWatcher: lw, + store: store, + expectedType: reflect.TypeOf(expectedType), + period: time.Second, } - return gc + return r } // Run starts a watch and handles watch events. Will restart the watch if it is closed. // Run starts a goroutine and returns immediately. -func (gc *Reflector) Run() { +func (r *Reflector) Run() { + go util.Forever(func() { r.listAndWatch() }, r.period) +} + +func (r *Reflector) listAndWatch() { var resourceVersion uint64 - go util.Forever(func() { - w, err := gc.watchFactory(resourceVersion) + + list, err := r.listerWatcher.List() + if err != nil { + glog.Errorf("Failed to list %v: %v", r.expectedType, err) + return + } + jsonBase, err := runtime.FindJSONBase(list) + if err != nil { + glog.Errorf("Unable to understand list result %#v", list) + return + } + resourceVersion = jsonBase.ResourceVersion() + items, err := runtime.ExtractList(list) + if err != nil { + glog.Errorf("Unable to understand list result %#v (%v)", list, err) + return + } + err = r.syncWith(items) + if err != nil { + glog.Errorf("Unable to sync list result: %v", err) + return + } + + for { + w, err := r.listerWatcher.Watch(resourceVersion) if err != nil { - glog.Errorf("failed to watch %v: %v", gc.expectedType, err) + glog.Errorf("failed to watch %v: %v", r.expectedType, err) return } - gc.watchHandler(w, &resourceVersion) - }, gc.period) + r.watchHandler(w, &resourceVersion) + } +} + +// syncWith replaces the store's items with the given list. +func (r *Reflector) syncWith(items []runtime.Object) error { + found := map[string]interface{}{} + for _, item := range items { + jsonBase, err := runtime.FindJSONBase(item) + if err != nil { + return fmt.Errorf("unexpected item in list: %v", err) + } + found[jsonBase.ID()] = item + } + + r.store.Replace(found) + return nil } // watchHandler watches w and keeps *resourceVersion up to date. -func (gc *Reflector) watchHandler(w watch.Interface, resourceVersion *uint64) { +func (r *Reflector) watchHandler(w watch.Interface, resourceVersion *uint64) { for { event, ok := <-w.ResultChan() if !ok { glog.Errorf("unexpected watch close") return } - if e, a := gc.expectedType, reflect.TypeOf(event.Object); e != a { + if e, a := r.expectedType, reflect.TypeOf(event.Object); e != a { glog.Errorf("expected type %v, but watch event object had type %v", e, a) continue } @@ -88,14 +137,14 @@ func (gc *Reflector) watchHandler(w watch.Interface, resourceVersion *uint64) { } switch event.Type { case watch.Added: - gc.store.Add(jsonBase.ID(), event.Object) + r.store.Add(jsonBase.ID(), event.Object) case watch.Modified: - gc.store.Update(jsonBase.ID(), event.Object) + r.store.Update(jsonBase.ID(), event.Object) case watch.Deleted: // TODO: Will any consumers need access to the "last known // state", which is passed in event.Object? If so, may need // to change this. - gc.store.Delete(jsonBase.ID()) + r.store.Delete(jsonBase.ID()) default: glog.Errorf("unable to understand watch event %#v", event) } diff --git a/pkg/client/cache/reflector_test.go b/pkg/client/cache/reflector_test.go index 89e5b7386da..60963d793af 100644 --- a/pkg/client/cache/reflector_test.go +++ b/pkg/client/cache/reflector_test.go @@ -17,15 +17,27 @@ limitations under the License. package cache import ( + "fmt" "testing" "github.com/GoogleCloudPlatform/kubernetes/pkg/api" + "github.com/GoogleCloudPlatform/kubernetes/pkg/runtime" "github.com/GoogleCloudPlatform/kubernetes/pkg/watch" ) +type testLW struct { + ListFunc func() (runtime.Object, error) + WatchFunc func(resourceVersion uint64) (watch.Interface, error) +} + +func (t *testLW) List() (runtime.Object, error) { return t.ListFunc() } +func (t *testLW) Watch(resourceVersion uint64) (watch.Interface, error) { + return t.WatchFunc(resourceVersion) +} + func TestReflector_watchHandler(t *testing.T) { s := NewStore() - g := NewReflector(nil, &api.Pod{}, s) + g := NewReflector(&testLW{}, &api.Pod{}, s) fw := watch.NewFake() s.Add("foo", &api.Pod{JSONBase: api.JSONBase{ID: "foo"}}) s.Add("bar", &api.Pod{JSONBase: api.JSONBase{ID: "bar"}}) @@ -68,27 +80,32 @@ func TestReflector_watchHandler(t *testing.T) { } } -func TestReflector_Run(t *testing.T) { +func TestReflector_listAndWatch(t *testing.T) { createdFakes := make(chan *watch.FakeWatcher) - // Expect our starter to get called at the beginning of the watch with 0, and again with 3 when we - // inject an error at 2. - expectedRVs := []uint64{0, 3} - watchStarter := func(rv uint64) (watch.Interface, error) { - fw := watch.NewFake() - if e, a := expectedRVs[0], rv; e != a { - t.Errorf("Expected rv %v, but got %v", e, a) - } - expectedRVs = expectedRVs[1:] - // channel is not buffered because the for loop below needs to block. But - // we don't want to block here, so report the new fake via a go routine. - go func() { createdFakes <- fw }() - return fw, nil + // The ListFunc says that it's at revision 1. Therefore, we expect our WatchFunc + // to get called at the beginning of the watch with 1, and again with 4 when we + // inject an error at 3. + expectedRVs := []uint64{1, 4} + lw := &testLW{ + WatchFunc: func(rv uint64) (watch.Interface, error) { + fw := watch.NewFake() + if e, a := expectedRVs[0], rv; e != a { + t.Errorf("Expected rv %v, but got %v", e, a) + } + expectedRVs = expectedRVs[1:] + // channel is not buffered because the for loop below needs to block. But + // we don't want to block here, so report the new fake via a go routine. + go func() { createdFakes <- fw }() + return fw, nil + }, + ListFunc: func() (runtime.Object, error) { + return &api.PodList{JSONBase: api.JSONBase{ResourceVersion: 1}}, nil + }, } s := NewFIFO() - r := NewReflector(watchStarter, &api.Pod{}, s) - r.period = 0 - r.Run() + r := NewReflector(lw, &api.Pod{}, s) + go r.listAndWatch() ids := []string{"foo", "bar", "baz", "qux", "zoo"} var fw *watch.FakeWatcher @@ -96,9 +113,9 @@ func TestReflector_Run(t *testing.T) { if fw == nil { fw = <-createdFakes } - sendingRV := uint64(i + 1) + sendingRV := uint64(i + 2) fw.Add(&api.Pod{JSONBase: api.JSONBase{ID: id, ResourceVersion: sendingRV}}) - if sendingRV == 2 { + if sendingRV == 3 { // Inject a failure. fw.Stop() fw = nil @@ -111,7 +128,7 @@ func TestReflector_Run(t *testing.T) { if e, a := id, pod.ID; e != a { t.Errorf("%v: Expected %v, got %v", i, e, a) } - if e, a := uint64(i+1), pod.ResourceVersion; e != a { + if e, a := uint64(i+2), pod.ResourceVersion; e != a { t.Errorf("%v: Expected %v, got %v", i, e, a) } } @@ -120,3 +137,91 @@ func TestReflector_Run(t *testing.T) { t.Error("called watchStarter an unexpected number of times") } } + +func TestReflector_listAndWatchWithErrors(t *testing.T) { + mkPod := func(id string, rv uint64) *api.Pod { + return &api.Pod{JSONBase: api.JSONBase{ID: id, ResourceVersion: rv}} + } + mkList := func(rv uint64, pods ...*api.Pod) *api.PodList { + list := &api.PodList{JSONBase: api.JSONBase{ResourceVersion: rv}} + for _, pod := range pods { + list.Items = append(list.Items, *pod) + } + return list + } + table := []struct { + list *api.PodList + listErr error + events []watch.Event + watchErr error + }{ + { + list: mkList(1), + events: []watch.Event{ + {watch.Added, mkPod("foo", 2)}, + {watch.Added, mkPod("bar", 3)}, + }, + }, { + list: mkList(3, mkPod("foo", 2), mkPod("bar", 3)), + events: []watch.Event{ + {watch.Deleted, mkPod("foo", 4)}, + {watch.Added, mkPod("qux", 5)}, + }, + }, { + listErr: fmt.Errorf("a list error"), + }, { + list: mkList(5, mkPod("bar", 3), mkPod("qux", 5)), + watchErr: fmt.Errorf("a watch error"), + }, { + list: mkList(5, mkPod("bar", 3), mkPod("qux", 5)), + events: []watch.Event{ + {watch.Added, mkPod("baz", 6)}, + }, + }, { + list: mkList(6, mkPod("bar", 3), mkPod("qux", 5), mkPod("baz", 6)), + }, + } + + s := NewFIFO() + for line, item := range table { + if item.list != nil { + // Test that the list is what currently exists in the store. + current := s.List() + checkMap := map[string]uint64{} + for _, item := range current { + pod := item.(*api.Pod) + checkMap[pod.ID] = pod.ResourceVersion + } + for _, pod := range item.list.Items { + if e, a := pod.ResourceVersion, checkMap[pod.ID]; e != a { + t.Errorf("%v: expected %v, got %v for pod %v", line, e, a, pod.ID) + } + } + if e, a := len(item.list.Items), len(checkMap); e != a { + t.Errorf("%v: expected %v, got %v", line, e, a) + } + } + watchRet, watchErr := item.events, item.watchErr + lw := &testLW{ + WatchFunc: func(rv uint64) (watch.Interface, error) { + if watchErr != nil { + return nil, watchErr + } + watchErr = fmt.Errorf("second watch") + fw := watch.NewFake() + go func() { + for _, e := range watchRet { + fw.Action(e.Type, e.Object) + } + fw.Stop() + }() + return fw, nil + }, + ListFunc: func() (runtime.Object, error) { + return item.list, item.listErr + }, + } + r := NewReflector(lw, &api.Pod{}, s) + r.listAndWatch() + } +} diff --git a/pkg/client/cache/store.go b/pkg/client/cache/store.go index 4b27aa29099..ebf0a2ce953 100644 --- a/pkg/client/cache/store.go +++ b/pkg/client/cache/store.go @@ -33,6 +33,11 @@ type Store interface { List() []interface{} Contains() util.StringSet Get(id string) (item interface{}, exists bool) + + // Replace will delete the contents of the store, using instead the + // given map. Store takes ownership of the map, you should not reference + // it after calling this function. + Replace(idToObj map[string]interface{}) } type cache struct { @@ -95,6 +100,15 @@ func (c *cache) Get(id string) (item interface{}, exists bool) { return item, exists } +// Replace will delete the contents of 'c', using instead the given map. +// 'c' takes ownership of the map, you should not reference the map again +// after calling this function. +func (c *cache) Replace(idToObj map[string]interface{}) { + c.lock.Lock() + defer c.lock.Unlock() + c.items = idToObj +} + // NewStore returns a Store implemented simply with a map and a lock. func NewStore() Store { return &cache{items: map[string]interface{}{}} diff --git a/pkg/client/cache/store_test.go b/pkg/client/cache/store_test.go index 4220d284f4a..7e9af6eef91 100644 --- a/pkg/client/cache/store_test.go +++ b/pkg/client/cache/store_test.go @@ -45,28 +45,58 @@ func doTestStore(t *testing.T, store Store) { t.Errorf("found deleted item??") } - // Test List + // Test List. store.Add("a", "b") store.Add("c", "d") store.Add("e", "e") - found := util.StringSet{} - for _, item := range store.List() { - found.Insert(item.(string)) - } - if !found.HasAll("b", "d", "e") { - t.Errorf("missing items") - } - if len(found) != 3 { - t.Errorf("extra items") + { + found := util.StringSet{} + for _, item := range store.List() { + found.Insert(item.(string)) + } + if !found.HasAll("b", "d", "e") { + t.Errorf("missing items") + } + if len(found) != 3 { + t.Errorf("extra items") + } + + // Check that ID list is correct. + ids := store.Contains() + if !ids.HasAll("a", "c", "e") { + t.Errorf("missing items") + } + if len(ids) != 3 { + t.Errorf("extra items") + } } - // Check that ID list is correct. - ids := store.Contains() - if !ids.HasAll("a", "c", "e") { - t.Errorf("missing items") - } - if len(ids) != 3 { - t.Errorf("extra items") + // Test Replace. + store.Replace(map[string]interface{}{ + "foo": "foo", + "bar": "bar", + }) + + { + found := util.StringSet{} + for _, item := range store.List() { + found.Insert(item.(string)) + } + if !found.HasAll("foo", "bar") { + t.Errorf("missing items") + } + if len(found) != 2 { + t.Errorf("extra items") + } + + // Check that ID list is correct. + ids := store.Contains() + if !ids.HasAll("foo", "bar") { + t.Errorf("missing items") + } + if len(ids) != 2 { + t.Errorf("extra items") + } } }