diff --git a/plugin/cmd/scheduler/scheduler.go b/plugin/cmd/scheduler/scheduler.go index c2725ca539a..f7727693f87 100644 --- a/plugin/cmd/scheduler/scheduler.go +++ b/plugin/cmd/scheduler/scheduler.go @@ -30,6 +30,8 @@ import ( verflag "github.com/GoogleCloudPlatform/kubernetes/pkg/version/flag" "github.com/GoogleCloudPlatform/kubernetes/pkg/watch" "github.com/GoogleCloudPlatform/kubernetes/plugin/pkg/scheduler" + + "github.com/golang/glog" ) var ( @@ -38,11 +40,11 @@ var ( // storeToMinionLister turns a store into a minion lister. The store must contain (only) minions. type storeToMinionLister struct { - s cache.Store + cache.Store } -func (s storeToMinionLister) List() (machines []string, err error) { - for _, m := range s.s.List() { +func (s *storeToMinionLister) List() (machines []string, err error) { + for _, m := range s.Store.List() { machines = append(machines, m.(*api.Minion).ID) } return machines, nil @@ -50,11 +52,11 @@ func (s storeToMinionLister) List() (machines []string, err error) { // storeToPodLister turns a store into a pod lister. The store must contain (only) pods. type storeToPodLister struct { - s cache.Store + cache.Store } -func (s storeToPodLister) ListPods(selector labels.Selector) (pods []api.Pod, err error) { - for _, m := range s.s.List() { +func (s *storeToPodLister) ListPods(selector labels.Selector) (pods []api.Pod, err error) { + for _, m := range s.List() { pod := m.(*api.Pod) if selector.Matches(labels.Set(pod.Labels)) { pods = append(pods, *pod) @@ -63,13 +65,31 @@ func (s storeToPodLister) ListPods(selector labels.Selector) (pods []api.Pod, er return pods, nil } +// minionEnumerator allows a cache.Poller to enumerate items in an api.PodList +type minionEnumerator struct { + *api.MinionList +} + +// Returns the number of items in the pod list. +func (me *minionEnumerator) Len() int { + if me.MinionList == nil { + return 0 + } + return len(me.Items) +} + +// Returns the item (and ID) with the particular index. +func (me *minionEnumerator) Get(index int) (string, interface{}) { + return me.Items[index].ID, &me.Items[index] +} + type binder struct { - kubeClient *client.Client + *client.Client } // Bind just does a POST binding RPC. -func (b binder) Bind(binding *api.Binding) error { - return b.kubeClient.Post().Path("bindings").Body(binding).Do().Error() +func (b *binder) Bind(binding *api.Binding) error { + return b.Post().Path("bindings").Body(binding).Do().Error() } func main() { @@ -90,8 +110,8 @@ func main() { // This query will only find pods with no assigned host. return kubeClient. Get(). - Path("pods"). Path("watch"). + Path("pods"). SelectorParam("fields", labels.Set{"DesiredState.Host": ""}.AsSelector()). UintParam("resourceVersion", resourceVersion). Watch() @@ -104,8 +124,8 @@ func main() { // This query will only find pods that do have an assigned host. return kubeClient. Get(). - Path("pods"). Path("watch"). + Path("pods"). ParseSelectorParam("fields", "DesiredState.Host!="). UintParam("resourceVersion", resourceVersion). Watch() @@ -114,25 +134,44 @@ func main() { // Watch minions. // Minions may be listed frequently, so provide a local up-to-date cache. minionCache := cache.NewStore() - cache.NewReflector(func(resourceVersion uint64) (watch.Interface, error) { - // This query will only find pods that do have an assigned host. - return kubeClient. - Get(). - Path("minions"). - Path("watch"). - UintParam("resourceVersion", resourceVersion). - Watch() - }, &api.Minion{}, minionCache).Run() + if false { + // Disable this code until minions support watches. + cache.NewReflector(func(resourceVersion uint64) (watch.Interface, error) { + // This query will only find pods that do have an assigned host. + return kubeClient. + Get(). + Path("watch"). + Path("minions"). + UintParam("resourceVersion", resourceVersion). + Watch() + }, &api.Minion{}, minionCache).Run() + } else { + cache.NewPoller(func() (cache.Enumerator, error) { + // This query will only find pods that do have an assigned host. + list := &api.MinionList{} + err := kubeClient.Get().Path("minions").Do().Into(list) + if err != nil { + return nil, err + } + return &minionEnumerator{list}, nil + }, 10*time.Second, minionCache).Run() + } r := rand.New(rand.NewSource(time.Now().UnixNano())) algo := algorithm.NewRandomFitScheduler( - storeToPodLister{podCache}, r) + &storeToPodLister{podCache}, r) s := scheduler.New(&scheduler.Config{ - MinionLister: storeToMinionLister{minionCache}, + MinionLister: &storeToMinionLister{minionCache}, Algorithm: algo, - NextPod: func() *api.Pod { return podQueue.Pop().(*api.Pod) }, - Binder: binder{kubeClient}, + Binder: &binder{kubeClient}, + NextPod: func() *api.Pod { + return podQueue.Pop().(*api.Pod) + }, + Error: func(pod *api.Pod, err error) { + glog.Errorf("Error scheduling %v: %v; retrying", pod.ID, err) + podQueue.Add(pod.ID, pod) + }, }) s.Run() diff --git a/plugin/pkg/scheduler/scheduler.go b/plugin/pkg/scheduler/scheduler.go index 0b5234ec700..2c6f81483e2 100644 --- a/plugin/pkg/scheduler/scheduler.go +++ b/plugin/pkg/scheduler/scheduler.go @@ -31,7 +31,7 @@ type Binder interface { // Scheduler watches for new unscheduled pods. It attempts to find // minions that they fit on and writes bindings back to the api server. type Scheduler struct { - c *Config + config *Config } type Config struct { @@ -53,28 +53,28 @@ type Config struct { // New returns a new scheduler. func New(c *Config) *Scheduler { s := &Scheduler{ - c: c, + config: c, } return s } -// Run begins watching and scheduling. +// Run begins watching and scheduling. It starts a goroutine and returns immediately. func (s *Scheduler) Run() { go util.Forever(s.scheduleOne, 0) } func (s *Scheduler) scheduleOne() { - pod := s.c.NextPod() - dest, err := s.c.Algorithm.Schedule(*pod, s.c.MinionLister) + pod := s.config.NextPod() + dest, err := s.config.Algorithm.Schedule(*pod, s.config.MinionLister) if err != nil { - s.c.Error(pod, err) + s.config.Error(pod, err) return } b := &api.Binding{ PodID: pod.ID, Host: dest, } - if err := s.c.Binder.Bind(b); err != nil { - s.c.Error(pod, err) + if err := s.config.Binder.Bind(b); err != nil { + s.config.Error(pod, err) } }