Merge pull request #14064 from brendandburns/flakes

Increase a bunch of timeouts to reduce flakes

pkg/api/rest/resttest/resttest.go

@@ -32,6 +32,7 @@ import (
 	"k8s.io/kubernetes/pkg/labels"
 	"k8s.io/kubernetes/pkg/runtime"
 	"k8s.io/kubernetes/pkg/tools"
+	"k8s.io/kubernetes/pkg/util"
 )
 
 type Tester struct {
@@ -961,7 +962,7 @@ func (t *Tester) testWatchFields(obj runtime.Object, initWatchFn InitWatchFunc,
 				if !ok {
 					t.Errorf("watch channel should be open")
 				}
-			case <-time.After(time.Millisecond * 100):
+			case <-time.After(util.ForeverTestTimeout):
 				t.Errorf("unexpected timeout from result channel")
 			}
 			watcher.Stop()
@@ -982,7 +983,7 @@ func (t *Tester) testWatchFields(obj runtime.Object, initWatchFn InitWatchFunc,
 			select {
 			case <-watcher.ResultChan():
 				t.Errorf("unexpected result from result channel")
-			case <-time.After(time.Millisecond * 100):
+			case <-time.After(time.Millisecond * 500):
 				// expected case
 			}
 			watcher.Stop()
@@ -1009,7 +1010,7 @@ func (t *Tester) testWatchLabels(obj runtime.Object, initWatchFn InitWatchFunc,
 				if !ok {
 					t.Errorf("watch channel should be open")
 				}
-			case <-time.After(time.Millisecond * 100):
+			case <-time.After(util.ForeverTestTimeout):
 				t.Errorf("unexpected timeout from result channel")
 			}
 			watcher.Stop()
@@ -1030,7 +1031,7 @@ func (t *Tester) testWatchLabels(obj runtime.Object, initWatchFn InitWatchFunc,
 			select {
 			case <-watcher.ResultChan():
 				t.Errorf("unexpected result from result channel")
-			case <-time.After(time.Millisecond * 100):
+			case <-time.After(time.Millisecond * 500):
 				// expected case
 			}
 			watcher.Stop()

pkg/apiserver/watch_test.go

@@ -33,6 +33,7 @@ import (
 	"k8s.io/kubernetes/pkg/fields"
 	"k8s.io/kubernetes/pkg/labels"
 	"k8s.io/kubernetes/pkg/runtime"
+	"k8s.io/kubernetes/pkg/util"
 	"k8s.io/kubernetes/pkg/watch"
 )
 
@@ -379,8 +380,8 @@ func TestWatchHTTPTimeout(t *testing.T) {
 		if !watcher.Stopped {
 			t.Errorf("Leaked watch on timeout")
 		}
-	case <-time.After(100 * time.Millisecond):
+	case <-time.After(util.ForeverTestTimeout):
-		t.Errorf("Failed to stop watcher after 100ms of timeout signal")
+		t.Errorf("Failed to stop watcher after %s of timeout signal", util.ForeverTestTimeout.String())
 	}
 
 	// Make sure we can't receive any more events through the timeout watch

pkg/client/cache/expiration_cache_test.go

@@ -55,7 +55,7 @@ func TestTTLExpirationBasic(t *testing.T) {
 		if delKey != key {
 			t.Errorf("Unexpected delete for key %s", key)
 		}
-	case <-time.After(time.Millisecond * 100):
+	case <-time.After(util.ForeverTestTimeout):
 		t.Errorf("Unexpected timeout waiting on delete")
 	}
 	close(deleteChan)
@@ -100,7 +100,7 @@ func TestTTLList(t *testing.T) {
 				t.Errorf("Unexpected delete for key %s", delKey)
 			}
 			expireKeys.Delete(delKey)
-		case <-time.After(time.Millisecond * 100):
+		case <-time.After(util.ForeverTestTimeout):
 			t.Errorf("Unexpected timeout waiting on delete")
 			return
 		}

pkg/client/cache/reflector_test.go

@@ -59,8 +59,8 @@ func TestCloseWatchChannelOnError(t *testing.T) {
 		if ok {
 			t.Errorf("Watch channel left open after cancellation")
 		}
-	case <-time.After(100 * time.Millisecond):
+	case <-time.After(util.ForeverTestTimeout):
-		t.Errorf("the cancellation is at least 99 milliseconds late")
+		t.Errorf("the cancellation is at least %s late", util.ForeverTestTimeout.String())
 		break
 	}
 }
@@ -88,8 +88,8 @@ func TestRunUntil(t *testing.T) {
 		if ok {
 			t.Errorf("Watch channel left open after stopping the watch")
 		}
-	case <-time.After(100 * time.Millisecond):
+	case <-time.After(util.ForeverTestTimeout):
-		t.Errorf("the cancellation is at least 99 milliseconds late")
+		t.Errorf("the cancellation is at least %s late", util.ForeverTestTimeout.String())
 		break
 	}
 }
