Merge pull request #92 from codeblooded/bug/serial-sessions

Fix blocking issue with multiple sessions

Author: Benjamin Reed

testctrl/go.sum

@@ -52,6 +52,7 @@ github.com/golang/protobuf v1.3.5 h1:F768QJ1E9tib+q5Sc8MkdJi1RxLTbRcTf8LJV56aRls
 github.com/golang/protobuf v1.3.5/go.mod h1:6O5/vntMXwX2lRkT1hjjk0nAC1IDOTvTlVgjlRvqsdk=
 github.com/google/btree v0.0.0-20180813153112-4030bb1f1f0c/go.mod h1:lNA+9X1NB3Zf8V7Ke586lFgjr2dZNuvo3lPJSGZ5JPQ=
 github.com/google/btree v1.0.0/go.mod h1:lNA+9X1NB3Zf8V7Ke586lFgjr2dZNuvo3lPJSGZ5JPQ=
+github.com/google/glog v0.4.0 h1:WV2GdGOpRcDyRt1i9LHUcpATSfmbxDOHL/I5OtjndLI=
 github.com/google/go-cmp v0.2.0/go.mod h1:oXzfMopK8JAjlY9xF4vHSVASa0yLyX7SntLO5aqRK0M=
 github.com/google/go-cmp v0.3.0 h1:crn/baboCvb5fXaQ0IJ1SGTsTVrWpDsCWC8EGETZijY=
 github.com/google/go-cmp v0.3.0/go.mod h1:8QqcDgzrUqlUb/G2PQTWiueGozuR1884gddMywk6iLU=

testctrl/svc/orch/controller.go

@@ -228,7 +228,10 @@ func (c *Controller) spawnExecutor(session *types.Session) {
 	c.incExecutors()
 
 	go func() {
-		defer c.decExecutors()
+		defer func() {
+			c.decExecutors()
+			c.waitQueue.Done(session)
+		}()
 
 		if err := executor.Execute(session); err != nil {
 			glog.Infof("%v", err)

testctrl/svc/orch/watcher.go

@@ -71,7 +71,7 @@ func (w *Watcher) Subscribe(sessionName string) (<-chan *PodWatchEvent, error) {
 		return nil, fmt.Errorf("session %v already has a follower", sessionName)
 	}
 
-	eventChan := make(chan *PodWatchEvent)
+	eventChan := make(chan *PodWatchEvent, eventBufferSize)
 	w.eventChans[sessionName] = eventChan
 	return eventChan, nil
 }
@@ -133,7 +133,16 @@ func (w *Watcher) publish(event *PodWatchEvent) {
 	defer w.mux.Unlock()
 
 	eventChan := w.eventChans[event.SessionName]
-	eventChan <- event
+	if eventChan == nil {
+		glog.Warningf("watcher: received event for session without subscriber: %v", event)
+		return
+	}
+
+	if len(eventChan) < cap(eventChan) {
+		eventChan <- event
+	} else {
+		glog.Warningf("watcher: too many events unread in subscriber channel, dropping: %v", event)
+	}
 }
 
 func (w *Watcher) diagnose(pod *corev1.Pod) (Health, error) {
@@ -192,3 +201,5 @@ type PodWatchEvent struct {
 	// Error may provide the error details that led to the failing health.
 	Error error
 }
+
+const eventBufferSize = 32