scheduling policies pluggable from commandline

schedulers/base.go

@@ -7,7 +7,14 @@ import (
 	"log"
 )
 
+// Implements mesos scheduler.
+type ElectronScheduler interface {
+	sched.Scheduler
+	init(opts ...schedPolicyOption)
+}
+
 type base struct {
+	ElectronScheduler
 	tasksCreated     int
 	tasksRunning     int
 	tasks            []def.Task
@@ -18,7 +25,7 @@ type base struct {
 
 	// First set of PCP values are garbage values, signal to logger to start recording when we're
 	// about to schedule a new task
-	RecordPCP bool
+	RecordPCP *bool
 
 	// This channel is closed when the program receives an interrupt,
 	// signalling that the program should shut down.
@@ -33,6 +40,16 @@ type base struct {
 	schedTrace *log.Logger
 }
 
+func (s *base) init(opts ...schedPolicyOption) {
+	for _, opt := range opts {
+		// applying options
+		if err := opt(s); err != nil {
+			log.Fatal(err)
+		}
+	}
+	s.running = make(map[string]map[string]bool)
+}
+
 func (s *base) OfferRescinded(_ sched.SchedulerDriver, offerID *mesos.OfferID) {
 	log.Printf("Offer %s rescinded", offerID)
 }

schedulers/bin-packing.go

@@ -10,7 +10,6 @@ import (
 	"github.com/mesos/mesos-go/mesosutil"
 	sched "github.com/mesos/mesos-go/scheduler"
 	"log"
-	"os"
 	"time"
 )
 
@@ -37,38 +36,20 @@ type BinPacking struct {
 	base // Type embedded to inherit common functions
 }
 
-// New elektron scheduler
-func NewBinPacking(tasks []def.Task, wattsAsAResource bool, schedTracePrefix string, classMapWatts bool) *BinPacking {
-	def.SortTasks(tasks, def.SortByWatts)
-
-	logFile, err := os.Create("./" + schedTracePrefix + "_schedTrace.log")
-	if err != nil {
-		log.Fatal(err)
-	}
-
-	s := &BinPacking{
-		base: base{
-			tasks:            tasks,
-			wattsAsAResource: wattsAsAResource,
-			classMapWatts:    classMapWatts,
-			Shutdown:         make(chan struct{}),
-			Done:             make(chan struct{}),
-			PCPLog:           make(chan struct{}),
-			running:          make(map[string]map[string]bool),
-			RecordPCP:        false,
-			schedTrace:       log.New(logFile, "", log.LstdFlags),
-		},
-	}
-	return s
+// Initialization
+func (s *BinPacking) init(opts ...schedPolicyOption) {
+	s.base.init(opts...)
+	// sorting the tasks based on watts
+	def.SortTasks(s.tasks, def.SortByWatts)
 }
 
 func (s *BinPacking) newTask(offer *mesos.Offer, task def.Task) *mesos.TaskInfo {
 	taskName := fmt.Sprintf("%s-%d", task.Name, *task.Instances)
 	s.tasksCreated++
 
-	if !s.RecordPCP {
+	if !*s.RecordPCP {
 		// Turn on logging
-		s.RecordPCP = true
+		*s.RecordPCP = true
 		time.Sleep(1 * time.Second) // Make sure we're recording by the time the first task starts
 	}
 

schedulers/first-fit.go

@@ -10,7 +10,6 @@ import (
 	"github.com/mesos/mesos-go/mesosutil"
 	sched "github.com/mesos/mesos-go/scheduler"
 	"log"
-	"os"
 	"time"
 )
 
@@ -38,37 +37,18 @@ type FirstFit struct {
 	base // Type embedded to inherit common functions
 }
 
-// New elektron scheduler
-func NewFirstFit(tasks []def.Task, wattsAsAResource bool, schedTracePrefix string, classMapWatts bool) *FirstFit {
-
-	logFile, err := os.Create("./" + schedTracePrefix + "_schedTrace.log")
-	if err != nil {
-		log.Fatal(err)
-	}
-
-	s := &FirstFit{
-		base: base{
-			tasks:            tasks,
-			wattsAsAResource: wattsAsAResource,
-			classMapWatts:    classMapWatts,
-			Shutdown:         make(chan struct{}),
-			Done:             make(chan struct{}),
-			PCPLog:           make(chan struct{}),
-			running:          make(map[string]map[string]bool),
-			RecordPCP:        false,
-			schedTrace:       log.New(logFile, "", log.LstdFlags),
-		},
-	}
-	return s
+// Initialization
+func (s *FirstFit) init(opts ...schedPolicyOption) {
+	s.base.init(opts...)
 }
 
 func (s *FirstFit) newTask(offer *mesos.Offer, task def.Task) *mesos.TaskInfo {
 	taskName := fmt.Sprintf("%s-%d", task.Name, *task.Instances)
 	s.tasksCreated++
 
-	if !s.RecordPCP {
+	if !*s.RecordPCP {
 		// Turn on logging
-		s.RecordPCP = true
+		*s.RecordPCP = true
 		time.Sleep(1 * time.Second) // Make sure we're recording by the time the first task starts
 	}
 

schedulers/helpers.go

@@ -2,8 +2,11 @@ package schedulers
 
 import (
 	"bitbucket.org/sunybingcloud/elektron/constants"
+	"bitbucket.org/sunybingcloud/elektron/def"
+	"errors"
 	"fmt"
 	"log"
+	"os"
 )
 
 func coLocated(tasks map[string]bool) {
@@ -24,3 +27,83 @@ func hostToPowerClass(hostName string) string {
 	}
 	return ""
 }
+
+// scheduler policy options to help initialize schedulers
+type schedPolicyOption func(e ElectronScheduler) error
+
+func WithTasks(ts []def.Task) schedPolicyOption {
+	return func(s ElectronScheduler) error {
+		if ts == nil {
+			return errors.New("Task[] is empty.")
+		} else {
+			s.(*base).tasks = ts
+			return nil
+		}
+	}
+}
+
+func WithWattsAsAResource(waar bool) schedPolicyOption {
+	return func(s ElectronScheduler) error {
+		s.(*base).wattsAsAResource = waar
+		return nil
+	}
+}
+
+func WithClassMapWatts(cmw bool) schedPolicyOption {
+	return func(s ElectronScheduler) error {
+		s.(*base).classMapWatts = cmw
+		return nil
+	}
+}
+
+func WithRecordPCP(recordPCP *bool) schedPolicyOption {
+	return func(s ElectronScheduler) error {
+		s.(*base).RecordPCP = recordPCP
+		return nil
+	}
+}
+
+func WithSchedTracePrefix(schedTracePrefix string) schedPolicyOption {
+	return func(s ElectronScheduler) error {
+		logFile, err := os.Create("./" + schedTracePrefix + "_schedTrace.log")
+		if err != nil {
+			return err
+		} else {
+			s.(*base).schedTrace = log.New(logFile, "", log.LstdFlags)
+			return nil
+		}
+	}
+}
+
+func WithShutdown(shutdown chan struct{}) schedPolicyOption {
+	return func(s ElectronScheduler) error {
+		if shutdown == nil {
+			return errors.New("Shutdown channel is nil.")
+		} else {
+			s.(*base).Shutdown = shutdown
+			return nil
+		}
+	}
+}
+
+func WithDone(done chan struct{}) schedPolicyOption {
+	return func(s ElectronScheduler) error {
+		if done == nil {
+			return errors.New("Done channel is nil.")
+		} else {
+			s.(*base).Done = done
+			return nil
+		}
+	}
+}
+
+func WithPCPLog(pcpLog chan struct{}) schedPolicyOption {
+	return func(s ElectronScheduler) error {
+		if pcpLog == nil {
+			return errors.New("PCPLog channel is nil.")
+		} else {
+			s.(*base).PCPLog = pcpLog
+			return nil
+		}
+	}
+}

schedulers/max-greedymins.go

@@ -10,7 +10,6 @@ import (
 	"github.com/mesos/mesos-go/mesosutil"
 	sched "github.com/mesos/mesos-go/scheduler"
 	"log"
-	"os"
 	"time"
 )
 
@@ -38,29 +37,9 @@ type MaxGreedyMins struct {
 	base //Type embedding to inherit common functions
 }
 
-// New elektron scheduler
-func NewMaxGreedyMins(tasks []def.Task, wattsAsAResource bool, schedTracePrefix string, classMapWatts bool) *MaxGreedyMins {
-	def.SortTasks(tasks, def.SortByWatts)
-
-	logFile, err := os.Create("./" + schedTracePrefix + "_schedTrace.log")
-	if err != nil {
-		log.Fatal(err)
-	}
-
-	s := &MaxGreedyMins{
-		base: base{
-			tasks:            tasks,
-			wattsAsAResource: wattsAsAResource,
-			classMapWatts:    classMapWatts,
-			Shutdown:         make(chan struct{}),
-			Done:             make(chan struct{}),
-			PCPLog:           make(chan struct{}),
-			running:          make(map[string]map[string]bool),
-			RecordPCP:        false,
-			schedTrace:       log.New(logFile, "", log.LstdFlags),
-		},
-	}
-	return s
+// Initialization
+func (s *MaxGreedyMins) init(opts ...schedPolicyOption) {
+	s.base.init(opts...)
 }
 
 func (s *MaxGreedyMins) newTask(offer *mesos.Offer, task def.Task) *mesos.TaskInfo {
@@ -68,9 +47,9 @@ func (s *MaxGreedyMins) newTask(offer *mesos.Offer, task def.Task) *mesos.TaskIn
 	s.tasksCreated++
 
 	// Start recording only when we're creating the first task
-	if !s.RecordPCP {
+	if !*s.RecordPCP {
 		// Turn on logging
-		s.RecordPCP = true
+		*s.RecordPCP = true
 		time.Sleep(1 * time.Second) // Make sure we're recording by the time the first task starts
 	}
 

schedulers/max-min.go

@@ -10,7 +10,6 @@ import (
 	"github.com/mesos/mesos-go/mesosutil"
 	sched "github.com/mesos/mesos-go/scheduler"
 	"log"
-	"os"
 	"time"
 )
 
@@ -38,29 +37,9 @@ type MaxMin struct {
 	base //Type embedding to inherit common functions
 }
 
-// New elektron scheduler
-func NewMaxMin(tasks []def.Task, wattsAsAResource bool, schedTracePrefix string, classMapWatts bool) *MaxMin {
-	def.SortTasks(tasks, def.SortByWatts)
-
-	logFile, err := os.Create("./" + schedTracePrefix + "_schedTrace.log")
-	if err != nil {
-		log.Fatal(err)
-	}
-
-	s := &MaxMin{
-		base: base{
-			tasks:            tasks,
-			wattsAsAResource: wattsAsAResource,
-			classMapWatts:    classMapWatts,
-			Shutdown:         make(chan struct{}),
-			Done:             make(chan struct{}),
-			PCPLog:           make(chan struct{}),
-			running:          make(map[string]map[string]bool),
-			RecordPCP:        false,
-			schedTrace:       log.New(logFile, "", log.LstdFlags),
-		},
-	}
-	return s
+// Initialization
+func (s *MaxMin) init(opts ...schedPolicyOption) {
+	s.base.init(opts...)
 }
 
 func (s *MaxMin) newTask(offer *mesos.Offer, task def.Task) *mesos.TaskInfo {
@@ -68,9 +47,9 @@ func (s *MaxMin) newTask(offer *mesos.Offer, task def.Task) *mesos.TaskInfo {
 	s.tasksCreated++
 
 	// Start recording only when we're creating the first task
-	if !s.RecordPCP {
+	if !*s.RecordPCP {
 		// Turn on logging
-		s.RecordPCP = true
+		*s.RecordPCP = true
 		time.Sleep(1 * time.Second) // Make sure we're recording by the time the first task starts
 	}
 

schedulers/store.go

@@ -0,0 +1,30 @@
+package schedulers
+
+import "github.com/mesos/mesos-go/scheduler"
+
+// Names of different scheduling policies.
+const (
+	ff  = "first-fit"
+	bp  = "bin-packing"
+	mgm = "max-greedymins"
+	mm  = "max-min"
+)
+
+// Scheduler class factory
+var Schedulers map[string]scheduler.Scheduler = map[string]scheduler.Scheduler{
+	ff:  &FirstFit{base: base{}},
+	bp:  &BinPacking{base: base{}},
+	mgm: &MaxGreedyMins{base: base{}},
+	mm:  &MaxMin{base: base{}},
+}
+
+// build the scheduling policy with the options being applied
+func BuildSchedPolicy(s scheduler.Scheduler, opts ...schedPolicyOption) {
+	s.(ElectronScheduler).init(opts...)
+}
+
+func SchedFactory(schedPolicyName string, opts ...schedPolicyOption) scheduler.Scheduler {
+	s := Schedulers[schedPolicyName]
+	BuildSchedPolicy(s, opts...)
+	return s
+}