Commit eace2d08 authored by Jonathan Basseri's avatar Jonathan Basseri

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

This reverts commit 4386751b.
parent 029c1d8b
...@@ -17,12 +17,8 @@ limitations under the License. ...@@ -17,12 +17,8 @@ limitations under the License.
package predicates package predicates
import ( import (
"github.com/golang/glog"
"k8s.io/api/core/v1" "k8s.io/api/core/v1"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/labels" "k8s.io/apimachinery/pkg/labels"
"k8s.io/apimachinery/pkg/util/sets"
"k8s.io/kubernetes/pkg/scheduler/algorithm"
schedutil "k8s.io/kubernetes/pkg/scheduler/util" schedutil "k8s.io/kubernetes/pkg/scheduler/util"
) )
...@@ -70,69 +66,6 @@ func CreateSelectorFromLabels(aL map[string]string) labels.Selector { ...@@ -70,69 +66,6 @@ func CreateSelectorFromLabels(aL map[string]string) labels.Selector {
return labels.Set(aL).AsSelector() 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 // portsConflict check whether existingPorts and wantPorts conflict with each other
// return true if we have a conflict // return true if we have a conflict
func portsConflict(existingPorts schedutil.HostPortInfo, wantPorts []*v1.ContainerPort) bool { func portsConflict(existingPorts schedutil.HostPortInfo, wantPorts []*v1.ContainerPort) bool {
......
...@@ -78,9 +78,6 @@ type PredicateFailureReason interface { ...@@ -78,9 +78,6 @@ type PredicateFailureReason interface {
GetReason() string 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. // NodeLister interface represents anything that can list nodes for a scheduler.
type NodeLister interface { type NodeLister interface {
// We explicitly return []*v1.Node, instead of v1.NodeList, to avoid // We explicitly return []*v1.Node, instead of v1.NodeList, to avoid
......
...@@ -77,13 +77,6 @@ func init() { ...@@ -77,13 +77,6 @@ func init() {
// Fit is determined by node selector query. // Fit is determined by node selector query.
factory.RegisterFitPredicate(predicates.MatchNodeSelectorPred, predicates.PodMatchNodeSelector) 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 // ServiceSpreadingPriority is a priority config factory that spreads pods by minimizing
// the number of pods (belonging to the same service) on the same node. // 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 // Register the factory so that it's available, but do not include it as part of the default priorities
......
...@@ -23,6 +23,7 @@ import ( ...@@ -23,6 +23,7 @@ import (
"k8s.io/api/core/v1" "k8s.io/api/core/v1"
"k8s.io/apimachinery/pkg/util/sets" "k8s.io/apimachinery/pkg/util/sets"
"k8s.io/kubernetes/pkg/scheduler/algorithm" "k8s.io/kubernetes/pkg/scheduler/algorithm"
"k8s.io/kubernetes/pkg/scheduler/algorithm/predicates"
hashutil "k8s.io/kubernetes/pkg/util/hash" hashutil "k8s.io/kubernetes/pkg/util/hash"
"github.com/golang/glog" "github.com/golang/glog"
...@@ -37,8 +38,7 @@ const maxCacheEntries = 100 ...@@ -37,8 +38,7 @@ const maxCacheEntries = 100
// 2. function to get equivalence pod // 2. function to get equivalence pod
type EquivalenceCache struct { type EquivalenceCache struct {
sync.RWMutex sync.RWMutex
getEquivalencePod algorithm.GetEquivalencePodFunc algorithmCache map[string]AlgorithmCache
algorithmCache map[string]AlgorithmCache
} }
// The AlgorithmCache stores PredicateMap with predicate name as key // The AlgorithmCache stores PredicateMap with predicate name as key
...@@ -62,11 +62,11 @@ func newAlgorithmCache() AlgorithmCache { ...@@ -62,11 +62,11 @@ func newAlgorithmCache() AlgorithmCache {
} }
} }
// NewEquivalenceCache creates a EquivalenceCache object. // NewEquivalenceCache returns EquivalenceCache to speed up predicates by caching
func NewEquivalenceCache(getEquivalencePodFunc algorithm.GetEquivalencePodFunc) *EquivalenceCache { // result from previous scheduling.
func NewEquivalenceCache() *EquivalenceCache {
return &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, ...@@ -191,21 +191,21 @@ func (ec *EquivalenceCache) InvalidateCachedPredicateItemForPodAdd(pod *v1.Pod,
// it will also fits to equivalence class of existing pods // it will also fits to equivalence class of existing pods
// GeneralPredicates: will always be affected by adding a new pod // 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. // MaxPDVolumeCountPredicate: we check the volumes of pod to make decision.
for _, vol := range pod.Spec.Volumes { for _, vol := range pod.Spec.Volumes {
if vol.PersistentVolumeClaim != nil { if vol.PersistentVolumeClaim != nil {
invalidPredicates.Insert("MaxEBSVolumeCount", "MaxGCEPDVolumeCount", "MaxAzureDiskVolumeCount") invalidPredicates.Insert(predicates.MaxEBSVolumeCountPred, predicates.MaxGCEPDVolumeCountPred, predicates.MaxAzureDiskVolumeCountPred)
} else { } else {
if vol.AWSElasticBlockStore != nil { if vol.AWSElasticBlockStore != nil {
invalidPredicates.Insert("MaxEBSVolumeCount") invalidPredicates.Insert(predicates.MaxEBSVolumeCountPred)
} }
if vol.GCEPersistentDisk != nil { if vol.GCEPersistentDisk != nil {
invalidPredicates.Insert("MaxGCEPDVolumeCount") invalidPredicates.Insert(predicates.MaxGCEPDVolumeCountPred)
} }
if vol.AzureDisk != nil { if vol.AzureDisk != nil {
invalidPredicates.Insert("MaxAzureDiskVolumeCount") invalidPredicates.Insert(predicates.MaxAzureDiskVolumeCountPred)
} }
} }
} }
...@@ -219,9 +219,11 @@ type equivalenceClassInfo struct { ...@@ -219,9 +219,11 @@ type equivalenceClassInfo struct {
hash uint64 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 { func (ec *EquivalenceCache) getEquivalenceClassInfo(pod *v1.Pod) *equivalenceClassInfo {
equivalencePod := ec.getEquivalencePod(pod) equivalencePod := getEquivalenceHash(pod)
if equivalencePod != nil { if equivalencePod != nil {
hash := fnv.New32a() hash := fnv.New32a()
hashutil.DeepHashObject(hash, equivalencePod) hashutil.DeepHashObject(hash, equivalencePod)
...@@ -231,3 +233,60 @@ func (ec *EquivalenceCache) getEquivalenceClassInfo(pod *v1.Pod) *equivalenceCla ...@@ -231,3 +233,60 @@ func (ec *EquivalenceCache) getEquivalenceClassInfo(pod *v1.Pod) *equivalenceCla
} }
return nil 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
}
...@@ -1058,14 +1058,8 @@ func (c *configFactory) CreateFromKeys(predicateKeys, priorityKeys sets.String, ...@@ -1058,14 +1058,8 @@ func (c *configFactory) CreateFromKeys(predicateKeys, priorityKeys sets.String,
} }
// Init equivalence class cache // Init equivalence class cache
if c.enableEquivalenceClassCache && getEquivalencePodFuncFactory != nil { if c.enableEquivalenceClassCache {
pluginArgs, err := c.getPluginArgs() c.equivalencePodCache = core.NewEquivalenceCache()
if err != nil {
return nil, err
}
c.equivalencePodCache = core.NewEquivalenceCache(
getEquivalencePodFuncFactory(*pluginArgs),
)
glog.Info("Created equivalence class cache") glog.Info("Created equivalence class cache")
} }
......
...@@ -75,9 +75,6 @@ type PriorityConfigFactory struct { ...@@ -75,9 +75,6 @@ type PriorityConfigFactory struct {
Weight int Weight int
} }
// EquivalencePodFuncFactory produces a function to get equivalence class for given pod.
type EquivalencePodFuncFactory func(PluginFactoryArgs) algorithm.GetEquivalencePodFunc
var ( var (
schedulerFactoryMutex sync.Mutex schedulerFactoryMutex sync.Mutex
...@@ -90,9 +87,6 @@ var ( ...@@ -90,9 +87,6 @@ var (
// Registered metadata producers // Registered metadata producers
priorityMetadataProducer PriorityMetadataProducerFactory priorityMetadataProducer PriorityMetadataProducerFactory
predicateMetadataProducer PredicateMetadataProducerFactory predicateMetadataProducer PredicateMetadataProducerFactory
// get equivalence pod function
getEquivalencePodFuncFactory EquivalencePodFuncFactory
) )
const ( const (
...@@ -346,11 +340,6 @@ func RegisterCustomPriorityFunction(policy schedulerapi.PriorityPolicy) string { ...@@ -346,11 +340,6 @@ func RegisterCustomPriorityFunction(policy schedulerapi.PriorityPolicy) string {
return RegisterPriorityConfigFactory(policy.Name, *pcf) 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. // IsPriorityFunctionRegistered is useful for testing providers.
func IsPriorityFunctionRegistered(name string) bool { func IsPriorityFunctionRegistered(name string) bool {
schedulerFactoryMutex.Lock() schedulerFactoryMutex.Lock()
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment