Initial version of the Orchard orchestration system (#3)

* Initial version of the Orchard orchestration system

* Update README.md

Co-authored-by: Fedor Korotkov <fedor.korotkov@gmail.com>

Co-authored-by: Fedor Korotkov <fedor.korotkov@gmail.com>

internal/controller/api.go

@@ -0,0 +1,77 @@
+package controller
+
+import (
+	storepkg "github.com/cirruslabs/orchard/internal/controller/store"
+	"github.com/cirruslabs/orchard/internal/responder"
+	"github.com/gin-gonic/gin"
+	"net/http"
+)
+
+type storeTxFunc func(cb func(txn *storepkg.Txn) error) error
+type apiTxFunc func(txn *storepkg.Txn) responder.Responder
+
+func (controller *Controller) initAPI() *gin.Engine {
+	gin.SetMode(gin.DebugMode)
+	ginEngine := gin.Default()
+
+	// v1 API
+	v1 := ginEngine.Group("/v1")
+
+	// Workers
+	v1.POST("/workers", func(c *gin.Context) {
+		controller.createWorker(c).Respond(c)
+	})
+	v1.PUT("/workers/:name", func(c *gin.Context) {
+		controller.updateWorker(c).Respond(c)
+	})
+	v1.GET("/workers/:name", func(c *gin.Context) {
+		controller.getWorker(c).Respond(c)
+	})
+	v1.GET("/workers", func(c *gin.Context) {
+		controller.listWorkers(c).Respond(c)
+	})
+	v1.DELETE("/workers/:name", func(c *gin.Context) {
+		controller.deleteWorker(c).Respond(c)
+	})
+
+	// VMs
+	v1.POST("/vms", func(c *gin.Context) {
+		controller.createVM(c).Respond(c)
+	})
+	v1.PUT("/vms/:name", func(c *gin.Context) {
+		controller.updateVM(c).Respond(c)
+	})
+	v1.GET("/vms/:name", func(c *gin.Context) {
+		controller.getVM(c).Respond(c)
+	})
+	v1.GET("/vms", func(c *gin.Context) {
+		controller.listVMs(c).Respond(c)
+	})
+	v1.DELETE("/vms/:name", func(c *gin.Context) {
+		controller.deleteVM(c).Respond(c)
+	})
+
+	return ginEngine
+}
+
+func (controller *Controller) storeView(cb apiTxFunc) responder.Responder {
+	return mapTxFuncs(controller.store.View, cb)
+}
+
+func (controller *Controller) storeUpdate(cb apiTxFunc) responder.Responder {
+	return mapTxFuncs(controller.store.Update, cb)
+}
+
+func mapTxFuncs(txFunc storeTxFunc, cb apiTxFunc) responder.Responder {
+	var result responder.Responder
+
+	if err := txFunc(func(txn *storepkg.Txn) error {
+		result = cb(txn)
+
+		return nil
+	}); err != nil {
+		return responder.Code(http.StatusInternalServerError)
+	}
+
+	return result
+}

internal/controller/api_vms.go

@@ -0,0 +1,145 @@
+package controller
+
+import (
+	"errors"
+	storepkg "github.com/cirruslabs/orchard/internal/controller/store"
+	"github.com/cirruslabs/orchard/internal/responder"
+	"github.com/cirruslabs/orchard/pkg/resource/v1"
+	"github.com/gin-gonic/gin"
+	"github.com/google/uuid"
+	"net/http"
+	"time"
+)
+
+func (controller *Controller) createVM(ctx *gin.Context) responder.Responder {
+	var vm v1.VM
+
+	if err := ctx.ShouldBindJSON(&vm); err != nil {
+		return responder.Code(http.StatusBadRequest)
+	}
+
+	if vm.Name == "" || vm.Image == "" || vm.CPU == 0 || vm.Memory == 0 {
+		return responder.Code(http.StatusPreconditionFailed)
+	}
+
+	vm.Status = v1.VMStatusPending
+	vm.CreatedAt = time.Now()
+	vm.DeletedAt = time.Time{}
+	vm.UID = uuid.New().String()
+	vm.Generation = 0
+
+	return controller.storeUpdate(func(txn *storepkg.Txn) responder.Responder {
+		// Does the VM resource with this name already exists?
+		_, err := txn.GetVM(vm.Name)
+		if !errors.Is(err, storepkg.ErrNotFound) {
+			return responder.Code(http.StatusConflict)
+		}
+
+		if err := txn.SetVM(&vm); err != nil {
+			return responder.Code(http.StatusInternalServerError)
+		}
+
+		return responder.JSON(http.StatusOK, &vm)
+	})
+}
+
+func (controller *Controller) updateVM(ctx *gin.Context) responder.Responder {
+	var userVM v1.VM
+
+	if err := ctx.ShouldBindJSON(&userVM); err != nil {
+		return responder.Code(http.StatusBadRequest)
+	}
+
+	if userVM.Name == "" {
+		return responder.Code(http.StatusPreconditionFailed)
+	}
+
+	return controller.storeUpdate(func(txn *storepkg.Txn) responder.Responder {
+		dbVM, err := txn.GetVM(userVM.Name)
+		if err != nil {
+			if errors.Is(err, storepkg.ErrNotFound) {
+				return responder.Code(http.StatusNotFound)
+			}
+
+			return responder.Code(http.StatusInternalServerError)
+		}
+
+		dbVM.Status = userVM.Status
+		dbVM.Generation++
+
+		if err := txn.SetVM(dbVM); err != nil {
+			return responder.Code(http.StatusInternalServerError)
+		}
+
+		return responder.JSON(http.StatusOK, &dbVM)
+	})
+}
+
+func (controller *Controller) getVM(ctx *gin.Context) responder.Responder {
+	name := ctx.Param("name")
+
+	return controller.storeView(func(txn *storepkg.Txn) responder.Responder {
+		vm, err := txn.GetVM(name)
+		if err != nil {
+			if errors.Is(err, storepkg.ErrNotFound) {
+				return responder.Code(http.StatusNotFound)
+			}
+
+			return responder.Code(http.StatusInternalServerError)
+		}
+
+		return responder.JSON(http.StatusOK, &vm)
+	})
+}
+
+func (controller *Controller) listVMs(ctx *gin.Context) responder.Responder {
+	return controller.storeView(func(txn *storepkg.Txn) responder.Responder {
+		vms, err := txn.ListVMs()
+		if err != nil {
+			if errors.Is(err, storepkg.ErrNotFound) {
+				return responder.Code(http.StatusNotFound)
+			}
+
+			return responder.Code(http.StatusInternalServerError)
+		}
+
+		return responder.JSON(http.StatusOK, &vms)
+	})
+}
+
+func (controller *Controller) deleteVM(ctx *gin.Context) responder.Responder {
+	name := ctx.Param("name")
+
+	if ctx.Query("force") != "" {
+		return controller.storeUpdate(func(txn *storepkg.Txn) responder.Responder {
+			if err := txn.DeleteVM(name); err != nil {
+				if errors.Is(err, storepkg.ErrNotFound) {
+					return responder.Code(http.StatusNotFound)
+				}
+
+				return responder.Code(http.StatusInternalServerError)
+			}
+
+			return responder.Code(http.StatusOK)
+		})
+	}
+
+	return controller.storeUpdate(func(txn *storepkg.Txn) responder.Responder {
+		vm, err := txn.GetVM(name)
+		if err != nil {
+			if errors.Is(err, storepkg.ErrNotFound) {
+				return responder.Code(http.StatusNotFound)
+			}
+
+			return responder.Code(http.StatusInternalServerError)
+		}
+
+		vm.DeletedAt = time.Now()
+
+		if err := txn.SetVM(vm); err != nil {
+			return responder.Code(http.StatusInternalServerError)
+		}
+
+		return responder.Code(http.StatusOK)
+	})
+}

internal/controller/api_workers.go

@@ -0,0 +1,123 @@
+package controller
+
+import (
+	"errors"
+	storepkg "github.com/cirruslabs/orchard/internal/controller/store"
+	"github.com/cirruslabs/orchard/internal/responder"
+	v1 "github.com/cirruslabs/orchard/pkg/resource/v1"
+	"github.com/gin-gonic/gin"
+	"github.com/google/uuid"
+	"net/http"
+	"time"
+)
+
+func (controller *Controller) createWorker(ctx *gin.Context) responder.Responder {
+	var worker v1.Worker
+
+	if err := ctx.ShouldBindJSON(&worker); err != nil {
+		return responder.Code(http.StatusBadRequest)
+	}
+
+	if worker.Name == "" {
+		return responder.Code(http.StatusPreconditionFailed)
+	}
+
+	currentTime := time.Now()
+	if worker.LastSeen.IsZero() {
+		worker.LastSeen = currentTime
+	}
+	worker.CreatedAt = currentTime
+	worker.DeletedAt = time.Time{}
+	worker.UID = uuid.New().String()
+	worker.Generation = 0
+
+	return controller.storeUpdate(func(txn *storepkg.Txn) responder.Responder {
+		// Does the worker resource with this name already exists?
+		_, err := txn.GetWorker(worker.Name)
+		if !errors.Is(err, storepkg.ErrNotFound) {
+			return responder.Code(http.StatusConflict)
+		}
+
+		if err := txn.SetWorker(&worker); err != nil {
+			return responder.Code(http.StatusInternalServerError)
+		}
+
+		return responder.JSON(200, &worker)
+	})
+}
+
+func (controller *Controller) updateWorker(ctx *gin.Context) responder.Responder {
+	var userWorker v1.Worker
+
+	if err := ctx.ShouldBindJSON(&userWorker); err != nil {
+		return responder.Code(http.StatusBadRequest)
+	}
+
+	return controller.storeUpdate(func(txn *storepkg.Txn) responder.Responder {
+		dbWorker, err := txn.GetWorker(userWorker.Name)
+		if err != nil {
+			if errors.Is(err, storepkg.ErrNotFound) {
+				return responder.Code(http.StatusNotFound)
+			}
+
+			return responder.Code(http.StatusInternalServerError)
+		}
+
+		dbWorker.LastSeen = userWorker.LastSeen
+		dbWorker.Generation++
+
+		if err := txn.SetWorker(dbWorker); err != nil {
+			return responder.Code(http.StatusInternalServerError)
+		}
+
+		return responder.JSON(200, &dbWorker)
+	})
+}
+
+func (controller *Controller) getWorker(ctx *gin.Context) responder.Responder {
+	name := ctx.Param("name")
+
+	return controller.storeView(func(txn *storepkg.Txn) responder.Responder {
+		worker, err := txn.GetWorker(name)
+		if err != nil {
+			if errors.Is(err, storepkg.ErrNotFound) {
+				return responder.Code(http.StatusNotFound)
+			}
+
+			return responder.Code(http.StatusInternalServerError)
+		}
+
+		return responder.JSON(http.StatusOK, &worker)
+	})
+}
+
+func (controller *Controller) listWorkers(ctx *gin.Context) responder.Responder {
+	return controller.storeView(func(txn *storepkg.Txn) responder.Responder {
+		workers, err := txn.ListWorkers()
+		if err != nil {
+			if errors.Is(err, storepkg.ErrNotFound) {
+				return responder.Code(http.StatusNotFound)
+			}
+
+			return responder.Code(http.StatusInternalServerError)
+		}
+
+		return responder.JSON(http.StatusOK, &workers)
+	})
+}
+
+func (controller *Controller) deleteWorker(ctx *gin.Context) responder.Responder {
+	name := ctx.Param("name")
+
+	return controller.storeUpdate(func(txn *storepkg.Txn) responder.Responder {
+		if err := txn.DeleteWorker(name); err != nil {
+			if errors.Is(err, storepkg.ErrNotFound) {
+				return responder.Code(http.StatusNotFound)
+			}
+
+			return responder.Code(http.StatusInternalServerError)
+		}
+
+		return responder.Code(http.StatusOK)
+	})
+}

internal/controller/controller.go

@@ -0,0 +1,96 @@
+package controller
+
+import (
+	"context"
+	"crypto/tls"
+	"fmt"
+	storepkg "github.com/cirruslabs/orchard/internal/controller/store"
+	"go.uber.org/zap"
+	"net"
+	"net/http"
+)
+
+type Controller struct {
+	dataDir    string
+	listenAddr string
+	tlsConfig  *tls.Config
+	listener   net.Listener
+	httpServer *http.Server
+	store      *storepkg.Store
+	logger     *zap.SugaredLogger
+}
+
+func New(opts ...Option) (*Controller, error) {
+	controller := &Controller{}
+
+	// Apply options
+	for _, opt := range opts {
+		opt(controller)
+	}
+
+	// Apply defaults
+	if controller.dataDir == "" {
+		return nil, fmt.Errorf("%w: please specify the data directory path with WithDataDir()")
+	}
+	if controller.listenAddr == "" {
+		controller.listenAddr = ":6120"
+	}
+	if controller.logger == nil {
+		controller.logger = zap.NewNop().Sugar()
+	}
+
+	// Instantiate controller
+	store, err := storepkg.New(controller.dbPath())
+	if err != nil {
+		return nil, err
+	}
+	controller.store = store
+
+	listener, err := net.Listen("tcp", controller.listenAddr)
+	if err != nil {
+		return nil, err
+	}
+	if controller.tlsConfig != nil {
+		controller.listener = tls.NewListener(listener, controller.tlsConfig)
+	} else {
+		controller.listener = listener
+	}
+
+	controller.httpServer = &http.Server{
+		Handler: controller.initAPI(),
+	}
+
+	return controller, nil
+}
+
+func (controller *Controller) Run(ctx context.Context) error {
+	// Run the scheduler so that each VM will eventually
+	// be assigned to a specific Worker
+	go func() {
+		err := runScheduler(controller.store)
+		if err != nil {
+			panic(err)
+		}
+	}()
+
+	// Run the janitor so that inactive workers
+	// will eventually be removed from the DB
+	go func() {
+		err := controller.runJanitor(controller.store)
+		if err != nil {
+			panic(err)
+		}
+	}()
+
+	// A helper function to shut down the HTTP server on context cancellation
+	go func() {
+		<-ctx.Done()
+		controller.httpServer.Shutdown(ctx)
+	}()
+
+	if err := controller.httpServer.Serve(controller.listener); err != nil {
+		return err
+	}
+
+	return nil
+}

internal/controller/datadir.go

@@ -0,0 +1,15 @@
+package controller
+
+import "path/filepath"
+
+func (controller *Controller) dbPath() string {
+	return filepath.Join(controller.dataDir, "db")
+}
+
+func (controller *Controller) caCertPath() string {
+	return filepath.Join(controller.dataDir, "ca.crt")
+}
+
+func (controller *Controller) caKeyPath() string {
+	return filepath.Join(controller.dataDir, "ca.key")
+}

internal/controller/janitor.go

@@ -0,0 +1,50 @@
+package controller
+
+import (
+	storepkg "github.com/cirruslabs/orchard/internal/controller/store"
+	"github.com/cirruslabs/orchard/pkg/resource/v1"
+	"time"
+)
+
+const janitorInterval = 5 * time.Second
+
+func (controller *Controller) runJanitor(store *storepkg.Store) error {
+	ticker := time.Tick(janitorInterval)
+
+	for {
+		if err := controller.runJanitorInner(store); err != nil {
+			return err
+		}
+
+		<-ticker
+	}
+}
+
+func (controller *Controller) runJanitorInner(store *storepkg.Store) error {
+	var workers []*v1.Worker
+	var err error
+
+	err = store.View(func(txn *storepkg.Txn) error {
+		workers, err = txn.ListWorkers()
+
+		return err
+	})
+	if err != nil {
+		return err
+	}
+
+	for _, worker := range workers {
+		if time.Now().Sub(worker.LastSeen).Minutes() > 1 {
+			controller.logger.Debugf("removing outdated worker %s", worker.Name)
+
+			err := store.Update(func(txn *storepkg.Txn) error {
+				return txn.DeleteWorker(worker.Name)
+			})
+			if err != nil {
+				return err
+			}
+		}
+	}
+
+	return nil
+}

internal/controller/option.go

@@ -0,0 +1,32 @@
+package controller
+
+import (
+	"crypto/tls"
+	"go.uber.org/zap"
+)
+
+type Option func(*Controller)
+
+func WithDataDir(dataDir string) Option {
+	return func(controller *Controller) {
+		controller.dataDir = dataDir
+	}
+}
+
+func WithListenAddr(listenAddr string) Option {
+	return func(controller *Controller) {
+		controller.listenAddr = listenAddr
+	}
+}
+
+func WithTLSConfig(tlsConfig *tls.Config) Option {
+	return func(controller *Controller) {
+		controller.tlsConfig = tlsConfig
+	}
+}
+
+func WithLogger(logger *zap.Logger) Option {
+	return func(controller *Controller) {
+		controller.logger = logger.Sugar()
+	}
+}

internal/controller/scheduler.go

@@ -0,0 +1,72 @@
+package controller
+
+import (
+	storepkg "github.com/cirruslabs/orchard/internal/controller/store"
+	"github.com/cirruslabs/orchard/pkg/resource/v1"
+	"sort"
+	"time"
+)
+
+const schedulerInterval = 5 * time.Second
+
+func runScheduler(store *storepkg.Store) error {
+	ticker := time.Tick(schedulerInterval)
+
+	for {
+		if err := runSchedulerInner(store); err != nil {
+			return err
+		}
+
+		<-ticker
+	}
+}
+
+func runSchedulerInner(store *storepkg.Store) error {
+	var vms []*v1.VM
+	var workers []*v1.Worker
+	var err error
+
+	err = store.View(func(txn *storepkg.Txn) error {
+		vms, err = txn.ListVMs()
+		if err != nil {
+			return err
+		}
+
+		workers, err = txn.ListWorkers()
+		if err != nil {
+			return err
+		}
+
+		return nil
+	})
+	if err != nil {
+		return err
+	}
+
+	// Sort VMs by date of creation
+	sort.Slice(vms, func(i, j int) bool {
+		return vms[i].CreatedAt.Before(vms[j].CreatedAt)
+	})
+
+	for _, vm := range vms {
+		if vm.Worker != "" {
+			continue
+		}
+
+		// Find an appropriate worker to run this VM on
+		for _, worker := range workers {
+			vm.Worker = worker.Name
+
+			err := store.Update(func(txn *storepkg.Txn) error {
+				return txn.SetVM(vm)
+			})
+			if err != nil {
+				return err
+			}
+
+			break
+		}
+	}
+
+	return nil
+}

internal/controller/store/errors.go

@@ -0,0 +1,20 @@
+package store
+
+import (
+	"errors"
+	"fmt"
+	"github.com/dgraph-io/badger/v3"
+)
+
+var (
+	ErrNotFound     = errors.New("DB entry not found")
+	ErrBadgerFailed = errors.New("BadgerDB failed")
+)
+
+func mapErr(err error) error {
+	if errors.Is(err, badger.ErrKeyNotFound) {
+		return ErrNotFound
+	}
+
+	return fmt.Errorf("%w: %v", ErrBadgerFailed, err)
+}

internal/controller/store/store.go

@@ -0,0 +1,24 @@
+package store
+
+import (
+	"github.com/dgraph-io/badger/v3"
+)
+
+type Store struct {
+	db *badger.DB
+}
+
+func New(dbPath string) (*Store, error) {
+	opts := badger.DefaultOptions(dbPath)
+
+	opts.SyncWrites = true
+
+	db, err := badger.Open(opts)
+	if err != nil {
+		return nil, err
+	}
+
+	return &Store{
+		db: db,
+	}, nil
+}

internal/controller/store/txn.go

@@ -0,0 +1,23 @@
+package store
+
+import "github.com/dgraph-io/badger/v3"
+
+type Txn struct {
+	badgerTxn *badger.Txn
+}
+
+func (store *Store) View(cb func(txn *Txn) error) error {
+	return store.db.View(func(txn *badger.Txn) error {
+		return cb(&Txn{
+			badgerTxn: txn,
+		})
+	})
+}
+
+func (store *Store) Update(cb func(txn *Txn) error) error {
+	return store.db.Update(func(txn *badger.Txn) error {
+		return cb(&Txn{
+			badgerTxn: txn,
+		})
+	})
+}

internal/controller/store/vm.go

@@ -0,0 +1,96 @@
+package store
+
+import (
+	"encoding/json"
+	"github.com/cirruslabs/orchard/pkg/resource/v1"
+	"github.com/dgraph-io/badger/v3"
+	"path"
+)
+
+const SpaceVMs = "/vms"
+
+func VMKey(name string) []byte {
+	return []byte(path.Join(SpaceVMs, name))
+}
+
+func (txn *Txn) GetVM(name string) (result *v1.VM, err error) {
+	defer func() {
+		err = mapErr(err)
+	}()
+
+	key := VMKey(name)
+
+	item, err := txn.badgerTxn.Get(key)
+	if err != nil {
+		return nil, err
+	}
+
+	valueBytes, err := item.ValueCopy(nil)
+	if err != nil {
+		return nil, err
+	}
+
+	var vm v1.VM
+
+	err = json.Unmarshal(valueBytes, &vm)
+	if err != nil {
+		return nil, err
+	}
+
+	return &vm, nil
+}
+
+func (txn *Txn) SetVM(vm *v1.VM) (err error) {
+	defer func() {
+		err = mapErr(err)
+	}()
+
+	key := VMKey(vm.Name)
+
+	valueBytes, err := json.Marshal(vm)
+	if err != nil {
+		return err
+	}
+
+	return txn.badgerTxn.Set(key, valueBytes)
+}
+
+func (txn *Txn) DeleteVM(name string) (err error) {
+	defer func() {
+		err = mapErr(err)
+	}()
+
+	key := VMKey(name)
+
+	return txn.badgerTxn.Delete(key)
+}
+
+func (txn *Txn) ListVMs() (result []*v1.VM, err error) {
+	defer func() {
+		err = mapErr(err)
+	}()
+
+	it := txn.badgerTxn.NewIterator(badger.IteratorOptions{
+		Prefix: []byte(SpaceVMs),
+	})
+	defer it.Close()
+
+	for it.Rewind(); it.Valid(); it.Next() {
+		item := it.Item()
+
+		vmBytes, err := item.ValueCopy(nil)
+		if err != nil {
+			return nil, err
+		}
+
+		var vm v1.VM
+
+		if err := json.Unmarshal(vmBytes, &vm); err != nil {
+			return nil, err
+		}
+
+		result = append(result, &vm)
+	}
+
+	return result, nil
+}

internal/controller/store/worker.go

@@ -0,0 +1,96 @@
+package store
+
+import (
+	"encoding/json"
+	"github.com/cirruslabs/orchard/pkg/resource/v1"
+	"github.com/dgraph-io/badger/v3"
+	"path"
+)
+
+const SpaceWorkers = "/workers"
+
+func WorkerKey(name string) []byte {
+	return []byte(path.Join(SpaceWorkers, name))
+}
+
+func (txn *Txn) GetWorker(name string) (result *v1.Worker, err error) {
+	defer func() {
+		err = mapErr(err)
+	}()
+
+	key := WorkerKey(name)
+
+	item, err := txn.badgerTxn.Get(key)
+	if err != nil {
+		return nil, err
+	}
+
+	valueBytes, err := item.ValueCopy(nil)
+	if err != nil {
+		return nil, err
+	}
+
+	var worker v1.Worker
+
+	err = json.Unmarshal(valueBytes, &worker)
+	if err != nil {
+		return nil, err
+	}
+
+	return &worker, nil
+}
+
+func (txn *Txn) SetWorker(worker *v1.Worker) (err error) {
+	defer func() {
+		err = mapErr(err)
+	}()
+
+	key := WorkerKey(worker.Name)
+
+	valueBytes, err := json.Marshal(worker)
+	if err != nil {
+		return err
+	}
+
+	return txn.badgerTxn.Set(key, valueBytes)
+}
+
+func (txn *Txn) DeleteWorker(name string) (err error) {
+	defer func() {
+		err = mapErr(err)
+	}()
+
+	key := WorkerKey(name)
+
+	return txn.badgerTxn.Delete(key)
+}
+
+func (txn *Txn) ListWorkers() (result []*v1.Worker, err error) {
+	defer func() {
+		err = mapErr(err)
+	}()
+
+	it := txn.badgerTxn.NewIterator(badger.IteratorOptions{
+		Prefix: []byte(SpaceWorkers),
+	})
+	defer it.Close()
+
+	for it.Rewind(); it.Valid(); it.Next() {
+		item := it.Item()
+
+		vmBytes, err := item.ValueCopy(nil)
+		if err != nil {
+			return nil, err
+		}
+
+		var worker v1.Worker
+
+		if err := json.Unmarshal(vmBytes, &worker); err != nil {
+			return nil, err
+		}
+
+		result = append(result, &worker)
+	}
+
+	return result, nil
+}