@@ -98,7 +98,7 @@ func TestReflector_resyncChan(t *testing.T) {
 	s := NewStore(MetaNamespaceKeyFunc)
 	g := NewReflector(&testLW{}, &api.Pod{}, s, time.Millisecond)
 	a, _ := g.resyncChan()
-	b := time.After(100 * time.Millisecond)
+	b := time.After(util.ForeverTestTimeout)
 	select {
 	case <-a:
 		t.Logf("got timeout as expected")

pkg/controller/replication/replication_controller_test.go

@@ -45,12 +45,6 @@ type FakePodControl struct {
 	err            error
 }
 
-// Give each test that starts a background controller up to 1/2 a second.
-// Since we need to start up a goroutine to test watch, this routine needs
-// to get cpu before the test can complete. If the test is starved of cpu,
-// the watch test will take up to 1/2 a second before timing out.
-const controllerTimeout = 500 * time.Millisecond
-
 var alwaysReady = func() bool { return true }
 
 func init() {
@@ -236,7 +230,7 @@ func TestDeleteFinalStateUnknown(t *testing.T) {
 		if key != expected {
 			t.Errorf("Unexpected sync all for rc %v, expected %v", key, expected)
 		}
-	case <-time.After(100 * time.Millisecond):
+	case <-time.After(util.ForeverTestTimeout):
 		t.Errorf("Processing DeleteFinalStateUnknown took longer than expected")
 	}
 }
@@ -495,7 +489,7 @@ func TestWatchControllers(t *testing.T) {
 
 	select {
 	case <-received:
-	case <-time.After(controllerTimeout):
+	case <-time.After(util.ForeverTestTimeout):
 		t.Errorf("Expected 1 call but got 0")
 	}
 }
@@ -540,7 +534,7 @@ func TestWatchPods(t *testing.T) {
 
 	select {
 	case <-received:
-	case <-time.After(controllerTimeout):
+	case <-time.After(util.ForeverTestTimeout):
 		t.Errorf("Expected 1 call but got 0")
 	}
 }
@@ -592,7 +586,7 @@ func TestUpdatePods(t *testing.T) {
 			if !expected.Has(got) {
 				t.Errorf("Expected keys %#v got %v", expected, got)
 			}
-		case <-time.After(controllerTimeout):
+		case <-time.After(util.ForeverTestTimeout):
 			t.Errorf("Expected update notifications for controllers within 100ms each")
 		}
 	}
@@ -632,7 +626,7 @@ func TestControllerUpdateRequeue(t *testing.T) {
 		if key != expectedKey {
 			t.Errorf("Expected requeue of controller with key %s got %s", expectedKey, key)
 		}
-	case <-time.After(controllerTimeout):
+	case <-time.After(util.ForeverTestTimeout):
 		manager.queue.ShutDown()
 		t.Errorf("Expected to find an rc in the queue, found none.")
 	}

pkg/kubelet/config/file_test.go

@@ -29,6 +29,7 @@ import (
 	"k8s.io/kubernetes/pkg/kubelet"
 	"k8s.io/kubernetes/pkg/runtime"
 	"k8s.io/kubernetes/pkg/securitycontext"
+	"k8s.io/kubernetes/pkg/util"
 )
 
 func TestExtractFromNonExistentFile(t *testing.T) {
@@ -51,7 +52,7 @@ func TestUpdateOnNonExistentFile(t *testing.T) {
 			t.Fatalf("Expected %#v, Got %#v", expected, update)
 		}
 
-	case <-time.After(time.Second):
+	case <-time.After(util.ForeverTestTimeout):
 		t.Errorf("Expected update, timeout instead")
 	}
 }
