Merge pull request #14881 from lavalamp/fix-14617

Auto commit by PR queue bot

pkg/controller/framework/controller_test.go

@@ -311,30 +311,6 @@ func TestUpdate(t *testing.T) {
 		pair{FROM, FROM}: true,
 	}
 
-	var testDoneWG sync.WaitGroup
-
-	// Make a controller that deletes things once it observes an update.
-	// It calls Done() on the wait group on deletions so we can tell when
-	// everything we've added has been deleted.
-	_, controller := framework.NewInformer(
-		source,
-		&api.Pod{},
-		time.Millisecond*1,
-		framework.ResourceEventHandlerFuncs{
-			UpdateFunc: func(oldObj, newObj interface{}) {
-				o, n := oldObj.(*api.Pod), newObj.(*api.Pod)
-				from, to := o.Labels["check"], n.Labels["check"]
-				if !allowedTransitions[pair{from, to}] {
-					t.Errorf("observed transition %q -> %q for %v", from, to, n.Name)
-				}
-				source.Delete(n)
-			},
-			DeleteFunc: func(obj interface{}) {
-				testDoneWG.Done()
-			},
-		},
-	)
-
 	pod := func(name, check string) *api.Pod {
 		return &api.Pod{
 			ObjectMeta: api.ObjectMeta{
@@ -368,8 +344,32 @@ func TestUpdate(t *testing.T) {
 	}
 
 	const threads = 3
+
+	var testDoneWG sync.WaitGroup
 	testDoneWG.Add(threads * len(tests))
 
+	// Make a controller that deletes things once it observes an update.
+	// It calls Done() on the wait group on deletions so we can tell when
+	// everything we've added has been deleted.
+	_, controller := framework.NewInformer(
+		source,
+		&api.Pod{},
+		time.Millisecond*1,
+		framework.ResourceEventHandlerFuncs{
+			UpdateFunc: func(oldObj, newObj interface{}) {
+				o, n := oldObj.(*api.Pod), newObj.(*api.Pod)
+				from, to := o.Labels["check"], n.Labels["check"]
+				if !allowedTransitions[pair{from, to}] {
+					t.Errorf("observed transition %q -> %q for %v", from, to, n.Name)
+				}
+				source.Delete(n)
+			},
+			DeleteFunc: func(obj interface{}) {
+				testDoneWG.Done()
+			},
+		},
+	)
+
 	// Run the controller and run it until we close stop.
 	// Once Run() is called, calls to testDoneWG.Done() might start, so
 	// all testDoneWG.Add() calls must happen before this point

pkg/controller/framework/fake_controller_source.go

@@ -103,7 +103,7 @@ func (f *FakeControllerSource) Change(e watch.Event, watchProbability float64) {
 		panic(err) // this is test code only
 	}
 
-	resourceVersion := len(f.changes)
+	resourceVersion := len(f.changes) + 1
 	objMeta.ResourceVersion = strconv.Itoa(resourceVersion)
 	f.changes = append(f.changes, e)
 	key := f.key(objMeta)
@@ -127,7 +127,7 @@ func (f *FakeControllerSource) List() (runtime.Object, error) {
 	for _, obj := range f.items {
 		// Must make a copy to allow clients to modify the object.
 		// Otherwise, if they make a change and write it back, they
-		// will inadvertently change the our canonical copy (in
+		// will inadvertently change our canonical copy (in
 		// addition to racing with other clients).
 		objCopy, err := api.Scheme.DeepCopy(obj)
 		if err != nil {
@@ -157,7 +157,6 @@ func (f *FakeControllerSource) Watch(resourceVersion string) (watch.Interface, e
 	if err != nil {
 		return nil, err
 	}
-	rc++ // Don't re-send them a change they already have.
 	if rc < len(f.changes) {
 		changes := []watch.Event{}
 		for _, c := range f.changes[rc:] {
@@ -178,3 +177,11 @@ func (f *FakeControllerSource) Watch(resourceVersion string) (watch.Interface, e
 	}
 	return f.broadcaster.Watch(), nil
 }
+
+// Shutdown closes the underlying broadcaster, waiting for events to be
+// delivered. It's an error to call any method after calling shutdown. This is
+// enforced by Shutdown() leaving f locked.
+func (f *FakeControllerSource) Shutdown() {
+	f.lock.Lock() // Purposely no unlock.
+	f.broadcaster.Shutdown()
+}

pkg/controller/framework/fake_controller_source_test.go

@@ -17,11 +17,36 @@ limitations under the License.
 package framework
 
 import (
+	"sync"
 	"testing"
 
 	"k8s.io/kubernetes/pkg/api"
+	"k8s.io/kubernetes/pkg/watch"
 )
 
+// ensure the watch delivers the requested and only the requested items.
+func consume(t *testing.T, w watch.Interface, rvs []string, done *sync.WaitGroup) {
+	defer done.Done()
+	for _, rv := range rvs {
+		got, ok := <-w.ResultChan()
+		if !ok {
+			t.Errorf("%#v: unexpected channel close, wanted %v", rvs, rv)
+			return
+		}
+		gotRV := got.Object.(*api.Pod).ObjectMeta.ResourceVersion
+		if e, a := rv, gotRV; e != a {
+			t.Errorf("wanted %v, got %v", e, a)
+		} else {
+			t.Logf("Got %v as expected", gotRV)
+		}
+	}
+	// We should not get anything else.
+	got, open := <-w.ResultChan()
+	if open {
+		t.Errorf("%#v: unwanted object %#v", rvs, got)
+	}
+}
+
 func TestRCNumber(t *testing.T) {
 	pod := func(name string) *api.Pod {
 		return &api.Pod{
@@ -31,6 +56,9 @@ func TestRCNumber(t *testing.T) {
 		}
 	}
 
+	wg := &sync.WaitGroup{}
+	wg.Add(3)
+
 	source := NewFakeControllerSource()
 	source.Add(pod("foo"))
 	source.Modify(pod("foo"))
@@ -40,9 +68,27 @@ func TestRCNumber(t *testing.T) {
 	if err != nil {
 		t.Fatalf("Unexpected error: %v", err)
 	}
-	defer w.Stop()
-	got := <-w.ResultChan()
-	if e, a := "2", got.Object.(*api.Pod).ObjectMeta.ResourceVersion; e != a {
+	go consume(t, w, []string{"2", "3"}, wg)
+
+	list, err := source.List()
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+	if e, a := "3", list.(*api.List).ResourceVersion; e != a {
 		t.Errorf("wanted %v, got %v", e, a)
 	}
+
+	w2, err := source.Watch("2")
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+	go consume(t, w2, []string{"3"}, wg)
+
+	w3, err := source.Watch("3")
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+	go consume(t, w3, []string{}, wg)
+	source.Shutdown()
+	wg.Wait()
 }