github/kubernetes
4
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Rename api.Scheduler -> types.Framework

Browse Source
This commit is contained in:
Dr. Stefan Schimanski
2015-10-26 14:42:25 -05:00
parent de8b958b2f
commit f4e1de55d6
10 changed files with 95 additions and 95 deletions
Download Patch File Download Diff File Expand all files Collapse all files

12
contrib/mesos/pkg/scheduler/mesos_scheduler.go
View File

@@ -957,7 +957,7 @@ func (k *MesosScheduler) NewPluginConfig(terminate <-chan struct{}, mux *http.Se
Config: &plugin.Config{
NodeLister: nil,
Algorithm: &schedulerApiAlgorithmAdapter{
scheduler: scheduler,
fw: scheduler,
podUpdates: podUpdates,
},
Binder: operations.NewBinder(scheduler),
@@ -965,10 +965,10 @@ func (k *MesosScheduler) NewPluginConfig(terminate <-chan struct{}, mux *http.Se
Error: eh.Error,
Recorder: eventBroadcaster.NewRecorder(api.EventSource{Component: "scheduler"}),
},
scheduler: scheduler,
client: k.client,
qr: q,
deleter: podDeleter,
starting: startLatch,
fw: scheduler,
client: k.client,
qr: q,
deleter: podDeleter,
starting: startLatch,
}
}

24
contrib/mesos/pkg/scheduler/operations/binder.go
View File

