Enhancements to scheduler priority functions

pkg/scheduler/listers.go

@@ -17,6 +17,8 @@ limitations under the License.
 package scheduler
 
 import (
+	"fmt"
+
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/api"
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/labels"
 )
@@ -52,3 +54,38 @@ func (f FakePodLister) List(s labels.Selector) (selected []api.Pod, err error) {
 	}
 	return selected, nil
 }
+
+// ServiceLister interface represents anything that can produce a list of services; the list is consumed by a scheduler.
+type ServiceLister interface {
+	// Lists all the services
+	ListServices() (api.ServiceList, error)
+	// Gets the service for the given pod
+	GetPodService(api.Pod) (api.Service, error)
+}
+
+// FakeServiceLister implements ServiceLister on []api.Service for test purposes.
+type FakeServiceLister []api.Service
+
+// FakeServiceLister returns api.ServiceList, the list of all services.
+func (f FakeServiceLister) ListServices() (api.ServiceList, error) {
+	return api.ServiceList{Items: f}, nil
+}
+
+// GetPodService gets the service that has the selector that can match the labels on the given pod
+// We are assuming a single service per pod.
+// In case of multiple services per pod, the first service found is returned
+func (f FakeServiceLister) GetPodService(pod api.Pod) (service api.Service, err error) {
+	var selector labels.Selector
+
+	for _, service := range f {
+		// consider only services that are in the same namespace as the pod
+		if service.Namespace != pod.Namespace {
+			continue
+		}
+		selector = labels.Set(service.Spec.Selector).AsSelector()
+		if selector.Matches(labels.Set(pod.Labels)) {
+			return service, nil
+		}
+	}
+	return service, fmt.Errorf("Could not find service for pod %s in namespace %s with labels: %v", pod.Name, pod.Namespace, pod.Labels)
+}

pkg/scheduler/spreading.go

@@ -17,28 +17,44 @@ limitations under the License.
 package scheduler
 
 import (
-	"math/rand"
-
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/api"
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/labels"
 )
 
+type ServiceSpread struct {
+	serviceLister ServiceLister
+}
+
+func NewServiceSpreadPriority(serviceLister ServiceLister) PriorityFunction {
+	serviceSpread := &ServiceSpread{
+		serviceLister: serviceLister,
+	}
+	return serviceSpread.CalculateSpreadPriority
+}
+
 // CalculateSpreadPriority spreads pods by minimizing the number of pods on the same machine with the same labels.
 // Importantly, if there are services in the system that span multiple heterogenous sets of pods, this spreading priority
 // may not provide optimal spreading for the members of that Service.
 // TODO: consider if we want to include Service label sets in the scheduling priority.
