Decouple executor initialization from kubelet

contrib/mesos/pkg/executor/executor.go

@@ -95,14 +95,14 @@ type podStatusFunc func() (*api.PodStatus, error)
 // KubernetesExecutor is an mesos executor that runs pods
 // in a minion machine.
 type KubernetesExecutor struct {
-	updateChan           chan<- interface{} // to send pod config updates to the kubelet
+	updateChan           chan<- kubetypes.PodUpdate // sent to the kubelet, closed on shutdown
 	state                stateType
 	tasks                map[string]*kuberTask
 	pods                 map[string]*api.Pod
 	lock                 sync.Mutex
-	sourcename           string
 	client               *client.Client
-	done                 chan struct{}                     // signals shutdown
+	terminate            chan struct{}                     // signals that the executor should shutdown
+	registered           chan struct{}                     // closed when registerd
 	outgoing             chan func() (mesos.Status, error) // outgoing queue to the mesos driver
 	dockerClient         dockertools.DockerInterface
 	suicideWatch         suicideWatcher
@@ -113,14 +113,12 @@ type KubernetesExecutor struct {
 	exitFunc             func(int)
 	podStatusFunc        func(*api.Pod) (*api.PodStatus, error)
 	staticPodsConfigPath string
-	initialRegComplete   chan struct{}
 	podController        *framework.Controller
 	launchGracePeriod    time.Duration
 }
 
 type Config struct {
-	Updates              chan<- interface{} // to send pod config updates to the kubelet
-	SourceName           string
+	Updates              chan<- kubetypes.PodUpdate // to send pod config updates to the kubelet
 	APIClient            *client.Client
 	Docker               dockertools.DockerInterface
 	ShutdownAlert        func()
@@ -144,9 +142,8 @@ func New(config Config) *KubernetesExecutor {
 		state:                disconnectedState,
 		tasks:                make(map[string]*kuberTask),
 		pods:                 make(map[string]*api.Pod),
-		sourcename:           config.SourceName,
 		client:               config.APIClient,
-		done:                 make(chan struct{}),
+		terminate:            make(chan struct{}),
 		outgoing:             make(chan func() (mesos.Status, error), 1024),
 		dockerClient:         config.Docker,
 		suicideTimeout:       config.SuicideTimeout,
@@ -155,7 +152,7 @@ func New(config Config) *KubernetesExecutor {
 		shutdownAlert:        config.ShutdownAlert,
 		exitFunc:             config.ExitFunc,
 		podStatusFunc:        config.PodStatusFunc,
-		initialRegComplete:   make(chan struct{}),
+		registered:           make(chan struct{}),
 		staticPodsConfigPath: config.StaticPodsConfigPath,
 		launchGracePeriod:    config.LaunchGracePeriod,
 	}
@@ -185,26 +182,24 @@ func New(config Config) *KubernetesExecutor {
 	return k
 }
 
-func (k *KubernetesExecutor) InitialRegComplete() <-chan struct{} {
-	return k.initialRegComplete
+// InitiallyRegistered returns a channel which is closed when the executor is
+// registered with the Mesos master.
+func (k *KubernetesExecutor) InitiallyRegistered() <-chan struct{} {
+	return k.registered
 }
 
 func (k *KubernetesExecutor) Init(driver bindings.ExecutorDriver) {
 	k.killKubeletContainers()
 	k.resetSuicideWatch(driver)
 
-	go k.podController.Run(k.done)
+	go k.podController.Run(k.terminate)
 	go k.sendLoop()
 	//TODO(jdef) monitor kubeletFinished and shutdown if it happens
 }
 
-func (k *KubernetesExecutor) Done() <-chan struct{} {
-	return k.done
-}
-
 func (k *KubernetesExecutor) isDone() bool {
 	select {
-	case <-k.done:
+	case <-k.terminate:
 		return true
 	default:
 		return false
@@ -234,7 +229,12 @@ func (k *KubernetesExecutor) Registered(driver bindings.ExecutorDriver,
 		}
 	}
 
-	k.initialRegistration.Do(k.onInitialRegistration)
+	k.updateChan <- kubetypes.PodUpdate{
+		Pods: []*api.Pod{},
+		Op:   kubetypes.SET,
+	}
+
+	close(k.registered)
 }
 
 // Reregistered is called when the executor is successfully re-registered with the slave.
@@ -255,12 +255,6 @@ func (k *KubernetesExecutor) Reregistered(driver bindings.ExecutorDriver, slaveI
 		}
 	}
 
-	k.initialRegistration.Do(k.onInitialRegistration)
-}
-
-func (k *KubernetesExecutor) onInitialRegistration() {
-	defer close(k.initialRegComplete)
-
 	// emit an empty update to allow the mesos "source" to be marked as seen
 	k.lock.Lock()
 	defer k.lock.Unlock()
@@ -268,9 +262,8 @@ func (k *KubernetesExecutor) onInitialRegistration() {
 		return
 	}
 	k.updateChan <- kubetypes.PodUpdate{
-		Pods:   []*api.Pod{},
-		Op:     kubetypes.SET,
-		Source: k.sourcename,
+		Pods: []*api.Pod{},
+		Op:   kubetypes.SET,
 	}
 }
 
@@ -818,7 +811,8 @@ func (k *KubernetesExecutor) doShutdown(driver bindings.ExecutorDriver) {
 	(&k.state).transitionTo(terminalState)
 
 	// signal to all listeners that this KubeletExecutor is done!
-	close(k.done)
+	close(k.terminate)
+	close(k.updateChan)
 
 	if k.shutdownAlert != nil {
 		func() {
@@ -892,7 +886,7 @@ func newStatus(taskId *mesos.TaskID, state mesos.TaskState, message string) *mes
 
 func (k *KubernetesExecutor) sendStatus(driver bindings.ExecutorDriver, status *mesos.TaskStatus) {
 	select {
-	case <-k.done:
+	case <-k.terminate:
 	default:
 		k.outgoing <- func() (mesos.Status, error) { return driver.SendStatusUpdate(status) }
 	}
@@ -900,7 +894,7 @@ func (k *KubernetesExecutor) sendStatus(driver bindings.ExecutorDriver, status *
 
 func (k *KubernetesExecutor) sendFrameworkMessage(driver bindings.ExecutorDriver, msg string) {
 	select {
-	case <-k.done:
+	case <-k.terminate:
 	default:
 		k.outgoing <- func() (mesos.Status, error) { return driver.SendFrameworkMessage(msg) }
 	}
@@ -910,12 +904,12 @@ func (k *KubernetesExecutor) sendLoop() {
 	defer log.V(1).Info("sender loop exiting")
 	for {
 		select {
-		case <-k.done:
+		case <-k.terminate:
 			return
 		default:
 			if !k.isConnected() {
 				select {
-				case <-k.done:
+				case <-k.terminate:
 				case <-time.After(1 * time.Second):
 				}
 				continue
@@ -935,7 +929,7 @@ func (k *KubernetesExecutor) sendLoop() {
 			}
 			// attempt to re-queue the sender
 			select {
-			case <-k.done:
+			case <-k.terminate:
 			case k.outgoing <- sender:
 			}
 		}

contrib/mesos/pkg/executor/executor_test.go

@@ -63,18 +63,14 @@ func TestExecutorRegister(t *testing.T) {
 	executor.Registered(mockDriver, nil, nil, nil)
 
 	initialPodUpdate := kubetypes.PodUpdate{
-		Pods:   []*api.Pod{},
-		Op:     kubetypes.SET,
-		Source: executor.sourcename,
+		Pods: []*api.Pod{},
+		Op:   kubetypes.SET,
 	}
 	receivedInitialPodUpdate := false
 	select {
-	case m := <-updates:
-		update, ok := m.(kubetypes.PodUpdate)
-		if ok {
-			if reflect.DeepEqual(initialPodUpdate, update) {
-				receivedInitialPodUpdate = true
-			}
+	case update := <-updates:
+		if reflect.DeepEqual(initialPodUpdate, update) {
+			receivedInitialPodUpdate = true
 		}
 	case <-time.After(time.Second):
 	}
@@ -136,7 +132,7 @@ func TestExecutorLaunchAndKillTask(t *testing.T) {
 	defer testApiServer.server.Close()
 
 	mockDriver := &MockExecutorDriver{}
-	updates := make(chan interface{}, 1024)
+	updates := make(chan kubetypes.PodUpdate, 1024)
 	config := Config{
 		Docker:  dockertools.ConnectToDockerOrDie("fake://"),
 		Updates: updates,
@@ -154,7 +150,7 @@ func TestExecutorLaunchAndKillTask(t *testing.T) {
 						},
 					},
 				},
-				Phase: api.PodRunning,
+				Phase:  api.PodRunning,
 				HostIP: "127.0.0.1",
 			}, nil
 		},
@@ -205,9 +201,8 @@ func TestExecutorLaunchAndKillTask(t *testing.T) {
 
 	gotPodUpdate := false
 	select {
-	case m := <-updates:
-		update, ok := m.(kubetypes.PodUpdate)
-		if ok && len(update.Pods) == 1 {
+	case update := <-updates:
+		if len(update.Pods) == 1 {
 			gotPodUpdate = true
 		}
 	case <-time.After(time.Second):
@@ -302,7 +297,7 @@ func TestExecutorStaticPods(t *testing.T) {
 	mockDriver := &MockExecutorDriver{}
 	config := Config{
 		Docker:  dockertools.ConnectToDockerOrDie("fake://"),
-		Updates: make(chan interface{}, 1), // allow kube-executor source to proceed past init
+		Updates: make(chan kubetypes.PodUpdate, 1), // allow kube-executor source to proceed past init
 		APIClient: client.NewOrDie(&client.Config{
 			Host:    testApiServer.server.URL,
 			Version: testapi.Default.Version(),
@@ -384,7 +379,7 @@ func TestExecutorFrameworkMessage(t *testing.T) {
 	kubeletFinished := make(chan struct{})
 	config := Config{
 		Docker:  dockertools.ConnectToDockerOrDie("fake://"),
-		Updates: make(chan interface{}, 1024),
+		Updates: make(chan kubetypes.PodUpdate, 1024),
 		APIClient: client.NewOrDie(&client.Config{
 			Host:    testApiServer.server.URL,
 			Version: testapi.Default.Version(),
@@ -399,7 +394,7 @@ func TestExecutorFrameworkMessage(t *testing.T) {
 						},
 					},
 				},
-				Phase: api.PodRunning,
+				Phase:  api.PodRunning,
 				HostIP: "127.0.0.1",
 			}, nil
 		},
@@ -560,9 +555,10 @@ func TestExecutorShutdown(t *testing.T) {
 	mockDriver := &MockExecutorDriver{}
 	kubeletFinished := make(chan struct{})
 	var exitCalled int32 = 0
+	updates := make(chan kubetypes.PodUpdate, 1024)
 	config := Config{
 		Docker:  dockertools.ConnectToDockerOrDie("fake://"),
-		Updates: make(chan interface{}, 1024),
+		Updates: updates,
 		ShutdownAlert: func() {
 			close(kubeletFinished)
 		},
@@ -586,11 +582,21 @@ func TestExecutorShutdown(t *testing.T) {
 	assert.Equal(t, true, executor.isDone(),
 		"executor should be in Done state after Shutdown")
 
-	select {
-	case <-executor.Done():
-	default:
-		t.Fatal("done channel should be closed after shutdown")
+	// channel should be closed now, only a constant number of updates left
+	num := len(updates)
+drainLoop:
+	for {
+		select {
+		case _, ok := <-updates:
+			if !ok {
+				break drainLoop
+			}
+			num -= 1
+		default:
+			t.Fatal("Updates chan should be closed after Shutdown")
+		}
 	}
+	assert.Equal(t, num, 0, "Updates chan should get no new updates after Shutdown")
 
 	assert.Equal(t, true, atomic.LoadInt32(&exitCalled) > 0,
 		"the executor should call its ExitFunc when it is ready to close down")

contrib/mesos/pkg/executor/mock_test.go

@@ -23,6 +23,7 @@ import (
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/mock"
 	"k8s.io/kubernetes/pkg/api"
+	kubetypes "k8s.io/kubernetes/pkg/kubelet/types"
 	"k8s.io/kubernetes/pkg/kubelet/dockertools"
 )
 
@@ -65,13 +66,12 @@ func (m *MockExecutorDriver) SendFrameworkMessage(msg string) (mesosproto.Status
 	return args.Get(0).(mesosproto.Status), args.Error(1)
 }
 
-func NewTestKubernetesExecutor() (*KubernetesExecutor, chan interface{}) {
-	updates := make(chan interface{}, 1024)
+func NewTestKubernetesExecutor() (*KubernetesExecutor, chan kubetypes.PodUpdate) {
+	updates := make(chan kubetypes.PodUpdate, 1024)
 	return New(Config{
 		Docker:     dockertools.ConnectToDockerOrDie("fake://"),
 		Updates:    updates,
 		PodLW:      &NewMockPodsListWatch(api.PodList{}).ListWatch,
-		SourceName: "executor_test",
 	}), updates
 }