Storable scheduler

api/instance_service.go

@@ -105,13 +105,21 @@ func (s *InstanceAPI) Start(ctx context.Context, in *StartRequest) (*StartReply,
 		"instance_id": in.GetInstanceId(),
 		"state":       lastState.String(),
 	})
+
 	switch lastState.GetState() {
 	case model.InstanceState_REGISTERED:
 		if err := lastState.ValidateGoalState(model.InstanceState_QUEUED); err != nil {
 			flog.Error(err)
 			// TODO: Investigate gRPC error response
 			return nil, err
 		}
+
+		err := s.api.instanceScheduler.Assign(inst)
+		if err != nil {
+			flog.Error(err)
+			return nil, err
+		}
+
 		if err := model.Instances(ctx).UpdateState(in.GetInstanceId(), model.InstanceState_QUEUED); err != nil {
 			flog.Error(err)
 			return nil, err

api/server.go

@@ -7,6 +7,7 @@ import (
 	log "github.com/Sirupsen/logrus"
 	"github.com/axsh/openvdc/model"
 	"github.com/axsh/openvdc/model/backend"
+	"github.com/axsh/openvdc/scheduler"
 	"github.com/gogo/protobuf/proto"
 	mesos "github.com/mesos/mesos-go/mesosproto"
 	sched "github.com/mesos/mesos-go/scheduler"
@@ -18,11 +19,12 @@ import (
 //go:generate protoc -I../proto -I${GOPATH}/src --go_out=plugins=grpc:${GOPATH}/src ../proto/v1.proto
 
 type APIServer struct {
-	server           *grpc.Server
-	modelStoreAddr   backend.ConnectionAddress
-	scheduler        sched.SchedulerDriver
-	mesosMasterAddr  *mesos.Address
-	mMesosMasterAddr sync.Mutex
+	server            *grpc.Server
+	modelStoreAddr    backend.ConnectionAddress
+	scheduler         sched.SchedulerDriver
+	mesosMasterAddr   *mesos.Address
+	mMesosMasterAddr  sync.Mutex
+	instanceScheduler scheduler.Schedule
 }
 
 type serverStreamWithContext struct {
@@ -34,7 +36,7 @@ func (ss *serverStreamWithContext) Context() context.Context {
 	return ss.ctx
 }
 
-func NewAPIServer(modelAddr backend.ConnectionAddress, driver sched.SchedulerDriver, ctx context.Context) *APIServer {
+func NewAPIServer(modelAddr backend.ConnectionAddress, driver sched.SchedulerDriver, instanceScheduler scheduler.Schedule, ctx context.Context) *APIServer {
 	// Assert the ctx has "cluster.backend" key
 	model.GetClusterBackendCtx(ctx)
 
@@ -66,9 +68,10 @@ func NewAPIServer(modelAddr backend.ConnectionAddress, driver sched.SchedulerDri
 		grpc.StreamInterceptor(insertFullMethodStream),
 	}
 	s := &APIServer{
-		server:         grpc.NewServer(sopts...),
-		modelStoreAddr: modelAddr,
-		scheduler:      driver,
+		server:            grpc.NewServer(sopts...),
+		modelStoreAddr:    modelAddr,
+		scheduler:         driver,
+		instanceScheduler: instanceScheduler,
 	}
 
 	RegisterInstanceServer(s.server, &InstanceAPI{api: s})

api/server_test.go

@@ -8,14 +8,18 @@ import (
 	"github.com/axsh/openvdc/internal/unittest"
 	"github.com/axsh/openvdc/model"
 	"github.com/axsh/openvdc/model/backend"
+	"github.com/axsh/openvdc/scheduler"
 	"golang.org/x/net/context"
 )
 
 func TestNewAPIServer(t *testing.T) {
 	ze := &backend.ZkEndpoint{}
 	ze.Set(unittest.TestZkServer)
+
+	sched := scheduler.Schedule{}
+
 	// TODO: Set mock SchedulerDriver
-	s := NewAPIServer(ze, nil, model.WithMockClusterBackendCtx(context.Background()))
+	s := NewAPIServer(ze, nil, sched, model.WithMockClusterBackendCtx(context.Background()))
 	if s == nil {
 		t.Error("NewAPIServer() returned nil")
 	}
@@ -28,8 +32,11 @@ func TestAPIServerRun(t *testing.T) {
 	}
 	ze := &backend.ZkEndpoint{}
 	ze.Set(unittest.TestZkServer)
+
+	sched := scheduler.Schedule{}
+
 	// TODO: Set mock SchedulerDriver
-	s := NewAPIServer(ze, nil, model.WithMockClusterBackendCtx(context.Background()))
+	s := NewAPIServer(ze, nil, sched, model.WithMockClusterBackendCtx(context.Background()))
 	go func() {
 		time.Sleep(2 * time.Second)
 		s.Stop()

ci/citest/acceptance-test/tests/storable_scheduler_test.go

@@ -0,0 +1,14 @@
+// +build acceptance
+
+package tests
+
+import (
+	"testing"
+)
+
+func TestStorable_scheduler_LxcHugeTemplate(t *testing.T) {
+	_, stderr := RunCmdAndExpectFail(t, "openvdc", "run", "centos/7/lxc_huge")
+	if stderr == nil {
+		t.Error("There is no machine can satisfy resource requirement but work")
+	}
+}

cmd/openvdc-scheduler/main.go

@@ -158,7 +158,10 @@ func execute(cmd *cobra.Command, args []string) {
 		log.Fatalln("Faild to bind address for gRPC API: ", viper.GetString("api.endpoint"))
 	}
 	log.Info("Listening gRPC API on: ", viper.GetString("api.endpoint"))
-	grpcServer := api.NewAPIServer(zkAddr, mesosDriver, ctx)
+
+	instanceScheduler := scheduler.Schedule{}
+
+	grpcServer := api.NewAPIServer(zkAddr, mesosDriver, instanceScheduler, ctx)
 
 	if err := detector.Detect(grpcServer); err != nil {
 		log.WithError(err).Fatal("Failed to start mesos detector")

handlers/types.go

@@ -4,9 +4,8 @@ import (
 	"encoding/json"
 	"fmt"
 	"io"
-	"strings"
-
 	"reflect"
+	"strings"
 
 	"github.com/axsh/openvdc/model"
 )
@@ -64,3 +63,7 @@ func FindByType(name string) (p ResourceHandler, ok bool) {
 	p, ok = resourceHandlers[name]
 	return
 }
+
+type InstanceScheduleHandler interface {
+	ScheduleInstance(model.InstanceResource, *model.VDCOffer) (bool, error) // compare with offer and resrouce request.
+}

handlers/vm/esxi/scheduler.go

@@ -0,0 +1,30 @@
+package esxi
+
+import (
+	"github.com/axsh/openvdc/handlers/vm"
+	"github.com/axsh/openvdc/model"
+	"github.com/axsh/openvdc/scheduler"
+)
+
+func init() {
+	scheduler.RegisterInstanceScheduleHandler("vm/esxi", &EsxiScheduler{})
+}
+
+type EsxiScheduler struct {
+}
+
+func NewScheduler() *EsxiScheduler {
+	return new(EsxiScheduler)
+}
+
+func (*EsxiScheduler) ScheduleInstance(ir model.InstanceResource, offer *model.VDCOffer) (bool, error) {
+	cpus := ir.GetVcpu()
+	mem := ir.GetMemoryGb()
+
+	offerCpus := int32(vm.GetOfferScalar(offer, "cpus"))
+	offerMem := int32(vm.GetOfferScalar(offer, "mem") / 1000)
+	if cpus < offerCpus && mem < offerMem {
+		return true, nil
+	}
+	return false, nil
+}

handlers/vm/lxc/lxc_test.go

@@ -16,7 +16,6 @@ func TestResourceName(t *testing.T) {
 
 func TestTypes(t *testing.T) {
 	assert := assert.New(t)
-	assert.Implements((*handlers.ResourceHandler)(nil), &LxcHandler{})
 	assert.Implements((*handlers.CLIHandler)(nil), &LxcHandler{})
 }
 

handlers/vm/lxc/scheduler.go

@@ -0,0 +1,30 @@
+package lxc
+
+import (
+	"github.com/axsh/openvdc/handlers/vm"
+	"github.com/axsh/openvdc/model"
+	"github.com/axsh/openvdc/scheduler"
+)
+
+func init() {
+	scheduler.RegisterInstanceScheduleHandler("vm/lxc", &LxcScheduler{})
+}
+
+type LxcScheduler struct {
+}
+
+func NewScheduler() *LxcScheduler {
+	return new(LxcScheduler)
+}
+
+func (*LxcScheduler) ScheduleInstance(ir model.InstanceResource, offer *model.VDCOffer) (bool, error) {
+	cpus := ir.GetVcpu()
+	mem := ir.GetMemoryGb()
+
+	offerCpus := int32(vm.GetOfferScalar(offer, "cpus"))
+	offerMem := int32(vm.GetOfferScalar(offer, "mem") / 1000)
+	if cpus < offerCpus && mem < offerMem {
+		return true, nil
+	}
+	return false, nil
+}

handlers/vm/null/null_test.go

@@ -14,6 +14,5 @@ func TestResourceName(t *testing.T) {
 
 func TestTypes(t *testing.T) {
 	assert := assert.New(t)
-	assert.Implements((*handlers.ResourceHandler)(nil), &NullHandler{})
 	assert.Implements((*handlers.CLIHandler)(nil), &NullHandler{})
 }

handlers/vm/null/scheduler.go

@@ -0,0 +1,21 @@
+package null
+
+import (
+	"github.com/axsh/openvdc/model"
+	"github.com/axsh/openvdc/scheduler"
+)
+
+func init() {
+	scheduler.RegisterInstanceScheduleHandler("vm/null", &NullScheduler{})
+}
+
+type NullScheduler struct {
+}
+
+func NewScheduler() *NullScheduler {
+	return new(NullScheduler)
+}
+
+func (*NullScheduler) ScheduleInstance(ir model.InstanceResource, offer *model.VDCOffer) (bool, error) {
+	return true, nil
+}

handlers/vm/qemu/qemu_test.go

@@ -16,7 +16,6 @@ func TestResourceName(t *testing.T) {
 
 func TestTypes(t *testing.T) {
 	assert := assert.New(t)
-	assert.Implements((*handlers.ResourceHandler)(nil), &QemuHandler{})
 	assert.Implements((*handlers.CLIHandler)(nil), &QemuHandler{})
 }
 

handlers/vm/qemu/scheduler.go

@@ -0,0 +1,30 @@
+package qemu
+
+import (
+	"github.com/axsh/openvdc/handlers/vm"
+	"github.com/axsh/openvdc/model"
+	"github.com/axsh/openvdc/scheduler"
+)
+
+func init() {
+	scheduler.RegisterInstanceScheduleHandler("vm/qemu", &QemuScheduler{})
+}
+
+type QemuScheduler struct {
+}
+
+func NewScheduler() *QemuScheduler {
+	return new(QemuScheduler)
+}
+
+func (*QemuScheduler) ScheduleInstance(ir model.InstanceResource, offer *model.VDCOffer) (bool, error) {
+	cpus := ir.GetVcpu()
+	mem := ir.GetMemoryGb()
+
+	offerCpus := int32(vm.GetOfferScalar(offer, "cpus"))
+	offerMem := int32(vm.GetOfferScalar(offer, "mem") / 1000)
+	if cpus < offerCpus && mem < offerMem {
+		return true, nil
+	}
+	return false, nil
+}

handlers/vm/util.go

@@ -0,0 +1,26 @@
+package vm
+
+import "github.com/axsh/openvdc/model"
+
+// Generic scheduling util functions
+// TODO change to dont allow non-nullable function
+func GetOfferScalar(offer *model.VDCOffer, name string) float64 {
+	resources := filterResources(offer.Resources, func(res model.VDCOfferResource) bool {
+		return res.Name == name
+	})
+
+	value := 0.0
+	for _, res := range resources {
+		value += res.Scalar
+	}
+	return value
+}
+
+func filterResources(resources []model.VDCOfferResource, filter func(model.VDCOfferResource) bool) (result []model.VDCOfferResource) {
+	for _, resource := range resources {
+		if filter(resource) {
+			result = append(result, resource)
+		}
+	}
+	return result
+}

model/resource_templates.go

@@ -24,6 +24,7 @@ func (*NullTemplate) ResourceName() string    { return "vm/null" }
 
 // InstanceResource is a marker interface for instance template structs.
 type InstanceResource interface {
+	ResourceTemplate
 	isInstanceResourceKind()
 	// protobuf message belongs to InstanceResource should have fields below:
 	//  int32 vcpu = xx;
@@ -72,3 +73,32 @@ func IsMatchingNodeGroups(res InstanceResource, offered []string) bool {
 	}
 	return true
 }
+
+// define openvdc's offer
+type VDCOffer struct {
+	SlaveID   string
+	Resources []VDCOfferResource
+}
+
+type VDCOfferResource struct {
+	Name   string
+	Type   VDCOfferValueType
+	Scalar float64
+	Ranges []VDCOfferValueRange
+	Set    []string
+	// Disk
+}
+
+type VDCOfferValueType int32
+
+const (
+	VDCOfferValueScalar VDCOfferValueType = 0
+	VDCOfferValueRanges VDCOfferValueType = 1
+	VDCOfferValueSet    VDCOfferValueType = 2
+	VDCOfferValueText   VDCOfferValueType = 3
+)
+
+type VDCOfferValueRange struct {
+	Begin uint64
+	End   uint64
+}