@@ -143,7 +144,7 @@ func TestReadPodsFromFile(t *testing.T) {
 				if !api.Semantic.DeepEqual(testCase.expected, update) {
 					t.Errorf("%s: Expected %#v, Got %#v", testCase.desc, testCase.expected, update)
 				}
-			case <-time.After(time.Second):
+			case <-time.After(util.ForeverTestTimeout):
 				t.Errorf("%s: Expected update, timeout instead", testCase.desc)
 			}
 		}()

pkg/kubelet/kubelet_test.go

@@ -3108,7 +3108,7 @@ func TestRegisterExistingNodeWithApiserver(t *testing.T) {
 		done <- struct{}{}
 	}()
 	select {
-	case <-time.After(5 * time.Second):
+	case <-time.After(util.ForeverTestTimeout):
 		t.Errorf("timed out waiting for registration")
 	case <-done:
 		return

pkg/kubelet/pod_workers_test.go

@@ -31,6 +31,7 @@ import (
 	"k8s.io/kubernetes/pkg/kubelet/dockertools"
 	"k8s.io/kubernetes/pkg/kubelet/network"
 	"k8s.io/kubernetes/pkg/types"
+	"k8s.io/kubernetes/pkg/util"
 )
 
 func newPod(uid, name string) *api.Pod {
@@ -146,7 +147,7 @@ func TestUpdateType(t *testing.T) {
 				if gotType != expectedTypes[i] {
 					t.Fatalf("Expected sync type %v got %v for pod with uid %v", expectedTypes[i], gotType, p.UID)
 				}
-			case <-time.After(100 * time.Millisecond):
+			case <-time.After(util.ForeverTestTimeout):
 				t.Errorf("Unexpected delay is running pod worker")
 			}
 		}

pkg/storage/etcd/etcd_watcher_test.go

@@ -29,6 +29,7 @@ import (
 	"k8s.io/kubernetes/pkg/storage"
 	"k8s.io/kubernetes/pkg/tools"
 	"k8s.io/kubernetes/pkg/tools/etcdtest"
+	"k8s.io/kubernetes/pkg/util"
 	"k8s.io/kubernetes/pkg/watch"
 )
 
@@ -700,7 +701,7 @@ func TestWatchFromOtherError(t *testing.T) {
 		if ok {
 			t.Fatalf("expected result channel to be closed")
 		}
-	case <-time.After(1 * time.Second):
+	case <-time.After(util.ForeverTestTimeout):
 		t.Fatalf("watch should have closed channel: %#v", watching)
 	}
 

pkg/util/atomic_value_test.go

@@ -38,7 +38,7 @@ func ExpectValue(t *testing.T, atomicValue *AtomicValue, expectedValue interface
 			t.Errorf("Expected to find %v, found %v", expectedValue, actualValue)
 			return
 		}
-	case <-time.After(time.Second * 5):
+	case <-time.After(ForeverTestTimeout):
 		t.Error("Value could not be read")
 		return
 	}

pkg/util/util.go

