Change all scheduler's Node and Pod list/lookups to be from the snapshot rather than the cache.

Author: ahg-g

pkg/scheduler/BUILD

@@ -28,6 +28,7 @@ go_library(
         "//pkg/scheduler/internal/queue:go_default_library",
         "//pkg/scheduler/listers:go_default_library",
         "//pkg/scheduler/metrics:go_default_library",
+        "//pkg/scheduler/nodeinfo/snapshot:go_default_library",
         "//pkg/scheduler/volumebinder:go_default_library",
         "//staging/src/k8s.io/api/core/v1:go_default_library",
         "//staging/src/k8s.io/api/storage/v1:go_default_library",

pkg/scheduler/algorithm/predicates/metadata_test.go

@@ -355,7 +355,7 @@ func TestPredicateMetadata_AddRemovePod(t *testing.T) {
 			// getMeta creates predicate meta data given the list of pods.
 			getMeta := func(lister fakelisters.PodLister) (*predicateMetadata, map[string]*schedulernodeinfo.NodeInfo) {
 				nodeInfoMap := schedulernodeinfo.CreateNodeNameToInfoMap(lister, test.nodes)
-				_, precompute := NewServiceAffinityPredicate(fakelisters.NodeLister(test.nodes), lister, fakelisters.ServiceLister(test.services), nil)
+				_, precompute := NewServiceAffinityPredicate(fakelisters.NewNodeInfoLister(test.nodes), lister, fakelisters.ServiceLister(test.services), nil)
 				RegisterPredicateMetadataProducer("ServiceAffinityMetaProducer", precompute)
 				meta := GetPredicateMetadata(test.pendingPod, nodeInfoMap)
 				return meta.(*predicateMetadata), nodeInfoMap

pkg/scheduler/algorithm/predicates/predicates.go

@@ -991,10 +991,10 @@ func (n *NodeLabelChecker) CheckNodeLabelPresence(pod *v1.Pod, meta PredicateMet
 
 // ServiceAffinity defines a struct used for creating service affinity predicates.
 type ServiceAffinity struct {
-	nodeLister    schedulerlisters.NodeLister
-	podLister     schedulerlisters.PodLister
-	serviceLister corelisters.ServiceLister
-	labels        []string
+	nodeInfoLister schedulerlisters.NodeInfoLister
+	podLister      schedulerlisters.PodLister
+	serviceLister  corelisters.ServiceLister
+	labels         []string
 }
 
 // serviceAffinityMetadataProducer should be run once by the scheduler before looping through the Predicate.  It is a helper function that
@@ -1024,12 +1024,12 @@ func (s *ServiceAffinity) serviceAffinityMetadataProducer(pm *predicateMetadata)
 }
 
 // NewServiceAffinityPredicate creates a ServiceAffinity.
-func NewServiceAffinityPredicate(nodeLister schedulerlisters.NodeLister, podLister schedulerlisters.PodLister, serviceLister corelisters.ServiceLister, labels []string) (FitPredicate, predicateMetadataProducer) {
+func NewServiceAffinityPredicate(nodeInfoLister schedulerlisters.NodeInfoLister, podLister schedulerlisters.PodLister, serviceLister corelisters.ServiceLister, labels []string) (FitPredicate, predicateMetadataProducer) {
 	affinity := &ServiceAffinity{
-		nodeLister:    nodeLister,
-		podLister:     podLister,
-		serviceLister: serviceLister,
-		labels:        labels,
+		nodeInfoLister: nodeInfoLister,
+		podLister:      podLister,
+		serviceLister:  serviceLister,
+		labels:         labels,
 	}
 	return affinity.checkServiceAffinity, affinity.serviceAffinityMetadataProducer
 }
@@ -1084,11 +1084,11 @@ func (s *ServiceAffinity) checkServiceAffinity(pod *v1.Pod, meta PredicateMetada
 	if len(s.labels) > len(affinityLabels) {
 		if len(services) > 0 {
 			if len(filteredPods) > 0 {
-				nodeWithAffinityLabels, err := s.nodeLister.GetNodeInfo(filteredPods[0].Spec.NodeName)
+				nodeWithAffinityLabels, err := s.nodeInfoLister.Get(filteredPods[0].Spec.NodeName)
 				if err != nil {
 					return false, nil, err
 				}
-				AddUnsetLabelsToMap(affinityLabels, s.labels, labels.Set(nodeWithAffinityLabels.Labels))
+				AddUnsetLabelsToMap(affinityLabels, s.labels, labels.Set(nodeWithAffinityLabels.Node().Labels))
 			}
 		}
 	}
@@ -1192,15 +1192,15 @@ func EssentialPredicates(pod *v1.Pod, meta PredicateMetadata, nodeInfo *schedule
 
 // PodAffinityChecker contains information to check pod affinity.
 type PodAffinityChecker struct {
-	nodeLister schedulerlisters.NodeLister
-	podLister  schedulerlisters.PodLister
+	nodeInfoLister schedulerlisters.NodeInfoLister
+	podLister      schedulerlisters.PodLister
 }
 
 // NewPodAffinityPredicate creates a PodAffinityChecker.
-func NewPodAffinityPredicate(nodeLister schedulerlisters.NodeLister, podLister schedulerlisters.PodLister) FitPredicate {
+func NewPodAffinityPredicate(nodeInfoLister schedulerlisters.NodeInfoLister, podLister schedulerlisters.PodLister) FitPredicate {
 	checker := &PodAffinityChecker{
-		nodeLister: nodeLister,
-		podLister:  podLister,
+		nodeInfoLister: nodeInfoLister,
+		podLister:      podLister,
 	}
 	return checker.InterPodAffinityMatches
 }
@@ -1254,15 +1254,15 @@ func (c *PodAffinityChecker) podMatchesPodAffinityTerms(pod, targetPod *v1.Pod,
 		return false, false, nil
 	}
 	// Namespace and selector of the terms have matched. Now we check topology of the terms.
-	targetPodNodeInfo, err := c.nodeLister.GetNodeInfo(targetPod.Spec.NodeName)
+	targetPodNodeInfo, err := c.nodeInfoLister.Get(targetPod.Spec.NodeName)
 	if err != nil {
 		return false, false, err
 	}
 	for _, term := range terms {
 		if len(term.TopologyKey) == 0 {
 			return false, false, fmt.Errorf("empty topologyKey is not allowed except for PreferredDuringScheduling pod anti-affinity")
 		}
-		if !priorityutil.NodesHaveSameTopologyKey(nodeInfo.Node(), targetPodNodeInfo, term.TopologyKey) {
+		if !priorityutil.NodesHaveSameTopologyKey(nodeInfo.Node(), targetPodNodeInfo.Node(), term.TopologyKey) {
 			return false, true, nil
 		}
 	}
@@ -1327,12 +1327,12 @@ func (c *PodAffinityChecker) getMatchingAntiAffinityTopologyPairsOfPods(pod *v1.
 	topologyMaps := newTopologyPairsMaps()
 
 	for _, existingPod := range existingPods {
-		existingPodNode, err := c.nodeLister.GetNodeInfo(existingPod.Spec.NodeName)
+		existingPodNodeInfo, err := c.nodeInfoLister.Get(existingPod.Spec.NodeName)
 		if err != nil {
 			klog.Errorf("Pod %s has NodeName %q but node is not found", podName(existingPod), existingPod.Spec.NodeName)
 			continue
 		}
-		existingPodTopologyMaps, err := getMatchingAntiAffinityTopologyPairsOfPod(pod, existingPod, existingPodNode)
+		existingPodTopologyMaps, err := getMatchingAntiAffinityTopologyPairsOfPod(pod, existingPod, existingPodNodeInfo.Node())
 		if err != nil {
 			return nil, err
 		}

pkg/scheduler/algorithm/predicates/predicates_test.go

@@ -1858,7 +1858,7 @@ func TestServiceAffinity(t *testing.T) {
 				nodeInfo.SetNode(test.node)
 				nodeInfoMap := map[string]*schedulernodeinfo.NodeInfo{test.node.Name: nodeInfo}
 				// Reimplementing the logic that the scheduler implements: Any time it makes a predicate, it registers any precomputations.
-				predicate, precompute := NewServiceAffinityPredicate(fakelisters.NodeLister(nodes), fakelisters.PodLister(test.pods), fakelisters.ServiceLister(test.services), test.labels)
+				predicate, precompute := NewServiceAffinityPredicate(fakelisters.NewNodeInfoLister(nodes), fakelisters.PodLister(test.pods), fakelisters.ServiceLister(test.services), test.labels)
 				// Register a precomputation or Rewrite the precomputation to a no-op, depending on the state we want to test.
 				RegisterPredicateMetadataProducer("ServiceAffinityMetaProducer", func(pm *predicateMetadata) {
 					if !skipPrecompute {
@@ -2931,8 +2931,8 @@ func TestInterPodAffinity(t *testing.T) {
 			}
 
 			fit := PodAffinityChecker{
-				nodeLister: fakelisters.NodeLister([]*v1.Node{node}),
-				podLister:  fakelisters.PodLister(test.pods),
+				nodeInfoLister: fakelisters.NewNodeInfoLister([]*v1.Node{node}),
+				podLister:      fakelisters.PodLister(test.pods),
 			}
 			nodeInfo := schedulernodeinfo.NewNodeInfo(podsOnNode...)
 			nodeInfo.SetNode(test.node)
@@ -4044,8 +4044,8 @@ func TestInterPodAffinityWithMultipleNodes(t *testing.T) {
 
 			for indexNode, node := range test.nodes {
 				testFit := PodAffinityChecker{
-					nodeLister: fakelisters.NodeLister(test.nodes),
-					podLister:  fakelisters.PodLister(test.pods),
+					nodeInfoLister: fakelisters.NewNodeInfoLister(test.nodes),
+					podLister:      fakelisters.PodLister(test.pods),
 				}
 
 				var meta PredicateMetadata

pkg/scheduler/algorithm/priorities/BUILD

@@ -80,6 +80,7 @@ go_test(
         "//pkg/scheduler/framework/v1alpha1:go_default_library",
         "//pkg/scheduler/listers/fake:go_default_library",
         "//pkg/scheduler/nodeinfo:go_default_library",
+        "//pkg/scheduler/nodeinfo/snapshot:go_default_library",
         "//pkg/scheduler/testing:go_default_library",
         "//pkg/util/parsers:go_default_library",
         "//staging/src/k8s.io/api/apps/v1:go_default_library",

pkg/scheduler/algorithm/priorities/interpod_affinity.go

@@ -34,14 +34,14 @@ import (
 
 // InterPodAffinity contains information to calculate inter pod affinity.
 type InterPodAffinity struct {
-	nodeLister            schedulerlisters.NodeLister
+	nodeInfoLister        schedulerlisters.NodeInfoLister
 	hardPodAffinityWeight int32
 }
 
 // NewInterPodAffinityPriority creates an InterPodAffinity.
-func NewInterPodAffinityPriority(nodeLister schedulerlisters.NodeLister, hardPodAffinityWeight int32) PriorityFunction {
+func NewInterPodAffinityPriority(nodeInfoLister schedulerlisters.NodeInfoLister, hardPodAffinityWeight int32) PriorityFunction {
 	interPodAffinity := &InterPodAffinity{
-		nodeLister:            nodeLister,
+		nodeInfoLister:        nodeInfoLister,
 		hardPodAffinityWeight: hardPodAffinityWeight,
 	}
 	return interPodAffinity.CalculateInterPodAffinityPriority
@@ -118,14 +118,15 @@ func (ipa *InterPodAffinity) CalculateInterPodAffinityPriority(pod *v1.Pod, node
 	var maxCount, minCount int64
 
 	processPod := func(existingPod *v1.Pod) error {
-		existingPodNode, err := ipa.nodeLister.GetNodeInfo(existingPod.Spec.NodeName)
+		existingPodNodeInfo, err := ipa.nodeInfoLister.Get(existingPod.Spec.NodeName)
 		if err != nil {
 			klog.Errorf("Node not found, %v", existingPod.Spec.NodeName)
 			return nil
 		}
 		existingPodAffinity := existingPod.Spec.Affinity
 		existingHasAffinityConstraints := existingPodAffinity != nil && existingPodAffinity.PodAffinity != nil
 		existingHasAntiAffinityConstraints := existingPodAffinity != nil && existingPodAffinity.PodAntiAffinity != nil
+		existingPodNode := existingPodNodeInfo.Node()
 
 		if hasAffinityConstraints {
 			// For every soft pod affinity term of <pod>, if <existingPod> matches the term,

pkg/scheduler/algorithm/priorities/interpod_affinity_test.go

@@ -23,8 +23,7 @@ import (
 	v1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	framework "k8s.io/kubernetes/pkg/scheduler/framework/v1alpha1"
-	fakelisters "k8s.io/kubernetes/pkg/scheduler/listers/fake"
-	schedulernodeinfo "k8s.io/kubernetes/pkg/scheduler/nodeinfo"
+	nodeinfosnapshot "k8s.io/kubernetes/pkg/scheduler/nodeinfo/snapshot"
 	st "k8s.io/kubernetes/pkg/scheduler/testing"
 )
 
@@ -515,12 +514,12 @@ func TestInterPodAffinityPriority(t *testing.T) {
 	}
 	for _, test := range tests {
 		t.Run(test.name, func(t *testing.T) {
-			nodeNameToInfo := schedulernodeinfo.CreateNodeNameToInfoMap(test.pods, test.nodes)
+			snapshot := nodeinfosnapshot.NewSnapshot(test.pods, test.nodes)
 			interPodAffinity := InterPodAffinity{
-				nodeLister:            fakelisters.NodeLister(test.nodes),
+				nodeInfoLister:        snapshot.NodeInfos(),
 				hardPodAffinityWeight: v1.DefaultHardPodAffinitySymmetricWeight,
 			}
-			list, err := interPodAffinity.CalculateInterPodAffinityPriority(test.pod, nodeNameToInfo, test.nodes)
+			list, err := interPodAffinity.CalculateInterPodAffinityPriority(test.pod, snapshot.NodeInfoMap, test.nodes)
 			if err != nil {
 				t.Errorf("unexpected error: %v", err)
 			}
@@ -603,12 +602,12 @@ func TestHardPodAffinitySymmetricWeight(t *testing.T) {
 	}
 	for _, test := range tests {
 		t.Run(test.name, func(t *testing.T) {
-			nodeNameToInfo := schedulernodeinfo.CreateNodeNameToInfoMap(test.pods, test.nodes)
+			snapshot := nodeinfosnapshot.NewSnapshot(test.pods, test.nodes)
 			ipa := InterPodAffinity{
-				nodeLister:            fakelisters.NodeLister(test.nodes),
+				nodeInfoLister:        snapshot.NodeInfos(),
 				hardPodAffinityWeight: test.hardPodAffinityWeight,
 			}
-			list, err := ipa.CalculateInterPodAffinityPriority(test.pod, nodeNameToInfo, test.nodes)
+			list, err := ipa.CalculateInterPodAffinityPriority(test.pod, snapshot.NodeInfoMap, test.nodes)
 			if err != nil {
 				t.Errorf("unexpected error: %v", err)
 			}
@@ -660,14 +659,14 @@ func BenchmarkInterPodAffinityPriority(b *testing.B) {
 	for _, tt := range tests {
 		b.Run(tt.name, func(b *testing.B) {
 			existingPods, allNodes := tt.prepFunc(tt.existingPodsNum, tt.allNodesNum)
-			nodeNameToInfo := schedulernodeinfo.CreateNodeNameToInfoMap(existingPods, allNodes)
+			snapshot := nodeinfosnapshot.NewSnapshot(existingPods, allNodes)
 			interPodAffinity := InterPodAffinity{
-				nodeLister:            fakelisters.NodeLister(allNodes),
+				nodeInfoLister:        snapshot.NodeInfos(),
 				hardPodAffinityWeight: v1.DefaultHardPodAffinitySymmetricWeight,
 			}
 			b.ResetTimer()
 			for i := 0; i < b.N; i++ {
-				interPodAffinity.CalculateInterPodAffinityPriority(tt.pod, nodeNameToInfo, allNodes)
+				interPodAffinity.CalculateInterPodAffinityPriority(tt.pod, snapshot.NodeInfoMap, allNodes)
 			}
 		})
 	}

pkg/scheduler/algorithm_factory.go

@@ -41,13 +41,13 @@ import (
 
 // PluginFactoryArgs are passed to all plugin factory functions.
 type PluginFactoryArgs struct {
+	NodeInfoLister                 schedulerlisters.NodeInfoLister
 	PodLister                      schedulerlisters.PodLister
 	ServiceLister                  corelisters.ServiceLister
 	ControllerLister               corelisters.ReplicationControllerLister
 	ReplicaSetLister               appslisters.ReplicaSetLister
 	StatefulSetLister              appslisters.StatefulSetLister
 	PDBLister                      policylisters.PodDisruptionBudgetLister
-	NodeLister                     schedulerlisters.NodeLister
 	CSINodeLister                  v1beta1storagelisters.CSINodeLister
 	PVLister                       corelisters.PersistentVolumeLister
 	PVCLister                      corelisters.PersistentVolumeClaimLister
@@ -270,7 +270,7 @@ func RegisterCustomFitPredicate(policy schedulerapi.PredicatePolicy) string {
 		if policy.Argument.ServiceAffinity != nil {
 			predicateFactory = func(args PluginFactoryArgs) predicates.FitPredicate {
 				predicate, precomputationFunction := predicates.NewServiceAffinityPredicate(
-					args.NodeLister,
+					args.NodeInfoLister,
 					args.PodLister,
 					args.ServiceLister,
 					policy.Argument.ServiceAffinity.Labels,

pkg/scheduler/algorithmprovider/defaults/register_predicates.go

@@ -92,7 +92,7 @@ func init() {
 	scheduler.RegisterFitPredicateFactory(
 		predicates.MatchInterPodAffinityPred,
 		func(args scheduler.PluginFactoryArgs) predicates.FitPredicate {
-			return predicates.NewPodAffinityPredicate(args.NodeLister, args.PodLister)
+			return predicates.NewPodAffinityPredicate(args.NodeInfoLister, args.PodLister)
 		},
 	)
 

pkg/scheduler/algorithmprovider/defaults/register_priorities.go

@@ -70,7 +70,7 @@ func init() {
 		priorities.InterPodAffinityPriority,
 		scheduler.PriorityConfigFactory{
 			Function: func(args scheduler.PluginFactoryArgs) priorities.PriorityFunction {
-				return priorities.NewInterPodAffinityPriority(args.NodeLister, args.HardPodAffinitySymmetricWeight)
+				return priorities.NewInterPodAffinityPriority(args.NodeInfoLister, args.HardPodAffinitySymmetricWeight)
 			},
 			Weight: 1,
 		},