scriptoriumadmin/featurebase
1
0
Fork 0
mirror of https://github.com/featurebasedb/featurebase.git synced 2026-08-28 10:54:59 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

WIP on worker_service and worker_service_provider

Browse source
This commit is contained in:
Matthew Jaffee 2023-04-03 09:34:45 -05:00
parent ea72396b4d
commit 5baee81eb5
33 changed files with 1321 additions and 106 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

5
ctl/dax.go
View file

@ -36,8 +36,13 @@ func BuildDAXFlags(cmd *cobra.Command, srv *server.Command) {
flags.BoolVar(&srv.Config.Queryer.Run, "queryer.run", srv.Config.Queryer.Run, "Run the Queryer service in process.")
flags.StringVar(&srv.Config.Queryer.Config.ControllerAddress, "queryer.config.controller-address", srv.Config.Queryer.Config.ControllerAddress, "Address of remote Controller process.")
// Worker Service Provider
flags.BoolVar(&srv.Config.WorkerServiceProvider.Run, "wsp.run", srv.Config.WorkerServiceProvider.Run, "Run the WSP service in process.")
flags.StringVar(&srv.Config.WorkerServiceProvider.Config.ID, "wsp.id", srv.Config.WorkerServiceProvider.Config.ID, "ID for worker service provider. Should be a string distinct from any other WSP in the deployment.")
flags.StringVar(&srv.Config.WorkerServiceProvider.Config.ControllerAddress, "wsp.config.controller-address", srv.Config.WorkerServiceProvider.Config.ControllerAddress, "Address of remote Controller process.")
// Computer
flags.BoolVar(&srv.Config.Computer.Run, "computer.run", srv.Config.Computer.Run, "Run the Computer service in process.")
flags.IntVar(&srv.Config.Computer.N, "computer.n", srv.Config.Computer.N, "The number of Computer services to run in process.")
flags.StringVar(&srv.Config.Computer.WorkerServiceID, "computer.worker_service_id", srv.Config.Computer.WorkerServiceID, "ID of WorkerService which spawned this computer.")
flags.AddFlagSet(serverFlagSet(&srv.Config.Computer.Config, "computer.config"))
}

3
dax/computer/service/computer.go
View file