@@ -21,20 +21,20 @@ import (
"strconv"
log "github.com/golang/glog"
schedapi "k8s.io/kubernetes/contrib/mesos/pkg/scheduler/api"
merrors "k8s.io/kubernetes/contrib/mesos/pkg/scheduler/errors"
annotation "k8s.io/kubernetes/contrib/mesos/pkg/scheduler/meta"
"k8s.io/kubernetes/contrib/mesos/pkg/scheduler/podtask"
types "k8s.io/kubernetes/contrib/mesos/pkg/scheduler/types"
"k8s.io/kubernetes/pkg/api"
)
type Binder struct {
scheduler schedapi.Scheduler
fw types.Framework
}
func NewBinder(scheduler schedapi.Scheduler) *Binder {
func NewBinder(fw types.Framework) *Binder {
return &Binder{
scheduler: scheduler,
fw: fw,
}
}
@@ -49,10 +49,10 @@ func (b *Binder) Bind(binding *api.Binding) error {
return err
}
b.scheduler.Lock()
defer b.scheduler.Unlock()
b.fw.Lock()
defer b.fw.Unlock()
switch task, state := b.scheduler.Tasks().ForPod(podKey); state {
switch task, state := b.fw.Tasks().ForPod(podKey); state {
case podtask.StatePending:
return b.bind(ctx, binding, task)
default:
@@ -66,7 +66,7 @@ func (b *Binder) Bind(binding *api.Binding) error {
func (b *Binder) rollback(task *podtask.T, err error) error {
task.Offer.Release()
task.Reset()
if err2 := b.scheduler.Tasks().Update(task); err2 != nil {
if err2 := b.fw.Tasks().Update(task); err2 != nil {
log.Errorf("failed to update pod task: %v", err2)
}
return err
@@ -88,7 +88,7 @@ func (b *Binder) bind(ctx api.Context, binding *api.Binding, task *podtask.T) (e
// By this time, there is a chance that the slave is disconnected.
offerId := task.GetOfferId()
if offer, ok := b.scheduler.Offers().Get(offerId); !ok || offer.HasExpired() {
if offer, ok := b.fw.Offers().Get(offerId); !ok || offer.HasExpired() {
// already rescinded or timed out or otherwise invalidated
return b.rollback(task, fmt.Errorf("failed prior to launchTask due to expired offer for task %v", task.ID))
}
@@ -96,10 +96,10 @@ func (b *Binder) bind(ctx api.Context, binding *api.Binding, task *podtask.T) (e
if err = b.prepareTaskForLaunch(ctx, binding.Target.Name, task, offerId); err == nil {
log.V(2).Infof("launching task: %q on target %q slave %q for pod \"%v/%v\", cpu %.2f, mem %.2f MB",
task.ID, binding.Target.Name, task.Spec.SlaveID, task.Pod.Namespace, task.Pod.Name, task.Spec.CPU, task.Spec.Memory)
if err = b.scheduler.LaunchTask(task); err == nil {
b.scheduler.Offers().Invalidate(offerId)
if err = b.fw.LaunchTask(task); err == nil {
b.fw.Offers().Invalidate(offerId)
task.Set(podtask.Launched)
if err = b.scheduler.Tasks().Update(task); err != nil {
if err = b.fw.Tasks().Update(task); err != nil {
// this should only happen if the task has been removed or has changed status,
// which SHOULD NOT HAPPEN as long as we're synchronizing correctly
log.Errorf("failed to update task w/ Launched status: %v", err)

26
contrib/mesos/pkg/scheduler/operations/deleter.go
View File

@@ -22,22 +22,22 @@ import (
log "github.com/golang/glog"
"k8s.io/kubernetes/contrib/mesos/pkg/queue"
"k8s.io/kubernetes/contrib/mesos/pkg/runtime"
schedapi "k8s.io/kubernetes/contrib/mesos/pkg/scheduler/api"
merrors "k8s.io/kubernetes/contrib/mesos/pkg/scheduler/errors"
"k8s.io/kubernetes/contrib/mesos/pkg/scheduler/podtask"
"k8s.io/kubernetes/contrib/mesos/pkg/scheduler/queuer"
types "k8s.io/kubernetes/contrib/mesos/pkg/scheduler/types"
"k8s.io/kubernetes/pkg/api"
)
type Deleter struct {
scheduler schedapi.Scheduler
qr *queuer.Queuer
fw types.Framework
qr *queuer.Queuer
}
func NewDeleter(scheduler schedapi.Scheduler, qr *queuer.Queuer) *Deleter {
func NewDeleter(fw types.Framework, qr *queuer.Queuer) *Deleter {
return &Deleter{
scheduler: scheduler,
qr: qr,
fw: fw,
qr: qr,
}
}
@@ -72,8 +72,8 @@ func (k *Deleter) DeleteOne(pod *queuer.Pod) error {
// removing the pod from the scheduling queue. this makes the concurrent
// execution of scheduler-error-handling and delete-handling easier to
// reason about.
k.scheduler.Lock()
defer k.scheduler.Unlock()
k.fw.Lock()
defer k.fw.Unlock()
// prevent the scheduler from attempting to pop this; it's also possible that
// it's concurrently being scheduled (somewhere between pod scheduling and
@@ -81,7 +81,7 @@ func (k *Deleter) DeleteOne(pod *queuer.Pod) error {
// will abort Bind()ing
k.qr.Dequeue(pod.GetUID())
switch task, state := k.scheduler.Tasks().ForPod(podKey); state {
switch task, state := k.fw.Tasks().ForPod(podKey); state {
case podtask.StateUnknown:
log.V(2).Infof("Could not resolve pod '%s' to task id", podKey)
return merrors.NoSuchPodErr
@@ -96,11 +96,11 @@ func (k *Deleter) DeleteOne(pod *queuer.Pod) error {
task.Reset()
task.Set(podtask.Deleted)
//TODO(jdef) probably want better handling here
if err := k.scheduler.Tasks().Update(task); err != nil {
if err := k.fw.Tasks().Update(task); err != nil {
return err
}
}
k.scheduler.Tasks().Unregister(task)
k.fw.Tasks().Unregister(task)
return nil
}
fallthrough
@@ -108,10 +108,10 @@ func (k *Deleter) DeleteOne(pod *queuer.Pod) error {
case podtask.StateRunning:
// signal to watchers that the related pod is going down
task.Set(podtask.Deleted)
if err := k.scheduler.Tasks().Update(task); err != nil {
if err := k.fw.Tasks().Update(task); err != nil {
log.Errorf("failed to update task w/ Deleted status: %v", err)
}
return k.scheduler.KillTask(task.ID)
return k.fw.KillTask(task.ID)
default:
log.Infof("cannot kill pod '%s': non-terminal task not found %v", podKey, task.ID)

13
contrib/mesos/pkg/scheduler/operations/deleter_test.go
View File

@@ -19,20 +19,19 @@ package operations
import (
"testing"
"k8s.io/kubernetes/pkg/api"
mesos "github.com/mesos/mesos-go/mesosproto"
"github.com/stretchr/testify/assert"
"k8s.io/kubernetes/contrib/mesos/pkg/queue"
schedapi "k8s.io/kubernetes/contrib/mesos/pkg/scheduler/api"
merrors "k8s.io/kubernetes/contrib/mesos/pkg/scheduler/errors"
"k8s.io/kubernetes/contrib/mesos/pkg/scheduler/podtask"
"k8s.io/kubernetes/contrib/mesos/pkg/scheduler/queuer"
types "k8s.io/kubernetes/contrib/mesos/pkg/scheduler/types"
"k8s.io/kubernetes/pkg/api"
)
func TestDeleteOne_NonexistentPod(t *testing.T) {
assert := assert.New(t)
obj := &schedapi.MockScheduler{}
obj := &types.MockScheduler{}
reg := podtask.NewInMemoryRegistry()
obj.On("Tasks").Return(reg)
@@ -51,7 +50,7 @@ func TestDeleteOne_NonexistentPod(t *testing.T) {
func TestDeleteOne_PendingPod(t *testing.T) {
assert := assert.New(t)
obj := &schedapi.MockScheduler{}
obj := &types.MockScheduler{}
reg := podtask.NewInMemoryRegistry()
obj.On("Tasks").Return(reg)
@@ -85,7 +84,7 @@ func TestDeleteOne_PendingPod(t *testing.T) {
func TestDeleteOne_Running(t *testing.T) {
assert := assert.New(t)
obj := &schedapi.MockScheduler{}
obj := &types.MockScheduler{}
reg := podtask.NewInMemoryRegistry()
obj.On("Tasks").Return(reg)
@@ -127,7 +126,7 @@ func TestDeleteOne_Running(t *testing.T) {
func TestDeleteOne_badPodNaming(t *testing.T) {
assert := assert.New(t)
obj := &schedapi.MockScheduler{}
obj := &types.MockScheduler{}
pod := &queuer.Pod{Pod: &api.Pod{}}
d := NewDeleter(obj, queuer.New(nil))

32
contrib/mesos/pkg/scheduler/operations/errorhandler.go
View File

@@ -21,26 +21,26 @@ import (
mesos "github.com/mesos/mesos-go/mesosproto"
"k8s.io/kubernetes/contrib/mesos/pkg/backoff"
"k8s.io/kubernetes/contrib/mesos/pkg/queue"
schedapi "k8s.io/kubernetes/contrib/mesos/pkg/scheduler/api"
merrors "k8s.io/kubernetes/contrib/mesos/pkg/scheduler/errors"
"k8s.io/kubernetes/contrib/mesos/pkg/scheduler/podschedulers"
"k8s.io/kubernetes/contrib/mesos/pkg/scheduler/podtask"
"k8s.io/kubernetes/contrib/mesos/pkg/scheduler/queuer"
types "k8s.io/kubernetes/contrib/mesos/pkg/scheduler/types"
"k8s.io/kubernetes/pkg/api"
"k8s.io/kubernetes/pkg/util"
)
type ErrorHandler struct {
scheduler schedapi.Scheduler
backoff *backoff.Backoff
qr *queuer.Queuer
fw types.Framework
backoff *backoff.Backoff
qr *queuer.Queuer
}
func NewErrorHandler(scheduler schedapi.Scheduler, backoff *backoff.Backoff, qr *queuer.Queuer) *ErrorHandler {
func NewErrorHandler(fw types.Framework, backoff *backoff.Backoff, qr *queuer.Queuer) *ErrorHandler {
return &ErrorHandler{
scheduler: scheduler,
backoff: backoff,
qr: qr,
fw: fw,
backoff: backoff,
qr: qr,
}
}
@@ -64,10 +64,10 @@ func (k *ErrorHandler) Error(pod *api.Pod, schedulingErr error) {
}
k.backoff.GC()
k.scheduler.Lock()
defer k.scheduler.Unlock()
k.fw.Lock()
defer k.fw.Unlock()
switch task, state := k.scheduler.Tasks().ForPod(podKey); state {
switch task, state := k.fw.Tasks().ForPod(podKey); state {
case podtask.StateUnknown:
// if we don't have a mapping here any more then someone deleted the pod
log.V(2).Infof("Could not resolve pod to task, aborting pod reschdule: %s", podKey)
@@ -81,16 +81,16 @@ func (k *ErrorHandler) Error(pod *api.Pod, schedulingErr error) {
breakoutEarly := queue.BreakChan(nil)
if schedulingErr == podschedulers.NoSuitableOffersErr {
log.V(3).Infof("adding backoff breakout handler for pod %v", podKey)
breakoutEarly = queue.BreakChan(k.scheduler.Offers().Listen(podKey, func(offer *mesos.Offer) bool {
k.scheduler.Lock()
defer k.scheduler.Unlock()
switch task, state := k.scheduler.Tasks().Get(task.ID); state {
breakoutEarly = queue.BreakChan(k.fw.Offers().Listen(podKey, func(offer *mesos.Offer) bool {
k.fw.Lock()
defer k.fw.Unlock()
switch task, state := k.fw.Tasks().Get(task.ID); state {
case podtask.StatePending:
// Assess fitness of pod with the current offer. The scheduler normally
// "backs off" when it can't find an offer that matches up with a pod.
// The backoff period for a pod can terminate sooner if an offer becomes
// available that matches up.
return !task.Has(podtask.Launched) && k.scheduler.PodScheduler().FitPredicate()(task, offer, nil)
return !task.Has(podtask.Launched) && k.fw.PodScheduler().FitPredicate()(task, offer, nil)
default:
// no point in continuing to check for matching offers
return true

66
contrib/mesos/pkg/scheduler/plugin.go
View File

@@ -27,12 +27,12 @@ import (
"k8s.io/kubernetes/contrib/mesos/pkg/offers"
"k8s.io/kubernetes/contrib/mesos/pkg/queue"
"k8s.io/kubernetes/contrib/mesos/pkg/runtime"
schedapi "k8s.io/kubernetes/contrib/mesos/pkg/scheduler/api"
merrors "k8s.io/kubernetes/contrib/mesos/pkg/scheduler/errors"
"k8s.io/kubernetes/contrib/mesos/pkg/scheduler/operations"
"k8s.io/kubernetes/contrib/mesos/pkg/scheduler/podschedulers"
"k8s.io/kubernetes/contrib/mesos/pkg/scheduler/podtask"
"k8s.io/kubernetes/contrib/mesos/pkg/scheduler/queuer"
types "k8s.io/kubernetes/contrib/mesos/pkg/scheduler/types"
"k8s.io/kubernetes/pkg/api"
apierrors "k8s.io/kubernetes/pkg/api/errors"
"k8s.io/kubernetes/pkg/client/cache"
@@ -99,7 +99,7 @@ func (k *mesosSchedulerApiAdapter) LaunchTask(task *podtask.T) error {
// k8smSchedulingAlgorithm implements the algorithm.ScheduleAlgorithm interface
type schedulerApiAlgorithmAdapter struct {
scheduler schedapi.Scheduler
fw types.Framework
podUpdates queue.FIFO
}
@@ -115,10 +115,10 @@ func (k *schedulerApiAlgorithmAdapter) Schedule(pod *api.Pod, unused algorithm.N
return "", err
}
k.scheduler.Lock()
defer k.scheduler.Unlock()
k.fw.Lock()
defer k.fw.Unlock()
switch task, state := k.scheduler.Tasks().ForPod(podKey); state {
switch task, state := k.fw.Tasks().ForPod(podKey); state {
case podtask.StateUnknown:
// There's a bit of a potential race here, a pod could have been yielded() and
// then before we get *here* it could be deleted.
@@ -133,7 +133,7 @@ func (k *schedulerApiAlgorithmAdapter) Schedule(pod *api.Pod, unused algorithm.N
log.Infof("aborting Schedule, pod has been deleted %+v", pod)
return "", merrors.NoSuchPodErr
}
return k.doSchedule(k.scheduler.Tasks().Register(k.scheduler.CreatePodTask(ctx, pod)))
return k.doSchedule(k.fw.Tasks().Register(k.fw.CreatePodTask(ctx, pod)))
//TODO(jdef) it's possible that the pod state has diverged from what
//we knew previously, we should probably update the task.Pod state here
@@ -163,19 +163,19 @@ func (k *schedulerApiAlgorithmAdapter) doSchedule(task *podtask.T, err error) (s
if task.HasAcceptedOffer() {
// verify that the offer is still on the table
offerId := task.GetOfferId()
if offer, ok := k.scheduler.Offers().Get(offerId); ok && !offer.HasExpired() {
if offer, ok := k.fw.Offers().Get(offerId); ok && !offer.HasExpired() {
// skip tasks that have already have assigned offers
offer = task.Offer
} else {
task.Offer.Release()
task.Reset()
if err = k.scheduler.Tasks().Update(task); err != nil {
if err = k.fw.Tasks().Update(task); err != nil {
return "", err
}
}
}
if err == nil && offer == nil {
offer, err = k.scheduler.PodScheduler().SchedulePod(k.scheduler.Offers(), k.scheduler, task)
offer, err = k.fw.PodScheduler().SchedulePod(k.fw.Offers(), k.fw, task)
}
if err != nil {
return "", err
@@ -185,10 +185,10 @@ func (k *schedulerApiAlgorithmAdapter) doSchedule(task *podtask.T, err error) (s
return "", fmt.Errorf("offer already invalid/expired for task %v", task.ID)
}
slaveId := details.GetSlaveId().GetValue()
if slaveHostName := k.scheduler.SlaveHostNameFor(slaveId); slaveHostName == "" {
if slaveHostName := k.fw.SlaveHostNameFor(slaveId); slaveHostName == "" {
// not much sense in Release()ing the offer here since its owner died
offer.Release()
k.scheduler.Offers().Invalidate(details.Id.GetValue())
k.fw.Offers().Invalidate(details.Id.GetValue())
return "", fmt.Errorf("Slave disappeared (%v) while scheduling task %v", slaveId, task.ID)
} else {
if task.Offer != nil && task.Offer != offer {
@@ -196,9 +196,9 @@ func (k *schedulerApiAlgorithmAdapter) doSchedule(task *podtask.T, err error) (s
}
task.Offer = offer
k.scheduler.PodScheduler().Procurement()(task, details) // TODO(jdef) why is nothing checking the error returned here?
k.fw.PodScheduler().Procurement()(task, details) // TODO(jdef) why is nothing checking the error returned here?
if err := k.scheduler.Tasks().Update(task); err != nil {
if err := k.fw.Tasks().Update(task); err != nil {
offer.Release()
return "", err
}
@@ -208,31 +208,31 @@ func (k *schedulerApiAlgorithmAdapter) doSchedule(task *podtask.T, err error) (s
type PluginConfig struct {
*plugin.Config
scheduler schedapi.Scheduler
client *client.Client
qr *queuer.Queuer
deleter *operations.Deleter
starting chan struct{} // startup latch
fw types.Framework
client *client.Client
qr *queuer.Queuer
deleter *operations.Deleter
starting chan struct{} // startup latch
}
func NewPlugin(c *PluginConfig) PluginInterface {
return &schedulerPlugin{
config: c.Config,
scheduler: c.scheduler,
client: c.client,
qr: c.qr,
deleter: c.deleter,
starting: c.starting,
config: c.Config,
fw: c.fw,
client: c.client,
qr: c.qr,
deleter: c.deleter,
starting: c.starting,
}
}
type schedulerPlugin struct {
config *plugin.Config
scheduler schedapi.Scheduler
client *client.Client
qr *queuer.Queuer
deleter *operations.Deleter
starting chan struct{}
config *plugin.Config
fw types.Framework
client *client.Client
qr *queuer.Queuer
deleter *operations.Deleter
starting chan struct{}
}
func (s *schedulerPlugin) Run(done <-chan struct{}) {
@@ -320,10 +320,10 @@ func (s *schedulerPlugin) reconcileTask(t *podtask.T) {
return
}
s.scheduler.Lock()
defer s.scheduler.Unlock()
s.fw.Lock()
defer s.fw.Unlock()
if _, state := s.scheduler.Tasks().ForPod(podKey); state != podtask.StateUnknown {
if _, state := s.fw.Tasks().ForPod(podKey); state != podtask.StateUnknown {
//TODO(jdef) reconcile the task
log.Errorf("task already registered for pod %v", pod.Name)
return

2
contrib/mesos/pkg/scheduler/plugin_test.go
View File

@@ -809,7 +809,7 @@ func TestPlugin_LifeCycle(t *testing.T) {
podKey, _ := podtask.MakePodKey(api.NewDefaultContext(), pod.Name)
assertext.EventuallyTrue(t, util.ForeverTestTimeout, func() bool {
t, _ := lt.plugin.scheduler.Tasks().ForPod(podKey)
t, _ := lt.plugin.fw.Tasks().ForPod(podKey)
return t == nil
})

6
contrib/mesos/pkg/scheduler/api/doc.go → contrib/mesos/pkg/scheduler/types/doc.go
View File

@@ -14,6 +14,6 @@ See the License for the specific language governing permissions and
limitations under the License.
*/
// Package api contains an abstract scheduler interface, implemented by the
// scheduler plugin and consumed by the scheduler operations.
package api
// Package types contains an abstract framework interface, implemented by the
// MesosScheduler and consumed by the scheduler operations.
package types

2
contrib/mesos/pkg/scheduler/api/mock.go → contrib/mesos/pkg/scheduler/types/mock.go
View File

@@ -14,7 +14,7 @@ See the License for the specific language governing permissions and
limitations under the License.
*/
package api
package types
import (
"sync"

7
contrib/mesos/pkg/scheduler/api/types.go → contrib/mesos/pkg/scheduler/types/types.go
View File

@@ -14,7 +14,7 @@ See the License for the specific language governing permissions and
limitations under the License.
*/
package api
package types
import (
"sync"
@@ -25,8 +25,9 @@ import (
"k8s.io/kubernetes/pkg/api"
)
// scheduler abstraction to allow for easier unit testing
type Scheduler interface {
// Framework abstracts everything other components of the scheduler need from
// the actual MesosScheduler implementation.
type Framework interface {
sync.Locker // synchronize scheduler plugin operations
podschedulers.SlaveIndex