Commit d14fe0f2 authored by Wojciech Tyczynski's avatar Wojciech Tyczynski

Change storeToNodeConditionLister to return []*api.Node instead of api.NodeList for performance

parent 9b74e24f
......@@ -151,11 +151,11 @@ type storeToNodeConditionLister struct {
}
// List returns a list of nodes that match the conditions defined by the predicate functions in the storeToNodeConditionLister.
func (s storeToNodeConditionLister) List() (nodes api.NodeList, err error) {
func (s storeToNodeConditionLister) List() (nodes []*api.Node, err error) {
for _, m := range s.store.List() {
node := m.(*api.Node)
if s.predicate(node) {
nodes.Items = append(nodes.Items, *node)
nodes = append(nodes, node)
} else {
glog.V(5).Infof("Node %s matches none of the conditions", node.Name)
}
......
......@@ -114,9 +114,9 @@ func TestStoreToNodeConditionLister(t *testing.T) {
if err != nil {
t.Fatalf("Unexpected error: %v", err)
}
got := make([]string, len(gotNodes.Items))
for ix := range gotNodes.Items {
got[ix] = gotNodes.Items[ix].Name
got := make([]string, len(gotNodes))
for ix := range gotNodes {
got[ix] = gotNodes[ix].Name
}
if !want.HasAll(got...) || len(got) != len(want) {
t.Errorf("Expected %v, got %v", want, got)
......
......@@ -629,13 +629,26 @@ func stringSlicesEqual(x, y []string) bool {
return true
}
func includeNodeFromNodeList(node *api.Node) bool {
return !node.Spec.Unschedulable
}
func hostsFromNodeList(list *api.NodeList) []string {
result := []string{}
for ix := range list.Items {
if list.Items[ix].Spec.Unschedulable {
continue
if includeNodeFromNodeList(&list.Items[ix]) {
result = append(result, list.Items[ix].Name)
}
}
return result
}
func hostsFromNodeSlice(nodes []*api.Node) []string {
result := []string{}
for _, node := range nodes {
if includeNodeFromNodeList(node) {
result = append(result, node.Name)
}
result = append(result, list.Items[ix].Name)
}
return result
}
......@@ -675,7 +688,7 @@ func (s *ServiceController) nodeSyncLoop(period time.Duration) {
glog.Errorf("Failed to retrieve current set of nodes from node lister: %v", err)
continue
}
newHosts := hostsFromNodeList(&nodes)
newHosts := hostsFromNodeSlice(nodes)
if stringSlicesEqual(newHosts, prevHosts) {
// The set of nodes in the cluster hasn't changed, but we can retry
// updating any services that we failed to update last time around.
......
......@@ -27,20 +27,23 @@ import (
// NodeLister interface represents anything that can list nodes for a scheduler.
type NodeLister interface {
List() (list api.NodeList, err error)
// We explicitly return []*api.Node, instead of api.NodeList, to avoid
// performing expensive copies that are unneded.
List() ([]*api.Node, error)
}
// FakeNodeLister implements NodeLister on a []string for test purposes.
type FakeNodeLister api.NodeList
type FakeNodeLister []*api.Node
// List returns nodes as a []string.
func (f FakeNodeLister) List() (api.NodeList, error) {
return api.NodeList(f), nil
func (f FakeNodeLister) List() ([]*api.Node, error) {
return f, nil
}
// PodLister interface represents anything that can list pods for a scheduler.
type PodLister interface {
// TODO: make this exactly the same as client's Pods(ns).List() method, by returning a api.PodList
// We explicitly return []*api.Pod, instead of api.PodList, to avoid
// performing expensive copies that are unneded.
List(labels.Selector) ([]*api.Pod, error)
}
......
......@@ -101,12 +101,12 @@ func (ipa *InterPodAffinity) CalculateInterPodAffinityPriority(pod *api.Pod, nod
var maxCount int
var minCount int
counts := map[string]int{}
for _, node := range nodes.Items {
for _, node := range nodes {
totalCount := 0
// count weights for the weighted pod affinity
if affinity.PodAffinity != nil {
for _, weightedTerm := range affinity.PodAffinity.PreferredDuringSchedulingIgnoredDuringExecution {
weightedCount, err := ipa.CountWeightByPodMatchAffinityTerm(pod, allPods, weightedTerm.Weight, weightedTerm.PodAffinityTerm, &node)
weightedCount, err := ipa.CountWeightByPodMatchAffinityTerm(pod, allPods, weightedTerm.Weight, weightedTerm.PodAffinityTerm, node)
if err != nil {
return nil, err
}
......@@ -117,7 +117,7 @@ func (ipa *InterPodAffinity) CalculateInterPodAffinityPriority(pod *api.Pod, nod
// count weights for the weighted pod anti-affinity
if affinity.PodAntiAffinity != nil {
for _, weightedTerm := range affinity.PodAntiAffinity.PreferredDuringSchedulingIgnoredDuringExecution {
weightedCount, err := ipa.CountWeightByPodMatchAffinityTerm(pod, allPods, (0 - weightedTerm.Weight), weightedTerm.PodAffinityTerm, &node)
weightedCount, err := ipa.CountWeightByPodMatchAffinityTerm(pod, allPods, (0 - weightedTerm.Weight), weightedTerm.PodAffinityTerm, node)
if err != nil {
return nil, err
}
......@@ -146,7 +146,7 @@ func (ipa *InterPodAffinity) CalculateInterPodAffinityPriority(pod *api.Pod, nod
//}
for _, epAffinityTerm := range podAffinityTerms {
match, err := ipa.failureDomains.CheckIfPodMatchPodAffinityTerm(pod, ep, epAffinityTerm,
func(pod *api.Pod) (*api.Node, error) { return &node, nil },
func(pod *api.Pod) (*api.Node, error) { return node, nil },
func(ep *api.Pod) (*api.Node, error) { return ipa.info.GetNodeInfo(ep.Spec.NodeName) },
)
if err != nil {
......@@ -161,7 +161,7 @@ func (ipa *InterPodAffinity) CalculateInterPodAffinityPriority(pod *api.Pod, nod
// count weight for the weighted pod affinity indicated by the existing pod.
for _, epWeightedTerm := range epAffinity.PodAffinity.PreferredDuringSchedulingIgnoredDuringExecution {
match, err := ipa.failureDomains.CheckIfPodMatchPodAffinityTerm(pod, ep, epWeightedTerm.PodAffinityTerm,
func(pod *api.Pod) (*api.Node, error) { return &node, nil },
func(pod *api.Pod) (*api.Node, error) { return node, nil },
func(ep *api.Pod) (*api.Node, error) { return ipa.info.GetNodeInfo(ep.Spec.NodeName) },
)
if err != nil {
......@@ -177,7 +177,7 @@ func (ipa *InterPodAffinity) CalculateInterPodAffinityPriority(pod *api.Pod, nod
if epAffinity.PodAntiAffinity != nil {
for _, epWeightedTerm := range epAffinity.PodAntiAffinity.PreferredDuringSchedulingIgnoredDuringExecution {
match, err := ipa.failureDomains.CheckIfPodMatchPodAffinityTerm(pod, ep, epWeightedTerm.PodAffinityTerm,
func(pod *api.Pod) (*api.Node, error) { return &node, nil },
func(pod *api.Pod) (*api.Node, error) { return node, nil },
func(ep *api.Pod) (*api.Node, error) { return ipa.info.GetNodeInfo(ep.Spec.NodeName) },
)
if err != nil {
......@@ -201,7 +201,7 @@ func (ipa *InterPodAffinity) CalculateInterPodAffinityPriority(pod *api.Pod, nod
// calculate final priority score for each node
result := []schedulerapi.HostPriority{}
for _, node := range nodes.Items {
for _, node := range nodes {
fScore := float64(0)
if (maxCount - minCount) > 0 {
fScore = 10 * (float64(counts[node.Name]-minCount) / float64(maxCount-minCount))
......
......@@ -37,7 +37,7 @@ func CalculateNodeAffinityPriority(pod *api.Pod, nodeNameToInfo map[string]*sche
}
var maxCount float64
counts := make(map[string]float64, len(nodes.Items))
counts := make(map[string]float64, len(nodes))
affinity, err := api.GetAffinityFromPodAnnotations(pod.Annotations)
if err != nil {
......@@ -59,7 +59,7 @@ func CalculateNodeAffinityPriority(pod *api.Pod, nodeNameToInfo map[string]*sche
return nil, err
}
for _, node := range nodes.Items {
for _, node := range nodes {
if nodeSelector.Matches(labels.Set(node.Labels)) {
counts[node.Name] += float64(preferredSchedulingTerm.Weight)
}
......@@ -71,9 +71,8 @@ func CalculateNodeAffinityPriority(pod *api.Pod, nodeNameToInfo map[string]*sche
}
}
result := make(schedulerapi.HostPriorityList, 0, len(nodes.Items))
for i := range nodes.Items {
node := &nodes.Items[i]
result := make(schedulerapi.HostPriorityList, 0, len(nodes))
for _, node := range nodes {
if maxCount > 0 {
fScore := 10 * (counts[node.Name] / maxCount)
result = append(result, schedulerapi.HostPriority{Host: node.Name, Score: int(fScore)})
......
......@@ -93,7 +93,7 @@ func TestNodeAffinityPriority(t *testing.T) {
tests := []struct {
pod *api.Pod
nodes []api.Node
nodes []*api.Node
expectedList schedulerapi.HostPriorityList
test string
}{
......@@ -103,7 +103,7 @@ func TestNodeAffinityPriority(t *testing.T) {
Annotations: map[string]string{},
},
},
nodes: []api.Node{
nodes: []*api.Node{
{ObjectMeta: api.ObjectMeta{Name: "machine1", Labels: label1}},
{ObjectMeta: api.ObjectMeta{Name: "machine2", Labels: label2}},
{ObjectMeta: api.ObjectMeta{Name: "machine3", Labels: label3}},
......@@ -117,7 +117,7 @@ func TestNodeAffinityPriority(t *testing.T) {
Annotations: affinity1,
},
},
nodes: []api.Node{
nodes: []*api.Node{
{ObjectMeta: api.ObjectMeta{Name: "machine1", Labels: label4}},
{ObjectMeta: api.ObjectMeta{Name: "machine2", Labels: label2}},
{ObjectMeta: api.ObjectMeta{Name: "machine3", Labels: label3}},
......@@ -131,7 +131,7 @@ func TestNodeAffinityPriority(t *testing.T) {
Annotations: affinity1,
},
},
nodes: []api.Node{
nodes: []*api.Node{
{ObjectMeta: api.ObjectMeta{Name: "machine1", Labels: label1}},
{ObjectMeta: api.ObjectMeta{Name: "machine2", Labels: label2}},
{ObjectMeta: api.ObjectMeta{Name: "machine3", Labels: label3}},
......@@ -145,7 +145,7 @@ func TestNodeAffinityPriority(t *testing.T) {
Annotations: affinity2,
},
},
nodes: []api.Node{
nodes: []*api.Node{
{ObjectMeta: api.ObjectMeta{Name: "machine1", Labels: label1}},
{ObjectMeta: api.ObjectMeta{Name: "machine5", Labels: label5}},
{ObjectMeta: api.ObjectMeta{Name: "machine2", Labels: label2}},
......@@ -156,7 +156,7 @@ func TestNodeAffinityPriority(t *testing.T) {
}
for _, test := range tests {
list, err := CalculateNodeAffinityPriority(test.pod, schedulercache.CreateNodeNameToInfoMap(nil), algorithm.FakeNodeLister(api.NodeList{Items: test.nodes}))
list, err := CalculateNodeAffinityPriority(test.pod, schedulercache.CreateNodeNameToInfoMap(nil), algorithm.FakeNodeLister(test.nodes))
if err != nil {
t.Errorf("unexpected error: %v", err)
}
......
......@@ -86,9 +86,8 @@ func LeastRequestedPriority(pod *api.Pod, nodeNameToInfo map[string]*schedulerca
return schedulerapi.HostPriorityList{}, err
}
list := make(schedulerapi.HostPriorityList, 0, len(nodes.Items))
for i := range nodes.Items {
node := &nodes.Items[i]
list := make(schedulerapi.HostPriorityList, 0, len(nodes))
for _, node := range nodes {
list = append(list, calculateResourceOccupancy(pod, node, nodeNameToInfo[node.Name]))
}
return list, nil
......@@ -118,12 +117,12 @@ func (n *NodeLabelPrioritizer) CalculateNodeLabelPriority(pod *api.Pod, nodeName
}
labeledNodes := map[string]bool{}
for _, node := range nodes.Items {
for _, node := range nodes {
exists := labels.Set(node.Labels).Has(n.label)
labeledNodes[node.Name] = (exists && n.presence) || (!exists && !n.presence)
}
result := make(schedulerapi.HostPriorityList, 0, len(nodes.Items))
result := make(schedulerapi.HostPriorityList, 0, len(nodes))
//score int - scale of 0-10
// 0 being the lowest priority and 10 being the highest
for nodeName, success := range labeledNodes {
......@@ -158,8 +157,7 @@ func ImageLocalityPriority(pod *api.Pod, nodeNameToInfo map[string]*schedulercac
}
for i := range pod.Spec.Containers {
for j := range nodes.Items {
node := &nodes.Items[j]
for _, node := range nodes {
// Check if this container's image is present and get its size.
imageSize := checkContainerImageOnNode(node, &pod.Spec.Containers[i])
// Add this size to the total result of this node.
......@@ -167,7 +165,7 @@ func ImageLocalityPriority(pod *api.Pod, nodeNameToInfo map[string]*schedulercac
}
}
result := make(schedulerapi.HostPriorityList, 0, len(nodes.Items))
result := make(schedulerapi.HostPriorityList, 0, len(nodes))
// score int - scale of 0-10
// 0 being the lowest priority and 10 being the highest.
for nodeName, sumSize := range sumSizeMap {
......@@ -222,9 +220,8 @@ func BalancedResourceAllocation(pod *api.Pod, nodeNameToInfo map[string]*schedul
return schedulerapi.HostPriorityList{}, err
}
list := make(schedulerapi.HostPriorityList, 0, len(nodes.Items))
for i := range nodes.Items {
node := &nodes.Items[i]
list := make(schedulerapi.HostPriorityList, 0, len(nodes))
for _, node := range nodes {
list = append(list, calculateBalancedResourceAllocation(pod, node, nodeNameToInfo[node.Name]))
}
return list, nil
......
......@@ -117,12 +117,13 @@ func (s *SelectorSpread) CalculateSpreadPriority(pod *api.Pod, nodeNameToInfo ma
// Create a number of go-routines that will be computing number
// of "similar" pods for given nodes.
workers := 16
toProcess := make(chan string, len(nodes.Items))
for i := range nodes.Items {
toProcess <- nodes.Items[i].Name
toProcess := make(chan string, len(nodes))
for i := range nodes {
toProcess <- nodes[i].Name
}
close(toProcess)
// TODO: Use Parallelize.
wg := sync.WaitGroup{}
wg.Add(workers)
for i := 0; i < workers; i++ {
......@@ -181,9 +182,7 @@ func (s *SelectorSpread) CalculateSpreadPriority(pod *api.Pod, nodeNameToInfo ma
// Count similar pods by zone, if zone information is present
countsByZone := map[string]int{}
for i := range nodes.Items {
node := &nodes.Items[i]
for _, node := range nodes {
count, found := countsByNodeName[node.Name]
if !found {
continue
......@@ -207,11 +206,10 @@ func (s *SelectorSpread) CalculateSpreadPriority(pod *api.Pod, nodeNameToInfo ma
}
}
result := make(schedulerapi.HostPriorityList, 0, len(nodes.Items))
result := make(schedulerapi.HostPriorityList, 0, len(nodes))
//score int - scale of 0-maxPriority
// 0 being the lowest priority and maxPriority being the highest
for i := range nodes.Items {
node := &nodes.Items[i]
for _, node := range nodes {
// initializing to the default/max node score of maxPriority
fScore := float32(maxPriority)
if maxCountByNodeName > 0 {
......@@ -281,7 +279,7 @@ func (s *ServiceAntiAffinity) CalculateAntiAffinityPriority(pod *api.Pod, nodeNa
// separate out the nodes that have the label from the ones that don't
otherNodes := []string{}
labeledNodes := map[string]string{}
for _, node := range nodes.Items {
for _, node := range nodes {
if labels.Set(node.Labels).Has(s.label) {
label := labels.Set(node.Labels).Get(s.label)
labeledNodes[node.Name] = label
......
......@@ -664,20 +664,18 @@ func TestZoneSpreadPriority(t *testing.T) {
}
}
func makeLabeledNodeList(nodeMap map[string]map[string]string) (result api.NodeList) {
nodes := []api.Node{}
func makeLabeledNodeList(nodeMap map[string]map[string]string) []*api.Node {
nodes := make([]*api.Node, 0, len(nodeMap))
for nodeName, labels := range nodeMap {
nodes = append(nodes, api.Node{ObjectMeta: api.ObjectMeta{Name: nodeName, Labels: labels}})
nodes = append(nodes, &api.Node{ObjectMeta: api.ObjectMeta{Name: nodeName, Labels: labels}})
}
return api.NodeList{Items: nodes}
return nodes
}
func makeNodeList(nodeNames []string) api.NodeList {
result := api.NodeList{
Items: make([]api.Node, len(nodeNames)),
func makeNodeList(nodeNames []string) []*api.Node {
nodes := make([]*api.Node, 0, len(nodeNames))
for _, nodeName := range nodeNames {
nodes = append(nodes, &api.Node{ObjectMeta: api.ObjectMeta{Name: nodeName}})
}
for ix := range nodeNames {
result.Items[ix].Name = nodeNames[ix]
}
return result
return nodes
}
......@@ -61,7 +61,7 @@ func ComputeTaintTolerationPriority(pod *api.Pod, nodeNameToInfo map[string]*sch
// the max value of counts
var maxCount float64
// counts hold the count of intolerable taints of a pod for a given node
counts := make(map[string]float64, len(nodes.Items))
counts := make(map[string]float64, len(nodes))
tolerations, err := api.GetTolerationsFromPodAnnotations(pod.Annotations)
if err != nil {
......@@ -71,8 +71,7 @@ func ComputeTaintTolerationPriority(pod *api.Pod, nodeNameToInfo map[string]*sch
tolerationList := getAllTolerationPreferNoSchedule(tolerations)
// calculate the intolerable taints for all the nodes
for i := range nodes.Items {
node := &nodes.Items[i]
for _, node := range nodes {
taints, err := api.GetTaintsFromNodeAnnotations(node.Annotations)
if err != nil {
return nil, err
......@@ -88,9 +87,8 @@ func ComputeTaintTolerationPriority(pod *api.Pod, nodeNameToInfo map[string]*sch
// The maximum priority value to give to a node
// Priority values range from 0 - maxPriority
const maxPriority = float64(10)
result := make(schedulerapi.HostPriorityList, 0, len(nodes.Items))
for i := range nodes.Items {
node := &nodes.Items[i]
result := make(schedulerapi.HostPriorityList, 0, len(nodes))
for _, node := range nodes {
fScore := maxPriority
if maxCount > 0 {
fScore = (1.0 - counts[node.Name]/maxCount) * 10
......
......@@ -27,9 +27,9 @@ import (
"k8s.io/kubernetes/plugin/pkg/scheduler/schedulercache"
)
func nodeWithTaints(nodeName string, taints []api.Taint) api.Node {
func nodeWithTaints(nodeName string, taints []api.Taint) *api.Node {
taintsData, _ := json.Marshal(taints)
return api.Node{
return &api.Node{
ObjectMeta: api.ObjectMeta{
Name: nodeName,
Annotations: map[string]string{
......@@ -57,7 +57,7 @@ func podWithTolerations(tolerations []api.Toleration) *api.Pod {
func TestTaintAndToleration(t *testing.T) {
tests := []struct {
pod *api.Pod
nodes []api.Node
nodes []*api.Node
expectedList schedulerapi.HostPriorityList
test string
}{
......@@ -70,7 +70,7 @@ func TestTaintAndToleration(t *testing.T) {
Value: "bar",
Effect: api.TaintEffectPreferNoSchedule,
}}),
nodes: []api.Node{
nodes: []*api.Node{
nodeWithTaints("nodeA", []api.Taint{{
Key: "foo",
Value: "bar",
......@@ -103,7 +103,7 @@ func TestTaintAndToleration(t *testing.T) {
Effect: api.TaintEffectPreferNoSchedule,
},
}),
nodes: []api.Node{
nodes: []*api.Node{
nodeWithTaints("nodeA", []api.Taint{}),
nodeWithTaints("nodeB", []api.Taint{
{
......@@ -139,7 +139,7 @@ func TestTaintAndToleration(t *testing.T) {
Value: "bar",
Effect: api.TaintEffectPreferNoSchedule,
}}),
nodes: []api.Node{
nodes: []*api.Node{
nodeWithTaints("nodeA", []api.Taint{}),
nodeWithTaints("nodeB", []api.Taint{
{
......@@ -182,7 +182,7 @@ func TestTaintAndToleration(t *testing.T) {
Effect: api.TaintEffectNoSchedule,
},
}),
nodes: []api.Node{
nodes: []*api.Node{
nodeWithTaints("nodeA", []api.Taint{}),
nodeWithTaints("nodeB", []api.Taint{
{
......@@ -215,7 +215,7 @@ func TestTaintAndToleration(t *testing.T) {
list, err := ComputeTaintTolerationPriority(
test.pod,
nodeNameToInfo,
algorithm.FakeNodeLister(api.NodeList{Items: test.nodes}))
algorithm.FakeNodeLister(test.nodes))
if err != nil {
t.Errorf("%s, unexpected error: %v", test.test, err)
}
......
......@@ -27,12 +27,12 @@ import (
type SchedulerExtender interface {
// Filter based on extender-implemented predicate functions. The filtered list is
// expected to be a subset of the supplied list.
Filter(pod *api.Pod, nodes *api.NodeList) (filteredNodes *api.NodeList, err error)
Filter(pod *api.Pod, nodes []*api.Node) (filteredNodes []*api.Node, err error)
// Prioritize based on extender-implemented priority functions. The returned scores & weight
// are used to compute the weighted score for an extender. The weighted scores are added to
// the scores computed by Kubernetes scheduler. The total scores are used to do the host selection.
Prioritize(pod *api.Pod, nodes *api.NodeList) (hostPriorities *schedulerapi.HostPriorityList, weight int, err error)
Prioritize(pod *api.Pod, nodes []*api.Node) (hostPriorities *schedulerapi.HostPriorityList, weight int, err error)
}
// ScheduleAlgorithm is an interface implemented by things that know how to schedule pods
......
......@@ -93,16 +93,20 @@ func NewHTTPExtender(config *schedulerapi.ExtenderConfig, apiVersion string) (al
// Filter based on extender implemented predicate functions. The filtered list is
// expected to be a subset of the supplied list.
func (h *HTTPExtender) Filter(pod *api.Pod, nodes *api.NodeList) (*api.NodeList, error) {
func (h *HTTPExtender) Filter(pod *api.Pod, nodes []*api.Node) ([]*api.Node, error) {
var result schedulerapi.ExtenderFilterResult
if h.filterVerb == "" {
return nodes, nil
}
nodeItems := make([]api.Node, 0, len(nodes))
for _, node := range nodes {
nodeItems = append(nodeItems, *node)
}
args := schedulerapi.ExtenderArgs{
Pod: *pod,
Nodes: *nodes,
Nodes: api.NodeList{Items: nodeItems},
}
if err := h.send(h.filterVerb, &args, &result); err != nil {
......@@ -111,26 +115,34 @@ func (h *HTTPExtender) Filter(pod *api.Pod, nodes *api.NodeList) (*api.NodeList,
if result.Error != "" {
return nil, fmt.Errorf(result.Error)
}
return &result.Nodes, nil
nodeResult := make([]*api.Node, 0, len(result.Nodes.Items))
for i := range result.Nodes.Items {
nodeResult = append(nodeResult, &result.Nodes.Items[i])
}
return nodeResult, nil
}
// Prioritize based on extender implemented priority functions. Weight*priority is added
// up for each such priority function. The returned score is added to the score computed
// by Kubernetes scheduler. The total score is used to do the host selection.
func (h *HTTPExtender) Prioritize(pod *api.Pod, nodes *api.NodeList) (*schedulerapi.HostPriorityList, int, error) {
func (h *HTTPExtender) Prioritize(pod *api.Pod, nodes []*api.Node) (*schedulerapi.HostPriorityList, int, error) {
var result schedulerapi.HostPriorityList
if h.prioritizeVerb == "" {
result := schedulerapi.HostPriorityList{}
for _, node := range nodes.Items {
for _, node := range nodes {
result = append(result, schedulerapi.HostPriority{Host: node.Name, Score: 0})
}
return &result, 0, nil
}
nodeItems := make([]api.Node, 0, len(nodes))
for _, node := range nodes {
nodeItems = append(nodeItems, *node)
}
args := schedulerapi.ExtenderArgs{
Pod: *pod,
Nodes: *nodes,
Nodes: api.NodeList{Items: nodeItems},
}
if err := h.send(h.prioritizeVerb, &args, &result); err != nil {
......
......@@ -28,7 +28,7 @@ import (
)
type fitPredicate func(pod *api.Pod, node *api.Node) (bool, error)
type priorityFunc func(pod *api.Pod, nodes *api.NodeList) (*schedulerapi.HostPriorityList, error)
type priorityFunc func(pod *api.Pod, nodes []*api.Node) (*schedulerapi.HostPriorityList, error)
type priorityConfig struct {
function priorityFunc
......@@ -61,13 +61,13 @@ func machine2PredicateExtender(pod *api.Pod, node *api.Node) (bool, error) {
return false, nil
}
func errorPrioritizerExtender(pod *api.Pod, nodes *api.NodeList) (*schedulerapi.HostPriorityList, error) {
func errorPrioritizerExtender(pod *api.Pod, nodes []*api.Node) (*schedulerapi.HostPriorityList, error) {
return &schedulerapi.HostPriorityList{}, fmt.Errorf("Some error")
}
func machine1PrioritizerExtender(pod *api.Pod, nodes *api.NodeList) (*schedulerapi.HostPriorityList, error) {
func machine1PrioritizerExtender(pod *api.Pod, nodes []*api.Node) (*schedulerapi.HostPriorityList, error) {
result := schedulerapi.HostPriorityList{}
for _, node := range nodes.Items {
for _, node := range nodes {
score := 1
if node.Name == "machine1" {
score = 10
......@@ -77,9 +77,9 @@ func machine1PrioritizerExtender(pod *api.Pod, nodes *api.NodeList) (*schedulera
return &result, nil
}
func machine2PrioritizerExtender(pod *api.Pod, nodes *api.NodeList) (*schedulerapi.HostPriorityList, error) {
func machine2PrioritizerExtender(pod *api.Pod, nodes []*api.Node) (*schedulerapi.HostPriorityList, error) {
result := schedulerapi.HostPriorityList{}
for _, node := range nodes.Items {
for _, node := range nodes {
score := 1
if node.Name == "machine2" {
score = 10
......@@ -96,7 +96,7 @@ func machine2Prioritizer(_ *api.Pod, nodeNameToInfo map[string]*schedulercache.N
}
result := []schedulerapi.HostPriority{}
for _, node := range nodes.Items {
for _, node := range nodes {
score := 1
if node.Name == "machine2" {
score = 10
......@@ -112,14 +112,14 @@ type FakeExtender struct {
weight int
}
func (f *FakeExtender) Filter(pod *api.Pod, nodes *api.NodeList) (*api.NodeList, error) {
filtered := []api.Node{}
for _, node := range nodes.Items {
func (f *FakeExtender) Filter(pod *api.Pod, nodes []*api.Node) ([]*api.Node, error) {
filtered := []*api.Node{}
for _, node := range nodes {
fits := true
for _, predicate := range f.predicates {
fit, err := predicate(pod, &node)
fit, err := predicate(pod, node)
if err != nil {
return &api.NodeList{}, err
return []*api.Node{}, err
}
if !fit {
fits = false
......@@ -130,10 +130,10 @@ func (f *FakeExtender) Filter(pod *api.Pod, nodes *api.NodeList) (*api.NodeList,
filtered = append(filtered, node)
}
}
return &api.NodeList{Items: filtered}, nil
return filtered, nil
}
func (f *FakeExtender) Prioritize(pod *api.Pod, nodes *api.NodeList) (*schedulerapi.HostPriorityList, int, error) {
func (f *FakeExtender) Prioritize(pod *api.Pod, nodes []*api.Node) (*schedulerapi.HostPriorityList, int, error) {
result := schedulerapi.HostPriorityList{}
combinedScores := map[string]int{}
for _, prioritizer := range f.prioritizers {
......
......@@ -82,7 +82,7 @@ func (g *genericScheduler) Schedule(pod *api.Pod, nodeLister algorithm.NodeListe
if err != nil {
return "", err
}
if len(nodes.Items) == 0 {
if len(nodes) == 0 {
return "", ErrNoNodesAvailable
}
......@@ -98,7 +98,7 @@ func (g *genericScheduler) Schedule(pod *api.Pod, nodeLister algorithm.NodeListe
return "", err
}
if len(filteredNodes.Items) == 0 {
if len(filteredNodes) == 0 {
return "", &FitError{
Pod: pod,
FailedPredicates: failedPredicateMap,
......@@ -136,19 +136,19 @@ func (g *genericScheduler) selectHost(priorityList schedulerapi.HostPriorityList
// Filters the nodes to find the ones that fit based on the given predicate functions
// Each node is passed through the predicate functions to determine if it is a fit
func findNodesThatFit(pod *api.Pod, nodeNameToInfo map[string]*schedulercache.NodeInfo, predicateFuncs map[string]algorithm.FitPredicate, nodes api.NodeList, extenders []algorithm.SchedulerExtender) (api.NodeList, FailedPredicateMap, error) {
func findNodesThatFit(pod *api.Pod, nodeNameToInfo map[string]*schedulercache.NodeInfo, predicateFuncs map[string]algorithm.FitPredicate, nodes []*api.Node, extenders []algorithm.SchedulerExtender) ([]*api.Node, FailedPredicateMap, error) {
// Create filtered list with enough space to avoid growing it.
filtered := make([]api.Node, 0, len(nodes.Items))
filtered := make([]*api.Node, 0, len(nodes))
failedPredicateMap := FailedPredicateMap{}
if len(predicateFuncs) == 0 {
filtered = nodes.Items
filtered = nodes
} else {
predicateResultLock := sync.Mutex{}
errs := []error{}
meta := predicates.PredicateMetadata(pod)
checkNode := func(i int) {
nodeName := nodes.Items[i].Name
nodeName := nodes[i].Name
fits, failedPredicate, err := podFitsOnNode(pod, meta, nodeNameToInfo[nodeName], predicateFuncs)
predicateResultLock.Lock()
......@@ -158,30 +158,30 @@ func findNodesThatFit(pod *api.Pod, nodeNameToInfo map[string]*schedulercache.No
return
}
if fits {
filtered = append(filtered, nodes.Items[i])
filtered = append(filtered, nodes[i])
} else {
failedPredicateMap[nodeName] = failedPredicate
}
}
workqueue.Parallelize(16, len(nodes.Items), checkNode)
workqueue.Parallelize(16, len(nodes), checkNode)
if len(errs) > 0 {
return api.NodeList{}, FailedPredicateMap{}, errors.NewAggregate(errs)
return []*api.Node{}, FailedPredicateMap{}, errors.NewAggregate(errs)
}
}
if len(filtered) > 0 && len(extenders) != 0 {
for _, extender := range extenders {
filteredList, err := extender.Filter(pod, &api.NodeList{Items: filtered})
filteredList, err := extender.Filter(pod, filtered)
if err != nil {
return api.NodeList{}, FailedPredicateMap{}, err
return []*api.Node{}, FailedPredicateMap{}, err
}
filtered = filteredList.Items
filtered = filteredList
if len(filtered) == 0 {
break
}
}
}
return api.NodeList{Items: filtered}, failedPredicateMap, nil
return filtered, failedPredicateMap, nil
}
// Checks whether node with a given name and NodeInfo satisfies all predicateFuncs.
......@@ -288,7 +288,7 @@ func PrioritizeNodes(
wg.Add(1)
go func(ext algorithm.SchedulerExtender) {
defer wg.Done()
prioritizedList, weight, err := ext.Prioritize(pod, &nodes)
prioritizedList, weight, err := ext.Prioritize(pod, nodes)
if err != nil {
// Prioritization errors from extender can be ignored, let k8s/other extenders determine the priorities
return
......@@ -320,8 +320,8 @@ func EqualPriority(_ *api.Pod, nodeNameToInfo map[string]*schedulercache.NodeInf
return []schedulerapi.HostPriority{}, err
}
result := []schedulerapi.HostPriority{}
for _, node := range nodes.Items {
result := make(schedulerapi.HostPriorityList, len(nodes))
for _, node := range nodes {
result = append(result, schedulerapi.HostPriority{
Host: node.Name,
Score: 1,
......
......@@ -66,7 +66,7 @@ func numericPriority(pod *api.Pod, nodeNameToInfo map[string]*schedulercache.Nod
if err != nil {
return nil, fmt.Errorf("failed to list nodes: %v", err)
}
for _, node := range nodes.Items {
for _, node := range nodes {
score, err := strconv.Atoi(node.Name)
if err != nil {
return nil, err
......@@ -102,12 +102,10 @@ func reverseNumericPriority(pod *api.Pod, nodeNameToInfo map[string]*schedulerca
return reverseResult, nil
}
func makeNodeList(nodeNames []string) api.NodeList {
result := api.NodeList{
Items: make([]api.Node, len(nodeNames)),
}
for ix := range nodeNames {
result.Items[ix].Name = nodeNames[ix]
func makeNodeList(nodeNames []string) []*api.Node {
result := make([]*api.Node, 0, len(nodeNames))
for _, nodeName := range nodeNames {
result = append(result, &api.Node{ObjectMeta: api.ObjectMeta{Name: nodeName}})
}
return result
}
......
......@@ -125,7 +125,7 @@ func TestScheduler(t *testing.T) {
},
},
NodeLister: algorithm.FakeNodeLister(
api.NodeList{Items: []api.Node{{ObjectMeta: api.ObjectMeta{Name: "machine1"}}}},
[]*api.Node{{ObjectMeta: api.ObjectMeta{Name: "machine1"}}},
),
Algorithm: item.algo,
Binder: fakeBinder{func(b *api.Binding) error {
......@@ -292,7 +292,7 @@ func setupTestSchedulerWithOnePod(t *testing.T, queuedPodStore *clientcache.FIFO
cfg := &Config{
SchedulerCache: scache,
NodeLister: algorithm.FakeNodeLister(
api.NodeList{Items: []api.Node{{ObjectMeta: api.ObjectMeta{Name: "machine1"}}}},
[]*api.Node{{ObjectMeta: api.ObjectMeta{Name: "machine1"}}},
),
Algorithm: algo,
Binder: fakeBinder{func(b *api.Binding) error {
......
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