@ -115,7 +115,8 @@ type CommandConfig struct {
// separate data directory for its holder.
Name string
ComputerConfig fbserver.Config
ComputerConfig fbserver.Config
WorkerServiceID dax.WorkerServiceID
Listener net.Listener
RootDataDir string

21
dax/controller.go
View file

@ -1,8 +1,21 @@
package dax
import "context"
type Controller interface {
Noder
Schemar
// RegisterWorkerServiceProvider makes the controller aware of a
// new WorkerServiceProvider.
RegisterWorkerServiceProvider(ctx context.Context, sp WorkerServiceProvider) (WorkerServices, error)
// RegisterWorkerService makes the controller aware of a new
// WorkerService, so that when workers of that service register
// themselves, the controller will have an entity to associate
// them with, which will ultimately correspond to what Database
// those workers get jobs for.
RegisterWorkerService(ctx context.Context, srv WorkerService) error
}
// Ensure type implements interface.
@ -18,3 +31,11 @@ type nopController struct {
func NewNopController() *nopController {
return &nopController{}
}
func (n *nopController) RegisterWorkerServiceProvider(ctx context.Context, sp WorkerServiceProvider) (WorkerServices, error) {
return nil, nil
}
func (n *nopController) RegisterWorkerService(ctx context.Context, srv WorkerService) error {
return nil
}

43
dax/controller/balancer.go
View file

@ -46,6 +46,32 @@ type Balancer interface {
// Nodes returns all nodes known by the Balancer.
Nodes(tx dax.Transaction) ([]*dax.Node, error)
// CreateWorkerServiceProvider adds a WorkerServiceProvider (WSP) to
// storage. Any WorkerService which registers must have a
// WorkerServiceProviderID that matches an existing WSP, and the
// Controller will ask for which WSPs are available when asked to
// create a database. It will then ask one of the WSPs for a
// WorkerService to assign to the database.
CreateWorkerServiceProvider(tx dax.Transaction, sp dax.WorkerServiceProvider) error
// CreateWorkerService adds a WorkerService to storage. Generally
// a WSP can create WorkerServices (which can create Workers)
// before they are requested. Because the workers will register
// themselves as soon as they come up, and they must be associated
// with a WorkerService, the WSP registers all Services with the
// Controller which stores the knowledge of their existence by
// calling this method.
CreateWorkerService(tx dax.Transaction, srv dax.WorkerService) error
WorkerServiceProviders(tx dax.Transaction /*, future optional filters */) (dax.WorkerServiceProviders, error)
AssignFreeServiceToDatabase(tx dax.Transaction, wspID dax.WorkerServiceProviderID, qdb *dax.QualifiedDatabase) (*dax.WorkerService, error)
// WorkerServices returns all worker services which came from the
// WorkerServiceProvider with the given ID. If that ID is empty,
// then all WorkerServices are returned.
WorkerServices(tx dax.Transaction, wspID dax.WorkerServiceProviderID) (dax.WorkerServices, error)
}
// Ensure type implements interface.
@ -94,3 +120,20 @@ func (b *NopBalancer) ReadNode(tx dax.Transaction, addr dax.Address) (*dax.Node,
func (b *NopBalancer) Nodes(tx dax.Transaction) ([]*dax.Node, error) {
return []*dax.Node{}, nil
}
func (b *NopBalancer) CreateWorkerServiceProvider(tx dax.Transaction, sp dax.WorkerServiceProvider) error {
return nil
}
func (b *NopBalancer) CreateWorkerService(tx dax.Transaction, srv dax.WorkerService) error {
return nil
}
func (b *NopBalancer) WorkerServices(tx dax.Transaction, wspID dax.WorkerServiceProviderID) (dax.WorkerServices, error) {
return nil, nil
}
func (b *NopBalancer) WorkerServiceProviders(tx dax.Transaction /*, future optional filters */) (dax.WorkerServiceProviders, error) {
return nil, nil
}
func (b *NopBalancer) AssignFreeServiceToDatabase(tx dax.Transaction, wspID dax.WorkerServiceProviderID, qdb *dax.QualifiedDatabase) (*dax.WorkerService, error)

35
dax/controller/balancer/balancer.go
View file

@ -40,17 +40,20 @@ type Balancer struct {
schemar schemar.Schemar
wsp WorkerServiceProviderService
logger logger.Logger
}
// New returns a new instance of Balancer.
func New(wr controller.WorkerRegistry, fjs FreeJobService, wjs WorkerJobService, fws FreeWorkerService, schemar schemar.Schemar, logger logger.Logger) *Balancer {
func New(wr controller.WorkerRegistry, fjs FreeJobService, wjs WorkerJobService, fws FreeWorkerService, schemar schemar.Schemar, wsp WorkerServiceProviderService, logger logger.Logger) *Balancer {
return &Balancer{
current: wjs,
workerRegistry: wr,
freeJobs: fjs,
freeWorkers: fws,
schemar: schemar,
wsp: wsp,
logger: logger,
}
}
@ -86,7 +89,7 @@ func (b *Balancer) assignMinWorkers(tx dax.Transaction, roleType dax.RoleType, q
b.logger.Debugf("assigning min workers for '%s', '%s'", roleType, qdbid)
// Find out how many free workers we have.
freeWorkers, err := b.freeWorkers.ListWorkers(tx, roleType)
freeWorkers, err := b.freeWorkers.ListWorkers(tx, qdbid, roleType)
if err != nil {
return nil, errors.Wrap(err, "getting free worker list")
}
@ -783,6 +786,24 @@ func (b *Balancer) Nodes(tx dax.Transaction) ([]*dax.Node, error) {
return b.workerRegistry.Workers(tx)
}
func (b *Balancer) CreateWorkerServiceProvider(tx dax.Transaction, sp dax.WorkerServiceProvider) error {
return b.wsp.CreateWorkerServiceProvider(tx, sp)
}
func (b *Balancer) CreateWorkerService(tx dax.Transaction, srv dax.WorkerService) error {
return b.wsp.CreateWorkerService(tx, srv)
}
func (b *Balancer) WorkerServiceProviders(tx dax.Transaction /*, future optional filters */) (dax.WorkerServiceProviders, error) {
return b.wsp.WorkerServiceProviders(tx)
}
func (b *Balancer) AssignFreeServiceToDatabase(tx dax.Transaction, wspID dax.WorkerServiceProviderID, qdb *dax.QualifiedDatabase) (*dax.WorkerService, error) {
return b.wsp.AssignFreeServiceToDatabase(tx, wspID, qdb)
}
func (b *Balancer) WorkerServices(tx dax.Transaction, wsp dax.WorkerServiceProviderID) (dax.WorkerServices, error) {
return b.wsp.WorkerServices(tx, wsp)
}
type WorkerJobService interface {
WorkersJobs(tx dax.Transaction, roleType dax.RoleType, qdbid dax.QualifiedDatabaseID) ([]dax.WorkerInfo, error)
@ -811,5 +832,13 @@ type FreeJobService interface {
type FreeWorkerService interface {
PopWorkers(tx dax.Transaction, roleType dax.RoleType, num int) ([]dax.Address, error)
ListWorkers(tx dax.Transaction, roleType dax.RoleType) (dax.Addresses, error)
ListWorkers(tx dax.Transaction, qdbid dax.QualifiedDatabaseID, roleType dax.RoleType) (dax.Addresses, error)
}
type WorkerServiceProviderService interface {
CreateWorkerServiceProvider(tx dax.Transaction, sp dax.WorkerServiceProvider) error
CreateWorkerService(tx dax.Transaction, srv dax.WorkerService) error
WorkerServiceProviders(tx dax.Transaction /*, future optional filters */) (dax.WorkerServiceProviders, error)
WorkerServices(tx dax.Transaction, wspID dax.WorkerServiceProviderID) (dax.WorkerServices, error)
AssignFreeServiceToDatabase(tx dax.Transaction, wspID dax.WorkerServiceProviderID, qdb *dax.QualifiedDatabase) (*dax.WorkerService, error)
}

100
dax/controller/balancer/balancer_test.go Normal file
View file

@ -0,0 +1,100 @@
package balancer_test
import (
"context"
"testing"
"github.com/featurebasedb/featurebase/v3/dax"
"github.com/featurebasedb/featurebase/v3/dax/controller/sqldb"
"github.com/featurebasedb/featurebase/v3/logger"
"github.com/stretchr/testify/require"
)
func TestWorkerServiceAndProvider(t *testing.T) {
tx, err := SQLTransactor.BeginTx(context.Background(), true)
require.NoError(t, err, "getting transaction")
defer func() {
err := tx.Rollback()
if err != nil {
t.Logf("rolling back: %v", err)
}
}()
balancer := sqldb.NewBalancer(logger.StderrLogger)
wsp := dax.WorkerServiceProvider{
ID: "wspID1",
Roles: []dax.RoleType{"compute"},
Address: "wsp1.example.com:8082",
Description: "the description",
}
wsp2 := dax.WorkerServiceProvider{
ID: "wspID2",
Roles: []dax.RoleType{"compute", "translate"},
Address: "wsp2.example.com:8082",
Description: "the description2",
}
t.Run("create and query a WSP", func(t *testing.T) {
err = balancer.CreateWorkerServiceProvider(tx, wsp)
require.NoError(t, err, "creating initial wsp")
wsps, err := balancer.WorkerServiceProviders(tx)
require.NoError(t, err, "reading wsps")
require.Equal(t, dax.WorkerServiceProviders{wsp}, wsps)
})
t.Run("create another wsp and query both", func(t *testing.T) {
err = balancer.CreateWorkerServiceProvider(tx, wsp2)
require.NoError(t, err, "creating wsp2")
wsps, err := balancer.WorkerServiceProviders(tx)
require.NoError(t, err, "reading wsps")
require.ElementsMatch(t, dax.WorkerServiceProviders{wsp, wsp2}, wsps)
})
t.Run("create 3 worker services across 2 wsps, and query", func(t *testing.T) {
ws := dax.WorkerService{
ID: "wsID1",
Roles: []dax.RoleType{"compute"},
WorkerServiceProviderID: "wspID1",
DatabaseID: "",
WorkersMin: 1,
WorkersMax: 1,
}
err = balancer.CreateWorkerService(tx, ws)
require.NoError(t, err, "creating worker service 1")
ws2 := dax.WorkerService{
ID: "wsID2",
Roles: []dax.RoleType{"compute"},
WorkerServiceProviderID: "wspID1",
DatabaseID: "",
WorkersMin: 1,
WorkersMax: 1,
}
err = balancer.CreateWorkerService(tx, ws2)
require.NoError(t, err, "creating worker service 2")
ws3 := dax.WorkerService{
ID: "wsID3",
Roles: []dax.RoleType{"compute", "translate"},
WorkerServiceProviderID: "wspID2",
DatabaseID: "",
WorkersMin: 1,
WorkersMax: 1,
}
err = balancer.CreateWorkerService(tx, ws3)
require.NoError(t, err, "creating worker service 3")
wsvcs, err := balancer.WorkerServices(tx, "")
require.NoError(t, err, "getting all worker services")
require.ElementsMatch(t, dax.WorkerServices{ws, ws2, ws3}, wsvcs)
wsvcs, err = balancer.WorkerServices(tx, "wspID1")
require.NoError(t, err, "getting all worker services")
require.ElementsMatch(t, dax.WorkerServices{ws, ws2}, wsvcs)
})
}

55
dax/controller/client/client.go
View file

@ -701,3 +701,58 @@ func (c *Client) SnapshotTable(ctx context.Context, qtid dax.QualifiedTableID) e
return nil
}
func (c *Client) RegisterWorkerServiceProvider(ctx context.Context, sp dax.WorkerServiceProvider) (dax.WorkerServices, error) {
url := fmt.Sprintf("%s/register-worker-service-provider", c.address.WithScheme(defaultScheme))
postBody, err := json.Marshal(sp)
if err != nil {
return nil, errors.Wrap(err, "marshaling WorkerServiceProvider")
}
reqBody := bytes.NewBuffer(postBody)
// Post the request.
resp, err := c.httpClient.Post(url, "application/json", reqBody)
if err != nil {
return nil, errors.Wrap(err, "posting translate-nodes request")
}
defer resp.Body.Close()
if resp.StatusCode != http.StatusOK {
b, _ := io.ReadAll(resp.Body)
return nil, errors.Errorf("status code: %d: %s", resp.StatusCode, b)
}
var svcs dax.WorkerServices
if err := json.NewDecoder(resp.Body).Decode(&svcs); err != nil {
return nil, errors.Wrap(err, "decoding worker services")
}
return svcs, nil
}
func (c *Client) RegisterWorkerService(ctx context.Context, srv dax.WorkerService) error {
url := fmt.Sprintf("%s/register-worker-service", c.address.WithScheme(defaultScheme))
postBody, err := json.Marshal(srv)
if err != nil {
return errors.Wrap(err, "marshaling WorkerService")
}
reqBody := bytes.NewBuffer(postBody)
// Post the request.
resp, err := c.httpClient.Post(url, "application/json", reqBody)
if err != nil {
return errors.Wrap(err, "posting translate-nodes request")
}
defer resp.Body.Close()
if resp.StatusCode != http.StatusOK {
b, _ := io.ReadAll(resp.Body)
return errors.Errorf("status code: %d: %s", resp.StatusCode, b)
}
return nil
}

72
dax/controller/controller.go
View file

@ -2,8 +2,11 @@
package controller
import (
"bytes"
"context"
"encoding/json"
"fmt"
"net/http"
"sort"
"time"
@ -147,6 +150,9 @@ func (c *Controller) RegisterNodes(ctx context.Context, nodes ...*dax.Node) erro
// Validate input.
for _, n := range nodes {
if n.ServiceID == "" {
return errors.New(errors.CodeTODO, "node must be associated with a WorkerService to register")
}
if n.Address == "" {
return NewErrNodeKeyInvalid(n.Address)
}
@ -257,6 +263,9 @@ func (c *Controller) RegisterNodes(ctx context.Context, nodes ...*dax.Node) erro
// used for anything or assigned any jobs.
func (c *Controller) RegisterNode(ctx context.Context, n *dax.Node) error {
// Validate input.
if n.ServiceID == "" {
return errors.New(errors.CodeTODO, "node must be associated with a WorkerService to register")
}
if n.Address == "" {
return NewErrNodeKeyInvalid(n.Address)
}
@ -632,6 +641,29 @@ func (c *Controller) CreateDatabase(ctx context.Context, qdb *dax.QualifiedDatab
if err := c.Schemar.CreateDatabase(tx, qdb); err != nil {
return errors.Wrap(err, "creating database in schemar")
}
wsps, err := c.Balancer.WorkerServiceProviders(tx)
if err != nil {
return errors.Wrap(err, "getting worker service providers")
}
if len(wsps) != 1 {
return errors.Errorf("unexpected number of worker service providers... should be exactly 1, got %d", len(wsps))
}
wsp := wsps[0]
svc, err := c.Balancer.AssignFreeServiceToDatabase(tx, wsp.ID, qdb)
if err != nil {
return errors.Wrap(err, "assigning free service to database")
}
// Encode the request.
postBody, err := json.Marshal(svc)
if err != nil {
return errors.Wrap(err, "marshalling post request")
}
buf := bytes.NewBuffer(postBody)
http.Post(fmt.Sprintf("%s/claim", wsp.Address), "application/json", buf)
return nil
}
@ -2222,6 +2254,46 @@ func (c *Controller) Workers(ctx context.Context) ([]*dax.Node, error) {
return c.Balancer.Nodes(tx)
}
// RegisterWorkerServiceProvider makes the controller aware of a new
// WorkerServiceProvider. When a database is created, the Controller
// can decided which WorkerServiceProvider should provide the
// WorkerService for that database. Different providers might exist in
// different geographic locations, or be earmarked for particular
// organizations, or whatever... the possibilities are endless.
func (c *Controller) RegisterWorkerServiceProvider(ctx context.Context, sp dax.WorkerServiceProvider) (dax.WorkerServices, error) {
var svcs dax.WorkerServices
fn := func(tx dax.Transaction, writable bool) error {
err := c.Balancer.CreateWorkerServiceProvider(tx, sp)
if err != nil {
return errors.Wrap(err, "creating worker service provider")
}
svcs, err = c.Balancer.WorkerServices(tx, sp.ID)
return err
}
if err := dax.RetryWithTx(ctx, c.Transactor, fn, true, txRetry); err != nil {
return nil, errors.Wrap(err, "retry with tx: write")
}
return svcs, nil
}
// RegisterWorkerService makes the controller aware of a new
// WorkerService, so that when workers of that service register
// themselves, the controller will have an entity to associate them
// with, which will ultimately correspond to what Database those
// workers get jobs for. The controller does not pick Services from
// those that are registered at this endpoint, rather it asks the
// WorkerServiceProvider to assign a WorkerService (which must already
// be registered). This way the WorkerServiceProvider knows which
// Services are used and can maintain enough free capacity to serve
// new requests.
func (c *Controller) RegisterWorkerService(ctx context.Context, srv dax.WorkerService) error {
return dax.NewErrUnimplemented("RegisterWorkerService")
}
func (c *Controller) Logger() logger.Logger {
return c.logger
}

56
dax/controller/http/handler.go
View file

@ -7,6 +7,7 @@ import (
"github.com/featurebasedb/featurebase/v3/dax"
"github.com/featurebasedb/featurebase/v3/dax/controller"
"github.com/featurebasedb/featurebase/v3/errors"
"github.com/featurebasedb/featurebase/v3/logger"
"github.com/gorilla/mux"
)
@ -50,6 +51,9 @@ func Handler(c *controller.Controller) http.Handler {
router.HandleFunc("/compute-nodes", server.postComputeNodes).Methods("POST").Name("PostComputeNodes")
router.HandleFunc("/translate-nodes", server.postTranslateNodes).Methods("POST").Name("PostTranslateNodes")
router.HandleFunc("/register-worker-service-provider", server.postRegisterWorkerServiceProvider).Methods("POST").Name("PostregisterWorkerServiceProvider")
router.HandleFunc("/register-worker-service", server.postRegisterWorkerService).Methods("POST").Name("PostregisterWorkerService")
// debug endpoints
router.HandleFunc("/debug/nodes", server.getDebugNodes).Methods("GET").Name("GetDebugNodes")
router.HandleFunc("/debug/balancer", server.getDebugBalancer).Methods("GET").Name("getDebugBalancer")
@ -790,11 +794,59 @@ func (s *server) postTranslateNodes(w http.ResponseWriter, r *http.Request) {
}
if err := json.NewEncoder(w).Encode(resp); err != nil {
s.Log().Errorf("Error writing response to /translate-nodes request: '%v'", err)
http.Error(w, err.Error(), http.StatusBadRequest)
return
}
}
// POST /register-worker-service-provider
func (s *server) postRegisterWorkerServiceProvider(w http.ResponseWriter, r *http.Request) {
body := r.Body
defer body.Close()
ctx := r.Context()
req := dax.WorkerServiceProvider{}
if err := json.NewDecoder(body).Decode(&req); err != nil {
http.Error(w, err.Error(), http.StatusBadRequest)
return
}
svcs, err := s.controller.RegisterWorkerServiceProvider(ctx, req)
if err != nil {
http.Error(w, err.Error(), http.StatusInternalServerError)
return
}
w.Header().Add("Content-Type", "application/json")
if err := json.NewEncoder(w).Encode(svcs); err != nil {
s.Log().Errorf("Error writing response to /register-worker-service-provider request: '%v'", err)
http.Error(w, err.Error(), http.StatusInternalServerError)
return
}
}
// POST /register-worker-service
func (s *server) postRegisterWorkerService(w http.ResponseWriter, r *http.Request) {
body := r.Body
defer body.Close()
ctx := r.Context()
req := dax.WorkerService{}
if err := json.NewDecoder(body).Decode(&req); err != nil {
http.Error(w, err.Error(), http.StatusBadRequest)
return
}
err := s.controller.RegisterWorkerService(ctx, req)
if err != nil {
http.Error(w, err.Error(), http.StatusInternalServerError)
return
}
}
// TranslateNodesRequest is used to specify the table/partitions to consider in
// the TranslateNodes method call.
type TranslateNodesRequest struct {
@ -810,3 +862,7 @@ type TranslateNodesRequest struct {
type TranslateNodesResponse struct {
TranslateNodes []dax.TranslateNode `json:"translate-nodes"`
}
func (s *server) Log() logger.Logger {
return s.controller.Logger()
}

3
dax/controller/sqldb/balancer.go
View file

@ -12,6 +12,7 @@ func NewBalancer(log logger.Logger) *balancer.Balancer {
wjs := NewWorkerJobService(log)
fws := NewFreeWorkerService(log)
ns := NewWorkerRegistry(log)
wsp := NewWorkerServiceProviderService(log)
return balancer.New(ns, fjs, wjs, fws, schemar, log)
return balancer.New(ns, fjs, wjs, fws, schemar, wsp, log)
}

2
dax/controller/sqldb/freeworker.go
View file

@ -49,7 +49,7 @@ func (fw *freeWorkerService) PopWorkers(tx dax.Transaction, roleType dax.RoleTyp
return ret, nil
}
func (fw *freeWorkerService) ListWorkers(tx dax.Transaction, roleType dax.RoleType) (dax.Addresses, error) {
func (fw *freeWorkerService) ListWorkers(tx dax.Transaction, qdbid dax.QualifiedDatabaseID, roleType dax.RoleType) (dax.Addresses, error) {
dt, ok := tx.(*DaxTransaction)
if !ok {
return nil, dax.NewErrInvalidTransaction("*sqldb.DaxTransaction")

29
dax/controller/sqldb/worker.go
View file

@ -29,34 +29,9 @@ func (w *workerRegistry) AddWorker(tx dax.Transaction, node *dax.Node) error {
return dax.NewErrInvalidTransaction("*sqldb.DaxTransaction")
}
workers := models.Workers{}
// Determine if a worker for this address already exists. We use `All()`
// here instead of `First()` because `First()` returns an error if there's
// no match.
if err := dt.C.Where("address = ?", node.Address).All(&workers); err != nil {
return errors.Wrapf(err, "getting workers by address: %s", node.Address)
}
switch len(workers) {
case 0:
// Continue on to create.
case 1:
// Since a worker for this address already exists, just update it and
// return.
worker := workers[0]
for _, roleType := range node.RoleTypes {
if err := worker.SetRole(roleType); err != nil {
return errors.Wrapf(err, "setting role: %s", roleType)
}
}
return dt.C.Update(worker)
default:
return errors.Errorf("found more than one worker for address: %s", node.Address)
}
worker := &models.Worker{
Address: node.Address,
Address: node.Address,
ServiceID: node.ServiceID,
}
for _, roleType := range node.RoleTypes {
if err := worker.SetRole(roleType); err != nil {

173
dax/controller/sqldb/worker_service_provider.go Normal file
View file

@ -0,0 +1,173 @@
package sqldb
import (
"github.com/featurebasedb/featurebase/v3/dax"
"github.com/featurebasedb/featurebase/v3/dax/models"
"github.com/featurebasedb/featurebase/v3/errors"
"github.com/featurebasedb/featurebase/v3/logger"
"github.com/gobuffalo/nulls"
)
func NewWorkerServiceProviderService(log logger.Logger) *workerServiceProviderService {
if log == nil {
log = logger.NopLogger
}
return &workerServiceProviderService{
log: log,
}
}
type workerServiceProviderService struct {
log logger.Logger
}
func (w *workerServiceProviderService) CreateWorkerServiceProvider(tx dax.Transaction, sp dax.WorkerServiceProvider) error {
dt, ok := tx.(*DaxTransaction)
if !ok {
return dax.NewErrInvalidTransaction("*sqldb.DaxTransaction")
}
mod := &models.WorkerServiceProvider{
ID: string(sp.ID),
Address: sp.Address,
RoleCompute: sp.Roles.Contains(dax.RoleTypeCompute),
RoleTranslate: sp.Roles.Contains(dax.RoleTypeTranslate),
Description: sp.Description,
}
err := dt.C.Create(mod)
return errors.Wrap(err, "inserting to DB")
}
func (w *workerServiceProviderService) CreateWorkerService(tx dax.Transaction, srv dax.WorkerService) error {
dt, ok := tx.(*DaxTransaction)
if !ok {
return dax.NewErrInvalidTransaction("*sqldb.DaxTransaction")
}
var dbID nulls.String
if srv.DatabaseID != "" {
dbID = nulls.NewString(string(srv.DatabaseID))
}
mod := &models.WorkerService{
ID: string(srv.ID),
WorkerServiceProviderID: string(srv.WorkerServiceProviderID),
DatabaseID: dbID,
RoleCompute: srv.Roles.Contains(dax.RoleTypeCompute),
RoleTranslate: srv.Roles.Contains(dax.RoleTypeTranslate),
WorkersMin: srv.WorkersMin,
WorkersMax: srv.WorkersMax,
}
err := dt.C.Create(mod)
return errors.Wrap(err, "inserting to DB")
}
func (w *workerServiceProviderService) WorkerServiceProviders(tx dax.Transaction /*, future optional filters */) (dax.WorkerServiceProviders, error) {
dt, ok := tx.(*DaxTransaction)
if !ok {
return nil, dax.NewErrInvalidTransaction("*sqldb.DaxTransaction")
}
wsps := make([]models.WorkerServiceProvider, 0, 1)
err := dt.C.All(&wsps)
ret := make(dax.WorkerServiceProviders, len(wsps))
for i, wsp := range wsps {
ret[i] = toDaxWSP(wsp)
}
return ret, errors.Wrap(err, "querying for all worker service providers")
}
func toDaxWSP(wsp models.WorkerServiceProvider) dax.WorkerServiceProvider {
roles := []dax.RoleType{}
if wsp.RoleCompute {
roles = append(roles, dax.RoleTypeCompute)
}
if wsp.RoleTranslate {
roles = append(roles, dax.RoleTypeTranslate)
}
return dax.WorkerServiceProvider{
ID: dax.WorkerServiceProviderID(wsp.ID),
Roles: roles,
Address: wsp.Address,
Description: wsp.Description,
}
}
// AssignFreeServiceToDatabase finds a WorkerService with the given service provider ID and
func (w *workerServiceProviderService) AssignFreeServiceToDatabase(tx dax.Transaction, wspID dax.WorkerServiceProviderID, qdb *dax.QualifiedDatabase) (*dax.WorkerService, error) {
dt, ok := tx.(*DaxTransaction)
if !ok {
return nil, dax.NewErrInvalidTransaction("*sqldb.DaxTransaction")
}
ws := models.WorkerService{}
err := dt.C.Where("database_id = NULL and worker_service_provider_id = ?", wspID).First(&ws)
if err != nil {
return nil, errors.Wrap(err, "querying for free worker_service")
}
ws.DatabaseID = nulls.NewString(string(qdb.Database.ID))
ws.WorkersMin = qdb.Database.Options.WorkersMin
ws.WorkersMax = qdb.Database.Options.WorkersMax
if err := dt.C.Update(ws); err != nil {
return nil, errors.Wrap(err, "updating worker service")
}
daxWS := toDaxWorkerService(ws)
return &daxWS, nil
}
func (w *workerServiceProviderService) WorkerServices(tx dax.Transaction, wspID dax.WorkerServiceProviderID) (dax.WorkerServices, error) {
dt, ok := tx.(*DaxTransaction)
if !ok {
return nil, dax.NewErrInvalidTransaction("*sqldb.DaxTransaction")
}
query := dt.C.Q()
workerServices := make([]models.WorkerService, 0)
if wspID != "" {
query = query.Where("worker_service_provider_id = ?", wspID)
}
if err := query.All(&workerServices); err != nil {
return nil, errors.Wrapf(err, "getting Worker Services for ID '%s'", wspID)
}
return toDaxWorkerServices(workerServices), nil
}
func toDaxWorkerServices(wss []models.WorkerService) dax.WorkerServices {
dws := make(dax.WorkerServices, len(wss))
for i, wsm := range wss {
dws[i] = toDaxWorkerService(wsm)
}
return dws
}
func toDaxWorkerService(ws models.WorkerService) dax.WorkerService {
roles := []dax.RoleType{}
if ws.RoleCompute {
roles = append(roles, dax.RoleTypeCompute)
}
if ws.RoleTranslate {
roles = append(roles, dax.RoleTypeTranslate)
}
dbID := ""
if byts, _ := ws.DatabaseID.MarshalJSON(); string(byts) != "null" {
dbID = string(byts)
}
return dax.WorkerService{
ID: dax.WorkerServiceID(ws.ID),
Roles: roles,
WorkerServiceProviderID: dax.WorkerServiceProviderID(ws.WorkerServiceProviderID),
DatabaseID: dax.DatabaseID(dbID),
WorkersMin: ws.WorkersMin,
WorkersMax: ws.WorkersMax,
}
}

5
dax/dax.go
View file

@ -6,6 +6,7 @@ const (
ServicePrefixComputer = "computer"
ServicePrefixController = "controller"
ServicePrefixQueryer = "queryer"
ServicePrefixSnapshotter = "snapshotter"
ServicePrefixWritelogger = "writelogger"
ServicePrefixWSP = "worker_service_provider"
ServicePrefixSnapshotter = "snapshotter" // TODO remove?
ServicePrefixWritelogger = "writelogger" // TODO remove?
)

7
dax/errors.go
View file

@ -129,3 +129,10 @@ func NewErrInvalidTransaction(txType string) error {
fmt.Sprintf("tx is not expected type: '%s'", txType),
)
}
func NewErrUnimplemented(what string) error {
return errors.New(
ErrUnimplemented,
fmt.Sprintf("%s is unimplemented", what),
)
}

19
dax/http/handler.go
View file

@ -19,11 +19,7 @@ import (
type Handler struct {
Handler http.Handler
bind string
ln net.Listener
// url is used to hold the advertise bind address for printing a log during startup.
url string
closeTimeout time.Duration
@ -42,13 +38,6 @@ type Handler struct {
// HandlerOption is a functional option type for Handler
type HandlerOption func(s *Handler) error
func OptHandlerBind(b string) HandlerOption {
return func(h *Handler) error {
h.bind = b
return nil
}
}
func OptHandlerController(c *controller.Controller) HandlerOption {
return func(h *Handler) error {
h.controller = c
@ -93,13 +82,11 @@ func OptHandlerCloseTimeout(d time.Duration) HandlerOption {
}
}
// OptHandlerListener set the listener that will be used by the HTTP server.
// Url must be the advertised URL. It will be used to show a log to the user
// about where the Web UI is. This option is mandatory.
func OptHandlerListener(ln net.Listener, url string) HandlerOption {
// OptHandlerListener set the listener that will be used by the HTTP
// server. This option is not optional.
func OptHandlerListener(ln net.Listener) HandlerOption {
return func(h *Handler) error {
h.ln = ln
h.url = url
return nil
}
}

4
dax/migrations/004_serviceprovider.down.fizz Normal file
View file

@ -0,0 +1,4 @@
drop_table("worker_service_providers")
drop_table("worker_services")
drop_foreign_key("workers", "service_id", {"if_exists": true})
drop_column("workers", "service_id")

27
dax/migrations/004_serviceprovider.up.fizz Normal file
View file

@ -0,0 +1,27 @@
create_table("worker_service_providers") {
t.Column("id", "string", {primary: true})
t.Column("role_compute", "bool")
t.Column("role_translate", "bool")
t.Column("address", "string")
t.Column("description", "string")
t.Timestamps()
}
create_table("worker_services") {
t.Column("id", "string", {primary: true})
t.Column("worker_service_provider_id", "string")
t.ForeignKey("worker_service_provider_id", {"worker_service_providers": ["id"]}, {"on_delete": "cascade"})
t.Column("database_id", "string", {"null": true})
t.ForeignKey("database_id", {"databases": ["id"]}, {"on_delete": "cascade"})
t.Column("role_compute", "bool")
t.Column("role_translate", "bool")
t.Column("workers_min", "integer", {})
t.Column("workers_max", "integer", {})
t.Timestamps()
}
add_column("workers", "service_id", "string", {})
add_foreign_key("workers", "service_id", {"worker_services": ["id"]}, {"on_delete": "cascade"})

19
dax/models/worker.go
View file

@ -16,15 +16,16 @@ import (
// Worker is a node plus a role that gets assigned to a database and
// can be assigned jobs for that database.
type Worker struct {
ID uuid.UUID `json:"id" db:"id"`
Address dax.Address `json:"address" db:"address"`
DatabaseID nulls.String `json:"database_id" db:"database_id"` // this can be empty which means the worker is unassigned
CreatedAt time.Time `json:"created_at" db:"created_at"`
UpdatedAt time.Time `json:"updated_at" db:"updated_at"`
Jobs Jobs `json:"jobs" has_many:"jobs" order_by:"name asc"`
RoleCompute bool `json:"role_compute" db:"role_compute"`
RoleTranslate bool `json:"role_translate" db:"role_translate"`
RoleQuery bool `json:"role_query" db:"role_query"`
ID uuid.UUID `json:"id" db:"id"`
Address dax.Address `json:"address" db:"address"`
ServiceID dax.WorkerServiceID `json:"service_id" db:"service_id"`
Jobs Jobs `json:"jobs" has_many:"jobs" order_by:"name asc"`
RoleCompute bool `json:"role_compute" db:"role_compute"`
RoleTranslate bool `json:"role_translate" db:"role_translate"`
RoleQuery bool `json:"role_query" db:"role_query"`
DatabaseID nulls.String `json:"database_id" db:"database_id"` // this can be empty which means the worker is unassigned. Probably get rid of this now that every worker is associated w/ a Service and every Service has an ID
CreatedAt time.Time `json:"created_at" db:"created_at"`
UpdatedAt time.Time `json:"updated_at" db:"updated_at"`
}
// String is not required by pop and may be deleted

62
dax/models/worker_service.go Normal file
View file

@ -0,0 +1,62 @@
package models
import (
"encoding/json"
"time"
"github.com/gobuffalo/nulls"
"github.com/gobuffalo/pop/v6"
"github.com/gobuffalo/validate/v3"
"github.com/gobuffalo/validate/v3/validators"
)
// WorkerService represents an entity which has registered with the
// Controller as being able to provide WorkerServices (a WorkerService being an
// isolated container for compute workers or queryer workers).
type WorkerService struct {
ID string `json:"id" db:"id"`
WorkerServiceProviderID string `json:"worker_service_provider_id" db:"worker_service_provider_id"`
DatabaseID nulls.String `json:"database_id" db:"database_id"`
RoleCompute bool `json:"role_compute" db:"role_compute"`
RoleTranslate bool `json:"role_translate" db:"role_translate"`
WorkersMin int `json:"workers_min" db:"workers_min"`
WorkersMax int `json:"workers_max" db:"workers_max"`
CreatedAt time.Time `json:"created_at" db:"created_at"`
UpdatedAt time.Time `json:"updated_at" db:"updated_at"`
}
// String is not required by pop and may be deleted
func (d WorkerService) String() string {
jd, _ := json.MarshalIndent(d, " ", " ") //nolint:errchkjson
return string(jd)
}
// WorkerServices is not required by pop and may be deleted
type WorkerServices []WorkerService
// String is not required by pop and may be deleted
func (d WorkerServices) String() string {
jd, _ := json.Marshal(d) //nolint:errchkjson
return string(jd)
}
// Validate gets run every time you call a "pop.Validate*" (pop.ValidateAndSave, pop.ValidateAndCreate, pop.ValidateAndUpdate) method.
// This method is not required and may be deleted.
func (d *WorkerService) Validate(tx *pop.Connection) (*validate.Errors, error) {
return validate.Validate(
&validators.StringIsPresent{Field: string(d.ID), Name: "ID"},
&validators.StringIsPresent{Field: d.WorkerServiceProviderID, Name: "WorkerServiceProviderID"},
), nil
}
// ValidateCreate gets run every time you call "pop.ValidateAndCreate" method.
// This method is not required and may be deleted.
func (d *WorkerService) ValidateCreate(tx *pop.Connection) (*validate.Errors, error) {
return validate.NewErrors(), nil
}
// ValidateUpdate gets run every time you call "pop.ValidateAndUpdate" method.
// This method is not required and may be deleted.
func (d *WorkerService) ValidateUpdate(tx *pop.Connection) (*validate.Errors, error) {
return validate.NewErrors(), nil
}

60
dax/models/worker_service_provider.go Normal file
View file

@ -0,0 +1,60 @@
package models
import (
"encoding/json"
"time"
"github.com/featurebasedb/featurebase/v3/dax"
"github.com/gobuffalo/pop/v6"
"github.com/gobuffalo/validate/v3"
"github.com/gobuffalo/validate/v3/validators"
)
// WorkerServiceProvider represents an entity which has registered with the
// Controller as being able to provide Services (a Service being an
// isolated container for compute workers or queryer workers).
type WorkerServiceProvider struct {
ID string `json:"id" db:"id"`
RoleCompute bool `json:"role_compute" db:"role_compute"`
RoleTranslate bool `json:"role_translate" db:"role_translate"`
Address dax.Address `json:"address" db:"address"`
Description string `json:"description" db:"description"`
CreatedAt time.Time `json:"created_at" db:"created_at"`
UpdatedAt time.Time `json:"updated_at" db:"updated_at"`
}
// String is not required by pop and may be deleted
func (d WorkerServiceProvider) String() string {
jd, _ := json.MarshalIndent(d, " ", " ") //nolint:errchkjson
return string(jd)
}
// WorkerServiceProviders is not required by pop and may be deleted
type WorkerServiceProviders []WorkerServiceProvider
// String is not required by pop and may be deleted
func (d WorkerServiceProviders) String() string {
jd, _ := json.Marshal(d) //nolint:errchkjson
return string(jd)
}
// Validate gets run every time you call a "pop.Validate*" (pop.ValidateAndSave, pop.ValidateAndCreate, pop.ValidateAndUpdate) method.
// This method is not required and may be deleted.
func (d *WorkerServiceProvider) Validate(tx *pop.Connection) (*validate.Errors, error) {
return validate.Validate(
&validators.StringIsPresent{Field: string(d.ID), Name: "ID"},
&validators.StringIsPresent{Field: string(d.Address), Name: "Address"},
), nil
}
// ValidateCreate gets run every time you call "pop.ValidateAndCreate" method.
// This method is not required and may be deleted.
func (d *WorkerServiceProvider) ValidateCreate(tx *pop.Connection) (*validate.Errors, error) {
return validate.NewErrors(), nil
}
// ValidateUpdate gets run every time you call "pop.ValidateAndUpdate" method.
// This method is not required and may be deleted.
func (d *WorkerServiceProvider) ValidateUpdate(tx *pop.Connection) (*validate.Errors, error) {
return validate.NewErrors(), nil
}

23
dax/role.go
View file

@ -1,5 +1,7 @@
package dax
import "github.com/featurebasedb/featurebase/v3/errors"
// RoleType represents a role type which a worker node can act as.
type RoleType string
@ -60,3 +62,24 @@ type TranslateRole struct {
func (cr *TranslateRole) Type() RoleType {
return RoleTypeTranslate
}
func RoleTypesFromStrings(roles []string) (RoleTypes, error) {
tmp := make(map[RoleType]struct{})
for _, role := range roles {
switch role {
case string(RoleTypeCompute):
tmp[RoleTypeCompute] = struct{}{}
case string(RoleTypeTranslate):
tmp[RoleTypeTranslate] = struct{}{}
case string(RoleTypeQuery):
tmp[RoleTypeQuery] = struct{}{}
default:
return nil, errors.Errorf("unknown role type: '%s'", role)
}
}
ret := make(RoleTypes, 0, len(tmp))
for k := range tmp {
ret = append(ret, k)
}
return ret, nil
}

23
dax/server/config.go
View file

@ -12,6 +12,7 @@ import (
"github.com/featurebasedb/featurebase/v3/dax/controller"
"github.com/featurebasedb/featurebase/v3/dax/queryer"
wsp "github.com/featurebasedb/featurebase/v3/dax/worker_service_provider"
"github.com/featurebasedb/featurebase/v3/errors"
fbserver "github.com/featurebasedb/featurebase/v3/server"
)
@ -43,9 +44,10 @@ type Config struct {
// LogPath configures where Pilosa will write logs.
LogPath string `toml:"log-path"`
Controller ControllerOptions `toml:"controller"`
Queryer QueryerOptions `toml:"queryer"`
Computer ComputerOptions `toml:"computer"`
Controller ControllerOptions `toml:"controller"`
Queryer QueryerOptions `toml:"queryer"`
WorkerServiceProvider WSPOptions `toml:"worker-service-provider"`
Computer ComputerOptions `toml:"computer"`
}
type ControllerOptions struct {
@ -59,9 +61,14 @@ type QueryerOptions struct {
}
type ComputerOptions struct {
Run bool `toml:"run"`
N int `toml:"n"`
Config fbserver.Config `toml:"config"`
Run bool `toml:"run"`
WorkerServiceID string `toml:"worker-service-id"`
Config fbserver.Config `toml:"config"`
}
type WSPOptions struct {
Run bool `toml:"run"`
Config wsp.Config `toml:"config"`
}
// NewConfig returns an instance of Config with default options.
@ -75,6 +82,10 @@ func NewConfig() *Config {
SnappingTurtleTimeout: time.Minute * 3,
},
},
WorkerServiceProvider: WSPOptions{
Run: false,
Config: wsp.NewConfig(),
},
Bind: ":" + defaultBindPort,
Computer: ComputerOptions{
Config: *fbserver.NewConfig(),

68
dax/server/server.go
View file

@ -29,6 +29,8 @@ import (
daxhttp "github.com/featurebasedb/featurebase/v3/dax/http"
"github.com/featurebasedb/featurebase/v3/dax/queryer"
queryersvc "github.com/featurebasedb/featurebase/v3/dax/queryer/service"
wsp "github.com/featurebasedb/featurebase/v3/dax/worker_service_provider"
wspsvc "github.com/featurebasedb/featurebase/v3/dax/worker_service_provider/service"
"github.com/featurebasedb/featurebase/v3/errors"
"github.com/featurebasedb/featurebase/v3/logger"
fbnet "github.com/featurebasedb/featurebase/v3/net"
@ -246,7 +248,9 @@ func (m *Command) setupServer() error {
//m.Config.FeatureBase.Config.Listener = ln
// Get advertise address as uri.
// Get advertise address as uri. TODO: What if you pass a
// non-default bind and then don't pass advertise? Looks like
// advertise will be set to the default? Seems like a bug.
m.advertiseURI, err = featurebase.AddressWithDefaults(m.Config.Advertise)
if err != nil {
return errors.Wrap(err, "processing advertise address")
@ -256,14 +260,15 @@ func (m *Command) setupServer() error {
}
handlerOpts := []daxhttp.HandlerOption{
daxhttp.OptHandlerBind(m.Config.Bind),
daxhttp.OptHandlerListener(m.ln, m.advertiseURI.String()),
daxhttp.OptHandlerListener(m.ln),
daxhttp.OptHandlerLogger(m.logger),
}
drouter := m.svcmgr.HTTPHandler()
// Set up Handler based on which services are running in process.
// TODO above comment seems wrong... not sure what this handler
// has to do with which services are running.
m.Handler, err = daxhttp.NewHandler(drouter, handlerOpts...)
if err != nil {
return errors.Wrap(err, "new handler")
@ -286,6 +291,7 @@ func (m *Command) setupServices() error {
})
m.svcmgr.Controller = controllersvc.New(m.advertiseURI, controllerCfg)
// TODO: why are we starting the controller (and later Queryer) in here?? We call svcmgr.StartAll right after setupServices
if err := m.svcmgr.ControllerStart(); err != nil {
return errors.Wrap(err, "starting controller service")
}
@ -325,34 +331,44 @@ func (m *Command) setupServices() error {
// every iteration and create the new Command based on the copy (which can
// have a unique DataDir).
rootDataDir := m.Config.Computer.Config.DataDir
baseComputerConfig := computersvc.CommandConfig{
ComputerConfig: m.Config.Computer.Config,
Listener: m.ln,
RootDataDir: rootDataDir,
Stderr: m.stderr,
Logger: m.logger,
}
if m.Config.WorkerServiceProvider.Run {
wspCfg := m.Config.WorkerServiceProvider.Config
wspCfg.Address = m.advertiseURI
if m.svcmgr.Controller != nil {
wspCfg.ControllerAddress = string(m.svcmgr.Controller.Address())
}
wspCfg.Logger = m.logger
m.svcmgr.WorkerServiceProvider = wspsvc.New(m.advertiseURI, wsp.New(m.svcmgr, wspCfg), m.logger)
if err := m.svcmgr.WorkerServiceProvider.Start(); err != nil {
return errors.Wrap(err, "starting worker service provider service")
}
}
// Set up Computer.
if m.Config.Computer.Run {
n := m.Config.Computer.N
if n == 0 {
n = 1
m.logger.Printf("Set up computer")
cfg := baseComputerConfig
cfg.WorkerServiceID = dax.WorkerServiceID(m.Config.Computer.WorkerServiceID)
if cfg.ComputerConfig.ControllerAddress == "" && m.svcmgr.Controller != nil {
cfg.ComputerConfig.ControllerAddress = string(m.svcmgr.Controller.Address())
}
for i := 0; i < n; i++ {
m.logger.Printf("Set up computer (%d)", i)
cfg := computersvc.CommandConfig{
ComputerConfig: m.Config.Computer.Config,
Listener: m.ln,
RootDataDir: rootDataDir,
Stderr: m.stderr,
Logger: m.logger,
}
if cfg.ComputerConfig.ControllerAddress == "" && m.svcmgr.Controller != nil {
cfg.ComputerConfig.ControllerAddress = string(m.svcmgr.Controller.Address())
}
// Add new computer service.
_ = m.svcmgr.AddComputer(
computersvc.New(dax.Address(m.advertiseURI.HostPort()), cfg, m.logger))
}
// Add new computer service.
_ = m.svcmgr.AddComputer(
computersvc.New(dax.Address(m.advertiseURI.HostPort()), cfg, m.logger), 0)
}
return nil

6
dax/server/test/managed.go
View file

@ -108,7 +108,7 @@ func (mc *ManagedCommand) NewQueryer(cfg queryer.Config) dax.ServiceKey {
}
// NewComputer adds a new ComputerService to the ManagedCommands ServiceManager.
func (mc *ManagedCommand) NewComputer() dax.ServiceKey {
func (mc *ManagedCommand) NewComputer(id int) dax.ServiceKey {
cfg := computersvc.CommandConfig{
ComputerConfig: mc.Config.Computer.Config,
@ -122,7 +122,7 @@ func (mc *ManagedCommand) NewComputer() dax.ServiceKey {
// Add new computer service.
return mc.svcmgr.AddComputer(
computersvc.New(mc.Address(), cfg, cfg.Logger))
computersvc.New(mc.Address(), cfg, cfg.Logger), id)
}
// Healthy returns true if the provided service's /health endpoint returns 200
@ -225,8 +225,6 @@ func DefaultConfig() *server.Config {
cfg.Controller.Config.RegistrationBatchTimeout = 0
cfg.Controller.Config.SQLDB = sqldb.GetTestConfig()
cfg.Queryer.Run = true
cfg.Computer.Run = true
cfg.Computer.N = 1
return cfg
}

37
dax/service_manager.go
View file

@ -31,9 +31,12 @@ type ServiceManager struct {
Queryer QueryerService
queryerStarted bool
// WorkerServiceProvider
WorkerServiceProvider WorkerServiceProviderService
wspStarted bool
// Computers
computerID int
computers map[ServiceKey]*computerServiceState
computers map[ServiceKey]*computerServiceState
drouter *dynamicRouter
@ -75,6 +78,10 @@ func (s *ServiceManager) StartAll() error {
return errors.Wrap(err, "starting queryer")
}
if err := s.WorkerServiceProvider.Start(); err != nil {
return errors.Wrap(err, "starting worker service provider")
}
// Computer(s)
for key := range s.computers {
if err := s.ComputerStart(key); err != nil {
@ -91,6 +98,10 @@ func (s *ServiceManager) StopAll() error {
s.Logger.Printf("stopping computer %s: %v", key, err)
}
}
if err := s.WorkerServiceProviderStop(); err != nil {
s.Logger.Printf("stopping WorkerServiceProvider: %v", err)
}
if err := s.QueryerStop(); err != nil {
s.Logger.Printf("stopping queryer: %v", err)
}
@ -199,6 +210,14 @@ func (s *ServiceManager) QueryerStop() error {
return nil
}
func (s *ServiceManager) WorkerServiceProviderStart() error {
return NewErrUnimplemented("WorkerServiceProviderStart")
}
func (s *ServiceManager) WorkerServiceProviderStop() error {
return NewErrUnimplemented("WorkerServiceProviderStop")
}
// Computer returns the ComputerService specified by the provided key.
func (s *ServiceManager) Computer(key ServiceKey) ComputerService {
serviceState, ok := s.computers[key]
@ -284,18 +303,19 @@ func (s *ServiceManager) Computers() map[ServiceKey]ComputerService {
// AddComputer adds the provided ComputerService to ServiceManager. It assigns
// the service a unique ServiceKey.
func (s *ServiceManager) AddComputer(cs ComputerService) ServiceKey {
func (s *ServiceManager) AddComputer(cs ComputerService, id int) ServiceKey {
s.mu.Lock()
defer s.mu.Unlock()
key := ServiceKey(fmt.Sprintf("%s%d", ServicePrefixComputer, s.computerID))
key := ServiceKey(fmt.Sprintf("%s%d", ServicePrefixComputer, id))
if _, ok := s.computers[key]; ok {
panic(fmt.Sprintf("attempt to add computer with key %s that already exists", key))
}
s.computers[key] = &computerServiceState{
service: cs,
}
cs.SetKey(key)
s.computerID++
return key
}
@ -394,6 +414,11 @@ type QueryerService interface {
SetController(Address) error
}
type WorkerServiceProviderService interface {
Service
SetController(Address) error
}
//////////////////////////////////////////
// dynamicRouter is used to dynamically swap out http routers as service states

8
dax/test/dax/dax_test.go
View file

@ -97,7 +97,7 @@ func TestDAXIntegration(t *testing.T) {
assert.False(t, mc.Healthy(computerKey1))
// New and Start Computer 0.
computerKey0 = mc.NewComputer()
computerKey0 = mc.NewComputer(0)
assert.NoError(t, svcmgr.ComputerStart(computerKey0))
assert.True(t, mc.Healthy(controllerKey))
assert.True(t, mc.Healthy(queryerKey))
@ -105,7 +105,7 @@ func TestDAXIntegration(t *testing.T) {
assert.False(t, mc.Healthy(computerKey1))
// New and Start Computer 1.
computerKey1 = mc.NewComputer()
computerKey1 = mc.NewComputer(1)
assert.NoError(t, svcmgr.ComputerStart(computerKey1))
assert.True(t, mc.Healthy(controllerKey))
assert.True(t, mc.Healthy(queryerKey))
@ -318,7 +318,7 @@ func TestDAXIntegration(t *testing.T) {
time.Sleep(5 * time.Second)
// New and Start Computer 1.
computerKey1 := mc.NewComputer()
computerKey1 := mc.NewComputer(1)
assert.NoError(t, svcmgr.ComputerStart(computerKey1))
assert.False(t, mc.Healthy(computerKey0))
assert.True(t, mc.Healthy(computerKey1))
@ -397,7 +397,7 @@ func TestDAXIntegration(t *testing.T) {
time.Sleep(5 * time.Second)
// New and Start Computer 1.
computerKey1 := mc.NewComputer()
computerKey1 := mc.NewComputer(1)
assert.NoError(t, svcmgr.ComputerStart(computerKey1))
assert.True(t, mc.Healthy(computerKey1))
mc.WaitForApplied(t, computerKey1, 60, time.Second)

6
dax/worker.go
View file

@ -11,8 +11,14 @@ import (
// Node is used in API requests, like RegisterNode (before being assigned
// roles).
type Node struct {
// Address is the node's network address
Address Address `json:"address"`
// WorkerServiceID identifies the service that created the
// node. That will affect which database this node/worker can be
// assigned to.
ServiceID WorkerServiceID `json:"service_id"`
// RoleTypes allows a registering node to specify which role type(s) it is
// capable of filling. The controller will not assign a role to this node
// with a type not included in RoleTypes.

25
dax/worker_service.go Normal file
View file

@ -0,0 +1,25 @@
package dax
type WorkerServiceProviderID string
type WorkerServiceProvider struct {
ID WorkerServiceProviderID `json:"id"`
Roles RoleTypes `json:"roles"`
Address Address `json:"address"`
Description string `json:"description"`
}
type WorkerServiceProviders []WorkerServiceProvider
type WorkerServiceID string
type WorkerService struct {
ID WorkerServiceID `json:"id"`
Roles RoleTypes `json:"roles"`
WorkerServiceProviderID WorkerServiceProviderID `json:"worker-service-provider-id"`
DatabaseID DatabaseID `json:"database-id"`
WorkersMin int `json:"workers-min"`
WorkersMax int `json:"workers-max"`
}
type WorkerServices []WorkerService

95
dax/worker_service_provider/http/handler.go Normal file
View file

@ -0,0 +1,95 @@
package http
import (
"encoding/json"
"net/http"
"strings"
"github.com/featurebasedb/featurebase/v3/dax"
"github.com/featurebasedb/featurebase/v3/dax/worker_service_provider"
"github.com/gorilla/mux"
)
func Handler(wsp *worker_service_provider.WSP) http.Handler {
svr := &server{
wsp: wsp,
}
logRequestMiddleWare := func(next http.Handler) http.Handler {
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
if !strings.Contains(r.URL.Path, "/health") {
wsp.Logger().Debugf("serving %s, %v", r.Method, r.URL)
}
next.ServeHTTP(w, r)
})
}
router := mux.NewRouter()
router.Use(logRequestMiddleWare)
router.HandleFunc("/health", svr.getHealth).Methods("GET").Name("GetHealth")
router.HandleFunc("/claim", svr.postClaim).Methods("POST").Name("PostClaim")
router.HandleFunc("/update", svr.postUpdate).Methods("POST").Name("PostUpdate")
router.HandleFunc("/drop", svr.postDrop).Methods("POST").Name("PostDrop")
return router
}
type server struct {
wsp *worker_service_provider.WSP
}
// GET /health
func (s *server) getHealth(w http.ResponseWriter, r *http.Request) {
w.WriteHeader(http.StatusOK)
}
func (s *server) postClaim(w http.ResponseWriter, r *http.Request) {
body := r.Body
defer body.Close()
req := dax.WorkerService{}
if err := json.NewDecoder(body).Decode(&req); err != nil {
http.Error(w, err.Error(), http.StatusBadRequest)
return
}
err := s.wsp.ClaimService(r.Context(), req)
if err != nil {
http.Error(w, err.Error(), http.StatusInternalServerError)
return
}
}
func (s *server) postUpdate(w http.ResponseWriter, r *http.Request) {
body := r.Body
defer body.Close()
req := dax.WorkerService{}
if err := json.NewDecoder(body).Decode(&req); err != nil {
http.Error(w, err.Error(), http.StatusBadRequest)
return
}
err := s.wsp.UpdateService(r.Context(), req)
if err != nil {
http.Error(w, err.Error(), http.StatusInternalServerError)
return
}
}
func (s *server) postDrop(w http.ResponseWriter, r *http.Request) {
body := r.Body
defer body.Close()
req := dax.WorkerService{}
if err := json.NewDecoder(body).Decode(&req); err != nil {
http.Error(w, err.Error(), http.StatusBadRequest)
return
}
err := s.wsp.DropService(r.Context(), req)
if err != nil {
http.Error(w, err.Error(), http.StatusInternalServerError)
return
}
}

56
dax/worker_service_provider/service/wsp.go Normal file
View file

@ -0,0 +1,56 @@
package service
import (
"net/http"
"github.com/featurebasedb/featurebase/v3/dax"
controllerclient "github.com/featurebasedb/featurebase/v3/dax/controller/client"
"github.com/featurebasedb/featurebase/v3/dax/worker_service_provider"
wsphttp "github.com/featurebasedb/featurebase/v3/dax/worker_service_provider/http"
"github.com/featurebasedb/featurebase/v3/errors"
"github.com/featurebasedb/featurebase/v3/logger"
fbnet "github.com/featurebasedb/featurebase/v3/net"
)
// Ensure type implements interface.
var _ dax.Service = (*wspService)(nil)
type wspService struct {
uri *fbnet.URI
wsp *worker_service_provider.WSP
logger logger.Logger
}
func New(uri *fbnet.URI, wsp *worker_service_provider.WSP, logger logger.Logger) *wspService {
return &wspService{
uri: uri,
wsp: wsp,
logger: logger.WithPrefix("WorkerSvcProvider: "),
}
}
func (w *wspService) Start() error {
// Start wsp service.
if err := w.wsp.Start(); err != nil {
return errors.Wrap(err, "starting wsp")
}
return nil
}
func (w *wspService) Stop() error {
return nil
}
func (w *wspService) Address() dax.Address {
return dax.Address(w.uri.HostPort() + "/" + dax.ServicePrefixWSP)
}
func (w *wspService) HTTPHandler() http.Handler {
return wsphttp.Handler(w.wsp)
}
func (w *wspService) SetController(addr dax.Address) error {
controllercli := controllerclient.New(addr, w.logger)
w.wsp.SetController(controllercli)
return nil
}

278
dax/worker_service_provider/worker_service_provider.go Normal file
View file

@ -0,0 +1,278 @@
package worker_service_provider
import (
"context"
"fmt"
"sync"
"github.com/featurebasedb/featurebase/v3/dax"
computersvc "github.com/featurebasedb/featurebase/v3/dax/computer/service"
controllerclient "github.com/featurebasedb/featurebase/v3/dax/controller/client"
"github.com/featurebasedb/featurebase/v3/errors"
"github.com/featurebasedb/featurebase/v3/logger"
fbnet "github.com/featurebasedb/featurebase/v3/net"
)
func NewConfig() Config {
return Config{
ID: "default",
Roles: []string{"compute", "translate"},
Logger: logger.StderrLogger,
}
}
type Config struct {
ID string `toml:"id"`
Address *fbnet.URI `toml:"-"`
ControllerAddress string `toml:"controller-address"`
Roles []string `toml:"roles"`
Logger logger.Logger `toml:"-"`
}
func New(svcmgr *dax.ServiceManager, cfg Config) *WSP {
if cfg.Logger == nil {
cfg.Logger = logger.StderrLogger
}
if cfg.Address == nil {
panic("can't new up a WSP without an Address")
}
if cfg.ID == "" {
panic("can't new up a WSP without and ID")
}
if len(cfg.Roles) == 0 {
panic("can't new up a WSP without roles")
}
if cfg.ControllerAddress == "" {
panic("can't new up WSP without specifying a controller address")
}
roles, err := dax.RoleTypesFromStrings(cfg.Roles)
if err != nil {
panic(errors.Wrap(err, "validating roles"))
}
return &WSP{
id: dax.WorkerServiceProviderID(cfg.ID),
cfg: &cfg,
roles: roles,
controller: controllerclient.New(dax.Address(cfg.ControllerAddress), cfg.Logger),
svcmgr: svcmgr,
services: make(map[dax.WorkerServiceID]*workerService),
logger: cfg.Logger,
}
}
type WSP struct {
cfg *Config
computerConfig computersvc.CommandConfig
id dax.WorkerServiceProviderID
controller dax.Controller
svcmgr *dax.ServiceManager
roles dax.RoleTypes
mu sync.Mutex
// svcNum is a monotonically increasing integer used to assign a
// unique service ID to each service... it is not the number of
// services which can be gotten by len(wsp.services).
svcNum int
// workerNum is a monotonically increasing integer used by service
// manager to assign a unique key to each worker. It is *not* the
// total number of active workers.
workerNum int
services map[dax.WorkerServiceID]*workerService
logger logger.Logger
}
func (w *WSP) Start() error {
w.mu.Lock()
defer w.mu.Unlock()
// register with controller
svcs, err := w.controller.RegisterWorkerServiceProvider(context.Background(), dax.WorkerServiceProvider{
ID: w.id,
Roles: w.roles,
Address: dax.Address(w.cfg.Address.Normalize()),
Description: "",
})
if err != nil {
return errors.Wrap(err, "registering")
}
if len(svcs) > 0 {
return errors.Errorf("wasn't expecting svcs, but got %d", len(svcs))
}
// start a service and add it to services
if err := w.addService(); err != nil {
return errors.Wrap(err, "adding initial worker service")
}
return nil
}
// addService creates a new empty Service and starts a worker.
func (w *WSP) addService() error {
cfg := w.computerConfig
cfg.WorkerServiceID = dax.WorkerServiceID(fmt.Sprintf("ServiceID-%d", w.svcNum))
// create service
workerSvc := &workerService{
WorkerService: dax.WorkerService{
ID: cfg.WorkerServiceID,
Roles: w.roles,
WorkerServiceProviderID: w.id,
DatabaseID: "",
WorkersMin: 1,
WorkersMax: 1,
},
keys: make([]string, 1),
}
w.services[workerSvc.ID] = workerSvc
// register service
if err := w.controller.RegisterWorkerService(context.Background(), workerSvc.WorkerService); err != nil {
return errors.Wrap(err, "registering worker service")
}
// create worker in service
key := w.svcmgr.AddComputer(
computersvc.New(dax.Address(w.cfg.Address.HostPort()), cfg, w.logger), w.workerNum)
// track workers on service
workerSvc.keys[0] = string(key)
w.svcNum++
w.workerNum++
return nil
}
func (w *WSP) ClaimService(ctx context.Context, svc dax.WorkerService) error {
w.mu.Lock()
defer w.mu.Unlock()
// check that svc is in services, then update it
mysvc, ok := w.services[svc.ID]
if !ok {
return errors.Errorf("can't claim/update service with ID %s because it doesn't exist", svc.ID)
}
if svc.DatabaseID == "" {
return errors.Errorf("service must have a database ID in order to be claimed/updated: %+v", svc)
}
if mysvc.DatabaseID != "" && mysvc.DatabaseID != svc.DatabaseID {
return errors.Errorf("can't claim svc ID %s, svc is not free and databases don't match", svc.ID)
}
isClaim := mysvc.DatabaseID == ""
mysvc.WorkerService = svc
w.services[svc.ID] = mysvc
// claiming a service might update min/max workers, so we scale.
err := w.scale(mysvc)
if err != nil {
return errors.Wrap(err, "scaling service")
}
// start a new service so we have a free one in reserve
if isClaim {
err := w.addService()
if err != nil {
return errors.Wrap(err, "adding service")
}
}
return nil
}
// scale must be called with lock held
func (w *WSP) scale(mysvc *workerService) error {
// scale up
for mysvc.WorkersMin > len(mysvc.keys) {
w.addWorker(mysvc)
}
// scale down
for mysvc.WorkersMin < len(mysvc.keys) {
err := w.removeWorker(mysvc)
if err != nil {
return errors.Wrap(err, "while scaling down")
}
}
return nil
}
func (w *WSP) addWorker(mysvc *workerService) {
// create worker in service
cfg := w.computerConfig
cfg.WorkerServiceID = dax.WorkerServiceID(fmt.Sprintf("ServiceID-%d", w.svcNum))
key := w.svcmgr.AddComputer(
computersvc.New(dax.Address(w.cfg.Address.HostPort()), cfg, w.logger), w.workerNum)
// track workers on service
mysvc.keys = append(mysvc.keys, string(key))
w.workerNum++
}
func (w *WSP) removeWorker(mysvc *workerService) error {
if len(mysvc.keys) == 0 {
return errors.Errorf("asked to remove a worker from empty service: %+v", mysvc)
}
key := dax.ServiceKey(mysvc.keys[len(mysvc.keys)-1])
if err := w.svcmgr.ComputerStop(key); err != nil {
return errors.Wrapf(err, "stopping computer '%s'", key)
}
if ok := w.svcmgr.RemoveComputer(key); !ok {
return errors.Errorf("computer not found by RemoveComputer after stopping??? key: %s", key)
}
mysvc.keys = mysvc.keys[:len(mysvc.keys)-1]
return nil
}
func (w *WSP) UpdateService(ctx context.Context, svc dax.WorkerService) error {
return w.ClaimService(ctx, svc)
}
func (w *WSP) DropService(ctx context.Context, svc dax.WorkerService) error {
w.mu.Lock()
defer w.mu.Unlock()
// check that svc is in services
mysvc, ok := w.services[svc.ID]
if !ok {
return errors.Errorf("can't drop service with ID %s because it doesn't exist", svc.ID)
}
// stop it
mysvc.WorkersMin = 0
mysvc.WorkersMax = 0
err := w.scale(mysvc)
if err != nil {
return errors.Wrapf(err, "scaling down service %s to drop it", svc.ID)
}
// remove it
delete(w.services, svc.ID)
return nil
}
func (w *WSP) SetController(controller dax.Controller) error {
w.controller = controller
return nil
}
func (w *WSP) Logger() logger.Logger {
return w.logger
}
type workerService struct {
dax.WorkerService
keys []string
}

2
errors/errors.go
View file

@ -13,6 +13,8 @@ import (
// example, see the Is() method.
type Code string
var CodeTODO Code = "TODOError"
func New(code Code, message string) error {
return errors.WithStack(codedError{
Code: code,