Remove Offers() dependency from ErrorHandler by moving out BreakChan factory

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

@@ -18,7 +18,6 @@ package errorhandler
 
 import (
 	log "github.com/golang/glog"
-	mesos "github.com/mesos/mesos-go/mesosproto"
 	"k8s.io/kubernetes/contrib/mesos/pkg/backoff"
 	"k8s.io/kubernetes/contrib/mesos/pkg/queue"
 	"k8s.io/kubernetes/contrib/mesos/pkg/scheduler"
@@ -38,15 +37,15 @@ type errorHandler struct {
 	sched        scheduler.Scheduler
 	backoff      *backoff.Backoff
 	qr           *queuer.Queuer
-	podScheduler podschedulers.PodScheduler
+	newBreakChan func(podKey string) queue.BreakChan
 }
 
-func New(sched scheduler.Scheduler, backoff *backoff.Backoff, qr *queuer.Queuer, podScheduler podschedulers.PodScheduler) ErrorHandler {
+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,
-		podScheduler: podScheduler,
+		newBreakChan: newBC,
 	}
 }
 
@@ -87,21 +86,7 @@ func (k *errorHandler) Error(pod *api.Pod, schedulingErr error) {
 		breakoutEarly := queue.BreakChan(nil)
 		if schedulingErr == podschedulers.NoSuitableOffersErr {
 			log.V(3).Infof("adding backoff breakout handler for pod %v", podKey)
-			breakoutEarly = queue.BreakChan(k.sched.Offers().Listen(podKey, func(offer *mesos.Offer) bool {
-				k.sched.Lock()
-				defer k.sched.Unlock()
-				switch task, state := k.sched.Tasks().Get(task.ID); 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) && k.podScheduler.FitPredicate()(task, offer, nil)
-				default:
-					// no point in continuing to check for matching offers
-					return true
-				}
-			}))
+			breakoutEarly = k.newBreakChan(podKey)
 		}
 		delay := k.backoff.Get(podKey)
 		log.V(3).Infof("requeuing pod %v with delay %v", podKey, delay)

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

@@ -20,6 +20,7 @@ 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"
@@ -75,7 +76,24 @@ func New(c *config.Config, fw framework.Framework, ps podschedulers.PodScheduler
 	core.podReconciler = podreconciler.New(core, client, q, podDeleter)
 
 	bo := backoff.New(c.InitialPodBackoff.Duration, c.MaxPodBackoff.Duration)
-	errorHandler := errorhandler.New(core, bo, q, ps)
+	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)