Revert "Revert revert of equivalence class hash calculation in scheduler"

pkg/scheduler/algorithm/predicates/utils.go

@@ -17,8 +17,12 @@ limitations under the License.
 package predicates
 
 import (
+	"github.com/golang/glog"
 	"k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/labels"
+	"k8s.io/apimachinery/pkg/util/sets"
+	"k8s.io/kubernetes/pkg/scheduler/algorithm"
 	schedutil "k8s.io/kubernetes/pkg/scheduler/util"
 )
 
@@ -66,6 +70,69 @@ func CreateSelectorFromLabels(aL map[string]string) labels.Selector {
 	return labels.Set(aL).AsSelector()
 }
 
+// EquivalencePodGenerator is a generator of equivalence class for pod with consideration of PVC info.
+type EquivalencePodGenerator struct {
+	pvcInfo PersistentVolumeClaimInfo
+}
+
+// NewEquivalencePodGenerator returns a getEquivalencePod method with consideration of PVC info.
+func NewEquivalencePodGenerator(pvcInfo PersistentVolumeClaimInfo) algorithm.GetEquivalencePodFunc {
+	g := &EquivalencePodGenerator{
+		pvcInfo: pvcInfo,
+	}
+	return g.getEquivalencePod
+}
+
+// GetEquivalencePod returns a EquivalencePod which contains a group of pod attributes which can be reused.
+func (e *EquivalencePodGenerator) getEquivalencePod(pod *v1.Pod) interface{} {
+	// For now we only consider pods:
+	// 1. OwnerReferences is Controller
+	// 2. with same OwnerReferences
+	// 3. with same PVC claim
+	// to be equivalent
+	for _, ref := range pod.OwnerReferences {
+		if ref.Controller != nil && *ref.Controller {
+			pvcSet, err := e.getPVCSet(pod)
+			if err == nil {
+				// A pod can only belongs to one controller, so let's return.
+				return &EquivalencePod{
+					ControllerRef: ref,
+					PVCSet:        pvcSet,
+				}
+			}
+
+			// If error encountered, log warning and return nil (i.e. no equivalent pod found)
+			glog.Warningf("[EquivalencePodGenerator] for pod: %v failed due to: %v", pod.GetName(), err)
+			return nil
+		}
+	}
+	return nil
+}
+
+// getPVCSet returns a set of PVC UIDs of given pod.
+func (e *EquivalencePodGenerator) getPVCSet(pod *v1.Pod) (sets.String, error) {
+	result := sets.NewString()
+	for _, volume := range pod.Spec.Volumes {
+		if volume.PersistentVolumeClaim == nil {
+			continue
+		}
+		pvcName := volume.PersistentVolumeClaim.ClaimName
+		pvc, err := e.pvcInfo.GetPersistentVolumeClaimInfo(pod.GetNamespace(), pvcName)
+		if err != nil {
+			return nil, err
+		}
+		result.Insert(string(pvc.UID))
+	}
+
+	return result, nil
+}
+
+// EquivalencePod is a group of pod attributes which can be reused as equivalence to schedule other pods.
+type EquivalencePod struct {
+	ControllerRef metav1.OwnerReference
+	PVCSet        sets.String
+}
+
 // portsConflict check whether existingPorts and wantPorts conflict with each other
 // return true if we have a conflict
 func portsConflict(existingPorts schedutil.HostPortInfo, wantPorts []*v1.ContainerPort) bool {

pkg/scheduler/algorithm/types.go

@@ -78,6 +78,9 @@ type PredicateFailureReason interface {
 	GetReason() string
 }
 
+// GetEquivalencePodFunc is a function that gets a EquivalencePod from a pod.
+type GetEquivalencePodFunc func(pod *v1.Pod) interface{}
+
 // NodeLister interface represents anything that can list nodes for a scheduler.
 type NodeLister interface {
 	// We explicitly return []*v1.Node, instead of v1.NodeList, to avoid

pkg/scheduler/algorithmprovider/defaults/defaults.go

@@ -77,6 +77,13 @@ func init() {
 	// Fit is determined by node selector query.
 	factory.RegisterFitPredicate(predicates.MatchNodeSelectorPred, predicates.PodMatchNodeSelector)
 
+	// Use equivalence class to speed up heavy predicates phase.
+	factory.RegisterGetEquivalencePodFunction(
+		func(args factory.PluginFactoryArgs) algorithm.GetEquivalencePodFunc {
+			return predicates.NewEquivalencePodGenerator(args.PVCInfo)
+		},
+	)
+
 	// ServiceSpreadingPriority is a priority config factory that spreads pods by minimizing
 	// the number of pods (belonging to the same service) on the same node.
 	// Register the factory so that it's available, but do not include it as part of the default priorities

pkg/scheduler/core/equivalence_cache.go

@@ -23,7 +23,6 @@ import (
 	"k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/util/sets"
 	"k8s.io/kubernetes/pkg/scheduler/algorithm"
-	"k8s.io/kubernetes/pkg/scheduler/algorithm/predicates"
 	hashutil "k8s.io/kubernetes/pkg/util/hash"
 
 	"github.com/golang/glog"
@@ -38,7 +37,8 @@ const maxCacheEntries = 100
 // 2. function to get equivalence pod
 type EquivalenceCache struct {
 	sync.RWMutex
-	algorithmCache map[string]AlgorithmCache
+	getEquivalencePod algorithm.GetEquivalencePodFunc
+	algorithmCache    map[string]AlgorithmCache
 }
 
 // The AlgorithmCache stores PredicateMap with predicate name as key
@@ -62,11 +62,11 @@ func newAlgorithmCache() AlgorithmCache {
 	}
 }
 
-// NewEquivalenceCache returns EquivalenceCache to speed up predicates by caching
-// result from previous scheduling.
-func NewEquivalenceCache() *EquivalenceCache {
+// NewEquivalenceCache creates a EquivalenceCache object.
+func NewEquivalenceCache(getEquivalencePodFunc algorithm.GetEquivalencePodFunc) *EquivalenceCache {
 	return &EquivalenceCache{
-		algorithmCache: make(map[string]AlgorithmCache),
+		getEquivalencePod: getEquivalencePodFunc,
+		algorithmCache:    make(map[string]AlgorithmCache),
 	}
 }
 
@@ -191,21 +191,21 @@ func (ec *EquivalenceCache) InvalidateCachedPredicateItemForPodAdd(pod *v1.Pod,
 	// it will also fits to equivalence class of existing pods
 
 	// GeneralPredicates: will always be affected by adding a new pod
-	invalidPredicates := sets.NewString(predicates.GeneralPred)
+	invalidPredicates := sets.NewString("GeneralPredicates")
 
 	// MaxPDVolumeCountPredicate: we check the volumes of pod to make decision.
 	for _, vol := range pod.Spec.Volumes {
 		if vol.PersistentVolumeClaim != nil {
-			invalidPredicates.Insert(predicates.MaxEBSVolumeCountPred, predicates.MaxGCEPDVolumeCountPred, predicates.MaxAzureDiskVolumeCountPred)
+			invalidPredicates.Insert("MaxEBSVolumeCount", "MaxGCEPDVolumeCount", "MaxAzureDiskVolumeCount")
 		} else {
 			if vol.AWSElasticBlockStore != nil {
-				invalidPredicates.Insert(predicates.MaxEBSVolumeCountPred)
+				invalidPredicates.Insert("MaxEBSVolumeCount")
 			}
 			if vol.GCEPersistentDisk != nil {
-				invalidPredicates.Insert(predicates.MaxGCEPDVolumeCountPred)
+				invalidPredicates.Insert("MaxGCEPDVolumeCount")
 			}
 			if vol.AzureDisk != nil {
-				invalidPredicates.Insert(predicates.MaxAzureDiskVolumeCountPred)
+				invalidPredicates.Insert("MaxAzureDiskVolumeCount")
 			}
 		}
 	}
@@ -219,11 +219,9 @@ type equivalenceClassInfo struct {
 	hash uint64
 }
 
-// getEquivalenceClassInfo returns a hash of the given pod.
-// The hashing function returns the same value for any two pods that are
-// equivalent from the perspective of scheduling.
+// getEquivalenceClassInfo returns the equivalence class of given pod.
 func (ec *EquivalenceCache) getEquivalenceClassInfo(pod *v1.Pod) *equivalenceClassInfo {
-	equivalencePod := getEquivalenceHash(pod)
+	equivalencePod := ec.getEquivalencePod(pod)
 	if equivalencePod != nil {
 		hash := fnv.New32a()
 		hashutil.DeepHashObject(hash, equivalencePod)
@@ -233,60 +231,3 @@ func (ec *EquivalenceCache) getEquivalenceClassInfo(pod *v1.Pod) *equivalenceCla
 	}
 	return nil
 }
-
-// equivalencePod is the set of pod attributes which must match for two pods to
-// be considered equivalent for scheduling purposes. For correctness, this must
-// include any Pod field which is used by a FitPredicate.
-//
-// NOTE: For equivalence hash to be formally correct, lists and maps in the
-// equivalencePod should be normalized. (e.g. by sorting them) However, the
-// vast majority of equivalent pod classes are expected to be created from a
-// single pod template, so they will all have the same ordering.
-type equivalencePod struct {
-	Namespace      *string
-	Labels         map[string]string
-	Affinity       *v1.Affinity
-	Containers     []v1.Container // See note about ordering
-	InitContainers []v1.Container // See note about ordering
-	NodeName       *string
-	NodeSelector   map[string]string
-	Tolerations    []v1.Toleration
-	Volumes        []v1.Volume // See note about ordering
-}
-
-// getEquivalenceHash returns the equivalencePod for a Pod.
-func getEquivalenceHash(pod *v1.Pod) *equivalencePod {
-	ep := &equivalencePod{
-		Namespace:      &pod.Namespace,
-		Labels:         pod.Labels,
-		Affinity:       pod.Spec.Affinity,
-		Containers:     pod.Spec.Containers,
-		InitContainers: pod.Spec.InitContainers,
-		NodeName:       &pod.Spec.NodeName,
-		NodeSelector:   pod.Spec.NodeSelector,
-		Tolerations:    pod.Spec.Tolerations,
-		Volumes:        pod.Spec.Volumes,
-	}
-	// DeepHashObject considers nil and empty slices to be different. Normalize them.
-	if len(ep.Containers) == 0 {
-		ep.Containers = nil
-	}
-	if len(ep.InitContainers) == 0 {
-		ep.InitContainers = nil
-	}
-	if len(ep.Tolerations) == 0 {
-		ep.Tolerations = nil
-	}
-	if len(ep.Volumes) == 0 {
-		ep.Volumes = nil
-	}
-	// Normalize empty maps also.
-	if len(ep.Labels) == 0 {
-		ep.Labels = nil
-	}
-	if len(ep.NodeSelector) == 0 {
-		ep.NodeSelector = nil
-	}
-	// TODO(misterikkit): Also normalize nested maps and slices.
-	return ep
-}

pkg/scheduler/core/generic_scheduler.go

@@ -455,6 +455,10 @@ func podFitsOnNode(
 	var (
 		eCacheAvailable  bool
 		failedPredicates []algorithm.PredicateFailureReason
+		invalid          bool
+		fit              bool
+		reasons          []algorithm.PredicateFailureReason
+		err              error
 	)
 	predicateResults := make(map[string]HostPredicate)
 
@@ -490,54 +494,38 @@ func podFitsOnNode(
 		// when pods are nominated or their nominations change.
 		eCacheAvailable = equivCacheInfo != nil && !podsAdded
 		for _, predicateKey := range predicates.Ordering() {
-			var (
-				fit     bool
-				reasons []algorithm.PredicateFailureReason
-				err     error
-			)
 			//TODO (yastij) : compute average predicate restrictiveness to export it as Prometheus metric
 			if predicate, exist := predicateFuncs[predicateKey]; exist {
-				// Use an in-line function to guarantee invocation of ecache.Unlock()
-				// when the in-line function returns.
-				func() {
-					var invalid bool
-					if eCacheAvailable {
-						// Lock ecache here to avoid a race condition against cache invalidation invoked
-						// in event handlers. This race has existed despite locks in equivClassCacheimplementation.
-						ecache.Lock()
-						defer ecache.Unlock()
-						// PredicateWithECache will return its cached predicate results.
-						fit, reasons, invalid = ecache.PredicateWithECache(
-							pod.GetName(), info.Node().GetName(),
-							predicateKey, equivCacheInfo.hash, false)
-					}
-
-					if !eCacheAvailable || invalid {
-						// we need to execute predicate functions since equivalence cache does not work
-						fit, reasons, err = predicate(pod, metaToUse, nodeInfoToUse)
-						if err != nil {
-							return
-						}
+				if eCacheAvailable {
+					// Lock ecache here to avoid a race condition against cache invalidation invoked
+					// in event handlers. This race has existed despite locks in eCache implementation.
+					ecache.Lock()
+					// PredicateWithECache will return its cached predicate results.
+					fit, reasons, invalid = ecache.PredicateWithECache(pod.GetName(), info.Node().GetName(), predicateKey, equivCacheInfo.hash, false)
+				}
 
-						if eCacheAvailable {
-							// Store data to update equivClassCacheafter this loop.
-							if res, exists := predicateResults[predicateKey]; exists {
-								res.Fit = res.Fit && fit
-								res.FailReasons = append(res.FailReasons, reasons...)
-								predicateResults[predicateKey] = res
-							} else {
-								predicateResults[predicateKey] = HostPredicate{Fit: fit, FailReasons: reasons}
-							}
-							result := predicateResults[predicateKey]
-							ecache.UpdateCachedPredicateItem(
-								pod.GetName(), info.Node().GetName(),
-								predicateKey, result.Fit, result.FailReasons, equivCacheInfo.hash, false)
+				if !eCacheAvailable || invalid {
+					// we need to execute predicate functions since equivalence cache does not work
+					fit, reasons, err = predicate(pod, metaToUse, nodeInfoToUse)
+					if err != nil {
+						return false, []algorithm.PredicateFailureReason{}, err
+					}
+					if eCacheAvailable {
+						// Store data to update eCache after this loop.
+						if res, exists := predicateResults[predicateKey]; exists {
+							res.Fit = res.Fit && fit
+							res.FailReasons = append(res.FailReasons, reasons...)
+							predicateResults[predicateKey] = res
+						} else {
+							predicateResults[predicateKey] = HostPredicate{Fit: fit, FailReasons: reasons}
 						}
+						result := predicateResults[predicateKey]
+						ecache.UpdateCachedPredicateItem(pod.GetName(), info.Node().GetName(), predicateKey, result.Fit, result.FailReasons, equivCacheInfo.hash, false)
 					}
-				}()
+				}
 
-				if err != nil {
-					return false, []algorithm.PredicateFailureReason{}, err
+				if eCacheAvailable {
+					ecache.Unlock()
 				}
 
 				if !fit {
@@ -545,9 +533,7 @@ func podFitsOnNode(
 					failedPredicates = append(failedPredicates, reasons...)
 					// if alwaysCheckAllPredicates is false, short circuit all predicates when one predicate fails.
 					if !alwaysCheckAllPredicates {
-						glog.V(5).Infoln("since alwaysCheckAllPredicates has not been set, the predicate" +
-							"evaluation is short circuited and there are chances" +
-							"of other predicates failing as well.")
+						glog.V(5).Infoln("since alwaysCheckAllPredicates has not been set, the predicate evaluation is short circuited and there are chances of other predicates failing as well.")
 						break
 					}
 				}

pkg/scheduler/factory/factory.go

@@ -136,8 +136,6 @@ type configFactory struct {
 	alwaysCheckAllPredicates bool
 }
 
-var _ scheduler.Configurator = &configFactory{}
-
 // NewConfigFactory initializes the default implementation of a Configurator To encourage eventual privatization of the struct type, we only
 // return the interface.
 func NewConfigFactory(
@@ -1073,8 +1071,14 @@ func (c *configFactory) CreateFromKeys(predicateKeys, priorityKeys sets.String,
 	}
 
 	// Init equivalence class cache
-	if c.enableEquivalenceClassCache {
-		c.equivalencePodCache = core.NewEquivalenceCache()
+	if c.enableEquivalenceClassCache && getEquivalencePodFuncFactory != nil {
+		pluginArgs, err := c.getPluginArgs()
+		if err != nil {
+			return nil, err
+		}
+		c.equivalencePodCache = core.NewEquivalenceCache(
+			getEquivalencePodFuncFactory(*pluginArgs),
+		)
 		glog.Info("Created equivalence class cache")
 	}
 

pkg/scheduler/factory/plugins.go

@@ -75,6 +75,9 @@ type PriorityConfigFactory struct {
 	Weight            int
 }
 
+// EquivalencePodFuncFactory produces a function to get equivalence class for given pod.
+type EquivalencePodFuncFactory func(PluginFactoryArgs) algorithm.GetEquivalencePodFunc
+
 var (
 	schedulerFactoryMutex sync.Mutex
 
@@ -87,6 +90,9 @@ var (
 	// Registered metadata producers
 	priorityMetadataProducer  PriorityMetadataProducerFactory
 	predicateMetadataProducer PredicateMetadataProducerFactory
+
+	// get equivalence pod function
+	getEquivalencePodFuncFactory EquivalencePodFuncFactory
 )
 
 const (
@@ -340,6 +346,11 @@ func RegisterCustomPriorityFunction(policy schedulerapi.PriorityPolicy) string {
 	return RegisterPriorityConfigFactory(policy.Name, *pcf)
 }
 
+// RegisterGetEquivalencePodFunction registers equivalenceFuncFactory to produce equivalence class for given pod.
+func RegisterGetEquivalencePodFunction(equivalenceFuncFactory EquivalencePodFuncFactory) {
+	getEquivalencePodFuncFactory = equivalenceFuncFactory
+}
+
 // IsPriorityFunctionRegistered is useful for testing providers.
 func IsPriorityFunctionRegistered(name string) bool {
 	schedulerFactoryMutex.Lock()