Merge pull request #16316 from mesosphere/scheduler-refactor

contrib/mesos/pkg/executor/executor_test.go

@@ -170,11 +170,10 @@ func TestExecutorLaunchAndKillTask(t *testing.T) {
 	}
 
 	pod := NewTestPod(1)
-	podTask, err := podtask.New(api.NewDefaultContext(), "",
-		*pod, &mesosproto.ExecutorInfo{})
+	podTask, err := podtask.New(api.NewDefaultContext(), "", pod)
 	assert.Equal(t, nil, err, "must be able to create a task from a pod")
 
-	taskInfo := podTask.BuildTaskInfo()
+	taskInfo := podTask.BuildTaskInfo(&mesosproto.ExecutorInfo{})
 	data, err := testapi.Default.Codec().Encode(pod)
 	assert.Equal(t, nil, err, "must be able to encode a pod's spec data")
 	taskInfo.Data = data
@@ -417,10 +416,8 @@ func TestExecutorFrameworkMessage(t *testing.T) {
 
 	// set up a pod to then lose
 	pod := NewTestPod(1)
-	podTask, _ := podtask.New(api.NewDefaultContext(), "foo",
-		*pod, &mesosproto.ExecutorInfo{})
-
-	taskInfo := podTask.BuildTaskInfo()
+	podTask, _ := podtask.New(api.NewDefaultContext(), "foo", pod)
+	taskInfo := podTask.BuildTaskInfo(&mesosproto.ExecutorInfo{})
 	data, _ := testapi.Default.Codec().Encode(pod)
 	taskInfo.Data = data
 

contrib/mesos/pkg/executor/mock_test.go

@@ -66,7 +66,7 @@ func (m *MockExecutorDriver) SendFrameworkMessage(msg string) (mesosproto.Status
 	return args.Get(0).(mesosproto.Status), args.Error(1)
 }
 
-func NewTestKubernetesExecutor() (*KubernetesExecutor, chan kubetypes.PodUpdate) {
+func NewTestKubernetesExecutor() (*Executor, chan kubetypes.PodUpdate) {
 	updates := make(chan kubetypes.PodUpdate, 1024)
 	return New(Config{
 		Docker:  dockertools.ConnectToDockerOrDie("fake://"),

contrib/mesos/pkg/minion/server.go

@@ -219,7 +219,7 @@ func (ms *MinionServer) launchHyperkubeServer(server string, args []string, logF
 			}
 			pwd, err := os.Getwd()
 			if err != nil {
-				log.Fatalf("Cannot get current directory: %v", err)
+				panic(fmt.Errorf("Cannot get current directory: %v", err))
 			}
 			kmEnv = append(kmEnv, fmt.Sprintf("%s:%s", e, path.Join(pwd, "bin")))
 		}

contrib/mesos/pkg/scheduler/components/algorithm/algorithm.go

+/*
+Copyright 2015 The Kubernetes Authors All rights reserved.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package algorithm
+
+import (
+	"fmt"
+
+	log "github.com/golang/glog"
+	"k8s.io/kubernetes/contrib/mesos/pkg/offers"
+	"k8s.io/kubernetes/contrib/mesos/pkg/queue"
+	"k8s.io/kubernetes/contrib/mesos/pkg/scheduler"
+	"k8s.io/kubernetes/contrib/mesos/pkg/scheduler/components/algorithm/podschedulers"
+	"k8s.io/kubernetes/contrib/mesos/pkg/scheduler/errors"
+	"k8s.io/kubernetes/contrib/mesos/pkg/scheduler/podtask"
+	"k8s.io/kubernetes/pkg/api"
+	"k8s.io/kubernetes/pkg/client/cache"
+)
+
+type SchedulerAlgorithm interface {
+	Schedule(pod *api.Pod) (string, error)
+}
+
+// SchedulerAlgorithm implements the algorithm.ScheduleAlgorithm interface
+type schedulerAlgorithm struct {
+	sched        scheduler.Scheduler
+	podUpdates   queue.FIFO
+	podScheduler podschedulers.PodScheduler
+}
+
+func New(sched scheduler.Scheduler, podUpdates queue.FIFO, podScheduler podschedulers.PodScheduler) SchedulerAlgorithm {
+	return &schedulerAlgorithm{
+		sched:        sched,
+		podUpdates:   podUpdates,
+		podScheduler: podScheduler,
+	}
+}
+
+// Schedule implements the Scheduler interface of Kubernetes.
+// It returns the selectedMachine's name and error (if there's any).
+func (k *schedulerAlgorithm) Schedule(pod *api.Pod) (string, error) {
+	log.Infof("Try to schedule pod %v\n", pod.Name)
+	ctx := api.WithNamespace(api.NewDefaultContext(), pod.Namespace)
+
+	// default upstream scheduler passes pod.Name as binding.PodID
+	podKey, err := podtask.MakePodKey(ctx, pod.Name)
+	if err != nil {
+		return "", err
+	}
+
+	k.sched.Lock()
+	defer k.sched.Unlock()
+
+	switch task, state := k.sched.Tasks().ForPod(podKey); state {
+	case podtask.StateUnknown:
+		// There's a bit of a potential race here, a pod could have been yielded() and
+		// then before we get *here* it could be deleted.
+		// We use meta to index the pod in the store since that's what k8s reflector does.
+		podName, err := cache.MetaNamespaceKeyFunc(pod)
+		if err != nil {
+			log.Warningf("aborting Schedule, unable to understand pod object %+v", pod)
+			return "", errors.NoSuchPodErr
+		}
+		if deleted := k.podUpdates.Poll(podName, queue.DELETE_EVENT); deleted {
+			// avoid scheduling a pod that's been deleted between yieldPod() and Schedule()
+			log.Infof("aborting Schedule, pod has been deleted %+v", pod)
+			return "", errors.NoSuchPodErr
+		}
+
+		podTask, err := podtask.New(ctx, "", pod)
+		if err != nil {
+			log.Warningf("aborting Schedule, unable to create podtask object %+v: %v", pod, err)
+			return "", err
+		}
+
+		podTask, err = k.sched.Tasks().Register(podTask)
+		if err != nil {
+			return "", err
+		}
+
+		return k.doSchedule(podTask)
+
+	//TODO(jdef) it's possible that the pod state has diverged from what
+	//we knew previously, we should probably update the task.Pod state here
+	//before proceeding with scheduling
+	case podtask.StatePending:
+		if pod.UID != task.Pod.UID {
+			// we're dealing with a brand new pod spec here, so the old one must have been
+			// deleted -- and so our task store is out of sync w/ respect to reality
+			//TODO(jdef) reconcile task
+			return "", fmt.Errorf("task %v spec is out of sync with pod %v spec, aborting schedule", task.ID, pod.Name)
+		} else if task.Has(podtask.Launched) {
+			// task has been marked as "launched" but the pod binding creation may have failed in k8s,
+			// but we're going to let someone else handle it, probably the mesos task error handler
+			return "", fmt.Errorf("task %s has already been launched, aborting schedule", task.ID)
+		} else {
+			return k.doSchedule(task)
+		}
+
+	default:
+		return "", fmt.Errorf("task %s is not pending, nothing to schedule", task.ID)
+	}
+}
+
+// Call ScheduleFunc and subtract some resources, returning the name of the machine the task is scheduled on
+func (k *schedulerAlgorithm) doSchedule(task *podtask.T) (string, error) {
+	var offer offers.Perishable
+	var err error
+
+	if task.HasAcceptedOffer() {
+		// verify that the offer is still on the table
+		var ok bool
+		offer, ok = k.sched.Offers().Get(task.GetOfferId())
+
+		if !ok || offer.HasExpired() {
+			task.Offer.Release()
+			task.Reset()
+			if err = k.sched.Tasks().Update(task); err != nil {
+				return "", err
+			}
+		}
+	}
+
+	if offer == nil {
+		offer, err = k.podScheduler.SchedulePod(k.sched.Offers(), task)
+	}
+
+	if err != nil {
+		return "", err
+	}
+
+	details := offer.Details()
+	if details == nil {
+		return "", fmt.Errorf("offer already invalid/expired for task %v", task.ID)
+	}
+
+	if task.Offer != nil && task.Offer != offer {
+		return "", fmt.Errorf("task.offer assignment must be idempotent, task %+v: offer %+v", task, offer)
+	}
+
+	task.Offer = offer
+	if err := k.podScheduler.Procurement()(task, details); err != nil {
+		offer.Release()
+		task.Reset()
+		return "", err
+	}
+
+	if err := k.sched.Tasks().Update(task); err != nil {
+		offer.Release()
+		return "", err
+	}
+
+	return details.GetHostname(), nil
+}

contrib/mesos/pkg/scheduler/slave/doc.go → contrib/mesos/pkg/scheduler/components/algorithm/doc.go

@@ -14,5 +14,5 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-// Package slave manages node hostnames for slave ids.
-package slave
+// Package algorithm implements the SchedulerAlgorithm
+package algorithm

contrib/mesos/pkg/scheduler/components/algorithm/podschedulers/doc.go

+/*
+Copyright 2015 The Kubernetes Authors All rights reserved.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+// Package podschedulers defines an interface (w/ implementations) for matching
+// pods against offers.
+package podschedulers

contrib/mesos/pkg/scheduler/fcfs.go → contrib/mesos/pkg/scheduler/components/algorithm/podschedulers/fcfs.go

@@ -14,7 +14,7 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-package scheduler
+package podschedulers
 
 import (
 	"fmt"
@@ -23,6 +23,7 @@ import (
 
 	"k8s.io/kubernetes/contrib/mesos/pkg/node"
 	"k8s.io/kubernetes/contrib/mesos/pkg/offers"
+	"k8s.io/kubernetes/contrib/mesos/pkg/scheduler/errors"
 	"k8s.io/kubernetes/contrib/mesos/pkg/scheduler/podtask"
 )
 
@@ -62,7 +63,7 @@ func NewFCFSPodScheduler(as AllocationStrategy, lookupNode node.LookupFunc) PodS
 }
 
 // A first-come-first-serve scheduler: acquires the first offer that can support the task
-func (fps *fcfsPodScheduler) SchedulePod(r offers.Registry, unused SlaveIndex, task *podtask.T) (offers.Perishable, error) {
+func (fps *fcfsPodScheduler) SchedulePod(r offers.Registry, task *podtask.T) (offers.Perishable, error) {
 	podName := fmt.Sprintf("%s/%s", task.Pod.Namespace, task.Pod.Name)
 	var acceptedOffer offers.Perishable
 	err := r.Walk(func(p offers.Perishable) (bool, error) {
@@ -101,5 +102,5 @@ func (fps *fcfsPodScheduler) SchedulePod(r offers.Registry, unused SlaveIndex, t
 		return nil, err
 	}
 	log.V(2).Infof("failed to find a fit for pod: %s", podName)
-	return nil, noSuitableOffersErr
+	return nil, errors.NoSuitableOffersErr
 }

contrib/mesos/pkg/scheduler/types.go → contrib/mesos/pkg/scheduler/components/algorithm/podschedulers/types.go

@@ -14,11 +14,9 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-package scheduler
+package podschedulers
 
 import (
-	"errors"
-
 	"k8s.io/kubernetes/contrib/mesos/pkg/offers"
 	"k8s.io/kubernetes/contrib/mesos/pkg/scheduler/podtask"
 )
@@ -37,25 +35,11 @@ type PodScheduler interface {
 	// SchedulePod implements how to schedule pods among slaves.
 	// We can have different implementation for different scheduling policy.
 	//
-	// The function accepts a group of slaves (each contains offers from
-	// that slave) and a single pod, which aligns well with the k8s scheduling
-	// algorithm. It returns an offerId that is acceptable for the pod, otherwise
-	// nil. The caller is responsible for filling in task state w/ relevant offer
-	// details.
+	// The function accepts a set of offers and a single pod, which aligns well
+	// with the k8s scheduling algorithm. It returns an offerId that is acceptable
+	// for the pod, otherwise nil. The caller is responsible for filling in task
+	// state w/ relevant offer details.
 	//
 	// See the FCFSPodScheduler for example.
-	SchedulePod(r offers.Registry, slaves SlaveIndex, task *podtask.T) (offers.Perishable, error)
-}
-
-// A minimal placeholder
-type empty struct{}
-
-var (
-	noSuitableOffersErr = errors.New("No suitable offers for pod/task")
-	noSuchPodErr        = errors.New("No such pod exists")
-	noSuchTaskErr       = errors.New("No such task exists")
-)
-
-type SlaveIndex interface {
-	slaveHostNameFor(id string) string
+	SchedulePod(r offers.Registry, task *podtask.T) (offers.Perishable, error)
 }

contrib/mesos/pkg/scheduler/components/binder/binder.go

+/*
+Copyright 2015 The Kubernetes Authors All rights reserved.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package binder
+
+import (
+	"fmt"
+	"strconv"
+
+	log "github.com/golang/glog"
+	"k8s.io/kubernetes/contrib/mesos/pkg/scheduler"
+	"k8s.io/kubernetes/contrib/mesos/pkg/scheduler/errors"
+	annotation "k8s.io/kubernetes/contrib/mesos/pkg/scheduler/meta"
+	"k8s.io/kubernetes/contrib/mesos/pkg/scheduler/podtask"
+	"k8s.io/kubernetes/pkg/api"
+)
+
+type Binder interface {
+	Bind(binding *api.Binding) error
+}
+
+type binder struct {
+	sched scheduler.Scheduler
+}
+
+func New(sched scheduler.Scheduler) Binder {
+	return &binder{
+		sched: sched,
+	}
+}
+
+// implements binding.Registry, launches the pod-associated-task in mesos
+func (b *binder) Bind(binding *api.Binding) error {
+
+	ctx := api.WithNamespace(api.NewContext(), binding.Namespace)
+
+	// default upstream scheduler passes pod.Name as binding.Name
+	podKey, err := podtask.MakePodKey(ctx, binding.Name)
+	if err != nil {
+		return err
+	}
+
+	b.sched.Lock()
+	defer b.sched.Unlock()
+
+	switch task, state := b.sched.Tasks().ForPod(podKey); state {
+	case podtask.StatePending:
+		return b.bind(ctx, binding, task)
+	default:
+		// in this case it's likely that the pod has been deleted between Schedule
+		// and Bind calls
+		log.Infof("No pending task for pod %s", podKey)
+		return errors.NoSuchPodErr //TODO(jdef) this error is somewhat misleading since the task could be running?!
+	}
+}
+
+func (b *binder) rollback(task *podtask.T, err error) error {
+	task.Offer.Release()
+	task.Reset()
+	if err2 := b.sched.Tasks().Update(task); err2 != nil {
+		log.Errorf("failed to update pod task: %v", err2)
+	}
+	return err
+}
+
+// assumes that: caller has acquired scheduler lock and that the task is still pending
+//
+// bind does not actually do the binding itself, but launches the pod as a Mesos task. The
+// kubernetes executor on the slave will finally do the binding. This is different from the
+// upstream scheduler in the sense that the upstream scheduler does the binding and the
+// kubelet will notice that and launches the pod.
+func (b *binder) bind(ctx api.Context, binding *api.Binding, task *podtask.T) (err error) {
+	// sanity check: ensure that the task hasAcceptedOffer(), it's possible that between
+	// Schedule() and now that the offer for this task was rescinded or invalidated.
+	// ((we should never see this here))
+	if !task.HasAcceptedOffer() {
+		return fmt.Errorf("task has not accepted a valid offer %v", task.ID)
+	}
+
+	// By this time, there is a chance that the slave is disconnected.
+	offerId := task.GetOfferId()
+	if offer, ok := b.sched.Offers().Get(offerId); !ok || offer.HasExpired() {
+		// already rescinded or timed out or otherwise invalidated
+		return b.rollback(task, fmt.Errorf("failed prior to launchTask due to expired offer for task %v", task.ID))
+	}
+
+	if err = b.prepareTaskForLaunch(ctx, binding.Target.Name, task, offerId); err == nil {
+		log.V(2).Infof("launching task: %q on target %q slave %q for pod \"%v/%v\", cpu %.2f, mem %.2f MB",
+			task.ID, binding.Target.Name, task.Spec.SlaveID, task.Pod.Namespace, task.Pod.Name, task.Spec.CPU, task.Spec.Memory)
+		if err = b.sched.LaunchTask(task); err == nil {
+			b.sched.Offers().Invalidate(offerId)
+			task.Set(podtask.Launched)
+			if err = b.sched.Tasks().Update(task); err != nil {
+				// this should only happen if the task has been removed or has changed status,
+				// which SHOULD NOT HAPPEN as long as we're synchronizing correctly
+				log.Errorf("failed to update task w/ Launched status: %v", err)
+			}
+			return
+		}
+	}
+	return b.rollback(task, fmt.Errorf("Failed to launch task %v: %v", task.ID, err))
+}
+
+//TODO(jdef) unit test this, ensure that task's copy of api.Pod is not modified
+func (b *binder) prepareTaskForLaunch(ctx api.Context, machine string, task *podtask.T, offerId string) error {
+	pod := task.Pod
+
+	// we make an effort here to avoid making changes to the task's copy of the pod, since
+	// we want that to reflect the initial user spec, and not the modified spec that we
+	// build for the executor to consume.
+	oemCt := pod.Spec.Containers
+	pod.Spec.Containers = append([]api.Container{}, oemCt...) // (shallow) clone before mod
+
+	if pod.Annotations == nil {
+		pod.Annotations = make(map[string]string)
+	}
+
+	task.SaveRecoveryInfo(pod.Annotations)
+	pod.Annotations[annotation.BindingHostKey] = task.Spec.AssignedSlave
+
+	for _, entry := range task.Spec.PortMap {
+		oemPorts := pod.Spec.Containers[entry.ContainerIdx].Ports
+		ports := append([]api.ContainerPort{}, oemPorts...)
+		p := &ports[entry.PortIdx]
+		p.HostPort = int(entry.OfferPort)
+		op := strconv.FormatUint(entry.OfferPort, 10)
+		pod.Annotations[fmt.Sprintf(annotation.PortMappingKeyFormat, p.Protocol, p.ContainerPort)] = op
+		if p.Name != "" {
+			pod.Annotations[fmt.Sprintf(annotation.PortNameMappingKeyFormat, p.Protocol, p.Name)] = op
+		}
+		pod.Spec.Containers[entry.ContainerIdx].Ports = ports
+	}
+
+	// the kubelet-executor uses this to instantiate the pod
+	log.V(3).Infof("prepared pod spec: %+v", pod)
+
+	data, err := api.Codec.Encode(&pod)
+	if err != nil {
+		log.V(2).Infof("Failed to marshal the pod spec: %v", err)
+		return err
+	}
+	task.Spec.Data = data
+	return nil
+}

contrib/mesos/pkg/scheduler/components/binder/doc.go

+/*
+Copyright 2015 The Kubernetes Authors All rights reserved.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+// Package binder implements the Binder which launched a task and let the
+// executor do the actual binding.
+package binder

contrib/mesos/pkg/scheduler/components/controller/controller.go

+/*
+Copyright 2015 The Kubernetes Authors All rights reserved.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package controller
+
+import (
+	"time"
+
+	log "github.com/golang/glog"
+	"k8s.io/kubernetes/contrib/mesos/pkg/runtime"
+	"k8s.io/kubernetes/contrib/mesos/pkg/scheduler/components/algorithm"
+	"k8s.io/kubernetes/contrib/mesos/pkg/scheduler/components/binder"
+	"k8s.io/kubernetes/pkg/api"
+	"k8s.io/kubernetes/pkg/client/record"
+	client "k8s.io/kubernetes/pkg/client/unversioned"
+)
+
+const (
+	recoveryDelay = 100 * time.Millisecond // delay after scheduler plugin crashes, before we resume scheduling
+
+	FailedScheduling = "FailedScheduling"
+	Scheduled        = "Scheduled"
+)
+
+type Controller interface {
+	Run(<-chan struct{})
+}
+
+type controller struct {
+	algorithm algorithm.SchedulerAlgorithm
+	binder    binder.Binder
+	nextPod   func() *api.Pod
+	error     func(*api.Pod, error)
+	recorder  record.EventRecorder
+	client    *client.Client
+	started   chan<- struct{} // startup latch
+}
+
+func New(client *client.Client, algorithm algorithm.SchedulerAlgorithm,
+	recorder record.EventRecorder, nextPod func() *api.Pod, error func(pod *api.Pod, schedulingErr error),
+	binder binder.Binder, started chan<- struct{}) Controller {
+	return &controller{
+		algorithm: algorithm,
+		binder:    binder,
+		nextPod:   nextPod,
+		error:     error,
+		recorder:  recorder,
+		client:    client,
+		started:   started,
+	}
+}
+
+func (s *controller) Run(done <-chan struct{}) {
+	defer close(s.started)
+	go runtime.Until(s.scheduleOne, recoveryDelay, done)
+}
+
+// hacked from GoogleCloudPlatform/kubernetes/plugin/pkg/scheduler/scheduler.go,
+// with the Modeler stuff removed since we don't use it because we have mesos.
+func (s *controller) scheduleOne() {
+	pod := s.nextPod()
+
+	// pods which are pre-scheduled (i.e. NodeName is set) are deleted by the kubelet
+	// in upstream. Not so in Mesos because the kubelet hasn't see that pod yet. Hence,
+	// the scheduler has to take care of this:
+	if pod.Spec.NodeName != "" && pod.DeletionTimestamp != nil {
+		log.V(3).Infof("deleting pre-scheduled, not yet running pod: %s/%s", pod.Namespace, pod.Name)
+		s.client.Pods(pod.Namespace).Delete(pod.Name, api.NewDeleteOptions(0))
+		return
+	}
+
+	log.V(3).Infof("Attempting to schedule: %+v", pod)
+	dest, err := s.algorithm.Schedule(pod)
+	if err != nil {
+		log.V(1).Infof("Failed to schedule: %+v", pod)
+		s.recorder.Eventf(pod, FailedScheduling, "Error scheduling: %v", err)
+		s.error(pod, err)
+		return
+	}
+	b := &api.Binding{
+		ObjectMeta: api.ObjectMeta{Namespace: pod.Namespace, Name: pod.Name},
+		Target: api.ObjectReference{
+			Kind: "Node",
+			Name: dest,
+		},
+	}
+	if err := s.binder.Bind(b); err != nil {
+		log.V(1).Infof("Failed to bind pod: %+v", err)
+		s.recorder.Eventf(pod, FailedScheduling, "Binding rejected: %v", err)
+		s.error(pod, err)
+		return
+	}
+	s.recorder.Eventf(pod, Scheduled, "Successfully assigned %v to %v", pod.Name, dest)
+}

contrib/mesos/pkg/scheduler/components/controller/doc.go

+/*
+Copyright 2015 The Kubernetes Authors All rights reserved.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+// Package controller implements the scheduling controller which waits for pod
+// events from the queuer (i.e. from the apiserver), passes them to the
+// SchedulerAlgorithm and in case of success to the binder which does the launch.
+package controller

contrib/mesos/pkg/scheduler/components/deleter/deleter.go

+/*
+Copyright 2015 The Kubernetes Authors All rights reserved.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package deleter
+
+import (
+	"time"
+
+	log "github.com/golang/glog"
+	"k8s.io/kubernetes/contrib/mesos/pkg/queue"
+	"k8s.io/kubernetes/contrib/mesos/pkg/runtime"
+	"k8s.io/kubernetes/contrib/mesos/pkg/scheduler"
+	"k8s.io/kubernetes/contrib/mesos/pkg/scheduler/errors"
+	"k8s.io/kubernetes/contrib/mesos/pkg/scheduler/podtask"
+	"k8s.io/kubernetes/contrib/mesos/pkg/scheduler/queuer"
+	"k8s.io/kubernetes/pkg/api"
+)
+
+type Deleter interface {
+	Run(updates <-chan queue.Entry, done <-chan struct{})
+	DeleteOne(pod *queuer.Pod) error
+}
+
+type deleter struct {
+	sched scheduler.Scheduler
+	qr    queuer.Queuer
+}
+
+func New(sched scheduler.Scheduler, qr queuer.Queuer) Deleter {
+	return &deleter{
+		sched: sched,
+		qr:    qr,
+	}
+}
+
+// currently monitors for "pod deleted" events, upon which handle()
+// is invoked.
+func (k *deleter) Run(updates <-chan queue.Entry, done <-chan struct{}) {
+	go runtime.Until(func() {
+		for {
+			entry := <-updates
+			pod := entry.Value().(*queuer.Pod)
+			if entry.Is(queue.DELETE_EVENT) {
+				if err := k.DeleteOne(pod); err != nil {
+					log.Error(err)
+				}
+			} else if !entry.Is(queue.POP_EVENT) {
+				k.qr.UpdatesAvailable()
+			}
+		}
+	}, 1*time.Second, done)
+}
+
+func (k *deleter) DeleteOne(pod *queuer.Pod) error {
+	ctx := api.WithNamespace(api.NewDefaultContext(), pod.Namespace)
+	podKey, err := podtask.MakePodKey(ctx, pod.Name)
+	if err != nil {
+		return err
+	}
+
+	log.V(2).Infof("pod deleted: %v", podKey)
+
+	// order is important here: we want to make sure we have the lock before
+	// removing the pod from the scheduling queue. this makes the concurrent
+	// execution of scheduler-error-handling and delete-handling easier to
+	// reason about.
+	k.sched.Lock()
+	defer k.sched.Unlock()
+
+	// prevent the scheduler from attempting to pop this; it's also possible that
+	// it's concurrently being scheduled (somewhere between pod scheduling and
+	// binding) - if so, then we'll end up removing it from taskRegistry which
+	// will abort Bind()ing
+	k.qr.Dequeue(pod.GetUID())
+
+	switch task, state := k.sched.Tasks().ForPod(podKey); state {
+	case podtask.StateUnknown:
+		log.V(2).Infof("Could not resolve pod '%s' to task id", podKey)
+		return errors.NoSuchPodErr
+
+	// determine if the task has already been launched to mesos, if not then
+	// cleanup is easier (unregister) since there's no state to sync
+	case podtask.StatePending:
+		if !task.Has(podtask.Launched) {
+			// we've been invoked in between Schedule() and Bind()
+			if task.HasAcceptedOffer() {
+				task.Offer.Release()
+				task.Reset()
+				task.Set(podtask.Deleted)
+				//TODO(jdef) probably want better handling here
+				if err := k.sched.Tasks().Update(task); err != nil {
+					return err
+				}
+			}
+			k.sched.Tasks().Unregister(task)
+			return nil
+		}
+		fallthrough
+
+	case podtask.StateRunning:
+		// signal to watchers that the related pod is going down
+		task.Set(podtask.Deleted)
+		if err := k.sched.Tasks().Update(task); err != nil {
+			log.Errorf("failed to update task w/ Deleted status: %v", err)
+		}
+		return k.sched.KillTask(task.ID)
+
+	default:
+		log.Infof("cannot kill pod '%s': non-terminal task not found %v", podKey, task.ID)
+		return errors.NoSuchTaskErr
+	}
+}

contrib/mesos/pkg/scheduler/components/deleter/deleter_test.go

+/*
+Copyright 2015 The Kubernetes Authors All rights reserved.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package deleter
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"k8s.io/kubernetes/contrib/mesos/pkg/queue"
+	types "k8s.io/kubernetes/contrib/mesos/pkg/scheduler"
+	"k8s.io/kubernetes/contrib/mesos/pkg/scheduler/errors"
+	"k8s.io/kubernetes/contrib/mesos/pkg/scheduler/podtask"
+	"k8s.io/kubernetes/contrib/mesos/pkg/scheduler/queuer"
+	"k8s.io/kubernetes/pkg/api"
+)
+
+func TestDeleteOne_NonexistentPod(t *testing.T) {
+	assert := assert.New(t)
+	obj := &types.MockScheduler{}
+	reg := podtask.NewInMemoryRegistry()
+	obj.On("Tasks").Return(reg)
+
+	q := queue.NewDelayFIFO()
+	qr := queuer.New(q, nil)
+	assert.Equal(0, len(q.List()))
+	d := New(obj, qr)
+	pod := &queuer.Pod{Pod: &api.Pod{
+		ObjectMeta: api.ObjectMeta{
+			Name:      "foo",
+			Namespace: api.NamespaceDefault,
+		}}}
+	err := d.DeleteOne(pod)
+	assert.Equal(err, errors.NoSuchPodErr)
+	obj.AssertExpectations(t)
+}
+
+func TestDeleteOne_PendingPod(t *testing.T) {
+	assert := assert.New(t)
+	obj := &types.MockScheduler{}
+	reg := podtask.NewInMemoryRegistry()
+	obj.On("Tasks").Return(reg)
+
+	pod := &queuer.Pod{Pod: &api.Pod{
+		ObjectMeta: api.ObjectMeta{
+			Name:      "foo",
+			UID:       "foo0",
+			Namespace: api.NamespaceDefault,
+		}}}
+	task, err := podtask.New(api.NewDefaultContext(), "bar", pod.Pod)
+	if err != nil {
+		t.Fatalf("failed to create task: %v", err)
+	}
+
+	_, err = reg.Register(task)
+	if err != nil {
+		t.Fatalf("failed to register task: %v", err)
+	}
+
+	// preconditions
+	q := queue.NewDelayFIFO()
+	qr := queuer.New(q, nil)
+	q.Add(pod, queue.ReplaceExisting)
+	assert.Equal(1, len(q.List()))
+	_, found := q.Get("default/foo")
+	assert.True(found)
+
+	// exec & post conditions
+	d := New(obj, qr)
+	err = d.DeleteOne(pod)
+	assert.Nil(err)
+	_, found = q.Get("foo0")
+	assert.False(found)
+	assert.Equal(0, len(q.List()))
+	obj.AssertExpectations(t)
+}
+
+func TestDeleteOne_Running(t *testing.T) {
+	assert := assert.New(t)
+	obj := &types.MockScheduler{}
+	reg := podtask.NewInMemoryRegistry()
+	obj.On("Tasks").Return(reg)
+
+	pod := &queuer.Pod{Pod: &api.Pod{
+		ObjectMeta: api.ObjectMeta{
+			Name:      "foo",
+			UID:       "foo0",
+			Namespace: api.NamespaceDefault,
+		}}}
+	task, err := podtask.New(api.NewDefaultContext(), "bar", pod.Pod)
+	if err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+
+	task, err = reg.Register(task)
+	if err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+
+	task.Set(podtask.Launched)
+	err = reg.Update(task)
+	if err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+
+	// preconditions
+	q := queue.NewDelayFIFO()
+	qr := queuer.New(q, nil)
+	q.Add(pod, queue.ReplaceExisting)
+	assert.Equal(1, len(q.List()))
+	_, found := q.Get("default/foo")
+	assert.True(found)
+
+	obj.On("KillTask", task.ID).Return(nil)
+
+	// exec & post conditions
+	d := New(obj, qr)
+	err = d.DeleteOne(pod)
+	assert.Nil(err)
+	_, found = q.Get("foo0")
+	assert.False(found)
+	assert.Equal(0, len(q.List()))
+	obj.AssertExpectations(t)
+}
+
+func TestDeleteOne_badPodNaming(t *testing.T) {
+	assert := assert.New(t)
+	obj := &types.MockScheduler{}
+	pod := &queuer.Pod{Pod: &api.Pod{}}
+	q := queue.NewDelayFIFO()
+	qr := queuer.New(q, nil)
+	d := New(obj, qr)
+
+	err := d.DeleteOne(pod)
+	assert.NotNil(err)
+
+	pod.Pod.ObjectMeta.Name = "foo"
+	err = d.DeleteOne(pod)
+	assert.NotNil(err)
+
+	pod.Pod.ObjectMeta.Name = ""
+	pod.Pod.ObjectMeta.Namespace = "bar"
+	err = d.DeleteOne(pod)
+	assert.NotNil(err)
+
+	obj.AssertExpectations(t)
+}

contrib/mesos/pkg/scheduler/components/deleter/doc.go

+/*
+Copyright 2015 The Kubernetes Authors All rights reserved.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+// Package deleter implements the deleter which listens for pod DELETE events
+// from the apiserver and kills tasks for deleted pods.
+package deleter

contrib/mesos/pkg/scheduler/components/doc.go

+/*
+Copyright 2015 The Kubernetes Authors All rights reserved.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+// Package components implements independent aspects of the scheduler which
+// do not use Framework or Scheduler internals, but rely solely on the Scheduler
+// interface.
+package components

contrib/mesos/pkg/scheduler/components/errorhandler/doc.go

+/*
+Copyright 2015 The Kubernetes Authors All rights reserved.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+// Package errorhandler implements the ErrorHandler which handles scheduer error
+// and possibly requeue pods for scheduling again.
+package errorhandler

contrib/mesos/pkg/scheduler/components/errorhandler/errorhandler.go

+/*
+Copyright 2015 The Kubernetes Authors All rights reserved.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package errorhandler
+
+import (
+	log "github.com/golang/glog"
+	"k8s.io/kubernetes/contrib/mesos/pkg/backoff"
+	"k8s.io/kubernetes/contrib/mesos/pkg/queue"
+	"k8s.io/kubernetes/contrib/mesos/pkg/scheduler"
+	"k8s.io/kubernetes/contrib/mesos/pkg/scheduler/errors"
+	"k8s.io/kubernetes/contrib/mesos/pkg/scheduler/podtask"
+	"k8s.io/kubernetes/contrib/mesos/pkg/scheduler/queuer"
+	"k8s.io/kubernetes/pkg/api"
+	"k8s.io/kubernetes/pkg/util"
+)
+
+type ErrorHandler interface {
+	Error(pod *api.Pod, schedulingErr error)
+}
+
+type errorHandler struct {
+	sched        scheduler.Scheduler
+	backoff      *backoff.Backoff
+	qr           queuer.Queuer
+	newBreakChan func(podKey string) queue.BreakChan
+}
+
+func New(sched scheduler.Scheduler, backoff *backoff.Backoff, qr queuer.Queuer, newBC func(podKey string) queue.BreakChan) ErrorHandler {
+	return &errorHandler{
+		sched:        sched,
+		backoff:      backoff,
+		qr:           qr,
+		newBreakChan: newBC,
+	}
+}
+
+// implementation of scheduling plugin's Error func; see plugin/pkg/scheduler
+func (k *errorHandler) Error(pod *api.Pod, schedulingErr error) {
+
+	if schedulingErr == errors.NoSuchPodErr {
+		log.V(2).Infof("Not rescheduling non-existent pod %v", pod.Name)
+		return
+	}
+
+	log.Infof("Error scheduling %v: %v; retrying", pod.Name, schedulingErr)
+	defer util.HandleCrash()
+
+	// default upstream scheduler passes pod.Name as binding.PodID
+	ctx := api.WithNamespace(api.NewDefaultContext(), pod.Namespace)
+	podKey, err := podtask.MakePodKey(ctx, pod.Name)
+	if err != nil {
+		log.Errorf("Failed to construct pod key, aborting scheduling for pod %v: %v", pod.Name, err)
+		return
+	}
+
+	k.backoff.GC()
+	k.sched.Lock()
+	defer k.sched.Unlock()
+
+	switch task, state := k.sched.Tasks().ForPod(podKey); state {
+	case podtask.StateUnknown:
+		// if we don't have a mapping here any more then someone deleted the pod
+		log.V(2).Infof("Could not resolve pod to task, aborting pod reschdule: %s", podKey)
+		return
+
+	case podtask.StatePending:
+		if task.Has(podtask.Launched) {
+			log.V(2).Infof("Skipping re-scheduling for already-launched pod %v", podKey)
+			return
+		}
+		breakoutEarly := queue.BreakChan(nil)
+		if schedulingErr == errors.NoSuitableOffersErr {
+			log.V(3).Infof("adding backoff breakout handler for pod %v", podKey)
+			breakoutEarly = k.newBreakChan(podKey)
+		}
+		delay := k.backoff.Get(podKey)
+		log.V(3).Infof("requeuing pod %v with delay %v", podKey, delay)
+		k.qr.Requeue(&queuer.Pod{Pod: pod, Delay: &delay, Notify: breakoutEarly})
+
+	default:
+		log.V(2).Infof("Task is no longer pending, aborting reschedule for pod %v", podKey)
+	}
+}

contrib/mesos/pkg/scheduler/components/framework/doc.go

+/*
+Copyright 2015 The Kubernetes Authors All rights reserved.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+// Package framework implements the Mesos scheduler.
+package framework

contrib/mesos/pkg/scheduler/mock_test.go → contrib/mesos/pkg/scheduler/components/framework/driver_mock.go

@@ -14,83 +14,13 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-package scheduler
+package framework
 
 import (
-	"sync"
-	"testing"
-
 	mesos "github.com/mesos/mesos-go/mesosproto"
 	"github.com/stretchr/testify/mock"
-	"k8s.io/kubernetes/contrib/mesos/pkg/offers"
-	"k8s.io/kubernetes/contrib/mesos/pkg/scheduler/podtask"
-	"k8s.io/kubernetes/pkg/api"
 )
 
-// implements SchedulerInterface
-type MockScheduler struct {
-	sync.RWMutex
-	mock.Mock
-}
-
-func (m *MockScheduler) slaveHostNameFor(id string) (hostName string) {
-	args := m.Called(id)
-	x := args.Get(0)
-	if x != nil {
-		hostName = x.(string)
-	}
-	return
-}
-func (m *MockScheduler) algorithm() (f PodScheduler) {
-	args := m.Called()
-	x := args.Get(0)
-	if x != nil {
-		f = x.(PodScheduler)
-	}
-	return
-}
-func (m *MockScheduler) createPodTask(ctx api.Context, pod *api.Pod) (task *podtask.T, err error) {
-	args := m.Called(ctx, pod)
-	x := args.Get(0)
-	if x != nil {
-		task = x.(*podtask.T)
-	}
-	err = args.Error(1)
-	return
-}
-func (m *MockScheduler) offers() (f offers.Registry) {
-	args := m.Called()
-	x := args.Get(0)
-	if x != nil {
-		f = x.(offers.Registry)
-	}
-	return
-}
-func (m *MockScheduler) tasks() (f podtask.Registry) {
-	args := m.Called()
-	x := args.Get(0)
-	if x != nil {
-		f = x.(podtask.Registry)
-	}
-	return
-}
-func (m *MockScheduler) killTask(taskId string) error {
-	args := m.Called(taskId)
-	return args.Error(0)
-}
-func (m *MockScheduler) launchTask(task *podtask.T) error {
-	args := m.Called(task)
-	return args.Error(0)
-}
-
-// @deprecated this is a placeholder for me to test the mock package
-func TestNoSlavesYet(t *testing.T) {
-	obj := &MockScheduler{}
-	obj.On("slaveHostNameFor", "foo").Return(nil)
-	obj.slaveHostNameFor("foo")
-	obj.AssertExpectations(t)
-}
-
 /*-----------------------------------------------------------------------------
  |
  |   this really belongs in the mesos-go package, but that's being updated soon
@@ -146,57 +76,84 @@ func (m *MockSchedulerDriver) Init() error {
 	args := m.Called()
 	return args.Error(0)
 }
+
 func (m *MockSchedulerDriver) Start() (mesos.Status, error) {
 	args := m.Called()
 	return status(args, 0), args.Error(1)
 }
+
 func (m *MockSchedulerDriver) Stop(b bool) (mesos.Status, error) {
 	args := m.Called(b)
 	return status(args, 0), args.Error(1)
 }
+
 func (m *MockSchedulerDriver) Abort() (mesos.Status, error) {
 	args := m.Called()
 	return status(args, 0), args.Error(1)
 }
+
 func (m *MockSchedulerDriver) Join() (mesos.Status, error) {
 	args := m.Called()
 	return status(args, 0), args.Error(1)
 }
+
 func (m *MockSchedulerDriver) Run() (mesos.Status, error) {
 	args := m.Called()
 	return status(args, 0), args.Error(1)
 }
+
 func (m *MockSchedulerDriver) RequestResources(r []*mesos.Request) (mesos.Status, error) {
 	args := m.Called(r)
 	return status(args, 0), args.Error(1)
 }
+
 func (m *MockSchedulerDriver) ReconcileTasks(statuses []*mesos.TaskStatus) (mesos.Status, error) {
 	args := m.Called(statuses)
 	return status(args, 0), args.Error(1)
 }
+
 func (m *MockSchedulerDriver) LaunchTasks(offerIds []*mesos.OfferID, ti []*mesos.TaskInfo, f *mesos.Filters) (mesos.Status, error) {
 	args := m.Called(offerIds, ti, f)
 	return status(args, 0), args.Error(1)
 }
+
 func (m *MockSchedulerDriver) KillTask(tid *mesos.TaskID) (mesos.Status, error) {
 	args := m.Called(tid)
 	return status(args, 0), args.Error(1)
 }
+
 func (m *MockSchedulerDriver) DeclineOffer(oid *mesos.OfferID, f *mesos.Filters) (mesos.Status, error) {
 	args := m.Called(oid, f)
 	return status(args, 0), args.Error(1)
 }
+
 func (m *MockSchedulerDriver) ReviveOffers() (mesos.Status, error) {
 	args := m.Called()
 	return status(args, 0), args.Error(0)
 }
+
 func (m *MockSchedulerDriver) SendFrameworkMessage(eid *mesos.ExecutorID, sid *mesos.SlaveID, s string) (mesos.Status, error) {
 	args := m.Called(eid, sid, s)
 	return status(args, 0), args.Error(1)
 }
+
 func (m *MockSchedulerDriver) Destroy() {
 	m.Called()
 }
+
 func (m *MockSchedulerDriver) Wait() {
 	m.Called()
 }
+
+type JoinableDriver struct {
+	MockSchedulerDriver
+	joinFunc func() (mesos.Status, error)
+}
+
+// Join invokes joinFunc if it has been set, otherwise blocks forever
+func (m *JoinableDriver) Join() (mesos.Status, error) {
+	if m.joinFunc != nil {
+		return m.joinFunc()
+	}
+	select {}
+}

contrib/mesos/pkg/scheduler/slave/slave.go → contrib/mesos/pkg/scheduler/components/framework/slaveregistry.go

@@ -14,25 +14,26 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-package slave
+package framework
 
 import (
 	"sync"
 )
 
-type Registry struct {
+// slaveRegistry manages node hostnames for slave ids.
+type slaveRegistry struct {
 	lock      sync.Mutex
 	hostNames map[string]string
 }
 
-func NewRegistry() *Registry {
-	return &Registry{
+func newSlaveRegistry() *slaveRegistry {
+	return &slaveRegistry{
 		hostNames: map[string]string{},
 	}
 }
 
 // Register creates a mapping between a slaveId and slave if not existing.
-func (st *Registry) Register(slaveId, slaveHostname string) {
+func (st *slaveRegistry) Register(slaveId, slaveHostname string) {
 	st.lock.Lock()
 	defer st.lock.Unlock()
 	_, exists := st.hostNames[slaveId]
@@ -42,7 +43,7 @@ func (st *Registry) Register(slaveId, slaveHostname string) {
 }
 
 // SlaveIDs returns the keys of the registry
-func (st *Registry) SlaveIDs() []string {
+func (st *slaveRegistry) SlaveIDs() []string {
 	st.lock.Lock()
 	defer st.lock.Unlock()
 	slaveIds := make([]string, 0, len(st.hostNames))
@@ -53,7 +54,7 @@ func (st *Registry) SlaveIDs() []string {
 }
 
 // HostName looks up a hostname for a given slaveId
-func (st *Registry) HostName(slaveId string) string {
+func (st *slaveRegistry) HostName(slaveId string) string {
 	st.lock.Lock()
 	defer st.lock.Unlock()
 	return st.hostNames[slaveId]

contrib/mesos/pkg/scheduler/slave/slave_test.go → contrib/mesos/pkg/scheduler/components/framework/slaveregistry_test.go

@@ -14,7 +14,7 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-package slave
+package framework
 
 import (
 	"testing"
@@ -26,7 +26,7 @@ import (
 func TestSlaveStorage_Register(t *testing.T) {
 	assert := assert.New(t)
 
-	slaveStorage := NewRegistry()
+	slaveStorage := newSlaveRegistry()
 	assert.Equal(0, len(slaveStorage.hostNames))
 
 	slaveId := "slave1"
@@ -42,7 +42,7 @@ func TestSlaveStorage_Register(t *testing.T) {
 func TestSlaveStorage_HostName(t *testing.T) {
 	assert := assert.New(t)
 
-	slaveStorage := NewRegistry()
+	slaveStorage := newSlaveRegistry()
 	assert.Equal(0, len(slaveStorage.hostNames))
 
 	slaveId := "slave1"
@@ -62,7 +62,7 @@ func TestSlaveStorage_HostName(t *testing.T) {
 func TestSlaveStorage_SlaveIds(t *testing.T) {
 	assert := assert.New(t)
 
-	slaveStorage := NewRegistry()
+	slaveStorage := newSlaveRegistry()
 	assert.Equal(0, len(slaveStorage.hostNames))
 
 	slaveId := "1"

contrib/mesos/pkg/scheduler/components/podreconciler/doc.go

+/*
+Copyright 2015 The Kubernetes Authors All rights reserved.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+// Package podreconciler implements pod reconcilation of pods which failed
+// to launch, i.e. before binding by the executor took place.
+package podreconciler

contrib/mesos/pkg/scheduler/components/podreconciler/podreconciler.go

+/*
+Copyright 2015 The Kubernetes Authors All rights reserved.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package podreconciler
+
+import (
+	"time"
+
+	log "github.com/golang/glog"
+	"k8s.io/kubernetes/contrib/mesos/pkg/scheduler"
+	"k8s.io/kubernetes/contrib/mesos/pkg/scheduler/components/deleter"
+	"k8s.io/kubernetes/contrib/mesos/pkg/scheduler/errors"
+	"k8s.io/kubernetes/contrib/mesos/pkg/scheduler/podtask"
+	"k8s.io/kubernetes/contrib/mesos/pkg/scheduler/queuer"
+	"k8s.io/kubernetes/pkg/api"
+	apierrors "k8s.io/kubernetes/pkg/api/errors"
+	client "k8s.io/kubernetes/pkg/client/unversioned"
+)
+
+// PodReconciler reconciles a pod with the apiserver
+type PodReconciler interface {
+	Reconcile(t *podtask.T)
+}
+
+type podReconciler struct {
+	sched   scheduler.Scheduler
+	client  *client.Client
+	qr      queuer.Queuer
+	deleter deleter.Deleter
+}
+
+func New(sched scheduler.Scheduler, client *client.Client, qr queuer.Queuer, deleter deleter.Deleter) PodReconciler {
+	return &podReconciler{
+		sched:   sched,
+		client:  client,
+		qr:      qr,
+		deleter: deleter,
+	}
+}
+
+// this pod may be out of sync with respect to the API server registry:
+//      this pod   |  apiserver registry
+//    -------------|----------------------
+//      host=.*    |  404           ; pod was deleted
+//      host=.*    |  5xx           ; failed to sync, try again later?
+//      host=""    |  host=""       ; perhaps no updates to process?
+//      host=""    |  host="..."    ; pod has been scheduled and assigned, is there a task assigned? (check TaskIdKey in binding?)
+//      host="..." |  host=""       ; pod is no longer scheduled, does it need to be re-queued?
+//      host="..." |  host="..."    ; perhaps no updates to process?
+//
+// TODO(jdef) this needs an integration test
+func (s *podReconciler) Reconcile(t *podtask.T) {
+	log.V(1).Infof("reconcile pod %v, assigned to slave %q", t.Pod.Name, t.Spec.AssignedSlave)
+	ctx := api.WithNamespace(api.NewDefaultContext(), t.Pod.Namespace)
+	pod, err := s.client.Pods(api.NamespaceValue(ctx)).Get(t.Pod.Name)
+	if err != nil {
+		if apierrors.IsNotFound(err) {
+			// attempt to delete
+			if err = s.deleter.DeleteOne(&queuer.Pod{Pod: &t.Pod}); err != nil && err != errors.NoSuchPodErr && err != errors.NoSuchTaskErr {
+				log.Errorf("failed to delete pod: %v: %v", t.Pod.Name, err)
+			}
+		} else {
+			//TODO(jdef) other errors should probably trigger a retry (w/ backoff).
+			//For now, drop the pod on the floor
+			log.Warning("aborting reconciliation for pod %v: %v", t.Pod.Name, err)
+		}
+		return
+	}
+
+	log.Infof("pod %v scheduled on %q according to apiserver", pod.Name, pod.Spec.NodeName)
+	if t.Spec.AssignedSlave != pod.Spec.NodeName {
+		if pod.Spec.NodeName == "" {
+			// pod is unscheduled.
+			// it's possible that we dropped the pod in the scheduler error handler
+			// because of task misalignment with the pod (task.Has(podtask.Launched) == true)
+
+			podKey, err := podtask.MakePodKey(ctx, pod.Name)
+			if err != nil {
+				log.Error(err)
+				return
+			}
+
+			s.sched.Lock()
+			defer s.sched.Unlock()
+
+			if _, state := s.sched.Tasks().ForPod(podKey); state != podtask.StateUnknown {
+				//TODO(jdef) reconcile the task
+				log.Errorf("task already registered for pod %v", pod.Name)
+				return
+			}
+
+			now := time.Now()
+			log.V(3).Infof("reoffering pod %v", podKey)
+			s.qr.Reoffer(queuer.NewPodWithDeadline(pod, &now))
+		} else {
+			// pod is scheduled.
+			// not sure how this happened behind our backs. attempt to reconstruct
+			// at least a partial podtask.T record.
+			//TODO(jdef) reconcile the task
+			log.Errorf("pod already scheduled: %v", pod.Name)
+		}
+	} else {
+		//TODO(jdef) for now, ignore the fact that the rest of the spec may be different
+		//and assume that our knowledge of the pod aligns with that of the apiserver
+		log.Error("pod reconciliation does not support updates; not yet implemented")
+	}
+}

contrib/mesos/pkg/scheduler/components/podstoreadapter.go

+/*
+Copyright 2015 The Kubernetes Authors All rights reserved.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package components
+
+import (
+	"k8s.io/kubernetes/contrib/mesos/pkg/queue"
+	"k8s.io/kubernetes/contrib/mesos/pkg/scheduler/queuer"
+	"k8s.io/kubernetes/pkg/api"
+)
+
+// Consumes *api.Pod, produces *Pod; the k8s reflector wants to push *api.Pod
+// objects at us, but we want to store more flexible (Pod) type defined in
+// this package. The adapter implementation facilitates this. It's a little
+// hackish since the object type going in is different than the object type
+// coming out -- you've been warned.
+type podStoreAdapter struct {
+	queue.FIFO
+}
+
+func (psa *podStoreAdapter) Add(obj interface{}) error {
+	pod := obj.(*api.Pod)
+	return psa.FIFO.Add(&queuer.Pod{Pod: pod})
+}
+
+func (psa *podStoreAdapter) Update(obj interface{}) error {
+	pod := obj.(*api.Pod)
+	return psa.FIFO.Update(&queuer.Pod{Pod: pod})
+}
+
+func (psa *podStoreAdapter) Delete(obj interface{}) error {
+	pod := obj.(*api.Pod)
+	return psa.FIFO.Delete(&queuer.Pod{Pod: pod})
+}
+
+func (psa *podStoreAdapter) Get(obj interface{}) (interface{}, bool, error) {
+	pod := obj.(*api.Pod)
+	return psa.FIFO.Get(&queuer.Pod{Pod: pod})
+}
+
+// Replace will delete the contents of the store, using instead the
+// given map. This store implementation does NOT take ownership of the map.
+func (psa *podStoreAdapter) Replace(objs []interface{}, resourceVersion string) error {
+	newobjs := make([]interface{}, len(objs))
+	for i, v := range objs {
+		pod := v.(*api.Pod)
+		newobjs[i] = &queuer.Pod{Pod: pod}
+	}
+	return psa.FIFO.Replace(newobjs, resourceVersion)
+}

contrib/mesos/pkg/scheduler/components/scheduler.go

+/*
+Copyright 2015 The Kubernetes Authors All rights reserved.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package components
+
+import (
+	"net/http"
+	"sync"
+
+	mesos "github.com/mesos/mesos-go/mesosproto"
+	"k8s.io/kubernetes/contrib/mesos/pkg/backoff"
+	"k8s.io/kubernetes/contrib/mesos/pkg/offers"
+	"k8s.io/kubernetes/contrib/mesos/pkg/queue"
+	"k8s.io/kubernetes/contrib/mesos/pkg/runtime"
+	"k8s.io/kubernetes/contrib/mesos/pkg/scheduler"
+	"k8s.io/kubernetes/contrib/mesos/pkg/scheduler/components/algorithm"
+	"k8s.io/kubernetes/contrib/mesos/pkg/scheduler/components/algorithm/podschedulers"
+	"k8s.io/kubernetes/contrib/mesos/pkg/scheduler/components/binder"
+	"k8s.io/kubernetes/contrib/mesos/pkg/scheduler/components/controller"
+	"k8s.io/kubernetes/contrib/mesos/pkg/scheduler/components/deleter"
+	"k8s.io/kubernetes/contrib/mesos/pkg/scheduler/components/errorhandler"
+	"k8s.io/kubernetes/contrib/mesos/pkg/scheduler/components/framework"
+	"k8s.io/kubernetes/contrib/mesos/pkg/scheduler/components/podreconciler"
+	"k8s.io/kubernetes/contrib/mesos/pkg/scheduler/config"
+	"k8s.io/kubernetes/contrib/mesos/pkg/scheduler/podtask"
+	"k8s.io/kubernetes/contrib/mesos/pkg/scheduler/queuer"
+	"k8s.io/kubernetes/pkg/api"
+	"k8s.io/kubernetes/pkg/client/cache"
+	"k8s.io/kubernetes/pkg/client/record"
+	client "k8s.io/kubernetes/pkg/client/unversioned"
+)
+
+// sched implements the Scheduler interface.
+type sched struct {
+	podReconciler podreconciler.PodReconciler
+	framework     framework.Framework
+	controller    controller.Controller
+
+	// unsafe state, needs to be guarded, especially changes to podtask.T objects
+	sync.RWMutex
+	taskRegistry podtask.Registry
+}
+
+func New(c *config.Config, fw framework.Framework, ps podschedulers.PodScheduler,
+	client *client.Client, recorder record.EventRecorder, terminate <-chan struct{}, mux *http.ServeMux, lw *cache.ListWatch) scheduler.Scheduler {
+
+	core := &sched{
+		framework:    fw,
+		taskRegistry: podtask.NewInMemoryRegistry(),
+	}
+
+	// Watch and queue pods that need scheduling.
+	podUpdatesBypass := make(chan queue.Entry, c.UpdatesBacklog)
+	podUpdates := &podStoreAdapter{queue.NewHistorical(podUpdatesBypass)}
+	reflector := cache.NewReflector(lw, &api.Pod{}, podUpdates, 0)
+
+	q := queuer.New(queue.NewDelayFIFO(), podUpdates)
+
+	algorithm := algorithm.New(core, podUpdates, ps)
+
+	podDeleter := deleter.New(core, q)
+
+	core.podReconciler = podreconciler.New(core, client, q, podDeleter)
+
+	bo := backoff.New(c.InitialPodBackoff.Duration, c.MaxPodBackoff.Duration)
+	newBC := func(podKey string) queue.BreakChan {
+		return queue.BreakChan(core.Offers().Listen(podKey, func(offer *mesos.Offer) bool {
+			core.Lock()
+			defer core.Unlock()
+			switch task, state := core.Tasks().ForPod(podKey); state {
+			case podtask.StatePending:
+				// Assess fitness of pod with the current offer. The scheduler normally
+				// "backs off" when it can't find an offer that matches up with a pod.
+				// The backoff period for a pod can terminate sooner if an offer becomes
+				// available that matches up.
+				return !task.Has(podtask.Launched) && ps.FitPredicate()(task, offer, nil)
+			default:
+				// no point in continuing to check for matching offers
+				return true
+			}
+		}))
+	}
+	errorHandler := errorhandler.New(core, bo, q, newBC)
+
+	binder := binder.New(core)
+
+	startLatch := make(chan struct{})
+
+	runtime.On(startLatch, func() {
+		reflector.Run() // TODO(jdef) should listen for termination
+		podDeleter.Run(podUpdatesBypass, terminate)
+		q.Run(terminate)
+
+		q.InstallDebugHandlers(mux)
+		podtask.InstallDebugHandlers(core.Tasks(), mux)
+	})
+
+	core.controller = controller.New(client, algorithm, recorder, q.Yield, errorHandler.Error, binder, startLatch)
+	return core
+}
+
+func (c *sched) Run(done <-chan struct{}) {
+	c.controller.Run(done)
+}
+
+func (c *sched) Reconcile(t *podtask.T) {
+	c.podReconciler.Reconcile(t)
+}
+
+func (c *sched) Tasks() podtask.Registry {
+	return c.taskRegistry
+}
+
+func (c *sched) Offers() offers.Registry {
+	return c.framework.Offers()
+}
+
+func (c *sched) KillTask(id string) error {
+	return c.framework.KillTask(id)
+}
+
+func (c *sched) LaunchTask(t *podtask.T) error {
+	return c.framework.LaunchTask(t)
+}

contrib/mesos/pkg/scheduler/components/tasksreconciler/doc.go

+/*
+Copyright 2015 The Kubernetes Authors All rights reserved.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+// Package taskreconciler implement Mesos task reconcilation.
+package taskreconciler

contrib/mesos/pkg/scheduler/components/tasksreconciler/tasksreconciler.go

+/*
+Copyright 2015 The Kubernetes Authors All rights reserved.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package taskreconciler
+
+import (
+	"fmt"
+	"time"
+
+	log "github.com/golang/glog"
+	mesos "github.com/mesos/mesos-go/mesosproto"
+	bindings "github.com/mesos/mesos-go/scheduler"
+	"k8s.io/kubernetes/contrib/mesos/pkg/proc"
+	"k8s.io/kubernetes/contrib/mesos/pkg/scheduler/errors"
+	"k8s.io/kubernetes/contrib/mesos/pkg/scheduler/metrics"
+)
+
+type Action func(driver bindings.SchedulerDriver, cancel <-chan struct{}) <-chan error
+
+type TasksReconciler interface {
+	RequestExplicit()
+	RequestImplicit()
+	Run(driver bindings.SchedulerDriver, done <-chan struct{})
+}
+
+type tasksReconciler struct {
+	proc.Doer
+	Action                             Action
+	explicit                           chan struct{} // send an empty struct to trigger explicit reconciliation
+	implicit                           chan struct{} // send an empty struct to trigger implicit reconciliation
+	cooldown                           time.Duration
+	explicitReconciliationAbortTimeout time.Duration
+}
+
+func New(doer proc.Doer, action Action,
+	cooldown, explicitReconciliationAbortTimeout time.Duration, done <-chan struct{}) TasksReconciler {
+	return &tasksReconciler{
+		Doer:     doer,
+		explicit: make(chan struct{}, 1),
+		implicit: make(chan struct{}, 1),
+		cooldown: cooldown,
+		explicitReconciliationAbortTimeout: explicitReconciliationAbortTimeout,
+		Action: func(driver bindings.SchedulerDriver, cancel <-chan struct{}) <-chan error {
+			// trigged the reconciler action in the doer's execution context,
+			// but it could take a while and the scheduler needs to be able to
+			// process updates, the callbacks for which ALSO execute in the SAME
+			// deferred execution context -- so the action MUST be executed async.
+			errOnce := proc.NewErrorOnce(cancel)
+			return errOnce.Send(doer.Do(func() {
+				// only triggers the action if we're the currently elected,
+				// registered master and runs the action async.
+				go func() {
+					var err <-chan error
+					defer errOnce.Send(err)
+					err = action(driver, cancel)
+				}()
+			})).Err()
+		},
+	}
+}
+
+func (r *tasksReconciler) RequestExplicit() {
+	select {
+	case r.explicit <- struct{}{}: // noop
+	default: // request queue full; noop
+	}
+}
+
+func (r *tasksReconciler) RequestImplicit() {
+	select {
+	case r.implicit <- struct{}{}: // noop
+	default: // request queue full; noop
+	}
+}
+
+// execute task reconciliation, returns when r.done is closed. intended to run as a goroutine.
+// if reconciliation is requested while another is in progress, the in-progress operation will be
+// cancelled before the new reconciliation operation begins.
+func (r *tasksReconciler) Run(driver bindings.SchedulerDriver, done <-chan struct{}) {
+	var cancel, finished chan struct{}
+requestLoop:
+	for {
+		select {
+		case <-done:
+			return
+		default: // proceed
+		}
+		select {
+		case <-r.implicit:
+			metrics.ReconciliationRequested.WithLabelValues("implicit").Inc()
+			select {
+			case <-done:
+				return
+			case <-r.explicit:
+				break // give preference to a pending request for explicit
+			default: // continue
+				// don't run implicit reconciliation while explicit is ongoing
+				if finished != nil {
+					select {
+					case <-finished: // continue w/ implicit
+					default:
+						log.Infoln("skipping implicit reconcile because explicit reconcile is ongoing")
+						continue requestLoop
+					}
+				}
+				errOnce := proc.NewErrorOnce(done)
+				errCh := r.Do(func() {
+					var err error
+					defer errOnce.Report(err)
+					log.Infoln("implicit reconcile tasks")
+					metrics.ReconciliationExecuted.WithLabelValues("implicit").Inc()
+					if _, err = driver.ReconcileTasks([]*mesos.TaskStatus{}); err != nil {
+						log.V(1).Infof("failed to request implicit reconciliation from mesos: %v", err)
+					}
+				})
+				proc.OnError(errOnce.Send(errCh).Err(), func(err error) {
+					log.Errorf("failed to run implicit reconciliation: %v", err)
+				}, done)
+				goto slowdown
+			}
+		case <-done:
+			return
+		case <-r.explicit: // continue
+			metrics.ReconciliationRequested.WithLabelValues("explicit").Inc()
+		}
+
+		if cancel != nil {
+			close(cancel)
+			cancel = nil
+
+			// play nice and wait for the prior operation to finish, complain
+			// if it doesn't
+			select {
+			case <-done:
+				return
+			case <-finished: // noop, expected
+			case <-time.After(r.explicitReconciliationAbortTimeout): // very unexpected
+				log.Error("reconciler action failed to stop upon cancellation")
+			}
+		}
+		// copy 'finished' to 'fin' here in case we end up with simultaneous go-routines,
+		// if cancellation takes too long or fails - we don't want to close the same chan
+		// more than once
+		cancel = make(chan struct{})
+		finished = make(chan struct{})
+		go func(fin chan struct{}) {
+			startedAt := time.Now()
+			defer func() {
+				metrics.ReconciliationLatency.Observe(metrics.InMicroseconds(time.Since(startedAt)))
+			}()
+
+			metrics.ReconciliationExecuted.WithLabelValues("explicit").Inc()
+			defer close(fin)
+			err := <-r.Action(driver, cancel)
+			if err == errors.ReconciliationCancelledErr {
+				metrics.ReconciliationCancelled.WithLabelValues("explicit").Inc()
+				log.Infoln(err.Error())
+			} else if err != nil {
+				log.Errorf("reconciler action failed: %v", err)
+			}
+		}(finished)
+	slowdown:
+		// don't allow reconciliation to run very frequently, either explicit or implicit
+		select {
+		case <-done:
+			return
+		case <-time.After(r.cooldown): // noop
+		}
+	} // for
+}
+
+// MakeComposite invokes the given ReconcilerAction funcs in sequence, aborting the sequence if reconciliation
+// is cancelled. if any other errors occur the composite reconciler will attempt to complete the
+// sequence, reporting only the last generated error.
+func MakeComposite(done <-chan struct{}, actions ...Action) Action {
+	if x := len(actions); x == 0 {
+		// programming error
+		panic("no actions specified for composite reconciler")
+	} else if x == 1 {
+		return actions[0]
+	}
+	chained := func(d bindings.SchedulerDriver, c <-chan struct{}, a, b Action) <-chan error {
+		ech := a(d, c)
+		ch := make(chan error, 1)
+		go func() {
+			select {
+			case <-done:
+			case <-c:
+			case e := <-ech:
+				if e != nil {
+					ch <- e
+					return
+				}
+				ech = b(d, c)
+				select {
+				case <-done:
+				case <-c:
+				case e := <-ech:
+					if e != nil {
+						ch <- e
+						return
+					}
+					close(ch)
+					return
+				}
+			}
+			ch <- fmt.Errorf("aborting composite reconciler action")
+		}()
+		return ch
+	}
+	result := func(d bindings.SchedulerDriver, c <-chan struct{}) <-chan error {
+		return chained(d, c, actions[0], actions[1])
+	}
+	for i := 2; i < len(actions); i++ {
+		i := i
+		next := func(d bindings.SchedulerDriver, c <-chan struct{}) <-chan error {
+			return chained(d, c, Action(result), actions[i])
+		}
+		result = next
+	}
+	return Action(result)
+}

contrib/mesos/pkg/scheduler/errors/doc.go

+/*
+Copyright 2015 The Kubernetes Authors All rights reserved.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+// Package errors contains all scheduler wide used errors
+package errors

contrib/mesos/pkg/scheduler/errors/errors.go

+/*
+Copyright 2015 The Kubernetes Authors All rights reserved.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package errors
+
+import (
+	"errors"
+)
+
+var (
+	NoSuchPodErr               = errors.New("No such pod exists")
+	NoSuchTaskErr              = errors.New("No such task exists")
+	ReconciliationCancelledErr = errors.New("explicit task reconciliation cancelled")
+	NoSuitableOffersErr        = errors.New("No suitable offers for pod/task")
+)

contrib/mesos/pkg/scheduler/ha/ha.go

@@ -112,10 +112,10 @@ type SchedulerProcess struct {
 	fin      chan struct{}
 }
 
-func New(sched bindings.Scheduler) *SchedulerProcess {
+func New(framework bindings.Scheduler) *SchedulerProcess {
 	p := &SchedulerProcess{
 		Process:   proc.New(),
-		Scheduler: sched,
+		Scheduler: framework,
 		stage:     initStage,
 		elected:   make(chan struct{}),
 		failover:  make(chan struct{}),

contrib/mesos/pkg/scheduler/integration/doc.go

+/*
+Copyright 2015 The Kubernetes Authors All rights reserved.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+// Package integration implements integration tests.
+package integration

contrib/mesos/pkg/scheduler/meta/annotations.go

@@ -25,7 +25,6 @@ const (
 	TaskIdKey                = "k8s.mesosphere.io/taskId"
 	SlaveIdKey               = "k8s.mesosphere.io/slaveId"
 	OfferIdKey               = "k8s.mesosphere.io/offerId"
-	ExecutorIdKey            = "k8s.mesosphere.io/executorId"
 	PortMappingKeyPrefix     = "k8s.mesosphere.io/port_"
 	PortMappingKeyFormat     = PortMappingKeyPrefix + "%s_%d"
 	PortNameMappingKeyPrefix = "k8s.mesosphere.io/portName_"

contrib/mesos/pkg/scheduler/podtask/pod_task.go

@@ -18,6 +18,7 @@ package podtask
 
 import (
 	"fmt"
+	"strings"
 	"time"
 
 	"github.com/gogo/protobuf/proto"
@@ -62,7 +63,6 @@ type T struct {
 	UpdatedTime time.Time // time of the most recent StatusUpdate we've seen from the mesos master
 
 	podStatus  api.PodStatus
-	executor   *mesos.ExecutorInfo // readonly
 	podKey     string
 	launchTime time.Time
 	bindTime   time.Time
@@ -130,21 +130,49 @@ func generateTaskName(pod *api.Pod) string {
 	return fmt.Sprintf("%s.%s.pods", pod.Name, ns)
 }
 
-func (t *T) BuildTaskInfo() *mesos.TaskInfo {
+func setCommandArgument(ei *mesos.ExecutorInfo, flag, value string, create bool) {
+	argv := []string{}
+	overwrite := false
+	if ei.Command != nil && ei.Command.Arguments != nil {
+		argv = ei.Command.Arguments
+		for i, arg := range argv {
+			if strings.HasPrefix(arg, flag+"=") {
+				overwrite = true
+				argv[i] = flag + "=" + value
+				break
+			}
+		}
+	}
+	if !overwrite && create {
+		argv = append(argv, flag+"="+value)
+		if ei.Command == nil {
+			ei.Command = &mesos.CommandInfo{}
+		}
+		ei.Command.Arguments = argv
+	}
+}
+
+func (t *T) BuildTaskInfo(prototype *mesos.ExecutorInfo) *mesos.TaskInfo {
 	info := &mesos.TaskInfo{
 		Name:     proto.String(generateTaskName(&t.Pod)),
 		TaskId:   mutil.NewTaskID(t.ID),
 		SlaveId:  mutil.NewSlaveID(t.Spec.SlaveID),
-		Executor: t.executor,
+		Executor: proto.Clone(prototype).(*mesos.ExecutorInfo),
 		Data:     t.Spec.Data,
 		Resources: []*mesos.Resource{
 			mutil.NewScalarResource("cpus", float64(t.Spec.CPU)),
 			mutil.NewScalarResource("mem", float64(t.Spec.Memory)),
 		},
 	}
+
 	if portsResource := rangeResource("ports", t.Spec.Ports); portsResource != nil {
 		info.Resources = append(info.Resources, portsResource)
 	}
+
+	// hostname needs of the executor needs to match that of the offer, otherwise
+	// the kubelet node status checker/updater is very unhappy
+	setCommandArgument(info.Executor, "--hostname-override", t.Spec.AssignedSlave, true)
+
 	return info
 }
 
@@ -170,10 +198,7 @@ func (t *T) Has(f FlagType) (exists bool) {
 	return
 }
 
-func New(ctx api.Context, id string, pod api.Pod, executor *mesos.ExecutorInfo) (*T, error) {
-	if executor == nil {
-		return nil, fmt.Errorf("illegal argument: executor was nil")
-	}
+func New(ctx api.Context, id string, pod *api.Pod) (*T, error) {
 	key, err := MakePodKey(ctx, pod.Name)
 	if err != nil {
 		return nil, err
@@ -182,13 +207,12 @@ func New(ctx api.Context, id string, pod api.Pod, executor *mesos.ExecutorInfo)
 		id = "pod." + uuid.NewUUID().String()
 	}
 	task := &T{
-		ID:       id,
-		Pod:      pod,
-		State:    StatePending,
-		podKey:   key,
-		mapper:   MappingTypeForPod(&pod),
-		Flags:    make(map[FlagType]struct{}),
-		executor: proto.Clone(executor).(*mesos.ExecutorInfo),
+		ID:     id,
+		Pod:    *pod,
+		State:  StatePending,
+		podKey: key,
+		mapper: MappingTypeForPod(pod),
+		Flags:  make(map[FlagType]struct{}),
 	}
 	task.CreateTime = time.Now()
 	return task, nil
@@ -198,7 +222,6 @@ func (t *T) SaveRecoveryInfo(dict map[string]string) {
 	dict[annotation.TaskIdKey] = t.ID
 	dict[annotation.SlaveIdKey] = t.Spec.SlaveID
 	dict[annotation.OfferIdKey] = t.Offer.Details().Id.GetValue()
-	dict[annotation.ExecutorIdKey] = t.executor.ExecutorId.GetValue()
 }
 
 // reconstruct a task from metadata stashed in a pod entry. there are limited pod states that
@@ -256,7 +279,6 @@ func RecoverFrom(pod api.Pod) (*T, bool, error) {
 		annotation.TaskIdKey,
 		annotation.SlaveIdKey,
 		annotation.OfferIdKey,
-		annotation.ExecutorIdKey,
 	} {
 		v, found := pod.Annotations[k]
 		if !found {
@@ -271,10 +293,6 @@ func RecoverFrom(pod api.Pod) (*T, bool, error) {
 			offerId = v
 		case annotation.TaskIdKey:
 			t.ID = v
-		case annotation.ExecutorIdKey:
-			// this is nowhere near sufficient to re-launch a task, but we really just
-			// want this for tracking
-			t.executor = &mesos.ExecutorInfo{ExecutorId: mutil.NewExecutorID(v)}
 		}
 	}
 	t.Offer = offers.Expired(offerId, t.Spec.AssignedSlave, 0)

contrib/mesos/pkg/scheduler/podtask/pod_task_test.go

@@ -35,12 +35,12 @@ const (
 )
 
 func fakePodTask(id string) (*T, error) {
-	return New(api.NewDefaultContext(), "", api.Pod{
+	return New(api.NewDefaultContext(), "", &api.Pod{
 		ObjectMeta: api.ObjectMeta{
 			Name:      id,
 			Namespace: api.NamespaceDefault,
 		},
-	}, &mesos.ExecutorInfo{})
+	})
 }
 
 func TestUnlimitedResources(t *testing.T) {

contrib/mesos/pkg/scheduler/podtask/port_mapping_test.go

@@ -52,7 +52,7 @@ func TestDefaultHostPortMatching(t *testing.T) {
 			}},
 		}},
 	}
-	task, err = New(api.NewDefaultContext(), "", *pod, &mesos.ExecutorInfo{})
+	task, err = New(api.NewDefaultContext(), "", pod)
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -100,7 +100,7 @@ func TestWildcardHostPortMatching(t *testing.T) {
 			}},
 		}},
 	}
-	task, err = New(api.NewDefaultContext(), "", *pod, &mesos.ExecutorInfo{})
+	task, err = New(api.NewDefaultContext(), "", pod)
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -123,7 +123,7 @@ func TestWildcardHostPortMatching(t *testing.T) {
 			}},
 		}},
 	}
-	task, err = New(api.NewDefaultContext(), "", *pod, &mesos.ExecutorInfo{})
+	task, err = New(api.NewDefaultContext(), "", pod)
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -144,7 +144,7 @@ func TestWildcardHostPortMatching(t *testing.T) {
 			}},
 		}},
 	}
-	task, err = New(api.NewDefaultContext(), "", *pod, &mesos.ExecutorInfo{})
+	task, err = New(api.NewDefaultContext(), "", pod)
 	if err != nil {
 		t.Fatal(err)
 	}

contrib/mesos/pkg/scheduler/podtask/procurement.go

@@ -17,8 +17,6 @@ limitations under the License.
 package podtask
 
 import (
-	"strings"
-
 	log "github.com/golang/glog"
 	mesos "github.com/mesos/mesos-go/mesosproto"
 	mresource "k8s.io/kubernetes/contrib/mesos/pkg/scheduler/resource"
@@ -74,31 +72,11 @@ func ValidateProcurement(t *T, offer *mesos.Offer) error {
 	return nil
 }
 
-func setCommandArgument(ei *mesos.ExecutorInfo, flag, value string, create bool) {
-	argv := ei.Command.Arguments
-	overwrite := false
-	for i, arg := range argv {
-		if strings.HasPrefix(arg, flag+"=") {
-			overwrite = true
-			argv[i] = flag + "=" + value
-			break
-		}
-	}
-	if !overwrite && create {
-		ei.Command.Arguments = append(argv, flag+"="+value)
-	}
-}
-
 // NodeProcurement updates t.Spec in preparation for the task to be launched on the
 // slave associated with the offer.
 func NodeProcurement(t *T, offer *mesos.Offer) error {
 	t.Spec.SlaveID = offer.GetSlaveId().GetValue()
 	t.Spec.AssignedSlave = offer.GetHostname()
-
-	// hostname needs of the executor needs to match that of the offer, otherwise
-	// the kubelet node status checker/updater is very unhappy
-	setCommandArgument(t.executor, "--hostname-override", offer.GetHostname(), true)
-
 	return nil
 }
 

contrib/mesos/pkg/scheduler/queuer/doc.go

+/*
+Copyright 2015 The Kubernetes Authors All rights reserved.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+// Package queuer implements a Pod Queuer which stores and yields pods waiting
+// being scheduled.
+package queuer

contrib/mesos/pkg/scheduler/pod.go → contrib/mesos/pkg/scheduler/queuer/pod.go

@@ -14,7 +14,7 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-package scheduler
+package queuer
 
 import (
 	"fmt"
@@ -29,8 +29,12 @@ import (
 type Pod struct {
 	*api.Pod
 	deadline *time.Time
-	delay    *time.Duration
-	notify   queue.BreakChan
+	Delay    *time.Duration
+	Notify   queue.BreakChan
+}
+
+func NewPodWithDeadline(pod *api.Pod, deadline *time.Time) *Pod {
+	return &Pod{Pod: pod, deadline: deadline}
 }
 
 // implements Copyable
@@ -54,21 +58,21 @@ func (p *Pod) GetUID() string {
 
 // implements Deadlined
 func (dp *Pod) Deadline() (time.Time, bool) {
-	if dp.deadline != nil {
+	if dp.Deadline != nil {
 		return *(dp.deadline), true
 	}
 	return time.Time{}, false
 }
 
 func (dp *Pod) GetDelay() time.Duration {
-	if dp.delay != nil {
-		return *(dp.delay)
+	if dp.Delay != nil {
+		return *(dp.Delay)
 	}
 	return 0
 }
 
 func (p *Pod) Breaker() queue.BreakChan {
-	return p.notify
+	return p.Notify
 }
 
 func (p *Pod) String() string {

contrib/mesos/pkg/scheduler/queuer/queuer.go

+/*
+Copyright 2015 The Kubernetes Authors All rights reserved.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package queuer
+
+import (
+	"fmt"
+	"io"
+	"net/http"
+	"sync"
+	"time"
+
+	log "github.com/golang/glog"
+	"k8s.io/kubernetes/contrib/mesos/pkg/queue"
+	"k8s.io/kubernetes/contrib/mesos/pkg/runtime"
+	annotation "k8s.io/kubernetes/contrib/mesos/pkg/scheduler/meta"
+	"k8s.io/kubernetes/pkg/api"
+	"k8s.io/kubernetes/pkg/client/cache"
+)
+
+const (
+	enqueuePopTimeout  = 200 * time.Millisecond
+	enqueueWaitTimeout = 1 * time.Second
+	yieldPopTimeout    = 200 * time.Millisecond
+	yieldWaitTimeout   = 1 * time.Second
+)
+
+type Queuer interface {
+	InstallDebugHandlers(mux *http.ServeMux)
+	UpdatesAvailable()
+	Dequeue(id string)
+	Requeue(pod *Pod)
+	Reoffer(pod *Pod)
+
+	Yield() *api.Pod
+
+	Run(done <-chan struct{})
+}
+
+type queuer struct {
+	lock            sync.Mutex       // shared by condition variables of this struct
+	updates         queue.FIFO       // queue of pod updates to be processed
+	queue           *queue.DelayFIFO // queue of pods to be scheduled
+	deltaCond       sync.Cond        // pod changes are available for processing
+	unscheduledCond sync.Cond        // there are unscheduled pods for processing
+}
+
+func New(queue *queue.DelayFIFO, updates queue.FIFO) Queuer {
+	q := &queuer{
+		queue:   queue,
+		updates: updates,
+	}
+	q.deltaCond.L = &q.lock
+	q.unscheduledCond.L = &q.lock
+	return q
+}
+
+func (q *queuer) InstallDebugHandlers(mux *http.ServeMux) {
+	mux.HandleFunc("/debug/scheduler/podqueue", func(w http.ResponseWriter, r *http.Request) {
+		for _, x := range q.queue.List() {
+			if _, err := io.WriteString(w, fmt.Sprintf("%+v\n", x)); err != nil {
+				break
+			}
+		}
+	})
+	mux.HandleFunc("/debug/scheduler/podstore", func(w http.ResponseWriter, r *http.Request) {
+		for _, x := range q.updates.List() {
+			if _, err := io.WriteString(w, fmt.Sprintf("%+v\n", x)); err != nil {
+				break
+			}
+		}
+	})
+}
+
+// signal that there are probably pod updates waiting to be processed
+func (q *queuer) UpdatesAvailable() {
+	q.deltaCond.Broadcast()
+}
+
+// delete a pod from the to-be-scheduled queue
+func (q *queuer) Dequeue(id string) {
+	q.queue.Delete(id)
+}
+
+// re-add a pod to the to-be-scheduled queue, will not overwrite existing pod data (that
+// may have already changed).
+func (q *queuer) Requeue(pod *Pod) {
+	// use KeepExisting in case the pod has already been updated (can happen if binding fails
+	// due to constraint voilations); we don't want to overwrite a newer entry with stale data.
+	q.queue.Add(pod, queue.KeepExisting)
+	q.unscheduledCond.Broadcast()
+}
+
+// same as Requeue but calls podQueue.Offer instead of podQueue.Add
+func (q *queuer) Reoffer(pod *Pod) {
+	// use KeepExisting in case the pod has already been updated (can happen if binding fails
+	// due to constraint voilations); we don't want to overwrite a newer entry with stale data.
+	if q.queue.Offer(pod, queue.KeepExisting) {
+		q.unscheduledCond.Broadcast()
+	}
+}
+
+// spawns a go-routine to watch for unscheduled pods and queue them up
+// for scheduling. returns immediately.
+func (q *queuer) Run(done <-chan struct{}) {
+	go runtime.Until(func() {
+		log.Info("Watching for newly created pods")
+		q.lock.Lock()
+		defer q.lock.Unlock()
+
+		for {
+			// limit blocking here for short intervals so that scheduling
+			// may proceed even if there have been no recent pod changes
+			p := q.updates.Await(enqueuePopTimeout)
+			if p == nil {
+				signalled := runtime.After(q.deltaCond.Wait)
+				// we've yielded the lock
+				select {
+				case <-time.After(enqueueWaitTimeout):
+					q.deltaCond.Broadcast() // abort Wait()
+					<-signalled             // wait for lock re-acquisition
+					log.V(4).Infoln("timed out waiting for a pod update")
+				case <-signalled:
+					// we've acquired the lock and there may be
+					// changes for us to process now
+				}
+				continue
+			}
+
+			pod := p.(*Pod)
+			if recoverAssignedSlave(pod.Pod) != "" {
+				log.V(3).Infof("dequeuing assigned pod for scheduling: %v", pod.Pod.Name)
+				q.Dequeue(pod.GetUID())
+			} else {
+				// use ReplaceExisting because we are always pushing the latest state
+				now := time.Now()
+				pod.deadline = &now
+				if q.queue.Offer(pod, queue.ReplaceExisting) {
+					q.unscheduledCond.Broadcast()
+					log.V(3).Infof("queued pod for scheduling: %v", pod.Pod.Name)
+				} else {
+					log.Warningf("failed to queue pod for scheduling: %v", pod.Pod.Name)
+				}
+			}
+		}
+	}, 1*time.Second, done)
+}
+
+// implementation of scheduling plugin's NextPod func; see k8s plugin/pkg/scheduler
+func (q *queuer) Yield() *api.Pod {
+	log.V(2).Info("attempting to yield a pod")
+	q.lock.Lock()
+	defer q.lock.Unlock()
+
+	for {
+		// limit blocking here to short intervals so that we don't block the
+		// enqueuer Run() routine for very long
+		kpod := q.queue.Await(yieldPopTimeout)
+		if kpod == nil {
+			signalled := runtime.After(q.unscheduledCond.Wait)
+			// lock is yielded at this point and we're going to wait for either
+			// a timeout, or a signal that there's data
+			select {
+			case <-time.After(yieldWaitTimeout):
+				q.unscheduledCond.Broadcast() // abort Wait()
+				<-signalled                   // wait for the go-routine, and the lock
+				log.V(4).Infoln("timed out waiting for a pod to yield")
+			case <-signalled:
+				// we have acquired the lock, and there
+				// may be a pod for us to pop now
+			}
+			continue
+		}
+
+		pod := kpod.(*Pod).Pod
+		if podName, err := cache.MetaNamespaceKeyFunc(pod); err != nil {
+			log.Warningf("yield unable to understand pod object %+v, will skip: %v", pod, err)
+		} else if !q.updates.Poll(podName, queue.POP_EVENT) {
+			log.V(1).Infof("yield popped a transitioning pod, skipping: %+v", pod)
+		} else if recoverAssignedSlave(pod) != "" {
+			// should never happen if enqueuePods is filtering properly
+			log.Warningf("yield popped an already-scheduled pod, skipping: %+v", pod)
+		} else {
+			return pod
+		}
+	}
+}
+
+// recoverAssignedSlave recovers the assigned Mesos slave from a pod by searching
+// the BindingHostKey. For tasks in the registry of the scheduler, the same
+// value is stored in T.Spec.AssignedSlave. Before launching, the BindingHostKey
+// annotation is added and the executor will eventually persist that to the
+// apiserver on binding.
+func recoverAssignedSlave(pod *api.Pod) string {
+	return pod.Annotations[annotation.BindingHostKey]
+}

contrib/mesos/pkg/scheduler/scheduler_mock.go

+/*
+Copyright 2015 The Kubernetes Authors All rights reserved.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package scheduler
+
+import (
+	"sync"
+
+	"github.com/stretchr/testify/mock"
+	"k8s.io/kubernetes/contrib/mesos/pkg/offers"
+	"k8s.io/kubernetes/contrib/mesos/pkg/runtime"
+	"k8s.io/kubernetes/contrib/mesos/pkg/scheduler/podtask"
+	"time"
+)
+
+// MockScheduler implements SchedulerApi
+type MockScheduler struct {
+	sync.RWMutex
+	mock.Mock
+}
+
+func (m *MockScheduler) Run(done <-chan struct{}) {
+	_ = m.Called()
+	runtime.Until(func() {
+		time.Sleep(time.Second)
+	}, time.Second, done)
+	return
+}
+
+func (m *MockScheduler) Offers() (f offers.Registry) {
+	args := m.Called()
+	x := args.Get(0)
+	if x != nil {
+		f = x.(offers.Registry)
+	}
+	return
+}
+
+func (m *MockScheduler) Tasks() (f podtask.Registry) {
+	args := m.Called()
+	x := args.Get(0)
+	if x != nil {
+		f = x.(podtask.Registry)
+	}
+	return
+}
+
+func (m *MockScheduler) KillTask(taskId string) error {
+	args := m.Called(taskId)
+	return args.Error(0)
+}
+
+func (m *MockScheduler) LaunchTask(task *podtask.T) error {
+	args := m.Called(task)
+	return args.Error(0)
+}
+
+func (m *MockScheduler) Reconcile(task *podtask.T) {
+	_ = m.Called()
+	return
+}

contrib/mesos/pkg/scheduler/service/publish.go

@@ -42,7 +42,7 @@ func (m *SchedulerServer) newServiceWriter(stop <-chan struct{}) func() {
 				glog.Errorf("Can't create scheduler service: %v", err)
 			}
 
-			if err := m.setEndpoints(SCHEDULER_SERVICE_NAME, net.IP(m.Address), m.Port); err != nil {
+			if err := m.setEndpoints(SCHEDULER_SERVICE_NAME, net.IP(m.address), m.port); err != nil {
 				glog.Errorf("Can't create scheduler endpoints: %v", err)
 			}
 
@@ -76,8 +76,8 @@ func (m *SchedulerServer) createSchedulerServiceIfNeeded(serviceName string, ser
 			SessionAffinity: api.ServiceAffinityNone,
 		},
 	}
-	if m.ServiceAddress != nil {
-		svc.Spec.ClusterIP = m.ServiceAddress.String()
+	if m.serviceAddress != nil {
+		svc.Spec.ClusterIP = m.serviceAddress.String()
 	}
 	_, err := m.client.Services(api.NamespaceValue(ctx)).Create(svc)
 	if err != nil && errors.IsAlreadyExists(err) {

contrib/mesos/pkg/scheduler/service/service_test.go

@@ -121,8 +121,8 @@ func Test_DefaultResourceLimits(t *testing.T) {
 	assert := assert.New(t)
 
 	s := NewSchedulerServer()
-	assert.Equal(s.DefaultContainerCPULimit, mresource.DefaultDefaultContainerCPULimit)
-	assert.Equal(s.DefaultContainerMemLimit, mresource.DefaultDefaultContainerMemLimit)
+	assert.Equal(s.defaultContainerCPULimit, mresource.DefaultDefaultContainerCPULimit)
+	assert.Equal(s.defaultContainerMemLimit, mresource.DefaultDefaultContainerMemLimit)
 }
 
 func Test_StaticPods(t *testing.T) {