Add options struct for Controller and Watcher

This change adds an options struct for the Controller and Watcher
types, allowing them to be configured. Previously, they used magic
constants.

Author: Ben Reed

testctrl/cmd/orchtest/main.go

@@ -49,7 +49,7 @@ func main() {
 		glog.Fatalf("Invalid config file specified by the KUBE_CONFIG_FILE env variable, unable to connect: %v", err)
 	}
 
-	c, _ := orch.NewController(clientset, nil)
+	c, _ := orch.NewController(clientset, nil, nil)
 	if err := c.Start(); err != nil {
 		panic(err)
 	}

testctrl/cmd/svc/main.go

@@ -82,7 +82,8 @@ func main() {
 
 	storageServer := store.NewStorageServer()
 
-	controller, err := orch.NewController(clientset, storageServer)
+	controllerOpts := &orch.ControllerOptions{}
+	controller, err := orch.NewController(clientset, storageServer, controllerOpts)
 	if err != nil {
 		glog.Fatalf("could not create a controller: %v", err)
 	}

testctrl/svc/orch/controller.go

@@ -28,9 +28,6 @@ import (
 	"github.com/grpc/grpc/testctrl/svc/types"
 )
 
-// executorCount specifies the maximum number of sessions that should be processed concurrently.
-const executorCount = 1
-
 // Controller serves as the coordinator for orchestrating sessions. It manages active and idle
 // sessions, as well as, interactions with Kubernetes through a set of a internal types.
 type Controller struct {
@@ -40,30 +37,72 @@ type Controller struct {
 	nl              NodeLister
 	watcher         *Watcher
 	waitQueue       *queue
+	executorCount   int
 	activeCount     int
 	running         bool
 	wg              sync.WaitGroup
 	mux             sync.Mutex
 	newExecutorFunc func() Executor
 }
 
-// NewController creates a controller using a Kubernetes clientset and a store. The clientset allows
-// the controller to interact with Kubernetes. The store is used to report significant orchestration
-// events, so progress can be reported. A nil clientset will result in an error.
-func NewController(clientset kubernetes.Interface, store store.Store) (*Controller, error) {
+// ControllerOptions overrides the defaults of the controller, allowing it to be
+// configured as needed.
+type ControllerOptions struct {
+	// ExecutorCount specifies the maximum number of sessions that should be
+	// processed at a time. It defaults to 1, disabling concurrent sessions.
+	ExecutorCount int
+
+	// WatchEventBufferSize specifies the size of the buffered channel for
+	// each channel the watcher creates. It allows the watcher to write
+	// additional kubernetes events without blocking for reads. It defaults
+	// to 32 events.
+	WatchEventBufferSize int
+}
+
+// NewController creates a controller using a Kubernetes clientset, store and an
+// optional ControllerOptions instance.
+//
+// The clientset allows the controller to interact with Kubernetes. If nil, an
+// error will be returned instead of a controller.
+//
+// The store is used to report orchestration events, so progress can be reported
+// to a user.
+//
+// The options value allows the controller to be customized. Specifying nil will
+// configure the controller to sane defaults described in the ControllerOptions
+// documentation.
+func NewController(clientset kubernetes.Interface, store store.Store, options *ControllerOptions) (*Controller, error) {
 	if clientset == nil {
 		return nil, errors.New("cannot create controller from nil kubernetes clientset")
 	}
 
+	opts := options
+	if opts == nil {
+		opts = &ControllerOptions{}
+	}
+
+	executorCount := opts.ExecutorCount
+	if executorCount == 0 {
+		executorCount = 1
+	}
+
+	watcherOpts := &WatcherOptions{
+		EventBufferSize: opts.WatchEventBufferSize,
+	}
+	if watcherOpts.EventBufferSize == 0 {
+		watcherOpts.EventBufferSize = 32
+	}
+
 	coreV1Interface := clientset.CoreV1()
 	podInterface := coreV1Interface.Pods(corev1.NamespaceDefault)
 
 	c := &Controller{
-		pcd:     podInterface,
-		pw:      podInterface,
-		nl:      coreV1Interface.Nodes(),
-		watcher: NewWatcher(podInterface),
-		store:   store,
+		pcd:           podInterface,
+		pw:            podInterface,
+		nl:            coreV1Interface.Nodes(),
+		watcher:       NewWatcher(podInterface, watcherOpts),
+		store:         store,
+		executorCount: executorCount,
 	}
 
 	c.newExecutorFunc = func() Executor {
@@ -197,7 +236,7 @@ func (c *Controller) next() (session *types.Session, quit bool) {
 		return nil, true
 	}
 
-	if c.activeCount > executorCount {
+	if c.activeCount > c.executorCount {
 		return nil, false
 	}
 

testctrl/svc/orch/controller_test.go

@@ -11,7 +11,7 @@ import (
 
 func TestNewController(t *testing.T) {
 	t.Run("nil clientset returns error", func(t *testing.T) {
-		controller, err := NewController(nil, nil)
+		controller, err := NewController(nil, nil, nil)
 		if err == nil {
 			t.Errorf("no error returned for nil clientset")
 		}
@@ -50,7 +50,7 @@ func TestControllerSchedule(t *testing.T) {
 
 	for _, tc := range cases {
 		t.Run(tc.description, func(t *testing.T) {
-			controller, _ := NewController(fake.NewSimpleClientset(), nil)
+			controller, _ := NewController(fake.NewSimpleClientset(), nil, nil)
 			executor := &executorMock{}
 			controller.newExecutorFunc = func() Executor {
 				return executor
@@ -84,7 +84,7 @@ func TestControllerSchedule(t *testing.T) {
 
 func TestControllerStart(t *testing.T) {
 	t.Run("sets running state", func(t *testing.T) {
-		controller, _ := NewController(fake.NewSimpleClientset(), nil)
+		controller, _ := NewController(fake.NewSimpleClientset(), nil, nil)
 		controller.Start()
 		defer controller.Stop(0)
 		if controller.Stopped() {
@@ -112,7 +112,7 @@ func TestControllerStart(t *testing.T) {
 
 	for _, tc := range cases {
 		t.Run(tc.description, func(t *testing.T) {
-			controller, _ := NewController(fake.NewSimpleClientset(), nil)
+			controller, _ := NewController(fake.NewSimpleClientset(), nil, nil)
 			controller.waitQueue = newQueue(limitlessTracker{})
 
 			if tc.mockNL != nil {
@@ -121,7 +121,7 @@ func TestControllerStart(t *testing.T) {
 
 			if tc.mockPW != nil {
 				controller.pw = tc.mockPW
-				controller.watcher = NewWatcher(tc.mockPW)
+				controller.watcher = NewWatcher(tc.mockPW, nil)
 			}
 
 			err := controller.Start()
@@ -170,7 +170,7 @@ func TestControllerStop(t *testing.T) {
 
 	for _, tc := range cases {
 		t.Run(tc.description, func(t *testing.T) {
-			controller, _ := NewController(fake.NewSimpleClientset(), nil)
+			controller, _ := NewController(fake.NewSimpleClientset(), nil, nil)
 			controller.running = true
 			controller.waitQueue = newQueue(limitlessTracker{})
 

testctrl/svc/orch/watcher.go

@@ -21,19 +21,40 @@ import (
 //
 // Create watcher instances with the NewWatcher constructor, not a literal.
 type Watcher struct {
-	eventChans map[string]chan *PodWatchEvent
-	pw         podWatcher
-	quit       chan struct{}
-	mux        sync.Mutex
-	wi         watch.Interface
+	eventChans      map[string]chan *PodWatchEvent
+	eventBufferSize int
+	pw              podWatcher
+	quit            chan struct{}
+	mux             sync.Mutex
+	wi              watch.Interface
+}
+
+// WatcherOptions overrides the defaults of the controller, allowing it to be
+// configured as needed.
+type WatcherOptions struct {
+	// EventBufferSize specifies the size of the buffered channel for each
+	// session. It allows the watcher to write additional kubernetes events
+	// without blocking for reads. It defaults to 32 events.
+	EventBufferSize int
 }
 
 // NewWatcher creates and prepares a new watcher instance.
-func NewWatcher(pw podWatcher) *Watcher {
+func NewWatcher(pw podWatcher, options *WatcherOptions) *Watcher {
+	opts := options
+	if opts == nil {
+		opts = &WatcherOptions{}
+	}
+
+	eventBufferSize := opts.EventBufferSize
+	if eventBufferSize == 0 {
+		eventBufferSize = 32
+	}
+
 	return &Watcher{
-		eventChans: make(map[string]chan *PodWatchEvent),
-		pw:         pw,
-		quit:       make(chan struct{}),
+		eventChans:      make(map[string]chan *PodWatchEvent),
+		eventBufferSize: eventBufferSize,
+		pw:              pw,
+		quit:            make(chan struct{}),
 	}
 }
 
@@ -71,7 +92,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, eventBufferSize)
+	eventChan := make(chan *PodWatchEvent, w.eventBufferSize)
 	w.eventChans[sessionName] = eventChan
 	return eventChan, nil
 }
@@ -201,5 +222,3 @@ type PodWatchEvent struct {
 	// Error may provide the error details that led to the failing health.
 	Error error
 }
-
-const eventBufferSize = 32

testctrl/svc/orch/watcher_test.go

@@ -28,7 +28,7 @@ func TestWatcherStart(t *testing.T) {
 
 	for _, tc := range cases {
 		t.Run(tc.description, func(t *testing.T) {
-			w := NewWatcher(tc.mock)
+			w := NewWatcher(tc.mock, nil)
 
 			defer w.Stop()
 			err := w.Start()
@@ -49,7 +49,7 @@ func TestWatcherStop(t *testing.T) {
 
 	wi := watch.NewRaceFreeFake()
 	mock := &podWatcherMock{wi: wi}
-	w := NewWatcher(mock)
+	w := NewWatcher(mock, nil)
 
 	if err = w.Start(); err != nil {
 		t.Fatalf("setup failed, Start returned error: %v", err)
@@ -76,7 +76,7 @@ func TestWatcherSubscribe(t *testing.T) {
 	var event *PodWatchEvent
 	timeout := 100 * time.Millisecond * timeMultiplier
 
-	w := NewWatcher(nil)
+	w := NewWatcher(nil, nil)
 	sharedSessionName := "double-subscription"
 	_, _ = w.Subscribe(sharedSessionName)
 	if _, err = w.Subscribe(sharedSessionName); err == nil {
@@ -85,7 +85,7 @@ func TestWatcherSubscribe(t *testing.T) {
 
 	wi := watch.NewRaceFreeFake()
 	mock := &podWatcherMock{wi: wi}
-	w = NewWatcher(mock)
+	w = NewWatcher(mock, nil)
 
 	if err = w.Start(); err != nil {
 		t.Fatalf("setup failed, Start returned error: %v", err)
@@ -140,7 +140,7 @@ func TestWatcherUnsubscribe(t *testing.T) {
 
 	// test an error is returned without subscription
 	t.Run("no subscription", func(t *testing.T) {
-		w := NewWatcher(nil)
+		w := NewWatcher(nil, nil)
 		if err := w.Unsubscribe("non-existent"); err == nil {
 			t.Errorf("did not return an error for Unsubscribe call without subscription")
 		}
@@ -150,7 +150,7 @@ func TestWatcherUnsubscribe(t *testing.T) {
 	t.Run("prevents further events", func(t *testing.T) {
 		wi := watch.NewRaceFreeFake()
 		mock := &podWatcherMock{wi: wi}
-		w := NewWatcher(mock)
+		w := NewWatcher(mock, nil)
 
 		if err = w.Start(); err != nil {
 			t.Fatalf("setup failed, Start returned error: %v", err)