Retrofitted all schedulers to now pick the hostname from the offer and add it to constants.Hosts

schedulers/binPackSortedWattsSortedOffers.go

@@ -1,6 +1,7 @@
 package schedulers
 
 import (
+	"bitbucket.org/sunybingcloud/electron/constants"
 	"bitbucket.org/sunybingcloud/electron/def"
 	"bitbucket.org/sunybingcloud/electron/utilities/mesosUtils"
 	"bitbucket.org/sunybingcloud/electron/utilities/offerUtils"
@@ -127,6 +128,10 @@ func (s *BinPackSortedWattsSortedOffers) ResourceOffers(driver sched.SchedulerDr
 	log.Println("Sorted Offers:")
 	for i := 0; i < len(offers); i++ {
 		offer := offers[i]
+		if _, ok := constants.Hosts[offer.GetHostname()]; !ok {
+			log.Printf("New host found. Adding host [%s]", offer.GetHostname())
+			constants.Hosts[offer.GetHostname()] = struct{}{}
+		}
 		offerCPU, _, _ := offerUtils.OfferAgg(offer)
 		log.Printf("Offer[%s].CPU = %f\n", offer.GetHostname(), offerCPU)
 	}

schedulers/binpackedpistoncapping.go

@@ -210,6 +210,10 @@ func (s *BinPackedPistonCapper) ResourceOffers(driver sched.SchedulerDriver, off
 
 	// retrieving the total power for each host in the offers
 	for _, offer := range offers {
+		if _, ok := constants.Hosts[offer.GetHostname()]; !ok {
+			log.Printf("New host found. Adding host [%s]", offer.GetHostname())
+			constants.Hosts[offer.GetHostname()] = struct{}{}
+		}
 		if _, ok := s.totalPower[*offer.Hostname]; !ok {
 			_, _, offerWatts := offerUtils.OfferAgg(offer)
 			s.totalPower[*offer.Hostname] = offerWatts

schedulers/binpacksortedwatts.go

@@ -1,6 +1,7 @@
 package schedulers
 
 import (
+	"bitbucket.org/sunybingcloud/electron/constants"
 	"bitbucket.org/sunybingcloud/electron/def"
 	"bitbucket.org/sunybingcloud/electron/utilities/mesosUtils"
 	"bitbucket.org/sunybingcloud/electron/utilities/offerUtils"
@@ -120,6 +121,10 @@ func (s *BinPackSortedWatts) ResourceOffers(driver sched.SchedulerDriver, offers
 	log.Printf("Received %d resource offers", len(offers))
 
 	for _, offer := range offers {
+		if _, ok := constants.Hosts[offer.GetHostname()]; !ok {
+			log.Printf("New host found. Adding host [%s]", offer.GetHostname())
+			constants.Hosts[offer.GetHostname()] = struct{}{}
+		}
 		select {
 		case <-s.Shutdown:
 			log.Println("Done scheduling tasks: declining offer on [", offer.GetHostname(), "]")

schedulers/bottomHeavy.go

@@ -282,6 +282,10 @@ func (s *BottomHeavy) ResourceOffers(driver sched.SchedulerDriver, offers []*mes
 	offersLightPowerClasses := []*mesos.Offer{}
 
 	for _, offer := range offers {
+		if _, ok := constants.Hosts[offer.GetHostname()]; !ok {
+			log.Printf("New host found. Adding host [%s]", offer.GetHostname())
+			constants.Hosts[offer.GetHostname()] = struct{}{}
+		}
 		select {
 		case <-s.Shutdown:
 			log.Println("Done scheduling tasks: declining offer on [", offer.GetHostname(), "]")

schedulers/bpswMaxMin.go

@@ -1,6 +1,7 @@
 package schedulers
 
 import (
+	"bitbucket.org/sunybingcloud/electron/constants"
 	"bitbucket.org/sunybingcloud/electron/def"
 	"bitbucket.org/sunybingcloud/electron/utilities/mesosUtils"
 	"bitbucket.org/sunybingcloud/electron/utilities/offerUtils"
@@ -164,6 +165,10 @@ func (s *BPSWMaxMinWatts) ResourceOffers(driver sched.SchedulerDriver, offers []
 	log.Printf("Received %d resource offers", len(offers))
 
 	for _, offer := range offers {
+		if _, ok := constants.Hosts[offer.GetHostname()]; !ok {
+			log.Printf("New host found. Adding host [%s]", offer.GetHostname())
+			constants.Hosts[offer.GetHostname()] = struct{}{}
+		}
 		select {
 		case <-s.Shutdown:
 			log.Println("Done scheduling tasks: declining offer on [", offer.GetHostname(), "]")

schedulers/bpswMaxMinPistonCapping.go

@@ -261,6 +261,10 @@ func (s *BPSWMaxMinPistonCapping) ResourceOffers(driver sched.SchedulerDriver, o
 	log.Printf("Received %d resource offers", len(offers))
 
 	for _, offer := range offers {
+		if _, ok := constants.Hosts[offer.GetHostname()]; !ok {
+			log.Printf("New host found. Adding host [%s]", offer.GetHostname())
+			constants.Hosts[offer.GetHostname()] = struct{}{}
+		}
 		select {
 		case <-s.Shutdown:
 			log.Println("Done scheduling tasks: declining offer on [", offer.GetHostname(), "]")

schedulers/bpswMaxMinProacCC.go

@@ -164,7 +164,7 @@ func (s *BPSWMaxMinProacCC) startCapping() {
 						// updating cap value
 						bpMaxMinProacCCCapValue = bpMaxMinProacCCNewCapValue
 						if bpMaxMinProacCCCapValue > 0.0 {
-							for _, host := range constants.Hosts {
+							for host, _ := range constants.Hosts {
 								// Rounding cap value to nearest int
 								if err := rapl.Cap(host, "rapl", float64(int(math.Floor(bpMaxMinProacCCCapValue+0.5)))); err != nil {
 									log.Println(err)
@@ -190,7 +190,7 @@ func (s *BPSWMaxMinProacCC) startRecapping() {
 				bpMaxMinProacCCMutex.Lock()
 				// If stopped performing cluster-wide capping, then we need to recap.
 				if s.isRecapping && bpMaxMinProacCCRecapValue > 0.0 {
-					for _, host := range constants.Hosts {
+					for host, _ := range constants.Hosts {
 						// Rounding the recap value to the nearest int
 						if err := rapl.Cap(host, "rapl", float64(int(math.Floor(bpMaxMinProacCCRecapValue+0.5)))); err != nil {
 							log.Println(err)
@@ -300,6 +300,10 @@ func (s *BPSWMaxMinProacCC) ResourceOffers(driver sched.SchedulerDriver, offers
 
 	// retrieving the available power for all the hosts in the offers.
 	for _, offer := range offers {
+		if _, ok := constants.Hosts[offer.GetHostname()]; !ok {
+			log.Printf("New host found. Adding host [%s]", offer.GetHostname())
+			constants.Hosts[offer.GetHostname()] = struct{}{}
+		}
 		_, _, offerWatts := offerUtils.OfferAgg(offer)
 		s.availablePower[*offer.Hostname] = offerWatts
 		// setting total power if the first time

schedulers/firstfit.go

@@ -1,6 +1,7 @@
 package schedulers
 
 import (
+	"bitbucket.org/sunybingcloud/electron/constants"
 	"bitbucket.org/sunybingcloud/electron/def"
 	"bitbucket.org/sunybingcloud/electron/utilities/mesosUtils"
 	"bitbucket.org/sunybingcloud/electron/utilities/offerUtils"
@@ -119,6 +120,10 @@ func (s *FirstFit) ResourceOffers(driver sched.SchedulerDriver, offers []*mesos.
 	log.Printf("Received %d resource offers", len(offers))
 
 	for _, offer := range offers {
+		if _, ok := constants.Hosts[offer.GetHostname()]; !ok {
+			log.Printf("New host found. Adding host [%s]", offer.GetHostname())
+			constants.Hosts[offer.GetHostname()] = struct{}{}
+		}
 		select {
 		case <-s.Shutdown:
 			log.Println("Done scheduling tasks: declining offer on [", offer.GetHostname(), "]")

schedulers/firstfitProacCC.go

@@ -164,7 +164,7 @@ func (s *FirstFitProacCC) startCapping() {
 				// Need to cap the cluster to the fcfsCurrentCapValue.
 				fcfsMutex.Lock()
 				if fcfsCurrentCapValue > 0.0 {
-					for _, host := range constants.Hosts {
+					for host, _ := range constants.Hosts {
 						// Rounding curreCapValue to the nearest int.
 						if err := rapl.Cap(host, "rapl", float64(int(math.Floor(fcfsCurrentCapValue+0.5)))); err != nil {
 							log.Println(err)
@@ -188,7 +188,7 @@ func (s *FirstFitProacCC) startRecapping() {
 				fcfsMutex.Lock()
 				// If stopped performing cluster wide capping then we need to explicitly cap the entire cluster.
 				if s.isRecapping && fcfsRecapValue > 0.0 {
-					for _, host := range constants.Hosts {
+					for host, _ := range constants.Hosts {
 						// Rounding curreCapValue to the nearest int.
 						if err := rapl.Cap(host, "rapl", float64(int(math.Floor(fcfsRecapValue+0.5)))); err != nil {
 							log.Println(err)
@@ -233,6 +233,10 @@ func (s *FirstFitProacCC) ResourceOffers(driver sched.SchedulerDriver, offers []
 
 	// retrieving the available power for all the hosts in the offers.
 	for _, offer := range offers {
+		if _, ok := constants.Hosts[offer.GetHostname()]; !ok {
+			log.Printf("New host found. Adding host [%s]", offer.GetHostname())
+			constants.Hosts[offer.GetHostname()] = struct{}{}
+		}
 		_, _, offer_watts := offerUtils.OfferAgg(offer)
 		s.availablePower[*offer.Hostname] = offer_watts
 		// setting total power if the first time.

schedulers/firstfitSortedOffers.go

@@ -1,6 +1,7 @@
 package schedulers
 
 import (
+	"bitbucket.org/sunybingcloud/electron/constants"
 	"bitbucket.org/sunybingcloud/electron/def"
 	"bitbucket.org/sunybingcloud/electron/utilities/mesosUtils"
 	"bitbucket.org/sunybingcloud/electron/utilities/offerUtils"
@@ -126,6 +127,10 @@ func (s *FirstFitSortedOffers) ResourceOffers(driver sched.SchedulerDriver, offe
 	log.Println("Sorted Offers:")
 	for i := 0; i < len(offers); i++ {
 		offer := offers[i]
+		if _, ok := constants.Hosts[offer.GetHostname()]; !ok {
+			log.Printf("New host found. Adding host [%s]", offer.GetHostname())
+			constants.Hosts[offer.GetHostname()] = struct{}{}
+		}
 		offerCPU, _, _ := offerUtils.OfferAgg(offer)
 		log.Printf("Offer[%s].CPU = %f\n", offer.GetHostname(), offerCPU)
 	}

schedulers/firstfitSortedWattsProacCC.go

@@ -177,7 +177,7 @@ func (s *FirstFitSortedWattsProacCC) startCapping() {
 				// Need to cap the cluster to the rankedCurrentCapValue.
 				rankedMutex.Lock()
 				if rankedCurrentCapValue > 0.0 {
-					for _, host := range constants.Hosts {
+					for host, _ := range constants.Hosts {
 						// Rounding currentCapValue to the nearest int.
 						if err := rapl.Cap(host, "rapl", float64(int(math.Floor(rankedCurrentCapValue+0.5)))); err != nil {
 							log.Println(err)
@@ -201,7 +201,7 @@ func (s *FirstFitSortedWattsProacCC) startRecapping() {
 				rankedMutex.Lock()
 				// If stopped performing cluster wide capping then we need to explicitly cap the entire cluster.
 				if s.isRecapping && rankedRecapValue > 0.0 {
-					for _, host := range constants.Hosts {
+					for host, _ := range constants.Hosts {
 						// Rounding currentCapValue to the nearest int.
 						if err := rapl.Cap(host, "rapl", float64(int(math.Floor(rankedRecapValue+0.5)))); err != nil {
 							log.Println(err)
@@ -246,6 +246,10 @@ func (s *FirstFitSortedWattsProacCC) ResourceOffers(driver sched.SchedulerDriver
 
 	// retrieving the available power for all the hosts in the offers.
 	for _, offer := range offers {
+		if _, ok := constants.Hosts[offer.GetHostname()]; !ok {
+			log.Printf("New host found. Adding host [%s]", offer.GetHostname())
+			constants.Hosts[offer.GetHostname()] = struct{}{}
+		}
 		_, _, offer_watts := offerUtils.OfferAgg(offer)
 		s.availablePower[*offer.Hostname] = offer_watts
 		// setting total power if the first time.

schedulers/firstfitSortedWattsSortedOffers.go

@@ -1,6 +1,7 @@
 package schedulers
 
 import (
+	"bitbucket.org/sunybingcloud/electron/constants"
 	"bitbucket.org/sunybingcloud/electron/def"
 	"bitbucket.org/sunybingcloud/electron/utilities/mesosUtils"
 	"bitbucket.org/sunybingcloud/electron/utilities/offerUtils"
@@ -128,6 +129,10 @@ func (s *FirstFitSortedWattsSortedOffers) ResourceOffers(driver sched.SchedulerD
 	log.Println("Sorted Offers:")
 	for i := 0; i < len(offers); i++ {
 		offer := offers[i]
+		if _, ok := constants.Hosts[offer.GetHostname()]; !ok {
+			log.Printf("New host found. Adding host [%s]", offer.GetHostname())
+			constants.Hosts[offer.GetHostname()] = struct{}{}
+		}
 		offerCPU, _, _ := offerUtils.OfferAgg(offer)
 		log.Printf("Offer[%s].CPU = %f\n", offer.GetHostname(), offerCPU)
 	}

schedulers/firstfitsortedwatts.go

@@ -1,6 +1,7 @@
 package schedulers
 
 import (
+	"bitbucket.org/sunybingcloud/electron/constants"
 	"bitbucket.org/sunybingcloud/electron/def"
 	"bitbucket.org/sunybingcloud/electron/utilities/mesosUtils"
 	"bitbucket.org/sunybingcloud/electron/utilities/offerUtils"
@@ -122,6 +123,10 @@ func (s *FirstFitSortedWatts) ResourceOffers(driver sched.SchedulerDriver, offer
 	log.Printf("Received %d resource offers", len(offers))
 
 	for _, offer := range offers {
+		if _, ok := constants.Hosts[offer.GetHostname()]; !ok {
+			log.Printf("New host found. Adding host [%s]", offer.GetHostname())
+			constants.Hosts[offer.GetHostname()] = struct{}{}
+		}
 		select {
 		case <-s.Shutdown:
 			log.Println("Done scheduling tasks: declining offer on [", offer.GetHostname(), "]")

schedulers/firstfitwattsonly.go

@@ -1,6 +1,7 @@
 package schedulers
 
 import (
+	"bitbucket.org/sunybingcloud/electron/constants"
 	"bitbucket.org/sunybingcloud/electron/def"
 	"bitbucket.org/sunybingcloud/electron/utilities/mesosUtils"
 	"bitbucket.org/sunybingcloud/electron/utilities/offerUtils"
@@ -112,6 +113,10 @@ func (s *FirstFitWattsOnly) ResourceOffers(driver sched.SchedulerDriver, offers
 	log.Printf("Received %d resource offers", len(offers))
 
 	for _, offer := range offers {
+		if _, ok := constants.Hosts[offer.GetHostname()]; !ok {
+			log.Printf("New host found. Adding host [%s]", offer.GetHostname())
+			constants.Hosts[offer.GetHostname()] = struct{}{}
+		}
 		select {
 		case <-s.Shutdown:
 			log.Println("Done scheduling tasks: declining offer on [", offer.GetHostname(), "]")