Stabilize volume unit tests by waiting for exact state

Wait for specific final state instead of waiting for specific number of
operations in volume unit tests. The tests are more readable and will survive
random goroutine ordering (PV and PVC controller have both their own
goroutine).

pkg/controller/volume/persistentvolume/controller_test.go

@@ -33,7 +33,6 @@ import (
 // can't reliably simulate periodic sync of volumes/claims - it would be
 // either very timing-sensitive or slow to wait for real periodic sync.
 func TestControllerSync(t *testing.T) {
-	expectedChanges := []int{4, 1, 1, 2, 1, 1, 1}
 	tests := []controllerTest{
 		// [Unit test set 5] - controller tests.
 		// We test the controller as if
@@ -157,7 +156,7 @@ func TestControllerSync(t *testing.T) {
 		},
 	}
 
-	for ix, test := range tests {
+	for _, test := range tests {
 		glog.V(4).Infof("starting test %q", test.name)
 
 		// Initialize the controller
@@ -176,7 +175,6 @@ func TestControllerSync(t *testing.T) {
 		}
 
 		// Start the controller
-		count := reactor.getChangeCount()
 		go ctrl.Run()
 
 		// Wait for the controller to pass initial sync and fill its caches.
@@ -199,13 +197,10 @@ func TestControllerSync(t *testing.T) {
 		ctrl.claims.Resync()
 		ctrl.volumes.store.Resync()
 
-		// Wait at least once, just in case expectedChanges[ix] == 0
+		err = reactor.waitTest(test)
-		reactor.waitTest()
+		if err != nil {
-		// Wait for expected number of operations.
+			t.Errorf("Failed to run test %s: %v", test.name, err)
-		for reactor.getChangeCount() < count+expectedChanges[ix] {
-			reactor.waitTest()
 		}
-
 		ctrl.Stop()
 
 		evaluateTestResults(ctrl, reactor, test, t)

pkg/controller/volume/persistentvolume/framework_test.go

@@ -43,6 +43,7 @@ import (
 	"k8s.io/kubernetes/pkg/runtime"
 	"k8s.io/kubernetes/pkg/types"
 	"k8s.io/kubernetes/pkg/util/diff"
+	"k8s.io/kubernetes/pkg/util/wait"
 	vol "k8s.io/kubernetes/pkg/volume"
 )
 
@@ -297,7 +298,7 @@ func (r *volumeReactor) injectReactError(action core.Action) error {
 
 // checkVolumes compares all expectedVolumes with set of volumes at the end of
 // the test and reports differences.
-func (r *volumeReactor) checkVolumes(t *testing.T, expectedVolumes []*api.PersistentVolume) error {
+func (r *volumeReactor) checkVolumes(expectedVolumes []*api.PersistentVolume) error {
 	r.lock.Lock()
 	defer r.lock.Unlock()
 
@@ -329,7 +330,7 @@ func (r *volumeReactor) checkVolumes(t *testing.T, expectedVolumes []*api.Persis
 
 // checkClaims compares all expectedClaims with set of claims at the end of the
 // test and reports differences.
-func (r *volumeReactor) checkClaims(t *testing.T, expectedClaims []*api.PersistentVolumeClaim) error {
+func (r *volumeReactor) checkClaims(expectedClaims []*api.PersistentVolumeClaim) error {
 	r.lock.Lock()
 	defer r.lock.Unlock()
 
@@ -454,9 +455,9 @@ func (r *volumeReactor) getChangeCount() int {
 	return r.changedSinceLastSync
 }
 
-// waitTest waits until all tests, controllers and other goroutines do their
+// waitForIdle waits until all tests, controllers and other goroutines do their
 // job and no new actions are registered for 10 milliseconds.
-func (r *volumeReactor) waitTest() {
+func (r *volumeReactor) waitForIdle() {
 	r.ctrl.runningOperations.Wait()
 	// Check every 10ms if the controller does something and stop if it's
 	// idle.
@@ -472,6 +473,32 @@ func (r *volumeReactor) waitTest() {
 	}
 }
 
+// waitTest waits until all tests, controllers and other goroutines do their
+// job and list of current volumes/claims is equal to list of expected
+// volumes/claims (with ~10 second timeout).
+func (r *volumeReactor) waitTest(test controllerTest) error {
+	// start with 10 ms, multiply by 2 each step, 10 steps = 10.23 seconds
+	backoff := wait.Backoff{
+		Duration: 10 * time.Millisecond,
+		Jitter:   0,
+		Factor:   2,
+		Steps:    10,
+	}
+	err := wait.ExponentialBackoff(backoff, func() (done bool, err error) {
+		// Finish all operations that are in progress
+		r.ctrl.runningOperations.Wait()
+
+		// Return 'true' if the reactor reached the expected state
+		err1 := r.checkClaims(test.expectedClaims)
+		err2 := r.checkVolumes(test.expectedVolumes)
+		if err1 == nil && err2 == nil {
+			return true, nil
+		}
+		return false, nil
+	})
+	return err
+}
+
 // deleteVolumeEvent simulates that a volume has been deleted in etcd and
 // the controller receives 'volume deleted' event.
 func (r *volumeReactor) deleteVolumeEvent(volume *api.PersistentVolume) {
@@ -806,11 +833,11 @@ func wrapTestWithInjectedOperation(toWrap testCall, injectBeforeOperation func(c
 
 func evaluateTestResults(ctrl *PersistentVolumeController, reactor *volumeReactor, test controllerTest, t *testing.T) {
 	// Evaluate results
-	if err := reactor.checkClaims(t, test.expectedClaims); err != nil {
+	if err := reactor.checkClaims(test.expectedClaims); err != nil {
 		t.Errorf("Test %q: %v", test.name, err)
 
 	}
-	if err := reactor.checkVolumes(t, test.expectedVolumes); err != nil {
+	if err := reactor.checkVolumes(test.expectedVolumes); err != nil {
 		t.Errorf("Test %q: %v", test.name, err)
 	}
 
@@ -848,8 +875,11 @@ func runSyncTests(t *testing.T, tests []controllerTest) {
 			t.Errorf("Test %q failed: %v", test.name, err)
 		}
 
-		// Wait for all goroutines to finish
+		// Wait for the target state
-		reactor.waitTest()
+		err = reactor.waitTest(test)
+		if err != nil {
+			t.Errorf("Test %q failed: %v", test.name, err)
+		}
 
 		evaluateTestResults(ctrl, reactor, test, t)
 	}
@@ -906,7 +936,7 @@ func runMultisyncTests(t *testing.T, tests []controllerTest) {
 			}
 
 			// Wait for all goroutines to finish
-			reactor.waitTest()
+			reactor.waitForIdle()
 
 			obj := reactor.popChange()
 			if obj == nil {