Merge pull request #1449 from brendandburns/scheduler

Refactor scheduler, use generic scheduler everywhere, delete old code.

pkg/scheduler/generic_scheduler.go

@@ -100,6 +100,24 @@ func getMinHosts(list HostPriorityList) []string {
 	return result
 }
 
+// EqualPriority is a prioritizer function that gives an equal weight of one to all nodes
+func EqualPriority(pod api.Pod, podLister PodLister, minionLister MinionLister) (HostPriorityList, error) {
+	nodes, err := minionLister.List()
+	result := []HostPriority{}
+
+	if err != nil {
+		fmt.Errorf("failed to list nodes: %v", err)
+		return []HostPriority{}, err
+	}
+	for _, minion := range nodes {
+		result = append(result, HostPriority{
+			host:  minion,
+			score: 1,
+		})
+	}
+	return result, nil
+}
+
 func NewGenericScheduler(predicates []FitPredicate, prioritizer PriorityFunction, pods PodLister, random *rand.Rand) Scheduler {
 	return &genericScheduler{
 		predicates:  predicates,

pkg/scheduler/generic_scheduler_test.go

@@ -25,25 +25,16 @@ import (
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/api"
 )
 
-func matchesPredicate(pod api.Pod, existingPods []api.Pod, node string) (bool, error) {
+func falsePredicate(pod api.Pod, existingPods []api.Pod, node string) (bool, error) {
-	return pod.ID == node, nil
+	return false, nil
 }
 
-func evenPriority(pod api.Pod, podLister PodLister, minionLister MinionLister) (HostPriorityList, error) {
+func truePredicate(pod api.Pod, existingPods []api.Pod, node string) (bool, error) {
-	nodes, err := minionLister.List()
+	return true, nil
-	result := []HostPriority{}
+}
 
-	if err != nil {
+func matchesPredicate(pod api.Pod, existingPods []api.Pod, node string) (bool, error) {
-		fmt.Errorf("failed to list nodes: %v", err)
+	return pod.ID == node, nil
-		return []HostPriority{}, err
-	}
-	for _, minion := range nodes {
-		result = append(result, HostPriority{
-			host:  minion,
-			score: 1,
-		})
-	}
-	return result, nil
 }
 
 func numericPriority(pod api.Pod, podLister PodLister, minionLister MinionLister) (HostPriorityList, error) {
@@ -78,13 +69,13 @@ func TestGenericScheduler(t *testing.T) {
 	}{
 		{
 			predicates:  []FitPredicate{falsePredicate},
-			prioritizer: evenPriority,
+			prioritizer: EqualPriority,
 			nodes:       []string{"machine1", "machine2"},
 			expectsErr:  true,
 		},
 		{
 			predicates:  []FitPredicate{truePredicate},
-			prioritizer: evenPriority,
+			prioritizer: EqualPriority,
 			nodes:       []string{"machine1", "machine2"},
 			// Random choice between both, the rand seeded above with zero, chooses "machine2"
 			expectedHost: "machine2",
@@ -92,7 +83,7 @@ func TestGenericScheduler(t *testing.T) {
 		{
 			// Fits on a machine where the pod ID matches the machine name
 			predicates:   []FitPredicate{matchesPredicate},
-			prioritizer:  evenPriority,
+			prioritizer:  EqualPriority,
 			nodes:        []string{"machine1", "machine2"},
 			pod:          api.Pod{JSONBase: api.JSONBase{ID: "machine2"}},
 			expectedHost: "machine2",

pkg/scheduler/predicates.go

@@ -0,0 +1,65 @@
+/*
+Copyright 2014 Google Inc. All rights reserved.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package scheduler
+
+import (
+	"github.com/GoogleCloudPlatform/kubernetes/pkg/api"
+	"github.com/GoogleCloudPlatform/kubernetes/pkg/labels"
+)
+
+func PodFitsPorts(pod api.Pod, existingPods []api.Pod, node string) (bool, error) {
+	for _, scheduledPod := range existingPods {
+		for _, container := range pod.DesiredState.Manifest.Containers {
+			for _, port := range container.Ports {
+				if port.HostPort == 0 {
+					continue
+				}
+				if containsPort(scheduledPod, port) {
+					return false, nil
+				}
+			}
+		}
+	}
+	return true, nil
+}
+
+func containsPort(pod api.Pod, port api.Port) bool {
+	for _, container := range pod.DesiredState.Manifest.Containers {
+		for _, podPort := range container.Ports {
+			if podPort.HostPort == port.HostPort {
+				return true
+			}
+		}
+	}
+	return false
+}
+
+// MapPodsToMachines obtains a list of pods and pivots that list into a map where the keys are host names
+// and the values are the list of pods running on that host.
+func MapPodsToMachines(lister PodLister) (map[string][]api.Pod, error) {
+	machineToPods := map[string][]api.Pod{}
+	// TODO: perform more targeted query...
+	pods, err := lister.ListPods(labels.Everything())
+	if err != nil {
+		return map[string][]api.Pod{}, err
+	}
+	for _, scheduledPod := range pods {
+		host := scheduledPod.CurrentState.Host
+		machineToPods[host] = append(machineToPods[host], scheduledPod)
+	}
+	return machineToPods, nil
+}

pkg/scheduler/predicates_test.go

@@ -0,0 +1,80 @@
+/*
+Copyright 2014 Google Inc. All rights reserved.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package scheduler
+
+import (
+	"testing"
+
+	"github.com/GoogleCloudPlatform/kubernetes/pkg/api"
+)
+
+func TestPodFitsPorts(t *testing.T) {
+	tests := []struct {
+		pod          api.Pod
+		existingPods []api.Pod
+		fits         bool
+		test         string
+	}{
+		{
+			pod:          api.Pod{},
+			existingPods: []api.Pod{},
+			fits:         true,
+			test:         "nothing running",
+		},
+		{
+			pod: newPod("m1", 8080),
+			existingPods: []api.Pod{
+				newPod("m1", 9090),
+			},
+			fits: true,
+			test: "other port",
+		},
+		{
+			pod: newPod("m1", 8080),
+			existingPods: []api.Pod{
+				newPod("m1", 8080),
+			},
+			fits: false,
+			test: "same port",
+		},
+		{
+			pod: newPod("m1", 8000, 8080),
+			existingPods: []api.Pod{
+				newPod("m1", 8080),
+			},
+			fits: false,
+			test: "second port",
+		},
+		{
+			pod: newPod("m1", 8000, 8080),
+			existingPods: []api.Pod{
+				newPod("m1", 8001, 8080),
+			},
+			fits: false,
+			test: "second port",
+		},
+	}
+	for _, test := range tests {
+		fits, err := PodFitsPorts(test.pod, test.existingPods, "machine")
+		if err != nil {
+			t.Errorf("unexpected error: %v")
+		}
+		if test.fits != fits {
+			t.Errorf("%s: expected %v, saw %v", test.test, test.fits, fits)
+		}
+	}
+}

pkg/scheduler/random.go

@@ -1,48 +0,0 @@
-/*
-Copyright 2014 Google Inc. All rights reserved.
-
-Licensed under the Apache License, Version 2.0 (the "License");
-you may not use this file except in compliance with the License.
-You may obtain a copy of the License at
-
-    http://www.apache.org/licenses/LICENSE-2.0
-
-Unless required by applicable law or agreed to in writing, software
-distributed under the License is distributed on an "AS IS" BASIS,
-WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-See the License for the specific language governing permissions and
-limitations under the License.
-*/
-
-package scheduler
-
-import (
-	"math/rand"
-	"sync"
-
-	"github.com/GoogleCloudPlatform/kubernetes/pkg/api"
-)
-
-// RandomScheduler chooses machines uniformly at random.
-type RandomScheduler struct {
-	random     *rand.Rand
-	randomLock sync.Mutex
-}
-
-func NewRandomScheduler(random *rand.Rand) Scheduler {
-	return &RandomScheduler{
-		random: random,
-	}
-}
-
-// Schedule schedules a given pod to a random machine.
-func (s *RandomScheduler) Schedule(pod api.Pod, minionLister MinionLister) (string, error) {
-	machines, err := minionLister.List()
-	if err != nil {
-		return "", err
-	}
-
-	s.randomLock.Lock()
-	defer s.randomLock.Unlock()
-	return machines[s.random.Int()%len(machines)], nil
-}

pkg/scheduler/random_test.go

@@ -1,34 +0,0 @@
-/*
-Copyright 2014 Google Inc. All rights reserved.
-
-Licensed under the Apache License, Version 2.0 (the "License");
-you may not use this file except in compliance with the License.
-You may obtain a copy of the License at
-
-    http://www.apache.org/licenses/LICENSE-2.0
-
-Unless required by applicable law or agreed to in writing, software
-distributed under the License is distributed on an "AS IS" BASIS,
-WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-See the License for the specific language governing permissions and
-limitations under the License.
-*/
-
-package scheduler
-
-import (
-	"math/rand"
-	"testing"
-
-	"github.com/GoogleCloudPlatform/kubernetes/pkg/api"
-)
-
-func TestRandomScheduler(t *testing.T) {
-	random := rand.New(rand.NewSource(0))
-	st := schedulerTester{
-		t:            t,
-		scheduler:    NewRandomScheduler(random),
-		minionLister: FakeMinionLister{"m1", "m2", "m3", "m4"},
-	}
-	st.expectSuccess(api.Pod{})
-}

pkg/scheduler/randomfit.go

@@ -1,127 +0,0 @@
-/*
-Copyright 2014 Google Inc. All rights reserved.
-
-Licensed under the Apache License, Version 2.0 (the "License");
-you may not use this file except in compliance with the License.
-You may obtain a copy of the License at
-
-    http://www.apache.org/licenses/LICENSE-2.0
-
-Unless required by applicable law or agreed to in writing, software
-distributed under the License is distributed on an "AS IS" BASIS,
-WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-See the License for the specific language governing permissions and
-limitations under the License.
-*/
-
-package scheduler
-
-import (
-	"fmt"
-	"math/rand"
-	"sync"
-
-	"github.com/GoogleCloudPlatform/kubernetes/pkg/api"
-	"github.com/GoogleCloudPlatform/kubernetes/pkg/labels"
-)
-
-// RandomFitScheduler is a Scheduler which schedules a Pod on a random machine which matches its requirement.
-type RandomFitScheduler struct {
-	podLister  PodLister
-	predicates []FitPredicate
-	random     *rand.Rand
-	randomLock sync.Mutex
-}
-
-// NewRandomFitScheduler creates a random fit scheduler with the default set of fit predicates
-func NewRandomFitScheduler(podLister PodLister, random *rand.Rand) Scheduler {
-	return NewRandomFitSchedulerWithPredicates(podLister, random, []FitPredicate{podFitsPorts})
-}
-
-// NewRandomFitScheduler creates a random fit scheduler with the specified set of fit predicates.
-// All predicates must be true for the pod to be considered a fit.
-func NewRandomFitSchedulerWithPredicates(podLister PodLister, random *rand.Rand, predicates []FitPredicate) Scheduler {
-	return &RandomFitScheduler{
-		podLister:  podLister,
-		random:     random,
-		predicates: predicates,
-	}
-}
-
-func podFitsPorts(pod api.Pod, existingPods []api.Pod, node string) (bool, error) {
-	for _, scheduledPod := range existingPods {
-		for _, container := range pod.DesiredState.Manifest.Containers {
-			for _, port := range container.Ports {
-				if port.HostPort == 0 {
-					continue
-				}
-				if containsPort(scheduledPod, port) {
-					return false, nil
-				}
-			}
-		}
-	}
-	return true, nil
-}
-
-func containsPort(pod api.Pod, port api.Port) bool {
-	for _, container := range pod.DesiredState.Manifest.Containers {
-		for _, podPort := range container.Ports {
-			if podPort.HostPort == port.HostPort {
-				return true
-			}
-		}
-	}
-	return false
-}
-
-// MapPodsToMachines obtains a list of pods and pivots that list into a map where the keys are host names
-// and the values are the list of pods running on that host.
-func MapPodsToMachines(lister PodLister) (map[string][]api.Pod, error) {
-	machineToPods := map[string][]api.Pod{}
-	// TODO: perform more targeted query...
-	pods, err := lister.ListPods(labels.Everything())
-	if err != nil {
-		return map[string][]api.Pod{}, err
-	}
-	for _, scheduledPod := range pods {
-		host := scheduledPod.CurrentState.Host
-		machineToPods[host] = append(machineToPods[host], scheduledPod)
-	}
-	return machineToPods, nil
-}
-
-// Schedule schedules a pod on a random machine which matches its requirement.
-func (s *RandomFitScheduler) Schedule(pod api.Pod, minionLister MinionLister) (string, error) {
-	machines, err := minionLister.List()
-	if err != nil {
-		return "", err
-	}
-	machineToPods, err := MapPodsToMachines(s.podLister)
-	if err != nil {
-		return "", err
-	}
-	var machineOptions []string
-	for _, machine := range machines {
-		podFits := true
-		for _, predicate := range s.predicates {
-			fits, err := predicate(pod, machineToPods[machine], machine)
-			if err != nil {
-				return "", err
-			}
-			if !fits {
-				podFits = false
-				break
-			}
-		}
-		if podFits {
-			machineOptions = append(machineOptions, machine)
-		}
-	}
-	if len(machineOptions) == 0 {
-		return "", fmt.Errorf("failed to find fit for %#v", pod)
-	}
-	s.randomLock.Lock()
-	defer s.randomLock.Unlock()
-	return machineOptions[s.random.Int()%len(machineOptions)], nil
-}

pkg/scheduler/randomfit_test.go

@@ -1,116 +0,0 @@
-/*
-Copyright 2014 Google Inc. All rights reserved.
-
-Licensed under the Apache License, Version 2.0 (the "License");
-you may not use this file except in compliance with the License.
-You may obtain a copy of the License at
-
-    http://www.apache.org/licenses/LICENSE-2.0
-
-Unless required by applicable law or agreed to in writing, software
-distributed under the License is distributed on an "AS IS" BASIS,
-WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-See the License for the specific language governing permissions and
-limitations under the License.
-*/
-
-package scheduler
-
-import (
-	"math/rand"
-	"testing"
-
-	"github.com/GoogleCloudPlatform/kubernetes/pkg/api"
-)
-
-func TestRandomFitSchedulerNothingScheduled(t *testing.T) {
-	fakeRegistry := FakePodLister{}
-	r := rand.New(rand.NewSource(0))
-	st := schedulerTester{
-		t:            t,
-		scheduler:    NewRandomFitScheduler(&fakeRegistry, r),
-		minionLister: FakeMinionLister{"m1", "m2", "m3"},
-	}
-	st.expectSchedule(api.Pod{}, "m3")
-}
-
-func TestRandomFitSchedulerFirstScheduled(t *testing.T) {
-	fakeRegistry := FakePodLister{
-		newPod("m1", 8080),
-	}
-	r := rand.New(rand.NewSource(0))
-	st := schedulerTester{
-		t:            t,
-		scheduler:    NewRandomFitScheduler(fakeRegistry, r),
-		minionLister: FakeMinionLister{"m1", "m2", "m3"},
-	}
-	st.expectSchedule(newPod("", 8080), "m3")
-}
-
-func TestRandomFitSchedulerFirstScheduledComplicated(t *testing.T) {
-	fakeRegistry := FakePodLister{
-		newPod("m1", 80, 8080),
-		newPod("m2", 8081, 8082, 8083),
-		newPod("m3", 80, 443, 8085),
-	}
-	r := rand.New(rand.NewSource(0))
-	st := schedulerTester{
-		t:            t,
-		scheduler:    NewRandomFitScheduler(fakeRegistry, r),
-		minionLister: FakeMinionLister{"m1", "m2", "m3"},
-	}
-	st.expectSchedule(newPod("", 8080, 8081), "m3")
-}
-
-func TestRandomFitSchedulerFirstScheduledImpossible(t *testing.T) {
-	fakeRegistry := FakePodLister{
-		newPod("m1", 8080),
-		newPod("m2", 8081),
-		newPod("m3", 8080),
-	}
-	r := rand.New(rand.NewSource(0))
-	st := schedulerTester{
-		t:            t,
-		scheduler:    NewRandomFitScheduler(fakeRegistry, r),
-		minionLister: FakeMinionLister{"m1", "m2", "m3"},
-	}
-	st.expectFailure(newPod("", 8080, 8081))
-}
-
-func falsePredicate(pod api.Pod, existingPods []api.Pod, node string) (bool, error) {
-	return false, nil
-}
-
-func truePredicate(pod api.Pod, existingPods []api.Pod, node string) (bool, error) {
-	return true, nil
-}
-
-func TestRandomFitSchedulerFirstScheduledComplicatedWithMultiplePredicates(t *testing.T) {
-	fakeRegistry := FakePodLister{
-		newPod("m1", 80, 8080),
-		newPod("m2", 8081, 8082, 8083),
-		newPod("m3", 80, 443, 8085),
-	}
-	r := rand.New(rand.NewSource(0))
-	st := schedulerTester{
-		t:            t,
-		scheduler:    NewRandomFitSchedulerWithPredicates(fakeRegistry, r, []FitPredicate{podFitsPorts, truePredicate}),
-		minionLister: FakeMinionLister{"m1", "m2", "m3"},
-	}
-	st.expectSchedule(newPod("", 8080, 8081), "m3")
-}
-
-func TestRandomFitSchedulerFailureManyPredicates(t *testing.T) {
-	fakeRegistry := FakePodLister{
-		newPod("m1", 8080),
-		newPod("m2", 8081),
-		newPod("m3", 8080),
-	}
-	r := rand.New(rand.NewSource(0))
-	st := schedulerTester{
-		t:            t,
-		scheduler:    NewRandomFitSchedulerWithPredicates(fakeRegistry, r, []FitPredicate{truePredicate, falsePredicate}),
-		minionLister: FakeMinionLister{"m1", "m2", "m3"},
-	}
-	st.expectFailure(newPod("", 8080, 8081))
-}

pkg/scheduler/roundrobin.go

@@ -1,43 +0,0 @@
-/*
-Copyright 2014 Google Inc. All rights reserved.
-
-Licensed under the Apache License, Version 2.0 (the "License");
-you may not use this file except in compliance with the License.
-You may obtain a copy of the License at
-
-    http://www.apache.org/licenses/LICENSE-2.0
-
-Unless required by applicable law or agreed to in writing, software
-distributed under the License is distributed on an "AS IS" BASIS,
-WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-See the License for the specific language governing permissions and
-limitations under the License.
-*/
-
-package scheduler
-
-import (
-	"github.com/GoogleCloudPlatform/kubernetes/pkg/api"
-)
-
-// RoundRobinScheduler chooses machines in order.
-type RoundRobinScheduler struct {
-	currentIndex int
-}
-
-func NewRoundRobinScheduler() Scheduler {
-	return &RoundRobinScheduler{
-		currentIndex: -1,
-	}
-}
-
-// Schedule schedules a pod on the machine next to the last scheduled machine.
-func (s *RoundRobinScheduler) Schedule(pod api.Pod, minionLister MinionLister) (string, error) {
-	machines, err := minionLister.List()
-	if err != nil {
-		return "", err
-	}
-	s.currentIndex = (s.currentIndex + 1) % len(machines)
-	result := machines[s.currentIndex]
-	return result, nil
-}

pkg/scheduler/roundrobin_test.go

@@ -1,35 +0,0 @@
-/*
-Copyright 2014 Google Inc. All rights reserved.
-
-Licensed under the Apache License, Version 2.0 (the "License");
-you may not use this file except in compliance with the License.
-You may obtain a copy of the License at
-
-    http://www.apache.org/licenses/LICENSE-2.0
-
-Unless required by applicable law or agreed to in writing, software
-distributed under the License is distributed on an "AS IS" BASIS,
-WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-See the License for the specific language governing permissions and
-limitations under the License.
-*/
-
-package scheduler
-
-import (
-	"testing"
-
-	"github.com/GoogleCloudPlatform/kubernetes/pkg/api"
-)
-
-func TestRoundRobinScheduler(t *testing.T) {
-	st := schedulerTester{
-		t:            t,
-		scheduler:    NewRoundRobinScheduler(),
-		minionLister: FakeMinionLister{"m1", "m2", "m3", "m4"},
-	}
-	st.expectSchedule(api.Pod{}, "m1")
-	st.expectSchedule(api.Pod{}, "m2")
-	st.expectSchedule(api.Pod{}, "m3")
-	st.expectSchedule(api.Pod{}, "m4")
-}

plugin/pkg/scheduler/factory/factory.go

@@ -62,7 +62,11 @@ func (factory *ConfigFactory) Create() *scheduler.Config {
 	}
 
 	r := rand.New(rand.NewSource(time.Now().UnixNano()))
-	algo := algorithm.NewRandomFitScheduler(
+	algo := algorithm.NewGenericScheduler(
+		// Fit is defined based on the absence of port conflicts.
+		[]algorithm.FitPredicate{algorithm.PodFitsPorts},
+		// All nodes where things fit are equally likely (Random)
+		algorithm.EqualPriority,
 		&storeToPodLister{podCache}, r)
 
 	return &scheduler.Config{