-func CalculateSpreadPriority(pod api.Pod, podLister PodLister, minionLister MinionLister) (HostPriorityList, error) {
-	pods, err := podLister.List(labels.SelectorFromSet(pod.Labels))
-	if err != nil {
-		return nil, err
+func (s *ServiceSpread) CalculateSpreadPriority(pod api.Pod, podLister PodLister, minionLister MinionLister) (HostPriorityList, error) {
+	var maxCount int
+	var pods []api.Pod
+	var err error
+
+	service, err := s.serviceLister.GetPodService(pod)
+	if err == nil {
+		selector := labels.SelectorFromSet(service.Spec.Selector)
+		pods, err = podLister.ListPods(selector)
+		if err != nil {
+			return nil, err
+		}
 	}
+
 	minions, err := minionLister.List()
 	if err != nil {
 		return nil, err
 	}
 
-	var maxCount int
-	var fScore float32 = 10.0
 	counts := map[string]int{}
 	if len(pods) > 0 {
 		for _, pod := range pods {
@@ -54,6 +70,8 @@ func CalculateSpreadPriority(pod api.Pod, podLister PodLister, minionLister Mini
 	//score int - scale of 0-10
 	// 0 being the lowest priority and 10 being the highest
 	for _, minion := range minions.Items {
+		// initializing to the default/max minion score of 10
+		fScore := float32(10)
 		if maxCount > 0 {
 			fScore = 10 * (float32(maxCount-counts[minion.Name]) / float32(maxCount))
 		}
@@ -62,6 +80,77 @@ func CalculateSpreadPriority(pod api.Pod, podLister PodLister, minionLister Mini
 	return result, nil
 }
 
-func NewSpreadingScheduler(podLister PodLister, minionLister MinionLister, predicates []FitPredicate, random *rand.Rand) Scheduler {
-	return NewGenericScheduler(predicates, []PriorityConfig{{Function: CalculateSpreadPriority, Weight: 1}}, podLister, random)
+type ZoneSpread struct {
+	serviceLister ServiceLister
+	zoneLabel     string
+}
+
+func NewZoneSpreadPriority(serviceLister ServiceLister, zoneLabel string) PriorityFunction {
+	zoneSpread := &ZoneSpread{
+		serviceLister: serviceLister,
+		zoneLabel:     zoneLabel,
+	}
+	return zoneSpread.ZoneSpreadPriority
+}
+
+func (z *ZoneSpread) ZoneSpreadPriority(pod api.Pod, podLister PodLister, minionLister MinionLister) (HostPriorityList, error) {
+	var service api.Service
+	var pods []api.Pod
+	var err error
+
+	service, err = z.serviceLister.GetPodService(pod)
+	if err == nil {
+		selector := labels.SelectorFromSet(service.Spec.Selector)
+		pods, err = podLister.ListPods(selector)
+		if err != nil {
+			return nil, err
+		}
+	}
+
+	minions, err := minionLister.List()
+	if err != nil {
+		return nil, err
+	}
+
+	// find the zones that the minions belong to
+	openMinions := []string{}
+	zonedMinions := map[string]string{}
+	for _, minion := range minions.Items {
+		if labels.Set(minion.Labels).Has(z.zoneLabel) {
+			zone := labels.Set(minion.Labels).Get(z.zoneLabel)
+			zonedMinions[minion.Name] = zone
+		} else {
+			openMinions = append(openMinions, minion.Name)
+		}
+	}
+
+	podCounts := map[string]int{}
+	numServicePods := len(pods)
+	if numServicePods > 0 {
+		for _, pod := range pods {
+			zone, exists := zonedMinions[pod.Status.Host]
+			if !exists {
+				continue
+			}
+			podCounts[zone]++
+		}
+	}
+
+	result := []HostPriority{}
+	//score int - scale of 0-10
+	// 0 being the lowest priority and 10 being the highest
+	for minion := range zonedMinions {
+		// initializing to the default/max minion score of 10
+		fScore := float32(10)
+		if numServicePods > 0 {
+			fScore = 10 * (float32(numServicePods-podCounts[zonedMinions[minion]]) / float32(numServicePods))
+		}
+		result = append(result, HostPriority{host: minion, score: int(fScore)})
+	}
+	// add the open minions with a score of 0
+	for _, minion := range openMinions {
+		result = append(result, HostPriority{host: minion, score: 0})
+	}
+
+	return result, nil
 }

plugin/pkg/scheduler/algorithmprovider/defaults/defaults.go

@@ -31,11 +31,11 @@ func defaultPredicates() util.StringSet {
 	return util.NewStringSet(
 		// Fit is defined based on the absence of port conflicts.
 		factory.RegisterFitPredicate("PodFitsPorts", algorithm.PodFitsPorts),
-		// Fit is determined by resource availability
+		// Fit is determined by resource availability.
 		factory.RegisterFitPredicate("PodFitsResources", algorithm.NewResourceFitPredicate(factory.MinionLister)),
-		// Fit is determined by non-conflicting disk volumes
+		// Fit is determined by non-conflicting disk volumes.
 		factory.RegisterFitPredicate("NoDiskConflict", algorithm.NoDiskConflict),
-		// Fit is determined by node selector query
+		// Fit is determined by node selector query.
 		factory.RegisterFitPredicate("MatchNodeSelector", algorithm.NewSelectorMatchPredicate(factory.MinionLister)),
 		// Fit is determined by the presence of the Host parameter and a string match
 		factory.RegisterFitPredicate("HostName", algorithm.PodFitsHost),
@@ -46,8 +46,11 @@ func defaultPriorities() util.StringSet {
 	return util.NewStringSet(
 		// Prioritize nodes by least requested utilization.
 		factory.RegisterPriorityFunction("LeastRequestedPriority", algorithm.LeastRequestedPriority, 1),
-		// spreads pods by minimizing the number of pods on the same minion with the same labels.
-		factory.RegisterPriorityFunction("SpreadingPriority", algorithm.CalculateSpreadPriority, 1),
+		// spreads pods by minimizing the number of pods (belonging to the same service) on the same minion.
+		factory.RegisterPriorityFunction("ServiceSpreadingPriority", algorithm.NewServiceSpreadPriority(factory.ServiceLister), 1),
+		// spreads pods belonging to the same service across minions in different zones
+		// TODO: remove the hardcoding of the "zone" label and move it to a constant
+		factory.RegisterPriorityFunction("ZoneSpreadingPriority", algorithm.NewZoneSpreadPriority(factory.ServiceLister, "zone"), 1),
 		// EqualPriority is a prioritizer function that gives an equal weight of one to all minions
 		factory.RegisterPriorityFunction("EqualPriority", algorithm.EqualPriority, 0),
 	)

plugin/pkg/scheduler/factory/factory.go

@@ -35,8 +35,9 @@ import (
 )
 
 var (
-	PodLister    = &cache.StoreToPodLister{cache.NewStore()}
-	MinionLister = &cache.StoreToNodeLister{cache.NewStore()}
+	PodLister     = &cache.StoreToPodLister{cache.NewStore()}
+	MinionLister  = &cache.StoreToNodeLister{cache.NewStore()}
+	ServiceLister = &cache.StoreToServiceLister{cache.NewStore()}
 )
 
 // ConfigFactory knows how to fill out a scheduler config with its support functions.
@@ -48,15 +49,18 @@ type ConfigFactory struct {
 	PodLister *cache.StoreToPodLister
 	// a means to list all minions
 	MinionLister *cache.StoreToNodeLister
+	// a means to list all services
+	ServiceLister *cache.StoreToServiceLister
 }
 
 // NewConfigFactory initializes the factory.
 func NewConfigFactory(client *client.Client) *ConfigFactory {
 	return &ConfigFactory{
-		Client:       client,
-		PodQueue:     cache.NewFIFO(),
-		PodLister:    PodLister,
-		MinionLister: MinionLister,
+		Client:        client,
+		PodQueue:      cache.NewFIFO(),
+		PodLister:     PodLister,
+		MinionLister:  MinionLister,
+		ServiceLister: ServiceLister,
 	}
 }
 
@@ -105,6 +109,11 @@ func (f *ConfigFactory) CreateFromKeys(predicateKeys, priorityKeys util.StringSe
 		cache.NewPoller(f.pollMinions, 10*time.Second, f.MinionLister.Store).Run()
 	}
 
+	// Watch and cache all service objects. Scheduler needs to find all pods
+	// created by the same service, so that it can spread them correctly.
+	// Cache this locally.
+	cache.NewReflector(f.createServiceLW(), &api.Service{}, f.ServiceLister.Store).Run()
+
 	r := rand.New(rand.NewSource(time.Now().UnixNano()))
 
 	algo := algorithm.NewGenericScheduler(predicateFuncs, priorityConfigs, f.PodLister, r)
@@ -178,6 +187,15 @@ func (factory *ConfigFactory) pollMinions() (cache.Enumerator, error) {
 	return &nodeEnumerator{list}, nil
 }
 
+// createServiceLW returns a listWatch that gets all changes to services.
+func (factory *ConfigFactory) createServiceLW() *listWatch {
+	return &listWatch{
+		client:        factory.Client,
+		fieldSelector: parseSelectorOrDie(""),
+		resource:      "services",
+	}
+}
+
 func (factory *ConfigFactory) makeDefaultErrorFunc(backoff *podBackoff, podQueue *cache.FIFO) func(pod *api.Pod, err error) {
 	return func(pod *api.Pod, err error) {
 		glog.Errorf("Error scheduling %v %v: %v; retrying", pod.Namespace, pod.Name, err)
@@ -208,6 +226,35 @@ type nodeEnumerator struct {
 	*api.NodeList
 }
 
+// storeToServiceLister turns a store into a service lister. The store must contain (only) services.
+type storeToServiceLister struct {
+	cache.Store
+}
+
+func (s *storeToServiceLister) ListServices() (services api.ServiceList, err error) {
+	for _, m := range s.List() {
+		services.Items = append(services.Items, *(m.(*api.Service)))
+	}
+	return services, nil
+}
+
+func (s *storeToServiceLister) GetPodService(pod api.Pod) (service api.Service, err error) {
+	var selector labels.Selector
+
+	for _, m := range s.List() {
+		service = *m.(*api.Service)
+		// consider only services that are in the same namespace as the pod
+		if service.Namespace != pod.Namespace {
+			continue
+		}
+		selector = labels.Set(service.Spec.Selector).AsSelector()
+		if selector.Matches(labels.Set(pod.Labels)) {
+			return service, nil
+		}
+	}
+	return service, fmt.Errorf("Could not find service for pod %s in namespace %s with labels: %v", pod.Name, pod.Namespace, pod.Labels)
+}
+
 // Len returns the number of items in the node list.
 func (ne *nodeEnumerator) Len() int {
 	if ne.NodeList == nil {