Merge pull request #3810 from ironcladlou/cache-namespacing

Support namespacing in cache.Store

pkg/client/cache/fifo.go

@@ -17,9 +17,8 @@ limitations under the License.
 package cache
 
 import (
+	"fmt"
 	"sync"
-
-	"github.com/GoogleCloudPlatform/kubernetes/pkg/util"
 )
 
 // FIFO receives adds and updates from a Reflector, and puts them in a queue for
@@ -33,11 +32,18 @@ type FIFO struct {
 	// We depend on the property that items in the set are in the queue and vice versa.
 	items map[string]interface{}
 	queue []string
+	// keyFunc is used to make the key used for queued item insertion and retrieval, and
+	// should be deterministic.
+	keyFunc KeyFunc
 }
 
 // Add inserts an item, and puts it in the queue. The item is only enqueued
 // if it doesn't already exist in the set.
-func (f *FIFO) Add(id string, obj interface{}) {
+func (f *FIFO) Add(obj interface{}) error {
+	id, err := f.keyFunc(obj)
+	if err != nil {
+		return fmt.Errorf("couldn't create key for object: %v", err)
+	}
 	f.lock.Lock()
 	defer f.lock.Unlock()
 	if _, exists := f.items[id]; !exists {
@@ -45,20 +51,26 @@ func (f *FIFO) Add(id string, obj interface{}) {
 	}
 	f.items[id] = obj
 	f.cond.Broadcast()
+	return nil
 }
 
 // Update is the same as Add in this implementation.
-func (f *FIFO) Update(id string, obj interface{}) {
-	f.Add(id, obj)
+func (f *FIFO) Update(obj interface{}) error {
+	return f.Add(obj)
 }
 
 // Delete removes an item. It doesn't add it to the queue, because
 // this implementation assumes the consumer only cares about the objects,
 // not the order in which they were created/added.
-func (f *FIFO) Delete(id string) {
+func (f *FIFO) Delete(obj interface{}) error {
+	id, err := f.keyFunc(obj)
+	if err != nil {
+		return fmt.Errorf("couldn't create key for object: %v", err)
+	}
 	f.lock.Lock()
 	defer f.lock.Unlock()
 	delete(f.items, id)
+	return err
 }
 
 // List returns a list of all the items.
@@ -72,25 +84,16 @@ func (f *FIFO) List() []interface{} {
 	return list
 }
 
-// ContainedIDs returns a util.StringSet containing all IDs of the stored items.
-// This is a snapshot of a moment in time, and one should keep in mind that
-// other go routines can add or remove items after you call this.
-func (c *FIFO) ContainedIDs() util.StringSet {
-	c.lock.RLock()
-	defer c.lock.RUnlock()
-	set := util.StringSet{}
-	for id := range c.items {
-		set.Insert(id)
-	}
-	return set
-}
-
 // Get returns the requested item, or sets exists=false.
-func (f *FIFO) Get(id string) (item interface{}, exists bool) {
+func (f *FIFO) Get(obj interface{}) (item interface{}, exists bool, err error) {
+	id, err := f.keyFunc(obj)
+	if err != nil {
+		return nil, false, fmt.Errorf("couldn't create key for object: %v", err)
+	}
 	f.lock.RLock()
 	defer f.lock.RUnlock()
 	item, exists = f.items[id]
-	return item, exists
+	return item, exists, nil
 }
 
 // Pop waits until an item is ready and returns it. If multiple items are
@@ -120,25 +123,36 @@ func (f *FIFO) Pop() interface{} {
 // '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{}) {
+func (f *FIFO) Replace(list []interface{}) error {
+	items := map[string]interface{}{}
+	for _, item := range list {
+		key, err := f.keyFunc(item)
+		if err != nil {
+			return fmt.Errorf("couldn't create key for object: %v", err)
+		}
+		items[key] = item
+	}
+
 	f.lock.Lock()
 	defer f.lock.Unlock()
-	f.items = idToObj
+	f.items = items
 	f.queue = f.queue[:0]
-	for id := range idToObj {
+	for id := range items {
 		f.queue = append(f.queue, id)
 	}
 	if len(f.queue) > 0 {
 		f.cond.Broadcast()
 	}
+	return nil
 }
 
 // NewFIFO returns a Store which can be used to queue up items to
 // process.
-func NewFIFO() *FIFO {
+func NewFIFO(keyFunc KeyFunc) *FIFO {
 	f := &FIFO{
-		items: map[string]interface{}{},
-		queue: []string{},
+		items:   map[string]interface{}{},
+		queue:   []string{},
+		keyFunc: keyFunc,
 	}
 	f.cond.L = &f.lock
 	return f

pkg/client/cache/fifo_test.go

@@ -21,30 +21,43 @@ import (
 	"time"
 )
 
+func testFifoObjectKeyFunc(obj interface{}) (string, error) {
+	return obj.(testFifoObject).name, nil
+}
+
+type testFifoObject struct {
+	name string
+	val  interface{}
+}
+
 func TestFIFO_basic(t *testing.T) {
-	f := NewFIFO()
+	mkObj := func(name string, val interface{}) testFifoObject {
+		return testFifoObject{name: name, val: val}
+	}
+
+	f := NewFIFO(testFifoObjectKeyFunc)
 	const amount = 500
 	go func() {
 		for i := 0; i < amount; i++ {
-			f.Add(string([]rune{'a', rune(i)}), i+1)
+			f.Add(mkObj(string([]rune{'a', rune(i)}), i+1))
 		}
 	}()
 	go func() {
-		for u := uint(0); u < amount; u++ {
-			f.Add(string([]rune{'b', rune(u)}), u+1)
+		for u := uint64(0); u < amount; u++ {
+			f.Add(mkObj(string([]rune{'b', rune(u)}), u+1))
 		}
 	}()
 
 	lastInt := int(0)
-	lastUint := uint(0)
+	lastUint := uint64(0)
 	for i := 0; i < amount*2; i++ {
-		switch obj := f.Pop().(type) {
+		switch obj := f.Pop().(testFifoObject).val.(type) {
 		case int:
 			if obj <= lastInt {
 				t.Errorf("got %v (int) out of order, last was %v", obj, lastInt)
 			}
 			lastInt = obj
-		case uint:
+		case uint64:
 			if obj <= lastUint {
 				t.Errorf("got %v (uint) out of order, last was %v", obj, lastUint)
 			} else {
@@ -57,81 +70,93 @@ func TestFIFO_basic(t *testing.T) {
 }
 
 func TestFIFO_addUpdate(t *testing.T) {
-	f := NewFIFO()
-	f.Add("foo", 10)
-	f.Update("foo", 15)
-	got := make(chan int, 2)
+	mkObj := func(name string, val interface{}) testFifoObject {
+		return testFifoObject{name: name, val: val}
+	}
+
+	f := NewFIFO(testFifoObjectKeyFunc)
+	f.Add(mkObj("foo", 10))
+	f.Update(mkObj("foo", 15))
+	got := make(chan testFifoObject, 2)
 	go func() {
 		for {
-			got <- f.Pop().(int)
+			got <- f.Pop().(testFifoObject)
 		}
 	}()
 
 	first := <-got
-	if e, a := 15, first; e != a {
+	if e, a := 15, first.val; 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)
+		t.Errorf("Got second value %v", unexpected.val)
 	case <-time.After(50 * time.Millisecond):
 	}
-	_, exists := f.Get("foo")
+	_, exists, _ := f.Get(mkObj("foo", ""))
 	if exists {
 		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)
+	mkObj := func(name string, val interface{}) testFifoObject {
+		return testFifoObject{name: name, val: val}
+	}
+
+	f := NewFIFO(testFifoObjectKeyFunc)
+	f.Add(mkObj("foo", 10))
+	f.Replace([]interface{}{mkObj("foo", 15)})
+	got := make(chan testFifoObject, 2)
 	go func() {
 		for {
-			got <- f.Pop().(int)
+			got <- f.Pop().(testFifoObject)
 		}
 	}()
 
 	first := <-got
-	if e, a := 15, first; e != a {
+	if e, a := 15, first.val; 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)
+		t.Errorf("Got second value %v", unexpected.val)
 	case <-time.After(50 * time.Millisecond):
 	}
-	_, exists := f.Get("foo")
+	_, exists, _ := f.Get(mkObj("foo", ""))
 	if exists {
 		t.Errorf("item did not get removed")
 	}
 }
 
 func TestFIFO_detectLineJumpers(t *testing.T) {
-	f := NewFIFO()
+	mkObj := func(name string, val interface{}) testFifoObject {
+		return testFifoObject{name: name, val: val}
+	}
 
-	f.Add("foo", 10)
-	f.Add("bar", 1)
-	f.Add("foo", 11)
-	f.Add("foo", 13)
-	f.Add("zab", 30)
+	f := NewFIFO(testFifoObjectKeyFunc)
 
-	if e, a := 13, f.Pop().(int); a != e {
+	f.Add(mkObj("foo", 10))
+	f.Add(mkObj("bar", 1))
+	f.Add(mkObj("foo", 11))
+	f.Add(mkObj("foo", 13))
+	f.Add(mkObj("zab", 30))
+
+	if e, a := 13, f.Pop().(testFifoObject).val; a != e {
 		t.Fatalf("expected %d, got %d", e, a)
 	}
 
-	f.Add("foo", 14) // ensure foo doesn't jump back in line
+	f.Add(mkObj("foo", 14)) // ensure foo doesn't jump back in line
 
-	if e, a := 1, f.Pop().(int); a != e {
+	if e, a := 1, f.Pop().(testFifoObject).val; a != e {
 		t.Fatalf("expected %d, got %d", e, a)
 	}
 
-	if e, a := 30, f.Pop().(int); a != e {
+	if e, a := 30, f.Pop().(testFifoObject).val; a != e {
 		t.Fatalf("expected %d, got %d", e, a)
 	}
 
-	if e, a := 14, f.Pop().(int); a != e {
+	if e, a := 14, f.Pop().(testFifoObject).val; a != e {
 		t.Fatalf("expected %d, got %d", e, a)
 	}
 }

pkg/client/cache/listers.go

@@ -69,10 +69,17 @@ func (s *StoreToNodeLister) List() (machines api.NodeList, err error) {
 // rather than a method of StoreToNodeLister.
 // GetNodeInfo returns cached data for the minion 'id'.
 func (s *StoreToNodeLister) GetNodeInfo(id string) (*api.Node, error) {
-	if minion, ok := s.Get(id); ok {
-		return minion.(*api.Node), nil
+	minion, exists, err := s.Get(&api.Node{ObjectMeta: api.ObjectMeta{Name: id}})
+
+	if err != nil {
+		return nil, fmt.Errorf("error retrieving minion '%v' from cache: %v", id, err)
 	}
-	return nil, fmt.Errorf("minion '%v' is not in cache", id)
+
+	if !exists {
+		return nil, fmt.Errorf("minion '%v' is not in cache", id)
+	}
+
+	return minion.(*api.Node), nil
 }
 
 // StoreToServiceLister makes a Store that has the List method of the client.ServiceInterface

pkg/client/cache/listers_test.go

@@ -25,10 +25,10 @@ import (
 )
 
 func TestStoreToMinionLister(t *testing.T) {
-	store := NewStore()
+	store := NewStore(MetaNamespaceKeyFunc)
 	ids := util.NewStringSet("foo", "bar", "baz")
 	for id := range ids {
-		store.Add(id, &api.Node{ObjectMeta: api.ObjectMeta{Name: id}})
+		store.Add(&api.Node{ObjectMeta: api.ObjectMeta{Name: id}})
 	}
 	sml := StoreToNodeLister{store}
 
@@ -46,10 +46,10 @@ func TestStoreToMinionLister(t *testing.T) {
 }
 
 func TestStoreToPodLister(t *testing.T) {
-	store := NewStore()
+	store := NewStore(MetaNamespaceKeyFunc)
 	ids := []string{"foo", "bar", "baz"}
 	for _, id := range ids {
-		store.Add(id, &api.Pod{
+		store.Add(&api.Pod{
 			ObjectMeta: api.ObjectMeta{
 				Name:   id,
 				Labels: map[string]string{"name": id},

pkg/client/cache/poller.go

@@ -27,7 +27,7 @@ import (
 // one object at a time.
 type Enumerator interface {
 	Len() int
-	Get(index int) (ID string, object interface{})
+	Get(index int) (object interface{})
 }
 
 // GetFunc should return an enumerator that you wish the Poller to proccess.
@@ -76,14 +76,11 @@ func (p *Poller) run() {
 }
 
 func (p *Poller) sync(e Enumerator) {
-	current := p.store.ContainedIDs()
+	items := []interface{}{}
 	for i := 0; i < e.Len(); i++ {
-		id, object := e.Get(i)
-		p.store.Update(id, object)
-		current.Delete(id)
-	}
-	// Delete all the objects not found.
-	for id := range current {
-		p.store.Delete(id)
+		object := e.Get(i)
+		items = append(items, object)
 	}
+
+	p.store.Replace(items)
 }

pkg/client/cache/poller_test.go

@@ -23,6 +23,10 @@ import (
 	"time"
 )
 
+func testPairKeyFunc(obj interface{}) (string, error) {
+	return obj.(testPair).id, nil
+}
+
 type testPair struct {
 	id  string
 	obj interface{}
@@ -30,8 +34,8 @@ type testPair struct {
 type testEnumerator []testPair
 
 func (t testEnumerator) Len() int { return len(t) }
-func (t testEnumerator) Get(i int) (string, interface{}) {
-	return t[i].id, t[i].obj
+func (t testEnumerator) Get(i int) interface{} {
+	return t[i]
 }
 
 func TestPoller_sync(t *testing.T) {
@@ -64,28 +68,35 @@ func TestPoller_sync(t *testing.T) {
 	}
 
 	for testCase, item := range table {
-		s := NewStore()
+		s := NewStore(testPairKeyFunc)
 		// This is a unit test for the sync function, hence the nil getFunc.
 		p := NewPoller(nil, 0, s)
 		for line, pairs := range item.steps {
 			p.sync(testEnumerator(pairs))
 
-			ids := s.ContainedIDs()
+			list := s.List()
 			for _, pair := range pairs {
-				if !ids.Has(pair.id) {
-					t.Errorf("%v, %v: expected to find entry for %v, but did not.", testCase, line, pair.id)
+				foundInList := false
+				for _, listItem := range list {
+					id, _ := testPairKeyFunc(listItem)
+					if pair.id == id {
+						foundInList = true
+					}
+				}
+				if !foundInList {
+					t.Errorf("%v, %v: expected to find list entry for %v, but did not.", testCase, line, pair.id)
 					continue
 				}
-				found, ok := s.Get(pair.id)
+				found, ok, _ := s.Get(pair)
 				if !ok {
 					t.Errorf("%v, %v: unexpected absent entry for %v", testCase, line, pair.id)
 					continue
 				}
-				if e, a := pair.obj, found; !reflect.DeepEqual(e, a) {
+				if e, a := pair.obj, found.(testPair).obj; !reflect.DeepEqual(e, a) {
 					t.Errorf("%v, %v: expected %v, got %v for %v", testCase, line, e, a, pair.id)
 				}
 			}
-			if e, a := len(pairs), len(ids); e != a {
+			if e, a := len(pairs), len(list); e != a {
 				t.Errorf("%v, %v: expected len %v, got %v", testCase, line, e, a)
 			}
 		}
@@ -93,7 +104,7 @@ func TestPoller_sync(t *testing.T) {
 }
 
 func TestPoller_Run(t *testing.T) {
-	s := NewStore()
+	s := NewStore(testPairKeyFunc)
 	const count = 10
 	var called = 0
 	done := make(chan struct{})
@@ -113,7 +124,7 @@ func TestPoller_Run(t *testing.T) {
 	<-done
 
 	// We never added anything, verify that.
-	if e, a := 0, len(s.ContainedIDs()); e != a {
+	if e, a := 0, len(s.List()); e != a {
 		t.Errorf("expected %v, got %v", e, a)
 	}
 }

pkg/client/cache/reflector.go

@@ -18,7 +18,6 @@ package cache
 
 import (
 	"errors"
-	"fmt"
 	"io"
 	"reflect"
 	"time"
@@ -97,8 +96,7 @@ func (r *Reflector) listAndWatch() {
 		glog.Errorf("Unable to understand list result %#v (%v)", list, err)
 		return
 	}
-	err = r.syncWith(items)
-	if err != nil {
+	if err := r.syncWith(items); err != nil {
 		glog.Errorf("Unable to sync list result: %v", err)
 		return
 	}
@@ -125,17 +123,12 @@ func (r *Reflector) listAndWatch() {
 
 // syncWith replaces the store's items with the given list.
 func (r *Reflector) syncWith(items []runtime.Object) error {
-	found := map[string]interface{}{}
+	found := make([]interface{}, 0, len(items))
 	for _, item := range items {
-		meta, err := meta.Accessor(item)
-		if err != nil {
-			return fmt.Errorf("unexpected item in list: %v", err)
-		}
-		found[meta.Name()] = item
+		found = append(found, item)
 	}
 
-	r.store.Replace(found)
-	return nil
+	return r.store.Replace(found)
 }
 
 // watchHandler watches w and keeps *resourceVersion up to date.
@@ -161,14 +154,14 @@ func (r *Reflector) watchHandler(w watch.Interface, resourceVersion *string) err
 		}
 		switch event.Type {
 		case watch.Added:
-			r.store.Add(meta.Name(), event.Object)
+			r.store.Add(event.Object)
 		case watch.Modified:
-			r.store.Update(meta.Name(), event.Object)
+			r.store.Update(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.
-			r.store.Delete(meta.Name())
+			r.store.Delete(event.Object)
 		default:
 			glog.Errorf("unable to understand watch event %#v", event)
 		}

pkg/client/cache/reflector_test.go

@@ -37,7 +37,7 @@ func (t *testLW) Watch(resourceVersion string) (watch.Interface, error) {
 }
 
 func TestReflector_watchHandlerError(t *testing.T) {
-	s := NewStore()
+	s := NewStore(MetaNamespaceKeyFunc)
 	g := NewReflector(&testLW{}, &api.Pod{}, s)
 	fw := watch.NewFake()
 	go func() {
@@ -51,11 +51,11 @@ func TestReflector_watchHandlerError(t *testing.T) {
 }
 
 func TestReflector_watchHandler(t *testing.T) {
-	s := NewStore()
+	s := NewStore(MetaNamespaceKeyFunc)
 	g := NewReflector(&testLW{}, &api.Pod{}, s)
 	fw := watch.NewFake()
-	s.Add("foo", &api.Pod{ObjectMeta: api.ObjectMeta{Name: "foo"}})
-	s.Add("bar", &api.Pod{ObjectMeta: api.ObjectMeta{Name: "bar"}})
+	s.Add(&api.Pod{ObjectMeta: api.ObjectMeta{Name: "foo"}})
+	s.Add(&api.Pod{ObjectMeta: api.ObjectMeta{Name: "bar"}})
 	go func() {
 		fw.Add(&api.Service{ObjectMeta: api.ObjectMeta{Name: "rejected"}})
 		fw.Delete(&api.Pod{ObjectMeta: api.ObjectMeta{Name: "foo"}})
@@ -69,26 +69,29 @@ func TestReflector_watchHandler(t *testing.T) {
 		t.Errorf("unexpected error %v", err)
 	}
 
+	mkPod := func(id string, rv string) *api.Pod {
+		return &api.Pod{ObjectMeta: api.ObjectMeta{Name: id, ResourceVersion: rv}}
+	}
+
 	table := []struct {
-		ID     string
-		RV     string
+		Pod    *api.Pod
 		exists bool
 	}{
-		{"foo", "", false},
-		{"rejected", "", false},
-		{"bar", "55", true},
-		{"baz", "32", true},
+		{mkPod("foo", ""), false},
+		{mkPod("rejected", ""), false},
+		{mkPod("bar", "55"), true},
+		{mkPod("baz", "32"), true},
 	}
 	for _, item := range table {
-		obj, exists := s.Get(item.ID)
+		obj, exists, _ := s.Get(item.Pod)
 		if e, a := item.exists, exists; e != a {
-			t.Errorf("%v: expected %v, got %v", item.ID, e, a)
+			t.Errorf("%v: expected %v, got %v", item.Pod, e, a)
 		}
 		if !exists {
 			continue
 		}
-		if e, a := item.RV, obj.(*api.Pod).ResourceVersion; e != a {
-			t.Errorf("%v: expected %v, got %v", item.ID, e, a)
+		if e, a := item.Pod.ResourceVersion, obj.(*api.Pod).ResourceVersion; e != a {
+			t.Errorf("%v: expected %v, got %v", item.Pod, e, a)
 		}
 	}
 
@@ -121,7 +124,7 @@ func TestReflector_listAndWatch(t *testing.T) {
 			return &api.PodList{ListMeta: api.ListMeta{ResourceVersion: "1"}}, nil
 		},
 	}
-	s := NewFIFO()
+	s := NewFIFO(MetaNamespaceKeyFunc)
 	r := NewReflector(lw, &api.Pod{}, s)
 	go r.listAndWatch()
 
@@ -200,7 +203,7 @@ func TestReflector_listAndWatchWithErrors(t *testing.T) {
 		},
 	}
 
-	s := NewFIFO()
+	s := NewFIFO(MetaNamespaceKeyFunc)
 	for line, item := range table {
 		if item.list != nil {
 			// Test that the list is what currently exists in the store.

pkg/client/cache/store.go

@@ -17,53 +17,89 @@ limitations under the License.
 package cache
 
 import (
+	"fmt"
 	"sync"
 
-	"github.com/GoogleCloudPlatform/kubernetes/pkg/util"
+	"github.com/GoogleCloudPlatform/kubernetes/pkg/api/meta"
 )
 
 // Store is a generic object storage interface. Reflector knows how to watch a server
 // and update a store. A generic store is provided, which allows Reflector to be used
 // as a local caching system, and an LRU store, which allows Reflector to work like a
 // queue of items yet to be processed.
+//
+// Store makes no assumptions about stored object identity; it is the responsibility
+// of a Store implementation to provide a mechanism to correctly key objects and to
+// define the contract for obtaining objects by some arbitrary key type.
 type Store interface {
-	Add(id string, obj interface{})
-	Update(id string, obj interface{})
-	Delete(id string)
+	Add(obj interface{}) error
+	Update(obj interface{}) error
+	Delete(obj interface{}) error
 	List() []interface{}
-	ContainedIDs() util.StringSet
-	Get(id string) (item interface{}, exists bool)
+	Get(obj interface{}) (item interface{}, exists bool, err error)
 
 	// Replace will delete the contents of the store, using instead the
-	// given map. Store takes ownership of the map, you should not reference
+	// given list. Store takes ownership of the list, you should not reference
 	// it after calling this function.
-	Replace(idToObj map[string]interface{})
+	Replace([]interface{}) error
+}
+
+// KeyFunc knows how to make a key from an object. Implementations should be deterministic.
+type KeyFunc func(obj interface{}) (string, error)
+
+// MetaNamespaceKeyFunc is a convenient default KeyFunc which knows how to make
+// keys for API objects which implement meta.Interface.
+// The key uses the format: <namespace>/<name>
+func MetaNamespaceKeyFunc(obj interface{}) (string, error) {
+	meta, err := meta.Accessor(obj)
+	if err != nil {
+		return "", fmt.Errorf("object has no meta: %v", err)
+	}
+	return meta.Namespace() + "/" + meta.Name(), nil
 }
 
 type cache struct {
 	lock  sync.RWMutex
 	items map[string]interface{}
+	// keyFunc is used to make the key for objects stored in and retrieved from items, and
+	// should be deterministic.
+	keyFunc KeyFunc
 }
 
 // Add inserts an item into the cache.
-func (c *cache) Add(id string, obj interface{}) {
+func (c *cache) Add(obj interface{}) error {
+	id, err := c.keyFunc(obj)
+	if err != nil {
+		return fmt.Errorf("couldn't create key for object: %v", err)
+	}
 	c.lock.Lock()
 	defer c.lock.Unlock()
 	c.items[id] = obj
+	return nil
 }
 
 // Update sets an item in the cache to its updated state.
-func (c *cache) Update(id string, obj interface{}) {
+func (c *cache) Update(obj interface{}) error {
+	id, err := c.keyFunc(obj)
+	if err != nil {
+		return fmt.Errorf("couldn't create key for object: %v", err)
+	}
 	c.lock.Lock()
 	defer c.lock.Unlock()
 	c.items[id] = obj
+	return nil
 }
 
 // Delete removes an item from the cache.
-func (c *cache) Delete(id string) {
+func (c *cache) Delete(obj interface{}) error {
+	id, err := c.keyFunc(obj)
+	if err != nil {
+		return fmt.Errorf("couldn't create key for object: %v", err)
+	}
 	c.lock.Lock()
 	defer c.lock.Unlock()
 	delete(c.items, id)
+	return nil
 }
 
 // List returns a list of all the items.
@@ -78,38 +114,39 @@ func (c *cache) List() []interface{} {
 	return list
 }
 
-// ContainedIDs returns a util.StringSet containing all IDs of the stored items.
-// This is a snapshot of a moment in time, and one should keep in mind that
-// other go routines can add or remove items after you call this.
-func (c *cache) ContainedIDs() util.StringSet {
-	c.lock.RLock()
-	defer c.lock.RUnlock()
-	set := util.StringSet{}
-	for id := range c.items {
-		set.Insert(id)
-	}
-	return set
-}
-
 // Get returns the requested item, or sets exists=false.
 // Get is completely threadsafe as long as you treat all items as immutable.
-func (c *cache) Get(id string) (item interface{}, exists bool) {
+func (c *cache) Get(obj interface{}) (item interface{}, exists bool, err error) {
+	id, _ := c.keyFunc(obj)
+	if err != nil {
+		return nil, false, fmt.Errorf("couldn't create key for object: %v", err)
+	}
 	c.lock.RLock()
 	defer c.lock.RUnlock()
 	item, exists = c.items[id]
-	return item, exists
+	return item, exists, nil
 }
 
-// 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
+// Replace will delete the contents of 'c', using instead the given list.
+// 'c' takes ownership of the list, you should not reference the list again
 // after calling this function.
-func (c *cache) Replace(idToObj map[string]interface{}) {
+func (c *cache) Replace(list []interface{}) error {
+	items := map[string]interface{}{}
+	for _, item := range list {
+		key, err := c.keyFunc(item)
+		if err != nil {
+			return fmt.Errorf("couldn't create key for object: %v", err)
+		}
+		items[key] = item
+	}
+
 	c.lock.Lock()
 	defer c.lock.Unlock()
-	c.items = idToObj
+	c.items = items
+	return nil
 }
 
 // NewStore returns a Store implemented simply with a map and a lock.
-func NewStore() Store {
-	return &cache{items: map[string]interface{}{}}
+func NewStore(keyFunc KeyFunc) Store {
+	return &cache{items: map[string]interface{}{}, keyFunc: keyFunc}
 }

pkg/client/cache/store_test.go

@@ -24,63 +24,58 @@ import (
 
 // Test public interface
 func doTestStore(t *testing.T, store Store) {
-	store.Add("foo", "bar")
-	if item, ok := store.Get("foo"); !ok {
+	mkObj := func(id string, val string) testStoreObject {
+		return testStoreObject{id: id, val: val}
+	}
+
+	store.Add(mkObj("foo", "bar"))
+	if item, ok, _ := store.Get(mkObj("foo", "")); !ok {
 		t.Errorf("didn't find inserted item")
 	} else {
-		if e, a := "bar", item.(string); e != a {
+		if e, a := "bar", item.(testStoreObject).val; e != a {
 			t.Errorf("expected %v, got %v", e, a)
 		}
 	}
-	store.Update("foo", "baz")
-	if item, ok := store.Get("foo"); !ok {
+	store.Update(mkObj("foo", "baz"))
+	if item, ok, _ := store.Get(mkObj("foo", "")); !ok {
 		t.Errorf("didn't find inserted item")
 	} else {
-		if e, a := "baz", item.(string); e != a {
+		if e, a := "baz", item.(testStoreObject).val; e != a {
 			t.Errorf("expected %v, got %v", e, a)
 		}
 	}
-	store.Delete("foo")
-	if _, ok := store.Get("foo"); ok {
+	store.Delete(mkObj("foo", ""))
+	if _, ok, _ := store.Get(mkObj("foo", "")); ok {
 		t.Errorf("found deleted item??")
 	}
 
 	// Test List.
-	store.Add("a", "b")
-	store.Add("c", "d")
-	store.Add("e", "e")
+	store.Add(mkObj("a", "b"))
+	store.Add(mkObj("c", "d"))
+	store.Add(mkObj("e", "e"))
 	{
 		found := util.StringSet{}
 		for _, item := range store.List() {
-			found.Insert(item.(string))
+			found.Insert(item.(testStoreObject).val)
 		}
 		if !found.HasAll("b", "d", "e") {
-			t.Errorf("missing items")
+			t.Errorf("missing items, found: %v", found)
 		}
 		if len(found) != 3 {
 			t.Errorf("extra items")
 		}
-
-		// Check that ID list is correct.
-		ids := store.ContainedIDs()
-		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",
+	store.Replace([]interface{}{
+		mkObj("foo", "foo"),
+		mkObj("bar", "bar"),
 	})
 
 	{
 		found := util.StringSet{}
 		for _, item := range store.List() {
-			found.Insert(item.(string))
+			found.Insert(item.(testStoreObject).val)
 		}
 		if !found.HasAll("foo", "bar") {
 			t.Errorf("missing items")
@@ -88,27 +83,27 @@ func doTestStore(t *testing.T, store Store) {
 		if len(found) != 2 {
 			t.Errorf("extra items")
 		}
-
-		// Check that ID list is correct.
-		ids := store.ContainedIDs()
-		if !ids.HasAll("foo", "bar") {
-			t.Errorf("missing items")
-		}
-		if len(ids) != 2 {
-			t.Errorf("extra items")
-		}
 	}
 }
 
+func testStoreKeyFunc(obj interface{}) (string, error) {
+	return obj.(testStoreObject).id, nil
+}
+
+type testStoreObject struct {
+	id  string
+	val string
+}
+
 func TestCache(t *testing.T) {
-	doTestStore(t, NewStore())
+	doTestStore(t, NewStore(testStoreKeyFunc))
 }
 
 func TestFIFOCache(t *testing.T) {
-	doTestStore(t, NewFIFO())
+	doTestStore(t, NewFIFO(testStoreKeyFunc))
 }
 
 func TestUndeltaStore(t *testing.T) {
 	nop := func([]interface{}) {}
-	doTestStore(t, NewUndeltaStore(nop))
+	doTestStore(t, NewUndeltaStore(nop, testStoreKeyFunc))
 }

pkg/client/cache/undelta_store.go

@@ -16,10 +16,6 @@ limitations under the License.
 
 package cache
 
-import (
-	"github.com/GoogleCloudPlatform/kubernetes/pkg/util"
-)
-
 // UndeltaStore listens to incremental updates and sends complete state on every change.
 // It implements the Store interface so that it can receive a stream of mirrored objects
 // from Reflector.  Whenever it receives any complete (Store.Replace) or incremental change
@@ -46,36 +42,45 @@ var _ Store = &UndeltaStore{}
 // 4               Store.List() -> [a,b]
 // 5                                         Store.List() -> [a,b]
 
-func (u *UndeltaStore) Add(id string, obj interface{}) {
-	u.ActualStore.Add(id, obj)
+func (u *UndeltaStore) Add(obj interface{}) error {
+	if err := u.ActualStore.Add(obj); err != nil {
+		return err
+	}
 	u.PushFunc(u.ActualStore.List())
+	return nil
 }
-func (u *UndeltaStore) Update(id string, obj interface{}) {
-	u.ActualStore.Update(id, obj)
+func (u *UndeltaStore) Update(obj interface{}) error {
+	if err := u.ActualStore.Update(obj); err != nil {
+		return err
+	}
 	u.PushFunc(u.ActualStore.List())
+	return nil
 }
-func (u *UndeltaStore) Delete(id string) {
-	u.ActualStore.Delete(id)
+func (u *UndeltaStore) Delete(obj interface{}) error {
+	if err := u.ActualStore.Delete(obj); err != nil {
+		return err
+	}
 	u.PushFunc(u.ActualStore.List())
+	return nil
 }
 func (u *UndeltaStore) List() []interface{} {
 	return u.ActualStore.List()
 }
-func (u *UndeltaStore) ContainedIDs() util.StringSet {
-	return u.ActualStore.ContainedIDs()
+func (u *UndeltaStore) Get(obj interface{}) (item interface{}, exists bool, err error) {
+	return u.ActualStore.Get(obj)
 }
-func (u *UndeltaStore) Get(id string) (item interface{}, exists bool) {
-	return u.ActualStore.Get(id)
-}
-func (u *UndeltaStore) Replace(idToObj map[string]interface{}) {
-	u.ActualStore.Replace(idToObj)
+func (u *UndeltaStore) Replace(list []interface{}) error {
+	if err := u.ActualStore.Replace(list); err != nil {
+		return err
+	}
 	u.PushFunc(u.ActualStore.List())
+	return nil
 }
 
 // NewUndeltaStore returns an UndeltaStore implemented with a Store.
-func NewUndeltaStore(pushFunc func([]interface{})) *UndeltaStore {
+func NewUndeltaStore(pushFunc func([]interface{}), keyFunc KeyFunc) *UndeltaStore {
 	return &UndeltaStore{
-		ActualStore: NewStore(),
+		ActualStore: NewStore(keyFunc),
 		PushFunc:    pushFunc,
 	}
 }

pkg/client/cache/undelta_store_test.go

@@ -24,15 +24,28 @@ import (
 // store_test.go checks that UndeltaStore conforms to the Store interface
 // behavior.  This test just tests that it calls the push func in addition.
 
-type t struct{ int }
+type testUndeltaObject struct {
+	name string
+	val  interface{}
+}
 
+func testUndeltaKeyFunc(obj interface{}) (string, error) {
+	return obj.(testUndeltaObject).name, nil
+}
+
+/*
 var (
 	o1 interface{}   = t{1}
 	o2 interface{}   = t{2}
 	l1 []interface{} = []interface{}{t{1}}
 )
+*/
 
 func TestUpdateCallsPush(t *testing.T) {
+	mkObj := func(name string, val interface{}) testUndeltaObject {
+		return testUndeltaObject{name: name, val: val}
+	}
+
 	var got []interface{}
 	var callcount int = 0
 	push := func(m []interface{}) {
@@ -40,19 +53,25 @@ func TestUpdateCallsPush(t *testing.T) {
 		got = m
 	}
 
-	u := NewUndeltaStore(push)
+	u := NewUndeltaStore(push, testUndeltaKeyFunc)
 
-	u.Add("a", o2)
-	u.Update("a", o1)
+	u.Add(mkObj("a", 2))
+	u.Update(mkObj("a", 1))
 	if callcount != 2 {
 		t.Errorf("Expected 2 calls, got %d", callcount)
 	}
-	if !reflect.DeepEqual(l1, got) {
-		t.Errorf("Expected %#v, Got %#v", l1, got)
+
+	l := []interface{}{mkObj("a", 1)}
+	if !reflect.DeepEqual(l, got) {
+		t.Errorf("Expected %#v, Got %#v", l, got)
 	}
 }
 
 func TestDeleteCallsPush(t *testing.T) {
+	mkObj := func(name string, val interface{}) testUndeltaObject {
+		return testUndeltaObject{name: name, val: val}
+	}
+
 	var got []interface{}
 	var callcount int = 0
 	push := func(m []interface{}) {
@@ -60,10 +79,10 @@ func TestDeleteCallsPush(t *testing.T) {
 		got = m
 	}
 
-	u := NewUndeltaStore(push)
+	u := NewUndeltaStore(push, testUndeltaKeyFunc)
 
-	u.Add("a", o2)
-	u.Delete("a")
+	u.Add(mkObj("a", 2))
+	u.Delete(mkObj("a", ""))
 	if callcount != 2 {
 		t.Errorf("Expected 2 calls, got %d", callcount)
 	}
@@ -78,15 +97,18 @@ func TestReadsDoNotCallPush(t *testing.T) {
 		t.Errorf("Unexpected call to push!")
 	}
 
-	u := NewUndeltaStore(push)
+	u := NewUndeltaStore(push, testUndeltaKeyFunc)
 
 	// These should not call push.
 	_ = u.List()
-	_ = u.ContainedIDs()
-	_, _ = u.Get("1")
+	_, _, _ = u.Get(testUndeltaObject{"a", ""})
 }
 
 func TestReplaceCallsPush(t *testing.T) {
+	mkObj := func(name string, val interface{}) testUndeltaObject {
+		return testUndeltaObject{name: name, val: val}
+	}
+
 	var got []interface{}
 	var callcount int = 0
 	push := func(m []interface{}) {
@@ -94,16 +116,15 @@ func TestReplaceCallsPush(t *testing.T) {
 		got = m
 	}
 
-	u := NewUndeltaStore(push)
+	u := NewUndeltaStore(push, testUndeltaKeyFunc)
 
-	m := make(map[string]interface{})
-	m["1"] = o1
+	m := []interface{}{mkObj("a", 1)}
 
 	u.Replace(m)
 	if callcount != 1 {
 		t.Errorf("Expected 1 calls, got %d", callcount)
 	}
-	expected := l1
+	expected := []interface{}{mkObj("a", 1)}
 	if !reflect.DeepEqual(expected, got) {
 		t.Errorf("Expected %#v, Got %#v", expected, got)
 	}

pkg/kubelet/config/apiserver.go

@@ -53,5 +53,5 @@ func newSourceApiserverFromLW(lw cache.ListerWatcher, updates chan<- interface{}
 		}
 		updates <- kubelet.PodUpdate{bpods, kubelet.SET, kubelet.ApiserverSource}
 	}
-	cache.NewReflector(lw, &api.Pod{}, cache.NewUndeltaStore(send)).Run()
+	cache.NewReflector(lw, &api.Pod{}, cache.NewUndeltaStore(send, cache.MetaNamespaceKeyFunc)).Run()
 }

pkg/kubelet/config/apiserver_test.go

@@ -139,6 +139,47 @@ func TestNewSourceApiserver_UpdatesAndMultiplePods(t *testing.T) {
 	}
 }
 
+func TestNewSourceApiserver_TwoNamespacesSameName(t *testing.T) {
+	pod1 := api.Pod{
+		ObjectMeta: api.ObjectMeta{Name: "p", Namespace: "one"},
+		Spec:       api.PodSpec{Containers: []api.Container{{Image: "image/one"}}}}
+	pod2 := api.Pod{
+		ObjectMeta: api.ObjectMeta{Name: "p", Namespace: "two"},
+		Spec:       api.PodSpec{Containers: []api.Container{{Image: "image/blah"}}}}
+
+	// Setup fake api client.
+	fakeWatch := watch.NewFake()
+	lw := fakePodLW{
+		listResp:  &api.PodList{Items: []api.Pod{pod1, pod2}},
+		watchResp: fakeWatch,
+	}
+
+	ch := make(chan interface{})
+
+	newSourceApiserverFromLW(lw, ch)
+
+	got, ok := <-ch
+	if !ok {
+		t.Errorf("Unable to read from channel when expected")
+	}
+	update := got.(kubelet.PodUpdate)
+	// Make sure that we get both pods.  Catches bug #2294.
+	if !(len(update.Pods) == 2) {
+		t.Errorf("Expected %d, Got %d", 2, len(update.Pods))
+	}
+
+	// Delete pod1
+	fakeWatch.Delete(&pod1)
+	got, ok = <-ch
+	if !ok {
+		t.Errorf("Unable to read from channel when expected")
+	}
+	update = got.(kubelet.PodUpdate)
+	if !(len(update.Pods) == 1) {
+		t.Errorf("Expected %d, Got %d", 1, len(update.Pods))
+	}
+}
+
 func TestNewSourceApiserverInitialEmptySendsEmptyPodUpdate(t *testing.T) {
 	// Setup fake api client.
 	fakeWatch := watch.NewFake()

pkg/kubelet/kubelet.go

@@ -93,7 +93,7 @@ func NewMainKubelet(
 		return nil, fmt.Errorf("invalid minimum GC age %d", minimumGCAge)
 	}
 
-	serviceStore := cache.NewStore()
+	serviceStore := cache.NewStore(cache.MetaNamespaceKeyFunc)
 	if kubeClient != nil {
 		cache.NewReflector(&cache.ListWatch{kubeClient, labels.Everything(), "services", api.NamespaceAll}, &api.Service{}, serviceStore).Run()
 	}

plugin/pkg/scheduler/factory/factory.go

@@ -35,9 +35,9 @@ import (
 )
 
 var (
-	PodLister     = &cache.StoreToPodLister{cache.NewStore()}
-	MinionLister  = &cache.StoreToNodeLister{cache.NewStore()}
-	ServiceLister = &cache.StoreToServiceLister{cache.NewStore()}
+	PodLister     = &cache.StoreToPodLister{cache.NewStore(cache.MetaNamespaceKeyFunc)}
+	MinionLister  = &cache.StoreToNodeLister{cache.NewStore(cache.MetaNamespaceKeyFunc)}
+	ServiceLister = &cache.StoreToServiceLister{cache.NewStore(cache.MetaNamespaceKeyFunc)}
 )
 
 // ConfigFactory knows how to fill out a scheduler config with its support functions.
@@ -57,7 +57,7 @@ type ConfigFactory struct {
 func NewConfigFactory(client *client.Client) *ConfigFactory {
 	return &ConfigFactory{
 		Client:        client,
-		PodQueue:      cache.NewFIFO(),
+		PodQueue:      cache.NewFIFO(cache.MetaNamespaceKeyFunc),
 		PodLister:     PodLister,
 		MinionLister:  MinionLister,
 		ServiceLister: ServiceLister,
@@ -241,7 +241,7 @@ func (factory *ConfigFactory) makeDefaultErrorFunc(backoff *podBackoff, podQueue
 				return
 			}
 			if pod.Status.Host == "" {
-				podQueue.Add(pod.Name, pod)
+				podQueue.Add(pod)
 			}
 		}()
 	}
@@ -261,8 +261,8 @@ func (ne *nodeEnumerator) Len() int {
 }
 
 // Get returns the item (and ID) with the particular index.
-func (ne *nodeEnumerator) Get(index int) (string, interface{}) {
-	return ne.Items[index].Name, &ne.Items[index]
+func (ne *nodeEnumerator) Get(index int) interface{} {
+	return &ne.Items[index]
 }
 
 type binder struct {

plugin/pkg/scheduler/factory/factory_test.go

@@ -208,7 +208,7 @@ func TestDefaultErrorFunc(t *testing.T) {
 	server := httptest.NewServer(mux)
 	defer server.Close()
 	factory := NewConfigFactory(client.NewOrDie(&client.Config{Host: server.URL, Version: testapi.Version()}))
-	queue := cache.NewFIFO()
+	queue := cache.NewFIFO(cache.MetaNamespaceKeyFunc)
 	podBackoff := podBackoff{
 		perPodBackoff:   map[string]*backoffEntry{},
 		clock:           &fakeClock{},
@@ -223,7 +223,7 @@ func TestDefaultErrorFunc(t *testing.T) {
 		// whole error handling system in the future. The test will time
 		// out if something doesn't work.
 		time.Sleep(10 * time.Millisecond)
-		got, exists := queue.Get("foo")
+		got, exists, _ := queue.Get(testPod)
 		if !exists {
 			continue
 		}
@@ -249,8 +249,8 @@ func TestMinionEnumerator(t *testing.T) {
 		t.Fatalf("expected %v, got %v", e, a)
 	}
 	for i := range testList.Items {
-		gotID, gotObj := me.Get(i)
-		if e, a := testList.Items[i].Name, gotID; e != a {
+		gotObj := me.Get(i)
+		if e, a := testList.Items[i].Name, gotObj.(*api.Node).Name; e != a {
 			t.Errorf("Expected %v, got %v", e, a)
 		}
 		if e, a := &testList.Items[i], gotObj; !reflect.DeepEqual(e, a) {