move SchedulingQueue to pkg/scheduler/internal/queue

This commit is contained in:
Wei Huang 2018-09-21 16:18:48 -07:00
parent 2e7461c087
commit 9da576f03c
No known key found for this signature in database
GPG Key ID: BE5E9752F8B6E005
10 changed files with 30 additions and 24 deletions

View File

@ -27,6 +27,7 @@ import (
"k8s.io/kubernetes/pkg/scheduler/algorithm" "k8s.io/kubernetes/pkg/scheduler/algorithm"
schedulerapi "k8s.io/kubernetes/pkg/scheduler/api" schedulerapi "k8s.io/kubernetes/pkg/scheduler/api"
schedulercache "k8s.io/kubernetes/pkg/scheduler/cache" schedulercache "k8s.io/kubernetes/pkg/scheduler/cache"
internalqueue "k8s.io/kubernetes/pkg/scheduler/internal/queue"
schedulertesting "k8s.io/kubernetes/pkg/scheduler/testing" schedulertesting "k8s.io/kubernetes/pkg/scheduler/testing"
"k8s.io/kubernetes/pkg/scheduler/util" "k8s.io/kubernetes/pkg/scheduler/util"
) )
@ -501,7 +502,7 @@ func TestGenericSchedulerWithExtenders(t *testing.T) {
for _, name := range test.nodes { for _, name := range test.nodes {
cache.AddNode(createNode(name)) cache.AddNode(createNode(name))
} }
queue := NewSchedulingQueue() queue := internalqueue.NewSchedulingQueue()
scheduler := NewGenericScheduler( scheduler := NewGenericScheduler(
cache, cache,
nil, nil,

View File

@ -41,6 +41,7 @@ import (
schedulerapi "k8s.io/kubernetes/pkg/scheduler/api" schedulerapi "k8s.io/kubernetes/pkg/scheduler/api"
schedulercache "k8s.io/kubernetes/pkg/scheduler/cache" schedulercache "k8s.io/kubernetes/pkg/scheduler/cache"
"k8s.io/kubernetes/pkg/scheduler/core/equivalence" "k8s.io/kubernetes/pkg/scheduler/core/equivalence"
internalqueue "k8s.io/kubernetes/pkg/scheduler/internal/queue"
"k8s.io/kubernetes/pkg/scheduler/metrics" "k8s.io/kubernetes/pkg/scheduler/metrics"
"k8s.io/kubernetes/pkg/scheduler/util" "k8s.io/kubernetes/pkg/scheduler/util"
"k8s.io/kubernetes/pkg/scheduler/volumebinder" "k8s.io/kubernetes/pkg/scheduler/volumebinder"
@ -96,7 +97,7 @@ func (f *FitError) Error() string {
type genericScheduler struct { type genericScheduler struct {
cache schedulercache.Cache cache schedulercache.Cache
equivalenceCache *equivalence.Cache equivalenceCache *equivalence.Cache
schedulingQueue SchedulingQueue schedulingQueue internalqueue.SchedulingQueue
predicates map[string]algorithm.FitPredicate predicates map[string]algorithm.FitPredicate
priorityMetaProducer algorithm.PriorityMetadataProducer priorityMetaProducer algorithm.PriorityMetadataProducer
predicateMetaProducer algorithm.PredicateMetadataProducer predicateMetaProducer algorithm.PredicateMetadataProducer
@ -493,7 +494,7 @@ func (g *genericScheduler) findNodesThatFit(pod *v1.Pod, nodes []*v1.Node) ([]*v
// to run on the node given in nodeInfo to meta and nodeInfo. It returns 1) whether // to run on the node given in nodeInfo to meta and nodeInfo. It returns 1) whether
// any pod was found, 2) augmented meta data, 3) augmented nodeInfo. // any pod was found, 2) augmented meta data, 3) augmented nodeInfo.
func addNominatedPods(podPriority int32, meta algorithm.PredicateMetadata, func addNominatedPods(podPriority int32, meta algorithm.PredicateMetadata,
nodeInfo *schedulercache.NodeInfo, queue SchedulingQueue) (bool, algorithm.PredicateMetadata, nodeInfo *schedulercache.NodeInfo, queue internalqueue.SchedulingQueue) (bool, algorithm.PredicateMetadata,
*schedulercache.NodeInfo) { *schedulercache.NodeInfo) {
if queue == nil || nodeInfo == nil || nodeInfo.Node() == nil { if queue == nil || nodeInfo == nil || nodeInfo.Node() == nil {
// This may happen only in tests. // This may happen only in tests.
@ -535,7 +536,7 @@ func podFitsOnNode(
info *schedulercache.NodeInfo, info *schedulercache.NodeInfo,
predicateFuncs map[string]algorithm.FitPredicate, predicateFuncs map[string]algorithm.FitPredicate,
nodeCache *equivalence.NodeCache, nodeCache *equivalence.NodeCache,
queue SchedulingQueue, queue internalqueue.SchedulingQueue,
alwaysCheckAllPredicates bool, alwaysCheckAllPredicates bool,
equivClass *equivalence.Class, equivClass *equivalence.Class,
) (bool, []algorithm.PredicateFailureReason, error) { ) (bool, []algorithm.PredicateFailureReason, error) {
@ -887,7 +888,7 @@ func selectNodesForPreemption(pod *v1.Pod,
potentialNodes []*v1.Node, potentialNodes []*v1.Node,
predicates map[string]algorithm.FitPredicate, predicates map[string]algorithm.FitPredicate,
metadataProducer algorithm.PredicateMetadataProducer, metadataProducer algorithm.PredicateMetadataProducer,
queue SchedulingQueue, queue internalqueue.SchedulingQueue,
pdbs []*policy.PodDisruptionBudget, pdbs []*policy.PodDisruptionBudget,
) (map[*v1.Node]*schedulerapi.Victims, error) { ) (map[*v1.Node]*schedulerapi.Victims, error) {
@ -976,7 +977,7 @@ func selectVictimsOnNode(
meta algorithm.PredicateMetadata, meta algorithm.PredicateMetadata,
nodeInfo *schedulercache.NodeInfo, nodeInfo *schedulercache.NodeInfo,
fitPredicates map[string]algorithm.FitPredicate, fitPredicates map[string]algorithm.FitPredicate,
queue SchedulingQueue, queue internalqueue.SchedulingQueue,
pdbs []*policy.PodDisruptionBudget, pdbs []*policy.PodDisruptionBudget,
) ([]*v1.Pod, int, bool) { ) ([]*v1.Pod, int, bool) {
potentialVictims := util.SortableList{CompFunc: util.HigherPriorityPod} potentialVictims := util.SortableList{CompFunc: util.HigherPriorityPod}
@ -1140,7 +1141,7 @@ func podPassesBasicChecks(pod *v1.Pod, pvcLister corelisters.PersistentVolumeCla
func NewGenericScheduler( func NewGenericScheduler(
cache schedulercache.Cache, cache schedulercache.Cache,
eCache *equivalence.Cache, eCache *equivalence.Cache,
podQueue SchedulingQueue, podQueue internalqueue.SchedulingQueue,
predicates map[string]algorithm.FitPredicate, predicates map[string]algorithm.FitPredicate,
predicateMetaProducer algorithm.PredicateMetadataProducer, predicateMetaProducer algorithm.PredicateMetadataProducer,
prioritizers []algorithm.PriorityConfig, prioritizers []algorithm.PriorityConfig,

View File

@ -41,6 +41,7 @@ import (
schedulerapi "k8s.io/kubernetes/pkg/scheduler/api" schedulerapi "k8s.io/kubernetes/pkg/scheduler/api"
schedulercache "k8s.io/kubernetes/pkg/scheduler/cache" schedulercache "k8s.io/kubernetes/pkg/scheduler/cache"
"k8s.io/kubernetes/pkg/scheduler/core/equivalence" "k8s.io/kubernetes/pkg/scheduler/core/equivalence"
internalqueue "k8s.io/kubernetes/pkg/scheduler/internal/queue"
schedulertesting "k8s.io/kubernetes/pkg/scheduler/testing" schedulertesting "k8s.io/kubernetes/pkg/scheduler/testing"
) )
@ -447,7 +448,7 @@ func TestGenericScheduler(t *testing.T) {
scheduler := NewGenericScheduler( scheduler := NewGenericScheduler(
cache, cache,
nil, nil,
NewSchedulingQueue(), internalqueue.NewSchedulingQueue(),
test.predicates, test.predicates,
algorithm.EmptyPredicateMetadataProducer, algorithm.EmptyPredicateMetadataProducer,
test.prioritizers, test.prioritizers,
@ -483,7 +484,7 @@ func makeScheduler(predicates map[string]algorithm.FitPredicate, nodes []*v1.Nod
s := NewGenericScheduler( s := NewGenericScheduler(
cache, cache,
nil, nil,
NewSchedulingQueue(), internalqueue.NewSchedulingQueue(),
predicates, predicates,
algorithm.EmptyPredicateMetadataProducer, algorithm.EmptyPredicateMetadataProducer,
prioritizers, prioritizers,
@ -1404,7 +1405,7 @@ func TestPreempt(t *testing.T) {
scheduler := NewGenericScheduler( scheduler := NewGenericScheduler(
cache, cache,
nil, nil,
NewSchedulingQueue(), internalqueue.NewSchedulingQueue(),
map[string]algorithm.FitPredicate{"matches": algorithmpredicates.PodFitsResources}, map[string]algorithm.FitPredicate{"matches": algorithmpredicates.PodFitsResources},
algorithm.EmptyPredicateMetadataProducer, algorithm.EmptyPredicateMetadataProducer,
[]algorithm.PriorityConfig{{Function: numericPriority, Weight: 1}}, []algorithm.PriorityConfig{{Function: numericPriority, Weight: 1}},
@ -1531,7 +1532,7 @@ func TestCacheInvalidationRace(t *testing.T) {
scheduler := NewGenericScheduler( scheduler := NewGenericScheduler(
mockCache, mockCache,
eCache, eCache,
NewSchedulingQueue(), internalqueue.NewSchedulingQueue(),
ps, ps,
algorithm.EmptyPredicateMetadataProducer, algorithm.EmptyPredicateMetadataProducer,
prioritizers, prioritizers,
@ -1614,7 +1615,7 @@ func TestCacheInvalidationRace2(t *testing.T) {
scheduler := NewGenericScheduler( scheduler := NewGenericScheduler(
cache, cache,
eCache, eCache,
NewSchedulingQueue(), internalqueue.NewSchedulingQueue(),
ps, ps,
algorithm.EmptyPredicateMetadataProducer, algorithm.EmptyPredicateMetadataProducer,
prioritizers, prioritizers,

View File

@ -25,14 +25,14 @@ import (
"k8s.io/apimachinery/pkg/labels" "k8s.io/apimachinery/pkg/labels"
corelisters "k8s.io/client-go/listers/core/v1" corelisters "k8s.io/client-go/listers/core/v1"
schedulercache "k8s.io/kubernetes/pkg/scheduler/cache" schedulercache "k8s.io/kubernetes/pkg/scheduler/cache"
"k8s.io/kubernetes/pkg/scheduler/core" internalqueue "k8s.io/kubernetes/pkg/scheduler/internal/queue"
) )
type cacheComparer struct { type cacheComparer struct {
nodeLister corelisters.NodeLister nodeLister corelisters.NodeLister
podLister corelisters.PodLister podLister corelisters.PodLister
cache schedulercache.Cache cache schedulercache.Cache
podQueue core.SchedulingQueue podQueue internalqueue.SchedulingQueue
compareStrategy compareStrategy
} }

View File

@ -60,6 +60,7 @@ import (
schedulercache "k8s.io/kubernetes/pkg/scheduler/cache" schedulercache "k8s.io/kubernetes/pkg/scheduler/cache"
"k8s.io/kubernetes/pkg/scheduler/core" "k8s.io/kubernetes/pkg/scheduler/core"
"k8s.io/kubernetes/pkg/scheduler/core/equivalence" "k8s.io/kubernetes/pkg/scheduler/core/equivalence"
internalqueue "k8s.io/kubernetes/pkg/scheduler/internal/queue"
"k8s.io/kubernetes/pkg/scheduler/util" "k8s.io/kubernetes/pkg/scheduler/util"
"k8s.io/kubernetes/pkg/scheduler/volumebinder" "k8s.io/kubernetes/pkg/scheduler/volumebinder"
) )
@ -81,7 +82,7 @@ var (
type configFactory struct { type configFactory struct {
client clientset.Interface client clientset.Interface
// queue for pods that need scheduling // queue for pods that need scheduling
podQueue core.SchedulingQueue podQueue internalqueue.SchedulingQueue
// a means to list all known scheduled pods. // a means to list all known scheduled pods.
scheduledPodLister corelisters.PodLister scheduledPodLister corelisters.PodLister
// a means to list all known scheduled pods and pods assumed to have been scheduled. // a means to list all known scheduled pods and pods assumed to have been scheduled.
@ -175,7 +176,7 @@ func NewConfigFactory(args *ConfigFactoryArgs) scheduler.Configurator {
c := &configFactory{ c := &configFactory{
client: args.Client, client: args.Client,
podLister: schedulerCache, podLister: schedulerCache,
podQueue: core.NewSchedulingQueue(), podQueue: internalqueue.NewSchedulingQueue(),
nodeLister: args.NodeInformer.Lister(), nodeLister: args.NodeInformer.Lister(),
pVLister: args.PvInformer.Lister(), pVLister: args.PvInformer.Lister(),
pVCLister: args.PvcInformer.Lister(), pVCLister: args.PvcInformer.Lister(),
@ -1348,7 +1349,7 @@ func NewPodInformer(client clientset.Interface, resyncPeriod time.Duration) core
} }
} }
func (c *configFactory) MakeDefaultErrorFunc(backoff *util.PodBackoff, podQueue core.SchedulingQueue) func(pod *v1.Pod, err error) { func (c *configFactory) MakeDefaultErrorFunc(backoff *util.PodBackoff, podQueue internalqueue.SchedulingQueue) func(pod *v1.Pod, err error) {
return func(pod *v1.Pod, err error) { return func(pod *v1.Pod, err error) {
if err == core.ErrNoNodesAvailable { if err == core.ErrNoNodesAvailable {
glog.V(4).Infof("Unable to schedule %v/%v: no nodes are registered to the cluster; waiting", pod.Namespace, pod.Name) glog.V(4).Infof("Unable to schedule %v/%v: no nodes are registered to the cluster; waiting", pod.Namespace, pod.Name)

View File

@ -41,7 +41,7 @@ import (
schedulerapi "k8s.io/kubernetes/pkg/scheduler/api" schedulerapi "k8s.io/kubernetes/pkg/scheduler/api"
latestschedulerapi "k8s.io/kubernetes/pkg/scheduler/api/latest" latestschedulerapi "k8s.io/kubernetes/pkg/scheduler/api/latest"
schedulercache "k8s.io/kubernetes/pkg/scheduler/cache" schedulercache "k8s.io/kubernetes/pkg/scheduler/cache"
"k8s.io/kubernetes/pkg/scheduler/core" internalqueue "k8s.io/kubernetes/pkg/scheduler/internal/queue"
schedulertesting "k8s.io/kubernetes/pkg/scheduler/testing" schedulertesting "k8s.io/kubernetes/pkg/scheduler/testing"
"k8s.io/kubernetes/pkg/scheduler/util" "k8s.io/kubernetes/pkg/scheduler/util"
) )
@ -296,7 +296,7 @@ func TestDefaultErrorFunc(t *testing.T) {
defer server.Close() defer server.Close()
client := clientset.NewForConfigOrDie(&restclient.Config{Host: server.URL, ContentConfig: restclient.ContentConfig{GroupVersion: &schema.GroupVersion{Group: "", Version: "v1"}}}) client := clientset.NewForConfigOrDie(&restclient.Config{Host: server.URL, ContentConfig: restclient.ContentConfig{GroupVersion: &schema.GroupVersion{Group: "", Version: "v1"}}})
factory := newConfigFactory(client, v1.DefaultHardPodAffinitySymmetricWeight) factory := newConfigFactory(client, v1.DefaultHardPodAffinitySymmetricWeight)
queue := &core.FIFO{FIFO: cache.NewFIFO(cache.MetaNamespaceKeyFunc)} queue := &internalqueue.FIFO{FIFO: cache.NewFIFO(cache.MetaNamespaceKeyFunc)}
podBackoff := util.CreatePodBackoff(1*time.Millisecond, 1*time.Second) podBackoff := util.CreatePodBackoff(1*time.Millisecond, 1*time.Second)
errFunc := factory.MakeDefaultErrorFunc(podBackoff, queue) errFunc := factory.MakeDefaultErrorFunc(podBackoff, queue)

View File

@ -24,7 +24,7 @@ limitations under the License.
// FIFO is here for flag-gating purposes and allows us to use the traditional // FIFO is here for flag-gating purposes and allows us to use the traditional
// scheduling queue when util.PodPriorityEnabled() returns false. // scheduling queue when util.PodPriorityEnabled() returns false.
package core package queue
import ( import (
"container/heap" "container/heap"

View File

@ -14,7 +14,7 @@ See the License for the specific language governing permissions and
limitations under the License. limitations under the License.
*/ */
package core package queue
import ( import (
"fmt" "fmt"
@ -27,6 +27,7 @@ import (
"k8s.io/kubernetes/pkg/scheduler/util" "k8s.io/kubernetes/pkg/scheduler/util"
) )
var negPriority, lowPriority, midPriority, highPriority, veryHighPriority = int32(-100), int32(0), int32(100), int32(1000), int32(10000)
var mediumPriority = (lowPriority + highPriority) / 2 var mediumPriority = (lowPriority + highPriority) / 2
var highPriorityPod, highPriNominatedPod, medPriorityPod, unschedulablePod = v1.Pod{ var highPriorityPod, highPriNominatedPod, medPriorityPod, unschedulablePod = v1.Pod{
ObjectMeta: metav1.ObjectMeta{ ObjectMeta: metav1.ObjectMeta{

View File

@ -34,6 +34,7 @@ import (
schedulercache "k8s.io/kubernetes/pkg/scheduler/cache" schedulercache "k8s.io/kubernetes/pkg/scheduler/cache"
"k8s.io/kubernetes/pkg/scheduler/core" "k8s.io/kubernetes/pkg/scheduler/core"
"k8s.io/kubernetes/pkg/scheduler/core/equivalence" "k8s.io/kubernetes/pkg/scheduler/core/equivalence"
internalqueue "k8s.io/kubernetes/pkg/scheduler/internal/queue"
"k8s.io/kubernetes/pkg/scheduler/metrics" "k8s.io/kubernetes/pkg/scheduler/metrics"
"k8s.io/kubernetes/pkg/scheduler/util" "k8s.io/kubernetes/pkg/scheduler/util"
"k8s.io/kubernetes/pkg/scheduler/volumebinder" "k8s.io/kubernetes/pkg/scheduler/volumebinder"
@ -85,7 +86,7 @@ type Configurator interface {
// Exposed for testing // Exposed for testing
GetHardPodAffinitySymmetricWeight() int32 GetHardPodAffinitySymmetricWeight() int32
// Exposed for testing // Exposed for testing
MakeDefaultErrorFunc(backoff *util.PodBackoff, podQueue core.SchedulingQueue) func(pod *v1.Pod, err error) MakeDefaultErrorFunc(backoff *util.PodBackoff, podQueue internalqueue.SchedulingQueue) func(pod *v1.Pod, err error)
// Predicate related accessors to be exposed for use by k8s.io/autoscaler/cluster-autoscaler // Predicate related accessors to be exposed for use by k8s.io/autoscaler/cluster-autoscaler
GetPredicateMetadataProducer() (algorithm.PredicateMetadataProducer, error) GetPredicateMetadataProducer() (algorithm.PredicateMetadataProducer, error)

View File

@ -25,7 +25,7 @@ import (
corelisters "k8s.io/client-go/listers/core/v1" corelisters "k8s.io/client-go/listers/core/v1"
"k8s.io/kubernetes/pkg/scheduler/algorithm" "k8s.io/kubernetes/pkg/scheduler/algorithm"
schedulerapi "k8s.io/kubernetes/pkg/scheduler/api" schedulerapi "k8s.io/kubernetes/pkg/scheduler/api"
"k8s.io/kubernetes/pkg/scheduler/core" internalqueue "k8s.io/kubernetes/pkg/scheduler/internal/queue"
"k8s.io/kubernetes/pkg/scheduler/util" "k8s.io/kubernetes/pkg/scheduler/util"
) )
@ -50,7 +50,7 @@ func (fc *FakeConfigurator) GetHardPodAffinitySymmetricWeight() int32 {
} }
// MakeDefaultErrorFunc is not implemented yet. // MakeDefaultErrorFunc is not implemented yet.
func (fc *FakeConfigurator) MakeDefaultErrorFunc(backoff *util.PodBackoff, podQueue core.SchedulingQueue) func(pod *v1.Pod, err error) { func (fc *FakeConfigurator) MakeDefaultErrorFunc(backoff *util.PodBackoff, podQueue internalqueue.SchedulingQueue) func(pod *v1.Pod, err error) {
return nil return nil
} }