Randomize apiserver watch timeouts

pkg/apiserver/watch.go

@@ -17,10 +17,12 @@ limitations under the License.
 package apiserver
 
 import (
+	"math/rand"
 	"net/http"
 	"reflect"
 	"regexp"
 	"strings"
+	"time"
 
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/httplog"
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/runtime"
@@ -32,19 +34,47 @@ import (
 	"golang.org/x/net/websocket"
 )
 
-var connectionUpgradeRegex = regexp.MustCompile("(^|.*,\\s*)upgrade($|\\s*,)")
+var (
+	connectionUpgradeRegex = regexp.MustCompile("(^|.*,\\s*)upgrade($|\\s*,)")
+
+	// nothing will ever be sent down this channel
+	neverExitWatch <-chan time.Time = make(chan time.Time)
+)
 
 func isWebsocketRequest(req *http.Request) bool {
 	return connectionUpgradeRegex.MatchString(strings.ToLower(req.Header.Get("Connection"))) && strings.ToLower(req.Header.Get("Upgrade")) == "websocket"
 }
 
+// timeoutFactory abstracts watch timeout logic for testing
+type timeoutFactory interface {
+	TimeoutCh() (<-chan time.Time, func() bool)
+}
+
+// realTimeoutFactory implements timeoutFactory
+type realTimeoutFactory struct {
+	timeout time.Duration
+}
+
+// TimeoutChan returns a channel which will receive something when the watch times out,
+// and a cleanup function to call when this happens.
+func (w *realTimeoutFactory) TimeoutCh() (<-chan time.Time, func() bool) {
+	if w.timeout == 0 {
+		return neverExitWatch, func() bool { return false }
+	}
+	t := time.NewTimer(w.timeout)
+	return t.C, t.Stop
+}
+
 // serveWatch handles serving requests to the server
 func serveWatch(watcher watch.Interface, scope RequestScope, w http.ResponseWriter, req *restful.Request) {
+	// Each watch gets a random timeout to avoid thundering herds. Rand is seeded once in the api installer.
+	timeout := time.Duration(MinTimeoutSecs+rand.Intn(MaxTimeoutSecs-MinTimeoutSecs)) * time.Second
+
 	watchServer := &WatchServer{watcher, scope.Codec, func(obj runtime.Object) {
 		if err := setSelfLink(obj, req, scope.Namer); err != nil {
 			glog.V(5).Infof("Failed to set self link for object %v: %v", reflect.TypeOf(obj), err)
 		}
-	}}
+	}, &realTimeoutFactory{timeout}}
 	if isWebsocketRequest(req.Request) {
 		websocket.Handler(watchServer.HandleWS).ServeHTTP(httplog.Unlogged(w), req.Request)
 	} else {
@@ -57,6 +87,7 @@ type WatchServer struct {
 	watching watch.Interface
 	codec    runtime.Codec
 	fixup    func(runtime.Object)
+	t        timeoutFactory
 }
 
 // HandleWS implements a websocket handler.
@@ -100,6 +131,9 @@ func (w *WatchServer) HandleWS(ws *websocket.Conn) {
 func (self *WatchServer) ServeHTTP(w http.ResponseWriter, req *http.Request) {
 	loggedW := httplog.LogOf(req, w)
 	w = httplog.Unlogged(w)
+	timeoutCh, cleanup := self.t.TimeoutCh()
+	defer cleanup()
+	defer self.watching.Stop()
 
 	cn, ok := w.(http.CloseNotifier)
 	if !ok {
@@ -113,16 +147,15 @@ func (self *WatchServer) ServeHTTP(w http.ResponseWriter, req *http.Request) {
 		http.NotFound(w, req)
 		return
 	}
-
 	w.Header().Set("Transfer-Encoding", "chunked")
 	w.WriteHeader(http.StatusOK)
 	flusher.Flush()
-
 	encoder := watchjson.NewEncoder(w, self.codec)
 	for {
 		select {
 		case <-cn.CloseNotify():
-			self.watching.Stop()
+			return
+		case <-timeoutCh:
 			return
 		case event, ok := <-self.watching.ResultChan():
 			if !ok {
@@ -132,7 +165,6 @@ func (self *WatchServer) ServeHTTP(w http.ResponseWriter, req *http.Request) {
 			self.fixup(event.Object)
 			if err := encoder.Encode(&event); err != nil {
 				// Client disconnect.
-				self.watching.Stop()
 				return
 			}
 			flusher.Flush()