@@ -41,6 +41,15 @@ import (
 // For testing, bypass HandleCrash.
 var ReallyCrash bool
 
+// For any test of the style:
+//   ...
+//   <- time.After(timeout):
+//      t.Errorf("Timed out")
+// The value for timeout should effectively be "forever." Obviously we don't want our tests to truly lock up forever, but 30s
+// is long enough that it is effectively forever for the things that can slow down a run on a heavily contended machine
+// (GC, seeks, etc), but not so long as to make a developer ctrl-c a test run if they do happen to break that test.
+var ForeverTestTimeout = time.Second * 30
+
 // PanicHandlers is a list of functions which will be invoked when a panic happens.
 var PanicHandlers = []func(interface{}){logPanic}
 

pkg/util/wait/wait_test.go

@@ -20,6 +20,8 @@ import (
 	"errors"
 	"testing"
 	"time"
+
+	"k8s.io/kubernetes/pkg/util"
 )
 
 func TestPoller(t *testing.T) {
@@ -34,7 +36,7 @@ DRAIN:
 				break DRAIN
 			}
 			count++
-		case <-time.After(time.Second):
+		case <-time.After(util.ForeverTestTimeout):
 			t.Errorf("unexpected timeout after poll")
 		}
 	}
@@ -109,7 +111,7 @@ func TestPollForever(t *testing.T) {
 			if !open {
 				t.Fatalf("did not expect channel to be closed")
 			}
-		case <-time.After(time.Second):
+		case <-time.After(util.ForeverTestTimeout):
 			t.Fatalf("channel did not return at least once within the poll interval")
 		}
 	}

pkg/watch/json/decoder_test.go

@@ -25,6 +25,7 @@ import (
 	"k8s.io/kubernetes/pkg/api"
 	"k8s.io/kubernetes/pkg/api/testapi"
 	"k8s.io/kubernetes/pkg/runtime"
+	"k8s.io/kubernetes/pkg/util"
 	"k8s.io/kubernetes/pkg/watch"
 )
 
@@ -98,7 +99,7 @@ func TestDecoder_SourceClose(t *testing.T) {
 	select {
 	case <-done:
 		break
-	case <-time.After(10 * time.Second):
+	case <-time.After(util.ForeverTestTimeout):
 		t.Error("Timeout")
 	}
 }

pkg/watch/mux_test.go

@@ -21,6 +21,8 @@ import (
 	"sync"
 	"testing"
 	"time"
+
+	"k8s.io/kubernetes/pkg/util"
 )
 
 type myType struct {
@@ -110,7 +112,7 @@ func TestBroadcasterWatcherStopDeadlock(t *testing.T) {
 	}(m.Watch(), m.Watch())
 	m.Action(Added, &myType{})
 	select {
-	case <-time.After(5 * time.Second):
+	case <-time.After(util.ForeverTestTimeout):
 		t.Error("timeout: deadlocked")
 	case <-done:
 	}

test/integration/client_test.go

@@ -32,6 +32,7 @@ import (
 	client "k8s.io/kubernetes/pkg/client/unversioned"
 	"k8s.io/kubernetes/pkg/fields"
 	"k8s.io/kubernetes/pkg/labels"
+	"k8s.io/kubernetes/pkg/util"
 	"k8s.io/kubernetes/pkg/version"
 	"k8s.io/kubernetes/pkg/watch"
 	"k8s.io/kubernetes/test/integration/framework"
@@ -160,8 +161,8 @@ func TestSingleWatch(t *testing.T) {
 	defer w.Stop()
 
 	select {
-	case <-time.After(5 * time.Second):
+	case <-time.After(util.ForeverTestTimeout):
-		t.Fatal("watch took longer than 15 seconds")
+		t.Fatalf("watch took longer than %s", util.ForeverTestTimeout.String())
 	case got, ok := <-w.ResultChan():
 		if !ok {
 			t.Fatal("Watch channel closed unexpectedly.")