Merge pull request #60753 from resouer/equiv-hash

pkg/scheduler/algorithm/predicates/utils.go

@@ -17,12 +17,8 @@ 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"
 )
 
@@ -70,69 +66,6 @@ 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,9 +78,6 @@ 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,13 +77,6 @@ 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,6 +23,7 @@ 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"
@@ -37,8 +38,7 @@ const maxCacheEntries = 100
 // 2. function to get equivalence pod
 type EquivalenceCache struct {
 	sync.RWMutex
-	getEquivalencePod algorithm.GetEquivalencePodFunc
-	algorithmCache    map[string]AlgorithmCache
+	algorithmCache map[string]AlgorithmCache
 }
 
 // The AlgorithmCache stores PredicateMap with predicate name as key
@@ -62,11 +62,11 @@ func newAlgorithmCache() AlgorithmCache {
 	}
 }
 
-// NewEquivalenceCache creates a EquivalenceCache object.
-func NewEquivalenceCache(getEquivalencePodFunc algorithm.GetEquivalencePodFunc) *EquivalenceCache {
+// NewEquivalenceCache returns EquivalenceCache to speed up predicates by caching
+// result from previous scheduling.
+func NewEquivalenceCache() *EquivalenceCache {
 	return &EquivalenceCache{
-		getEquivalencePod: getEquivalencePodFunc,
-		algorithmCache:    make(map[string]AlgorithmCache),
+		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("GeneralPredicates")
+	invalidPredicates := sets.NewString(predicates.GeneralPred)
 
 	// MaxPDVolumeCountPredicate: we check the volumes of pod to make decision.
 	for _, vol := range pod.Spec.Volumes {
 		if vol.PersistentVolumeClaim != nil {
-			invalidPredicates.Insert("MaxEBSVolumeCount", "MaxGCEPDVolumeCount", "MaxAzureDiskVolumeCount")
+			invalidPredicates.Insert(predicates.MaxEBSVolumeCountPred, predicates.MaxGCEPDVolumeCountPred, predicates.MaxAzureDiskVolumeCountPred)
 		} else {
 			if vol.AWSElasticBlockStore != nil {
-				invalidPredicates.Insert("MaxEBSVolumeCount")
+				invalidPredicates.Insert(predicates.MaxEBSVolumeCountPred)
 			}
 			if vol.GCEPersistentDisk != nil {
-				invalidPredicates.Insert("MaxGCEPDVolumeCount")
+				invalidPredicates.Insert(predicates.MaxGCEPDVolumeCountPred)
 			}
 			if vol.AzureDisk != nil {
-				invalidPredicates.Insert("MaxAzureDiskVolumeCount")
+				invalidPredicates.Insert(predicates.MaxAzureDiskVolumeCountPred)
 			}
 		}
 	}
@@ -219,9 +219,11 @@ type equivalenceClassInfo struct {
 	hash uint64
 }
 
-// getEquivalenceClassInfo returns the equivalence class of given pod.
+// 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.
 func (ec *EquivalenceCache) getEquivalenceClassInfo(pod *v1.Pod) *equivalenceClassInfo {
-	equivalencePod := ec.getEquivalencePod(pod)
+	equivalencePod := getEquivalenceHash(pod)
 	if equivalencePod != nil {
 		hash := fnv.New32a()
 		hashutil.DeepHashObject(hash, equivalencePod)
@@ -231,3 +233,60 @@ 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,10 +455,6 @@ func podFitsOnNode(
 	var (
 		eCacheAvailable  bool
 		failedPredicates []algorithm.PredicateFailureReason
-		invalid          bool
-		fit              bool
-		reasons          []algorithm.PredicateFailureReason
-		err              error
 	)
 	predicateResults := make(map[string]HostPredicate)
 
@@ -494,38 +490,54 @@ 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 {
-				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 || 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
-					}
+				// Use an in-line function to guarantee invocation of ecache.Unlock()
+				// when the in-line function returns.
+				func() {
+					var invalid bool
 					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}
+						// 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 {
+							// 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)
 						}
-						result := predicateResults[predicateKey]
-						ecache.UpdateCachedPredicateItem(pod.GetName(), info.Node().GetName(), predicateKey, result.Fit, result.FailReasons, equivCacheInfo.hash, false)
 					}
-				}
+				}()
 
-				if eCacheAvailable {
-					ecache.Unlock()
+				if err != nil {
+					return false, []algorithm.PredicateFailureReason{}, err
 				}
 
 				if !fit {
@@ -533,7 +545,9 @@ 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

@@ -134,6 +134,8 @@ 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(
@@ -1046,14 +1048,8 @@ func (c *configFactory) CreateFromKeys(predicateKeys, priorityKeys sets.String,
 	}
 
 	// Init equivalence class cache
-	if c.enableEquivalenceClassCache && getEquivalencePodFuncFactory != nil {
-		pluginArgs, err := c.getPluginArgs()
-		if err != nil {
-			return nil, err
-		}
-		c.equivalencePodCache = core.NewEquivalenceCache(
-			getEquivalencePodFuncFactory(*pluginArgs),
-		)
+	if c.enableEquivalenceClassCache {
+		c.equivalencePodCache = core.NewEquivalenceCache()
 		glog.Info("Created equivalence class cache")
 	}
 

pkg/scheduler/factory/plugins.go

@@ -75,9 +75,6 @@ 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
 
@@ -90,9 +87,6 @@ var (
 	// Registered metadata producers
 	priorityMetadataProducer  PriorityMetadataProducerFactory
 	predicateMetadataProducer PredicateMetadataProducerFactory
-
-	// get equivalence pod function
-	getEquivalencePodFuncFactory EquivalencePodFuncFactory
 )
 
 const (
@@ -346,11 +340,6 @@ 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()