Add timeouts to orch package

This change implements timeouts on orchestration events, preventing
rouge operations from blocking indefinitely. Specifically, it:

 - Switches the Executor.Execute method to accept a context.Context

 - Modifies (*kubeExecutor).provision and (*kubeExecutor).monitor to
   use a context.Context. They now return an error when the context
   is cancelled before completion.

 - Adds a TestTimeout field to ControllerOptions. This sets the
   timeout on the Executor's context.

It also performs a handful of clean-up tasks, including:

 - Using UUIDs to name executors

 - Making TestKubeExecutorProvision a table-driven test

Author: Ben Reed

testctrl/svc/orch/controller.go

@@ -21,6 +21,7 @@ import (
 	"sync"
 	"time"
 
+	"github.com/google/uuid"
 	"github.com/golang/glog"
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/client-go/kubernetes"
@@ -44,6 +45,7 @@ type Controller struct {
 	wg              sync.WaitGroup
 	mux             sync.Mutex
 	newExecutorFunc func() Executor
+	testTimeout	time.Duration
 }
 
 // ControllerOptions overrides the defaults of the controller, allowing it to be
@@ -58,6 +60,14 @@ type ControllerOptions struct {
 	// additional kubernetes events without blocking for reads. It defaults
 	// to 32 events.
 	WatchEventBufferSize int
+
+	// TestTimeout is the maximum duration to wait for component containers
+	// to provision and terminate with a successful exit code. If this
+	// timeout is reached before an exit, the session will error.
+	//
+	// The zero value provides unlimited time to the test, so it should be
+	// avoided in production.
+	TestTimeout time.Duration
 }
 
 // NewController creates a controller using a Kubernetes clientset, store and an
@@ -104,10 +114,16 @@ func NewController(clientset kubernetes.Interface, store store.Store, options *C
 		watcher:       NewWatcher(podInterface, watcherOpts),
 		store:         store,
 		executorCount: executorCount,
+		testTimeout:   opts.TestTimeout,
 	}
 
 	c.newExecutorFunc = func() Executor {
-		return newKubeExecutor(0, c.pcd, c.watcher, c.store)
+		return &kubeExecutor{
+			name:             uuid.New().String(),
+			pcd:              c.pcd,
+			watcher:          c.watcher,
+			store:            c.store,
+		}
 	}
 
 	return c, nil
@@ -269,12 +285,16 @@ func (c *Controller) spawnExecutor(session *types.Session) {
 	c.incExecutors()
 
 	go func() {
-		defer func() {
-			c.decExecutors()
-			c.waitQueue.Done(session)
-		}()
+		ctx, cancel := context.WithCancel(context.Background())
+		if c.testTimeout > 0 {
+			ctx, _ = context.WithTimeout(ctx, c.testTimeout)
+		}
+
+		defer cancel()
+		defer c.decExecutors()
+		defer c.waitQueue.Done(session)
 
-		if err := executor.Execute(session); err != nil {
+		if err := executor.Execute(ctx, session); err != nil {
 			glog.Infof("%v", err)
 		}
 	}()

testctrl/svc/orch/controller_test.go

@@ -59,7 +59,7 @@ func TestControllerSchedule(t *testing.T) {
 
 			if tc.start {
 				controller.Start()
-				defer controller.Stop(context.TODO())
+				defer controller.Stop(context.Background())
 			}
 
 			err := controller.Schedule(tc.session)
@@ -87,7 +87,7 @@ func TestControllerStart(t *testing.T) {
 	t.Run("sets running state", func(t *testing.T) {
 		controller, _ := NewController(fake.NewSimpleClientset(), nil, nil)
 		controller.Start()
-		defer controller.Stop(context.TODO())
+		defer controller.Stop(context.Background())
 		if controller.Stopped() {
 			t.Errorf("Stopped unexpectedly returned true after starting controller")
 		}
@@ -198,7 +198,7 @@ func TestControllerStop(t *testing.T) {
 			go controller.loop()
 			time.Sleep(timeout)
 
-			ctx, cancel := context.WithTimeout(context.TODO(), tc.stopTimeout)
+			ctx, cancel := context.WithTimeout(context.Background(), tc.stopTimeout)
 			defer cancel()
 
 			err := controller.Stop(ctx)

testctrl/svc/orch/executor.go

@@ -1,6 +1,7 @@
 package orch
 
 import (
+	"context"
 	"fmt"
 	"strings"
 	"time"
@@ -13,45 +14,40 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 )
 
-// Executor executes sessions, returning an error if there is a problem with the infrastructure.
-// Executors are expected to provision, monitor and clean up resources related to a session.
-// An error is not indicative of a success or failure of the tests, but signals an orchestration
-// issue.
+// Executors can run a test session by provisioning its components, monitoring
+// its health and cleaning up after its termination.
 type Executor interface {
-	Execute(*types.Session) error
+	// Execute runs a test session. It accepts a context that can prevent
+	// problematic sessions from running indefinitely.
+	//
+	// An error is returned if there is a problem regarding the test itself.
+	// This does not include internal errors that are not specific to the
+	// test.
+	Execute(context.Context, *types.Session) error
 }
 
 type kubeExecutor struct {
-	name      string
-	watcher   *Watcher
-	eventChan <-chan *PodWatchEvent
-	session   *types.Session
-	pcd       podCreateDeleter
-	store     store.Store
+	name             string
+	watcher          *Watcher
+	pcd              podCreateDeleter
+	store            store.Store
+	session          *types.Session
+	eventChan        <-chan *PodWatchEvent
 }
 
-func newKubeExecutor(index int, pcd podCreateDeleter, watcher *Watcher, store store.Store) *kubeExecutor {
-	return &kubeExecutor{
-		name:    fmt.Sprintf("%d", index),
-		watcher: watcher,
-		pcd:     pcd,
-		store:   store,
-	}
-}
-
-func (k *kubeExecutor) Execute(session *types.Session) error {
+func (k *kubeExecutor) Execute(ctx context.Context, session *types.Session) error {
 	k.setSession(session)
 	k.writeEvent(types.AcceptEvent, nil, "kubernetes executor %v assigned session %v", k.name, session.Name)
 
 	k.writeEvent(types.ProvisionEvent, nil, "started provisioning components for session")
-	err := k.provision()
+	err := k.provision(ctx)
 	if err != nil {
 		err = fmt.Errorf("failed to provision: %v", err)
 		goto endSession
 	}
 
 	k.writeEvent(types.RunEvent, nil, "all containers appear healthy, monitoring during tests")
-	err = k.monitor()
+	err = k.monitor(ctx)
 	if err != nil {
 		err = fmt.Errorf("failed during test: %v", err)
 	}
@@ -84,7 +80,7 @@ endSession:
 	return nil
 }
 
-func (k *kubeExecutor) provision() error {
+func (k *kubeExecutor) provision(ctx context.Context) error {
 	var components []*types.Component
 	var workerIPs []string
 
@@ -125,7 +121,8 @@ func (k *kubeExecutor) provision() error {
 				default:
 					continue
 				}
-			// TODO(#54): Add timeout/deadline for provisioning resources
+			case <-ctx.Done():
+				return fmt.Errorf("provision did not complete before timeout")
 			default:
 				time.Sleep(1 * time.Second)
 			}
@@ -138,7 +135,7 @@ func (k *kubeExecutor) provision() error {
 	return nil
 }
 
-func (k *kubeExecutor) monitor() error {
+func (k *kubeExecutor) monitor(ctx context.Context) error {
 	glog.Infof("kubeExecutor[%v]: monitoring components while session %v runs", k.name, k.session.Name)
 
 	for {
@@ -150,8 +147,8 @@ func (k *kubeExecutor) monitor() error {
 			case Failed:
 				return fmt.Errorf("component %v has failed: %v", event.ComponentName, event.Error)
 			}
-
-			// TODO(#54): Add timeout/deadline for test execution (see concerns on GitHub)
+		case <-ctx.Done():
+			return fmt.Errorf("test did not complete before timeout")
 		}
 	}
 }