From 31c746d9609d7104995dfbec66a954c6b0262d2a Mon Sep 17 00:00:00 2001 From: Jonathan Basseri Date: Wed, 23 May 2018 16:06:45 -0700 Subject: [PATCH] Move equivalence cache into new package. This moves the equivalence cache implementation out of the 'core' package and into k8s.io/kubernetes/pkg/scheduler/core/equivalence. Separating the equiv. cache from the genericScheduler implementation make their interaction points easier to follow, and prevents us from accidentally accessing unexported fields. --- pkg/scheduler/BUILD | 1 + pkg/scheduler/core/BUILD | 10 +- pkg/scheduler/core/equivalence/BUILD | 47 ++++++++ .../eqivalence.go} | 2 +- .../eqivalence_test.go} | 102 +----------------- pkg/scheduler/core/generic_scheduler.go | 11 +- pkg/scheduler/core/generic_scheduler_test.go | 99 +++++++++++++++++ pkg/scheduler/factory/BUILD | 1 + pkg/scheduler/factory/factory.go | 5 +- pkg/scheduler/scheduler.go | 3 +- 10 files changed, 167 insertions(+), 114 deletions(-) create mode 100644 pkg/scheduler/core/equivalence/BUILD rename pkg/scheduler/core/{equivalence_cache.go => equivalence/eqivalence.go} (99%) rename pkg/scheduler/core/{equivalence_cache_test.go => equivalence/eqivalence_test.go} (85%) diff --git a/pkg/scheduler/BUILD b/pkg/scheduler/BUILD index 1e14062c5a..33fd8ea30b 100644 --- a/pkg/scheduler/BUILD +++ b/pkg/scheduler/BUILD @@ -46,6 +46,7 @@ go_library( "//pkg/scheduler/api:go_default_library", "//pkg/scheduler/cache:go_default_library", "//pkg/scheduler/core:go_default_library", + "//pkg/scheduler/core/equivalence:go_default_library", "//pkg/scheduler/metrics:go_default_library", "//pkg/scheduler/util:go_default_library", "//pkg/scheduler/volumebinder:go_default_library", diff --git a/pkg/scheduler/core/BUILD b/pkg/scheduler/core/BUILD index 0e8cd4177c..6cd5b1bb10 100644 --- a/pkg/scheduler/core/BUILD +++ b/pkg/scheduler/core/BUILD @@ -9,7 +9,6 @@ load( go_test( name = "go_default_test", srcs = [ - "equivalence_cache_test.go", "extender_test.go", "generic_scheduler_test.go", "scheduling_queue_test.go", @@ -22,6 +21,7 @@ go_test( "//pkg/scheduler/algorithm/priorities/util:go_default_library", "//pkg/scheduler/api:go_default_library", "//pkg/scheduler/cache:go_default_library", + "//pkg/scheduler/core/equivalence:go_default_library", "//pkg/scheduler/testing:go_default_library", "//pkg/scheduler/util:go_default_library", "//vendor/k8s.io/api/apps/v1beta1:go_default_library", @@ -38,7 +38,6 @@ go_test( go_library( name = "go_default_library", srcs = [ - "equivalence_cache.go", "extender.go", "generic_scheduler.go", "scheduling_queue.go", @@ -51,10 +50,10 @@ go_library( "//pkg/scheduler/algorithm/priorities/util:go_default_library", "//pkg/scheduler/api:go_default_library", "//pkg/scheduler/cache:go_default_library", + "//pkg/scheduler/core/equivalence:go_default_library", "//pkg/scheduler/metrics:go_default_library", "//pkg/scheduler/util:go_default_library", "//pkg/scheduler/volumebinder:go_default_library", - "//pkg/util/hash:go_default_library", "//vendor/github.com/golang/glog:go_default_library", "//vendor/k8s.io/api/core/v1:go_default_library", "//vendor/k8s.io/api/policy/v1beta1:go_default_library", @@ -80,6 +79,9 @@ filegroup( filegroup( name = "all-srcs", - srcs = [":package-srcs"], + srcs = [ + ":package-srcs", + "//pkg/scheduler/core/equivalence:all-srcs", + ], tags = ["automanaged"], ) diff --git a/pkg/scheduler/core/equivalence/BUILD b/pkg/scheduler/core/equivalence/BUILD new file mode 100644 index 0000000000..c46e344c25 --- /dev/null +++ b/pkg/scheduler/core/equivalence/BUILD @@ -0,0 +1,47 @@ +load("@io_bazel_rules_go//go:def.bzl", "go_library", "go_test") + +go_library( + name = "go_default_library", + srcs = ["eqivalence.go"], + importpath = "k8s.io/kubernetes/pkg/scheduler/core/equivalence", + visibility = ["//visibility:public"], + deps = [ + "//pkg/scheduler/algorithm:go_default_library", + "//pkg/scheduler/algorithm/predicates:go_default_library", + "//pkg/scheduler/cache:go_default_library", + "//pkg/util/hash:go_default_library", + "//vendor/github.com/golang/glog:go_default_library", + "//vendor/k8s.io/api/core/v1:go_default_library", + "//vendor/k8s.io/apimachinery/pkg/util/sets:go_default_library", + ], +) + +go_test( + name = "go_default_test", + srcs = ["eqivalence_test.go"], + embed = [":go_default_library"], + deps = [ + "//pkg/scheduler/algorithm:go_default_library", + "//pkg/scheduler/algorithm/predicates:go_default_library", + "//pkg/scheduler/cache:go_default_library", + "//pkg/scheduler/testing:go_default_library", + "//vendor/k8s.io/api/core/v1:go_default_library", + "//vendor/k8s.io/apimachinery/pkg/api/resource:go_default_library", + "//vendor/k8s.io/apimachinery/pkg/apis/meta/v1:go_default_library", + "//vendor/k8s.io/apimachinery/pkg/util/sets:go_default_library", + ], +) + +filegroup( + name = "package-srcs", + srcs = glob(["**"]), + tags = ["automanaged"], + visibility = ["//visibility:private"], +) + +filegroup( + name = "all-srcs", + srcs = [":package-srcs"], + tags = ["automanaged"], + visibility = ["//visibility:public"], +) diff --git a/pkg/scheduler/core/equivalence_cache.go b/pkg/scheduler/core/equivalence/eqivalence.go similarity index 99% rename from pkg/scheduler/core/equivalence_cache.go rename to pkg/scheduler/core/equivalence/eqivalence.go index c845ddbdde..eee9324ceb 100644 --- a/pkg/scheduler/core/equivalence_cache.go +++ b/pkg/scheduler/core/equivalence/eqivalence.go @@ -14,7 +14,7 @@ See the License for the specific language governing permissions and limitations under the License. */ -package core +package equivalence import ( "fmt" diff --git a/pkg/scheduler/core/equivalence_cache_test.go b/pkg/scheduler/core/equivalence/eqivalence_test.go similarity index 85% rename from pkg/scheduler/core/equivalence_cache_test.go rename to pkg/scheduler/core/equivalence/eqivalence_test.go index 7df8097a17..c9353d3ac9 100644 --- a/pkg/scheduler/core/equivalence_cache_test.go +++ b/pkg/scheduler/core/equivalence/eqivalence_test.go @@ -14,20 +14,17 @@ See the License for the specific language governing permissions and limitations under the License. */ -package core +package equivalence import ( "errors" "reflect" - "sync" "testing" - "time" "k8s.io/api/core/v1" "k8s.io/apimachinery/pkg/api/resource" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/util/sets" - "k8s.io/apimachinery/pkg/util/wait" "k8s.io/kubernetes/pkg/scheduler/algorithm" "k8s.io/kubernetes/pkg/scheduler/algorithm/predicates" schedulercache "k8s.io/kubernetes/pkg/scheduler/cache" @@ -796,100 +793,3 @@ func BenchmarkEquivalenceHash(b *testing.B) { getEquivalencePod(pod) } } - -// syncingMockCache delegates method calls to an actual Cache, -// but calls to UpdateNodeNameToInfoMap synchronize with the test. -type syncingMockCache struct { - schedulercache.Cache - cycleStart, cacheInvalidated chan struct{} - once sync.Once -} - -// UpdateNodeNameToInfoMap delegates to the real implementation, but on the first call, it -// synchronizes with the test. -// -// Since UpdateNodeNameToInfoMap is one of the first steps of (*genericScheduler).Schedule, we use -// this point to signal to the test that a scheduling cycle has started. -func (c *syncingMockCache) UpdateNodeNameToInfoMap(infoMap map[string]*schedulercache.NodeInfo) error { - err := c.Cache.UpdateNodeNameToInfoMap(infoMap) - c.once.Do(func() { - c.cycleStart <- struct{}{} - <-c.cacheInvalidated - }) - return err -} - -// TestEquivalenceCacheInvalidationRace tests that equivalence cache invalidation is correctly -// handled when an invalidation event happens early in a scheduling cycle. Specifically, the event -// occurs after schedulercache is snapshotted and before equivalence cache lock is acquired. -func TestEquivalenceCacheInvalidationRace(t *testing.T) { - // Create a predicate that returns false the first time and true on subsequent calls. - podWillFit := false - var callCount int - testPredicate := func(pod *v1.Pod, - meta algorithm.PredicateMetadata, - nodeInfo *schedulercache.NodeInfo) (bool, []algorithm.PredicateFailureReason, error) { - callCount++ - if !podWillFit { - podWillFit = true - return false, []algorithm.PredicateFailureReason{predicates.ErrFakePredicate}, nil - } - return true, nil, nil - } - - // Set up the mock cache. - cache := schedulercache.New(time.Duration(0), wait.NeverStop) - cache.AddNode(&v1.Node{ObjectMeta: metav1.ObjectMeta{Name: "machine1"}}) - mockCache := &syncingMockCache{ - Cache: cache, - cycleStart: make(chan struct{}), - cacheInvalidated: make(chan struct{}), - } - - eCache := NewEquivalenceCache() - // Ensure that equivalence cache invalidation happens after the scheduling cycle starts, but before - // the equivalence cache would be updated. - go func() { - <-mockCache.cycleStart - pod := &v1.Pod{ - ObjectMeta: metav1.ObjectMeta{Name: "new-pod", UID: "new-pod"}, - Spec: v1.PodSpec{NodeName: "machine1"}} - if err := cache.AddPod(pod); err != nil { - t.Errorf("Could not add pod to cache: %v", err) - } - eCache.InvalidateAllPredicatesOnNode("machine1") - mockCache.cacheInvalidated <- struct{}{} - }() - - // Set up the scheduler. - ps := map[string]algorithm.FitPredicate{"testPredicate": testPredicate} - predicates.SetPredicatesOrdering([]string{"testPredicate"}) - prioritizers := []algorithm.PriorityConfig{{Map: EqualPriorityMap, Weight: 1}} - pvcLister := schedulertesting.FakePersistentVolumeClaimLister([]*v1.PersistentVolumeClaim{}) - scheduler := NewGenericScheduler( - mockCache, - eCache, - NewSchedulingQueue(), - ps, - algorithm.EmptyPredicateMetadataProducer, - prioritizers, - algorithm.EmptyPriorityMetadataProducer, - nil, nil, pvcLister, true, false) - - // First scheduling attempt should fail. - nodeLister := schedulertesting.FakeNodeLister(makeNodeList([]string{"machine1"})) - pod := &v1.Pod{ObjectMeta: metav1.ObjectMeta{Name: "test-pod"}} - machine, err := scheduler.Schedule(pod, nodeLister) - if machine != "" || err == nil { - t.Error("First scheduling attempt did not fail") - } - - // Second scheduling attempt should succeed because cache was invalidated. - _, err = scheduler.Schedule(pod, nodeLister) - if err != nil { - t.Errorf("Second scheduling attempt failed: %v", err) - } - if callCount != 2 { - t.Errorf("Predicate should have been called twice. Was called %d times.", callCount) - } -} diff --git a/pkg/scheduler/core/generic_scheduler.go b/pkg/scheduler/core/generic_scheduler.go index 54d340a8f5..35bf8e212a 100644 --- a/pkg/scheduler/core/generic_scheduler.go +++ b/pkg/scheduler/core/generic_scheduler.go @@ -39,6 +39,7 @@ import ( "k8s.io/kubernetes/pkg/scheduler/algorithm/predicates" schedulerapi "k8s.io/kubernetes/pkg/scheduler/api" schedulercache "k8s.io/kubernetes/pkg/scheduler/cache" + "k8s.io/kubernetes/pkg/scheduler/core/equivalence" "k8s.io/kubernetes/pkg/scheduler/metrics" "k8s.io/kubernetes/pkg/scheduler/util" "k8s.io/kubernetes/pkg/scheduler/volumebinder" @@ -85,7 +86,7 @@ func (f *FitError) Error() string { type genericScheduler struct { cache schedulercache.Cache - equivalenceCache *EquivalenceCache + equivalenceCache *equivalence.EquivalenceCache schedulingQueue SchedulingQueue predicates map[string]algorithm.FitPredicate priorityMetaProducer algorithm.PriorityMetadataProducer @@ -342,7 +343,7 @@ func (g *genericScheduler) findNodesThatFit(pod *v1.Pod, nodes []*v1.Node) ([]*v // We can use the same metadata producer for all nodes. meta := g.predicateMetaProducer(pod, g.cachedNodeInfoMap) - var equivCacheInfo *EquivalenceClassInfo + var equivCacheInfo *equivalence.EquivalenceClassInfo if g.equivalenceCache != nil { // getEquivalenceClassInfo will return immediately if no equivalence pod found equivCacheInfo = g.equivalenceCache.GetEquivalenceClassInfo(pod) @@ -459,10 +460,10 @@ func podFitsOnNode( info *schedulercache.NodeInfo, predicateFuncs map[string]algorithm.FitPredicate, cache schedulercache.Cache, - ecache *EquivalenceCache, + ecache *equivalence.EquivalenceCache, queue SchedulingQueue, alwaysCheckAllPredicates bool, - equivCacheInfo *EquivalenceClassInfo, + equivCacheInfo *equivalence.EquivalenceClassInfo, ) (bool, []algorithm.PredicateFailureReason, error) { var ( eCacheAvailable bool @@ -1056,7 +1057,7 @@ func podPassesBasicChecks(pod *v1.Pod, pvcLister corelisters.PersistentVolumeCla // NewGenericScheduler creates a genericScheduler object. func NewGenericScheduler( cache schedulercache.Cache, - eCache *EquivalenceCache, + eCache *equivalence.EquivalenceCache, podQueue SchedulingQueue, predicates map[string]algorithm.FitPredicate, predicateMetaProducer algorithm.PredicateMetadataProducer, diff --git a/pkg/scheduler/core/generic_scheduler_test.go b/pkg/scheduler/core/generic_scheduler_test.go index ce10d9fd3f..9b164cf6d6 100644 --- a/pkg/scheduler/core/generic_scheduler_test.go +++ b/pkg/scheduler/core/generic_scheduler_test.go @@ -22,6 +22,7 @@ import ( "reflect" "strconv" "strings" + "sync" "testing" "time" @@ -39,6 +40,7 @@ import ( priorityutil "k8s.io/kubernetes/pkg/scheduler/algorithm/priorities/util" schedulerapi "k8s.io/kubernetes/pkg/scheduler/api" schedulercache "k8s.io/kubernetes/pkg/scheduler/cache" + "k8s.io/kubernetes/pkg/scheduler/core/equivalence" schedulertesting "k8s.io/kubernetes/pkg/scheduler/testing" ) @@ -1342,3 +1344,100 @@ func TestPreempt(t *testing.T) { close(stop) } } + +// syncingMockCache delegates method calls to an actual Cache, +// but calls to UpdateNodeNameToInfoMap synchronize with the test. +type syncingMockCache struct { + schedulercache.Cache + cycleStart, cacheInvalidated chan struct{} + once sync.Once +} + +// UpdateNodeNameToInfoMap delegates to the real implementation, but on the first call, it +// synchronizes with the test. +// +// Since UpdateNodeNameToInfoMap is one of the first steps of (*genericScheduler).Schedule, we use +// this point to signal to the test that a scheduling cycle has started. +func (c *syncingMockCache) UpdateNodeNameToInfoMap(infoMap map[string]*schedulercache.NodeInfo) error { + err := c.Cache.UpdateNodeNameToInfoMap(infoMap) + c.once.Do(func() { + c.cycleStart <- struct{}{} + <-c.cacheInvalidated + }) + return err +} + +// TestEquivalenceCacheInvalidationRace tests that equivalence cache invalidation is correctly +// handled when an invalidation event happens early in a scheduling cycle. Specifically, the event +// occurs after schedulercache is snapshotted and before equivalence cache lock is acquired. +func TestEquivalenceCacheInvalidationRace(t *testing.T) { + // Create a predicate that returns false the first time and true on subsequent calls. + podWillFit := false + var callCount int + testPredicate := func(pod *v1.Pod, + meta algorithm.PredicateMetadata, + nodeInfo *schedulercache.NodeInfo) (bool, []algorithm.PredicateFailureReason, error) { + callCount++ + if !podWillFit { + podWillFit = true + return false, []algorithm.PredicateFailureReason{algorithmpredicates.ErrFakePredicate}, nil + } + return true, nil, nil + } + + // Set up the mock cache. + cache := schedulercache.New(time.Duration(0), wait.NeverStop) + cache.AddNode(&v1.Node{ObjectMeta: metav1.ObjectMeta{Name: "machine1"}}) + mockCache := &syncingMockCache{ + Cache: cache, + cycleStart: make(chan struct{}), + cacheInvalidated: make(chan struct{}), + } + + eCache := equivalence.NewEquivalenceCache() + // Ensure that equivalence cache invalidation happens after the scheduling cycle starts, but before + // the equivalence cache would be updated. + go func() { + <-mockCache.cycleStart + pod := &v1.Pod{ + ObjectMeta: metav1.ObjectMeta{Name: "new-pod", UID: "new-pod"}, + Spec: v1.PodSpec{NodeName: "machine1"}} + if err := cache.AddPod(pod); err != nil { + t.Errorf("Could not add pod to cache: %v", err) + } + eCache.InvalidateAllPredicatesOnNode("machine1") + mockCache.cacheInvalidated <- struct{}{} + }() + + // Set up the scheduler. + ps := map[string]algorithm.FitPredicate{"testPredicate": testPredicate} + algorithmpredicates.SetPredicatesOrdering([]string{"testPredicate"}) + prioritizers := []algorithm.PriorityConfig{{Map: EqualPriorityMap, Weight: 1}} + pvcLister := schedulertesting.FakePersistentVolumeClaimLister([]*v1.PersistentVolumeClaim{}) + scheduler := NewGenericScheduler( + mockCache, + eCache, + NewSchedulingQueue(), + ps, + algorithm.EmptyPredicateMetadataProducer, + prioritizers, + algorithm.EmptyPriorityMetadataProducer, + nil, nil, pvcLister, true, false) + + // First scheduling attempt should fail. + nodeLister := schedulertesting.FakeNodeLister(makeNodeList([]string{"machine1"})) + pod := &v1.Pod{ObjectMeta: metav1.ObjectMeta{Name: "test-pod"}} + machine, err := scheduler.Schedule(pod, nodeLister) + if machine != "" || err == nil { + t.Error("First scheduling attempt did not fail") + } + + // Second scheduling attempt should succeed because cache was invalidated. + _, err = scheduler.Schedule(pod, nodeLister) + if err != nil { + t.Errorf("Second scheduling attempt failed: %v", err) + } + if callCount != 2 { + t.Errorf("Predicate should have been called twice. Was called %d times.", callCount) + } +} diff --git a/pkg/scheduler/factory/BUILD b/pkg/scheduler/factory/BUILD index faf552687c..cc0ebf88ff 100644 --- a/pkg/scheduler/factory/BUILD +++ b/pkg/scheduler/factory/BUILD @@ -62,6 +62,7 @@ go_library( "//pkg/scheduler/api/validation:go_default_library", "//pkg/scheduler/cache:go_default_library", "//pkg/scheduler/core:go_default_library", + "//pkg/scheduler/core/equivalence:go_default_library", "//pkg/scheduler/util:go_default_library", "//pkg/scheduler/volumebinder:go_default_library", "//vendor/github.com/golang/glog:go_default_library", diff --git a/pkg/scheduler/factory/factory.go b/pkg/scheduler/factory/factory.go index c401f834b3..29323526fe 100644 --- a/pkg/scheduler/factory/factory.go +++ b/pkg/scheduler/factory/factory.go @@ -61,6 +61,7 @@ import ( "k8s.io/kubernetes/pkg/scheduler/api/validation" schedulercache "k8s.io/kubernetes/pkg/scheduler/cache" "k8s.io/kubernetes/pkg/scheduler/core" + "k8s.io/kubernetes/pkg/scheduler/core/equivalence" "k8s.io/kubernetes/pkg/scheduler/util" "k8s.io/kubernetes/pkg/scheduler/volumebinder" ) @@ -123,7 +124,7 @@ type configFactory struct { hardPodAffinitySymmetricWeight int32 // Equivalence class cache - equivalencePodCache *core.EquivalenceCache + equivalencePodCache *equivalence.EquivalenceCache // Enable equivalence class cache enableEquivalenceClassCache bool @@ -1074,7 +1075,7 @@ func (c *configFactory) CreateFromKeys(predicateKeys, priorityKeys sets.String, // Init equivalence class cache if c.enableEquivalenceClassCache { - c.equivalencePodCache = core.NewEquivalenceCache() + c.equivalencePodCache = equivalence.NewEquivalenceCache() glog.Info("Created equivalence class cache") } diff --git a/pkg/scheduler/scheduler.go b/pkg/scheduler/scheduler.go index 98b36678ab..1288758008 100644 --- a/pkg/scheduler/scheduler.go +++ b/pkg/scheduler/scheduler.go @@ -34,6 +34,7 @@ import ( schedulerapi "k8s.io/kubernetes/pkg/scheduler/api" schedulercache "k8s.io/kubernetes/pkg/scheduler/cache" "k8s.io/kubernetes/pkg/scheduler/core" + "k8s.io/kubernetes/pkg/scheduler/core/equivalence" "k8s.io/kubernetes/pkg/scheduler/metrics" "k8s.io/kubernetes/pkg/scheduler/util" "k8s.io/kubernetes/pkg/scheduler/volumebinder" @@ -104,7 +105,7 @@ type Config struct { SchedulerCache schedulercache.Cache // Ecache is used for optimistically invalid affected cache items after // successfully binding a pod - Ecache *core.EquivalenceCache + Ecache *equivalence.EquivalenceCache NodeLister algorithm.NodeLister Algorithm algorithm.ScheduleAlgorithm GetBinder func(pod *v1.Pod) Binder