Restructuring mesos scheduler driver outside of Cluster

Signed-off-by: Isabel Jimenez <contact@isabeljimenez.com>

cluster/mesos/cluster.go

@@ -27,12 +27,11 @@ import (
 type Cluster struct {
 	sync.RWMutex
 
-	driver              *mesosscheduler.MesosSchedulerDriver
 	dockerEnginePort    string
 	eventHandler        cluster.EventHandler
 	master              string
 	agents              map[string]*agent
-	scheduler           *scheduler.Scheduler
+	scheduler           *Scheduler
 	TLSConfig           *tls.Config
 	options             *cluster.DriverOpts
 	offerTimeout        time.Duration
@@ -69,7 +68,6 @@ func NewCluster(scheduler *scheduler.Scheduler, TLSConfig *tls.Config, master st
 		dockerEnginePort:    defaultDockerEnginePort,
 		master:              master,
 		agents:              make(map[string]*agent),
-		scheduler:           scheduler,
 		TLSConfig:           TLSConfig,
 		options:             &options,
 		offerTimeout:        defaultOfferTimeout,
@@ -89,7 +87,6 @@ func NewCluster(scheduler *scheduler.Scheduler, TLSConfig *tls.Config, master st
 	hostname, _ := os.Hostname()
 
 	driverConfig := mesosscheduler.DriverConfig{
-		Scheduler:        cluster,
 		Framework:        &mesosproto.FrameworkInfo{Name: proto.String(frameworkName), User: &user},
 		Master:           cluster.master,
 		HostnameOverride: hostname,
@@ -137,14 +134,13 @@ func NewCluster(scheduler *scheduler.Scheduler, TLSConfig *tls.Config, master st
 		cluster.refuseTimeout = d
 	}
 
-	driver, err := mesosscheduler.NewMesosSchedulerDriver(driverConfig)
+	sched, err := NewScheduler(driverConfig, cluster, scheduler)
 	if err != nil {
 		return nil, err
 	}
 
-	cluster.driver = driver
+	cluster.scheduler = sched
-
+	status, err := sched.driver.Start()
-	status, err := driver.Start()
 	if err != nil {
 		log.Debugf("Mesos driver started, status/err %v: %v", status, err)
 		return nil, err
@@ -430,7 +426,7 @@ func (c *Cluster) addOffer(offer *mesosproto.Offer) {
 		time.Sleep(c.offerTimeout)
 		// declining Mesos offers to make them available to other Mesos services
 		if c.removeOffer(offer) {
-			if _, err := c.driver.DeclineOffer(offer.Id, &mesosproto.Filters{}); err != nil {
+			if _, err := c.scheduler.driver.DeclineOffer(offer.Id, &mesosproto.Filters{}); err != nil {
 				log.WithFields(log.Fields{"name": "mesos"}).Errorf("Error while declining offer %q: %v", offer.Id.GetValue(), err)
 			} else {
 				log.WithFields(log.Fields{"name": "mesos"}).Debugf("Offer %q declined successfully", offer.Id.GetValue())
@@ -485,7 +481,7 @@ func (c *Cluster) scheduleTask(t *task) bool {
 	refuseSeconds := c.refuseTimeout.Seconds()
 	offerFilters.RefuseSeconds = &refuseSeconds
 
-	if _, err := c.driver.LaunchTasks(offerIDs, []*mesosproto.TaskInfo{&t.TaskInfo}, offerFilters); err != nil {
+	if _, err := c.scheduler.driver.LaunchTasks(offerIDs, []*mesosproto.TaskInfo{&t.TaskInfo}, offerFilters); err != nil {
 		// TODO: Do not erase all the offers, only the one used
 		for _, offer := range s.offers {
 			c.removeOffer(offer)

cluster/mesos/scheduler.go

@@ -5,32 +5,58 @@ import (
 
 	log "github.com/Sirupsen/logrus"
 	"github.com/docker/swarm/cluster"
+	"github.com/docker/swarm/scheduler"
 	"github.com/mesos/mesos-go/mesosproto"
 	mesosscheduler "github.com/mesos/mesos-go/scheduler"
 )
 
+// Scheduler structure for mesos driver
+type Scheduler struct {
+	scheduler.Scheduler
+
+	driver  *mesosscheduler.MesosSchedulerDriver
+	cluster *Cluster
+}
+
+// NewScheduler for Scheduler mesos driver creation
+func NewScheduler(config mesosscheduler.DriverConfig, cluster *Cluster, sched *scheduler.Scheduler) (*Scheduler, error) {
+	scheduler := Scheduler{
+		Scheduler: *sched,
+		cluster:   cluster,
+	}
+
+	config.Scheduler = &scheduler
+	driver, err := mesosscheduler.NewMesosSchedulerDriver(config)
+	if err != nil {
+		return nil, err
+	}
+	scheduler.driver = driver
+	return &scheduler, nil
+}
+
 // Registered method for registered mesos framework
-func (c *Cluster) Registered(driver mesosscheduler.SchedulerDriver, fwID *mesosproto.FrameworkID, masterInfo *mesosproto.MasterInfo) {
+func (s *Scheduler) Registered(driver mesosscheduler.SchedulerDriver, fwID *mesosproto.FrameworkID, masterInfo *mesosproto.MasterInfo) {
 	log.WithFields(log.Fields{"name": "mesos", "frameworkId": fwID.GetValue()}).Debug("Framework registered")
 }
 
 // Reregistered method for registered mesos framework
-func (c *Cluster) Reregistered(mesosscheduler.SchedulerDriver, *mesosproto.MasterInfo) {
+func (s *Scheduler) Reregistered(mesosscheduler.SchedulerDriver, *mesosproto.MasterInfo) {
 	log.WithFields(log.Fields{"name": "mesos"}).Debug("Framework re-registered")
 }
 
 // Disconnected method
-func (c *Cluster) Disconnected(mesosscheduler.SchedulerDriver) {
+func (s *Scheduler) Disconnected(mesosscheduler.SchedulerDriver) {
 	log.WithFields(log.Fields{"name": "mesos"}).Debug("Framework disconnected")
 }
 
 // ResourceOffers method
-func (c *Cluster) ResourceOffers(_ mesosscheduler.SchedulerDriver, offers []*mesosproto.Offer) {
+func (s *Scheduler) ResourceOffers(_ mesosscheduler.SchedulerDriver, offers []*mesosproto.Offer) {
 	log.WithFields(log.Fields{"name": "mesos", "offers": len(offers)}).Debug("Offers received")
 
 	for _, offer := range offers {
 		agentID := offer.SlaveId.GetValue()
-		dockerPort := c.dockerEnginePort
+		dockerPort := s.cluster.dockerEnginePort
+
 		for _, attribute := range offer.GetAttributes() {
 			if attribute.GetName() == dockerPortAttribute {
 				switch attribute.GetType() {
@@ -41,40 +67,43 @@ func (c *Cluster) ResourceOffers(_ mesosscheduler.SchedulerDriver, offers []*mes
 				}
 			}
 		}
-		s, ok := c.agents[agentID]
+
+		a, ok := s.cluster.agents[agentID]
 		if !ok {
-			engine := cluster.NewEngine(*offer.Hostname+":"+dockerPort, 0, c.engineOpts)
+			engine := cluster.NewEngine(*offer.Hostname+":"+dockerPort, 0, s.cluster.engineOpts)
-			if err := engine.Connect(c.TLSConfig); err != nil {
+			if err := engine.Connect(s.cluster.TLSConfig); err != nil {
 				log.Error(err)
 			} else {
 				// Set engine state to healthy and start refresh loop
 				engine.ValidationComplete()
-				s = newAgent(agentID, engine)
+				a = newAgent(agentID, engine)
-				c.agents[agentID] = s
+				s.cluster.agents[agentID] = a
-				if err := s.engine.RegisterEventHandler(c); err != nil {
+				if err := a.engine.RegisterEventHandler(s.cluster); err != nil {
 					log.Error(err)
 				}
 			}
+
 		}
-		c.addOffer(offer)
+		s.cluster.addOffer(offer)
+
 	}
-	go c.pendingTasks.Process()
+	go s.cluster.pendingTasks.Process()
 }
 
 // OfferRescinded method
-func (c *Cluster) OfferRescinded(mesosscheduler.SchedulerDriver, *mesosproto.OfferID) {
+func (s *Scheduler) OfferRescinded(mesosscheduler.SchedulerDriver, *mesosproto.OfferID) {
 }
 
 // StatusUpdate method
-func (c *Cluster) StatusUpdate(_ mesosscheduler.SchedulerDriver, taskStatus *mesosproto.TaskStatus) {
+func (s *Scheduler) StatusUpdate(_ mesosscheduler.SchedulerDriver, taskStatus *mesosproto.TaskStatus) {
 	log.WithFields(log.Fields{"name": "mesos", "state": taskStatus.State.String()}).Debug("Status update")
 	taskID := taskStatus.TaskId.GetValue()
 	agentID := taskStatus.SlaveId.GetValue()
-	s, ok := c.agents[agentID]
+	a, ok := s.cluster.agents[agentID]
 	if !ok {
 		return
 	}
-	if task, ok := s.tasks[taskID]; ok {
+	if task, ok := a.tasks[taskID]; ok {
 		task.sendStatus(taskStatus)
 	} else {
 		var reason = ""
@@ -92,18 +121,18 @@ func (c *Cluster) StatusUpdate(_ mesosscheduler.SchedulerDriver, taskStatus *mes
 }
 
 // FrameworkMessage method
-func (c *Cluster) FrameworkMessage(mesosscheduler.SchedulerDriver, *mesosproto.ExecutorID, *mesosproto.SlaveID, string) {
+func (s *Scheduler) FrameworkMessage(mesosscheduler.SchedulerDriver, *mesosproto.ExecutorID, *mesosproto.SlaveID, string) {
 }
 
 // SlaveLost method
-func (c *Cluster) SlaveLost(mesosscheduler.SchedulerDriver, *mesosproto.SlaveID) {
+func (s *Scheduler) SlaveLost(mesosscheduler.SchedulerDriver, *mesosproto.SlaveID) {
 }
 
 // ExecutorLost method
-func (c *Cluster) ExecutorLost(mesosscheduler.SchedulerDriver, *mesosproto.ExecutorID, *mesosproto.SlaveID, int) {
+func (s *Scheduler) ExecutorLost(mesosscheduler.SchedulerDriver, *mesosproto.ExecutorID, *mesosproto.SlaveID, int) {
 }
 
 // Error method
-func (c *Cluster) Error(d mesosscheduler.SchedulerDriver, msg string) {
+func (s *Scheduler) Error(d mesosscheduler.SchedulerDriver, msg string) {
 	log.WithFields(log.Fields{"name": "mesos"}).Error(msg)
 }