From 468461fbcf32648b009a6641b4408792c8f989ef Mon Sep 17 00:00:00 2001 From: Travis Turner Date: Mon, 23 Jan 2023 19:59:47 -0600 Subject: [PATCH] Add Drop Database and Drop Table support (#2208) * Support Drop Table in serverless (include Snapshotter, Writelogger) * Finish Database methods Things like: - `Databases` - `DatabaseByID` - `DatabaseByName` - `DropDatabase` * Change Poller to use NodeService instead of its own map Instead of the Poller maintaining its own map of Addresses to poll, this commit changes the Poller to use the NodeService interface to get all known nodes from the Controller. The next commit needs to: Next, the logic in the boltdb NodeService implementation was moved to the boltdb Balancer implementation. That way, the Balancer can be the source of truth for all things nodes/workers/jobs. * Move NodeService from Controller to Balancer This commit moves the implementation of the NodeService into the Balancer, and aligns `Balancer.AddWorker` with `NodeService.CreateNode` so that they stay in sync. (Same for `Balancer.RemoveWorker` and `NodeService.DeleteNode`). * fix import of private repo * Fix go vet issues * Fix bug in DeregisterNode We need to remove the node from the NodeService even if it's not assigned to a database. The logic had a bug in it. This also adds some no-op implementations for SnapshotService and WriteloggerService. If a directory was not configured for that, then the computer node would panic on trying to read from the Snapshotter upon receiving a Directive. * queryer response content-type: json * Add support for NULL to WriteloggerDir and SnapshotterDir configs This commit changes the way WriteLoggerDir and SnapshotterDir are handled. If value is empty `""`, an error will be returned on computer startup. If value is `"NULL"`, a no-op implementation of the service will be used. This would be for a case that wanted to run serverless on-prem with no durable storage. Finally, any other value will be used as the directory to use. Some things which aren't considered here and may result in unexpected behavior: - a value with spaces `" "` - any "null" which is not "NULL"... like lowercase. * Finish the DropTable test * Change "disable service" value to case-insensitive "off" This commit also removes an unnecessary sleep in the tests. * Fix docker-compose variables for IDK test Co-authored-by: Matthew Jaffee --- api.go | 8 + api_directive.go | 14 +- ctl/dax.go | 8 - ctl/server.go | 4 +- dax/computer/interfaces.go | 59 +++- dax/computer/service/computer.go | 87 +++--- dax/controller.go | 20 ++ dax/dax.go | 2 +- dax/docker-compose.yml | 6 +- dax/errors.go | 12 +- dax/http/handler.go | 4 +- dax/mds/client/client.go | 214 +++++++++---- dax/mds/controller/balancer.go | 12 + dax/mds/controller/balancer/balancer.go | 39 ++- .../controller/balancer/boltdb/balancer.go | 3 +- .../controller/balancer}/boltdb/node.go | 42 ++- .../controller/balancer}/boltdb/node_test.go | 4 +- dax/mds/controller/config.go | 3 + dax/mds/controller/controller.go | 285 +++++++++++------- dax/mds/controller/node.go | 36 +++ dax/mds/controller/sets.go | 6 + dax/mds/http/handler.go | 92 +++++- dax/mds/mds.go | 45 +-- dax/mds/poller/config.go | 1 + dax/mds/poller/poller.go | 43 +-- dax/mds/poller/poller_test.go | 102 +++++-- dax/mds/schemar/boltdb/schemar.go | 45 +++ dax/mds/schemar/schemar.go | 13 +- dax/node.go | 33 +- dax/queryer/http/handler.go | 2 + dax/queryer/orchestrator.go | 4 +- dax/queryer/queryer.go | 35 +-- dax/queryer/service/queryer.go | 3 +- dax/queryer/translator.go | 24 +- dax/schema.go | 23 +- dax/server/config.go | 20 +- dax/server/server.go | 27 +- dax/server/test/managed.go | 14 +- dax/snapshotter/config.go | 8 - dax/snapshotter/snapshotter.go | 16 +- dax/snapshotter/snapshotter_test.go | 2 +- dax/storage/storage.go | 54 +++- dax/storage/storage_test.go | 12 +- dax/table.go | 7 + dax/test/dax/dax_test.go | 135 ++++++++- dax/writelogger/config.go | 8 - dax/writelogger/writelogger.go | 67 ++-- dax/writelogger/writelogger_test.go | 10 +- idk/docker-compose.yml | 6 +- idk/ingest.go | 2 +- idk/ingest_test.go | 2 +- idk/mds/importer.go | 28 +- server/config.go | 14 +- server/server.go | 9 +- sql3/test/defs/defs_unkeyed.go | 2 + 55 files changed, 1219 insertions(+), 557 deletions(-) create mode 100644 dax/controller.go rename dax/{ => mds/controller/balancer}/boltdb/node.go (75%) rename dax/{ => mds/controller/balancer}/boltdb/node_test.go (89%) create mode 100644 dax/mds/controller/node.go delete mode 100644 dax/snapshotter/config.go delete mode 100644 dax/writelogger/config.go diff --git a/api.go b/api.go index 28bbf76f4..0e2bacfb9 100644 --- a/api.go +++ b/api.go @@ -338,6 +338,14 @@ func (api *API) DeleteIndex(ctx context.Context, indexName string) error { if err != nil { return errors.Wrap(err, "deleting index") } + + // Remove from writelogger/snapshotter if serverless. + if api.isComputeNode { + if err := api.serverlessStorage.RemoveTable(dax.TableKey(indexName).QualifiedTableID()); err != nil { + return errors.Wrapf(err, "removing table from serverless storage: %s", indexName) + } + } + // Send the delete index message to all nodes. err = api.server.SendSync( &DeleteIndexMessage{ diff --git a/api_directive.go b/api_directive.go index 1e321e781..1dafbb680 100644 --- a/api_directive.go +++ b/api_directive.go @@ -42,10 +42,6 @@ func (api *API) ApplyDirective(ctx context.Context, d *dax.Directive) error { if err := api.deleteAllIndexes(ctx); err != nil { return errors.Wrap(err, "deleting all indexes") } - if err := api.serverlessStorage.RemoveAll(); err != nil { - return errors.Wrap(err, "removing all managers") - } - // Set previousDirective to empty so the diff handles everything as new. previousDirective = dax.Directive{} @@ -64,7 +60,7 @@ func (api *API) ApplyDirective(ctx context.Context, d *dax.Directive) error { // the "enactDirective" stage of ApplyDirective which validates against this // cached Directive, so it's important that it be set before calling // enactDirective(). An example: when loading partition data from the - // WriteLogger, there are validations to ensure that the partition being + // Writelogger, there are validations to ensure that the partition being // loaded is meant to be handled by this node; that validation is done // against the cached Directive. // TODO(tlt): despite what this comment says, this logic is not sound; we @@ -250,7 +246,7 @@ func (api *API) enactTables(ctx context.Context, fromD, toD *dax.Directive) erro // Remove all indexes that are no longer part of the directive. for _, tkey := range sc.removed() { idx := string(tkey) - if err := api.holder.deleteIndex(idx); err != nil { + if err := api.DeleteIndex(ctx, idx); err != nil { return errors.Wrapf(err, "deleting index: %s", tkey) } } @@ -328,7 +324,7 @@ func (api *API) pushJobsTableKeys(ctx context.Context, jobs chan<- directiveJobT // Get the diff between from/to directive.partitions. partComp := newPartitionsComparer(fromD.TranslatePartitionsMap(), toPartitionsMap) - // Loop over the partition map and load from WriteLogger. + // Loop over the partition map and load from Writelogger. for tkey, partitions := range partComp.added() { // Get index in order to find the translate stores (by partition) for // the table. @@ -415,7 +411,7 @@ func (api *API) pushJobsFieldKeys(ctx context.Context, jobs chan<- directiveJobT // Get the diff between from/to directive.fields. fieldComp := newFieldsComparer(fromD.TranslateFieldsMap(), toD.TranslateFieldsMap()) - // Loop over the field map and load from WriteLogger. + // Loop over the field map and load from Writelogger. for tkey, fields := range fieldComp.added() { for _, field := range fields { jobs <- directiveJobFieldKeys{ @@ -499,7 +495,7 @@ func (api *API) pushJobsShards(ctx context.Context, jobs chan<- directiveJobType // Get the diff between from/to directive shards. shardComp := newShardsComparer(fromD.ComputeShardsMap(), shardMap) - // Loop over the shard map and load from WriteLogger. + // Loop over the shard map and load from Writelogger. for tkey, shards := range shardComp.added() { for _, shard := range shards { jobs <- directiveJobShards{ diff --git a/ctl/dax.go b/ctl/dax.go index 61759916a..cb607c5a2 100644 --- a/ctl/dax.go +++ b/ctl/dax.go @@ -20,14 +20,6 @@ func BuildDAXFlags(cmd *cobra.Command, srv *server.Command) { flags.StringVar(&srv.Config.MDS.Config.DataDir, "mds.config.data-dir", srv.Config.MDS.Config.DataDir, "MDS directory to use in process.") flags.DurationVar(&srv.Config.MDS.Config.SnappingTurtleTimeout, "mds.config.snapping-turtle-timeout", srv.Config.MDS.Config.SnappingTurtleTimeout, "Period for running automatic snapshotting routine.") - // WriteLogger - flags.BoolVar(&srv.Config.WriteLogger.Run, "writelogger.run", srv.Config.WriteLogger.Run, "Run the WriteLogger service in process.") - flags.StringVar(&srv.Config.WriteLogger.Config.DataDir, "writelogger.config.data-dir", srv.Config.WriteLogger.Config.DataDir, "WriteLogger directory to use in process.") - - // Snapshotter - flags.BoolVar(&srv.Config.Snapshotter.Run, "snapshotter.run", srv.Config.Snapshotter.Run, "Run the Snapshotter service in process.") - flags.StringVar(&srv.Config.Snapshotter.Config.DataDir, "snapshotter.config.data-dir", srv.Config.Snapshotter.Config.DataDir, "Snapshotter directory to use in process.") - // Queryer flags.BoolVar(&srv.Config.Queryer.Run, "queryer.run", srv.Config.Queryer.Run, "Run the Queryer service in process.") flags.StringVar(&srv.Config.Queryer.Config.MDSAddress, "queryer.config.mds-address", srv.Config.Queryer.Config.MDSAddress, "Address of remote MDS process.") diff --git a/ctl/server.go b/ctl/server.go index f1a17c8e3..ec9416bf9 100644 --- a/ctl/server.go +++ b/ctl/server.go @@ -34,8 +34,8 @@ func serverFlagSet(srv *server.Config, prefix string) *pflag.FlagSet { flags := pflag.NewFlagSet("featurebase", pflag.ExitOnError) flags.StringVar(&srv.Name, pre("name"), srv.Name, "Name of the node in the cluster.") flags.StringVar(&srv.MDSAddress, pre("mds-address"), srv.MDSAddress, "MDS service to register with.") - flags.StringVar(&srv.WriteLogger, pre("write-logger"), srv.WriteLogger, "WriteLogger to read/write append logs.") - flags.StringVar(&srv.Snapshotter, pre("snapshotter"), srv.Snapshotter, "Snapshotter to read/write snapshots.") + flags.StringVar(&srv.WriteloggerDir, pre("writelogger-dir"), srv.WriteloggerDir, "Writelogger directory to read/write append logs.") + flags.StringVar(&srv.SnapshotterDir, pre("snapshotter-dir"), srv.SnapshotterDir, "Snapshotter directory to read/write snapshots.") flags.StringVarP(&srv.DataDir, pre("data-dir"), short("d"), srv.DataDir, "Directory to store FeatureBase data files.") flags.StringVarP(&srv.Bind, pre("bind"), short("b"), srv.Bind, "Default URI on which FeatureBase should listen.") flags.StringVar(&srv.BindGRPC, pre("bind-grpc"), srv.BindGRPC, "URI on which FeatureBase should listen for gRPC requests.") diff --git a/dax/computer/interfaces.go b/dax/computer/interfaces.go index ee2a6dfa1..a79f9743c 100644 --- a/dax/computer/interfaces.go +++ b/dax/computer/interfaces.go @@ -14,9 +14,9 @@ type Registrar interface { CheckInNode(ctx context.Context, node *dax.Node) error } -// WriteLogService represents the WriteLogService methods which Computer uses. -// These are typically implemented by the WriteLogger client. -type WriteLogService interface { +// WritelogService represents the WritelogService methods which Computer uses. +// These are typically implemented by the Writelogger client. +type WritelogService interface { AppendMessage(bucket string, key string, version int, msg []byte) error LogReader(bucket string, key string, version int) (io.ReadCloser, error) LogReaderFrom(bucket string, key string, version int, offset int) (io.ReadCloser, error) @@ -45,5 +45,56 @@ type SnapInfo struct { type WriteLogInfo SnapInfo // LogMessage is implemented by a variety of types which can be serialized as -// messages to the WriteLogger. +// messages to the Writelogger. type LogMessage interface{} + +/////////// No-op implementations of the interfaces ////////////// + +// nopWritelogService is a no-op implementation of the WritelogService +// interface. +type nopWritelogService struct{} + +func NewNopWritelogService() *nopWritelogService { + return &nopWritelogService{} +} +func (b *nopWritelogService) AppendMessage(bucket string, key string, version int, msg []byte) error { + return nil +} +func (b *nopWritelogService) LogReader(bucket string, key string, version int) (io.ReadCloser, error) { + return nil, nil +} +func (b *nopWritelogService) LogReaderFrom(bucket string, key string, version int, offset int) (io.ReadCloser, error) { + return nil, nil +} +func (b *nopWritelogService) DeleteLog(bucket string, key string, version int) error { + return nil +} +func (b *nopWritelogService) List(bucket, key string) ([]WriteLogInfo, error) { + return nil, nil +} +func (b *nopWritelogService) Lock(bucket, key string) error { + return nil +} +func (b *nopWritelogService) Unlock(bucket, key string) error { + return nil +} + +// nopSnapshotterService is a no-op implementation of the WritelogService +// interface. +type nopSnapshotterService struct{} + +func NewNopSnapshotterService() *nopSnapshotterService { + return &nopSnapshotterService{} +} +func (b *nopSnapshotterService) Read(bucket string, key string, version int) (io.ReadCloser, error) { + return nil, nil +} +func (b *nopSnapshotterService) Write(bucket string, key string, version int, rc io.ReadCloser) error { + return nil +} +func (b *nopSnapshotterService) WriteTo(bucket string, key string, version int, wrTo io.WriterTo) error { + return nil +} +func (b *nopSnapshotterService) List(bucket, key string) ([]SnapInfo, error) { + return nil, nil +} diff --git a/dax/computer/service/computer.go b/dax/computer/service/computer.go index 3a8cd8127..f5f422686 100644 --- a/dax/computer/service/computer.go +++ b/dax/computer/service/computer.go @@ -5,6 +5,7 @@ import ( "io" "net" "net/http" + "strings" featurebase "github.com/featurebasedb/featurebase/v3" "github.com/featurebasedb/featurebase/v3/dax" @@ -47,13 +48,17 @@ func (c *computerService) Start() error { // happen in a reasonable order like we do with the other service types. if c.computer == nil { c.cfg.Name = string(c.Key()) - c.computer = newCommand(c.addr, c.cfg) + if cmd, err := newCommand(c.addr, c.cfg); err != nil { + return errors.Wrapf(err, "getting new command for computer config: %s", c.cfg.Name) + } else { + c.computer = cmd + } if c.cfg.ComputerConfig.MDSAddress != "" { mdsAddr := dax.Address(c.cfg.ComputerConfig.MDSAddress) // Set mds (registrar) on computer. if err := c.SetMDS(mdsAddr); err != nil { - return errors.Wrapf(err, "setting mds service on computer: %s, %v", c.cfg.Name, err) + return errors.Wrapf(err, "setting mds service on computer: %s", c.cfg.Name) } } } @@ -109,11 +114,7 @@ type CommandConfig struct { // separate data directory for its holder. Name string - WriteLoggerRun bool - WriteLoggerConfig writelogger.Config - SnapshotterRun bool - SnapshotterConfig snapshotter.Config - ComputerConfig fbserver.Config + ComputerConfig fbserver.Config Listener net.Listener RootDataDir string @@ -122,28 +123,36 @@ type CommandConfig struct { Logger logger.Logger } -func newCommand(addr dax.Address, cfg CommandConfig) *fbserver.Command { - // Set up WriteLogger. - // TODO(tlt): since WriteLogger is no longer a separate service (but - // rather just a directory path) its configuration could be moved under - // computer, and then get rid of WriteLogger.Run. This would become "if - // DataDir != ''". Let's do this after we get rid of the dax integration - // tests which start up separate writelogger and snapshotter containers. - var wlSvc *writelogger.WriteLogger - if cfg.WriteLoggerRun { - wlSvc = writelogger.New(writelogger.Config{ - DataDir: cfg.WriteLoggerConfig.DataDir, - Logger: cfg.Logger, - }) +// serviceOffValue is a reserved term used to explicitly disable a service. When +// used, the service will be set to use its no-op implementation. For the user, +// this is case-insensitive. +var serviceOffValue = "off" + +func newCommand(addr dax.Address, cfg CommandConfig) (*fbserver.Command, error) { + // Set up Writelogger. + var wlSvc computer.WritelogService + wlDirToCompare := strings.TrimSpace(strings.ToLower(cfg.ComputerConfig.WriteloggerDir)) + switch wlDirToCompare { + case "": + return nil, errors.New(errors.ErrUncoded, "no writelogger directory configured") + case serviceOffValue: + wlSvc = computer.NewNopWritelogService() + cfg.Logger.Warnf("No writelogger configured, dynamic scaling will not function properly.") + default: + wlSvc = writelogger.New(cfg.ComputerConfig.WriteloggerDir, cfg.Logger) } // Set up Snapshotter. - var ssSvc *snapshotter.Snapshotter - if cfg.SnapshotterRun { - ssSvc = snapshotter.New(snapshotter.Config{ - DataDir: cfg.SnapshotterConfig.DataDir, - Logger: cfg.Logger, - }) + var ssSvc computer.SnapshotService + ssDirToCompare := strings.TrimSpace(strings.ToLower(cfg.ComputerConfig.SnapshotterDir)) + switch ssDirToCompare { + case "": + return nil, errors.New(errors.ErrUncoded, "no snapshotter directory configured") + case serviceOffValue: + ssSvc = computer.NewNopSnapshotterService() + cfg.Logger.Warnf("No snapshotter configured, dynamic scaling will not function properly.") + default: + ssSvc = snapshotter.New(cfg.ComputerConfig.SnapshotterDir, cfg.Logger) } // Set the FeatureBase.Config values based on the top-level Config @@ -153,26 +162,6 @@ func newCommand(addr dax.Address, cfg CommandConfig) *fbserver.Command { cfg.ComputerConfig.GRPCListener = &nopListener{} cfg.ComputerConfig.DataDir = cfg.RootDataDir + "/" + cfg.Name - var writeLoggerImpl computer.WriteLogService - if cfg.ComputerConfig.WriteLogger != "" { - panic("running separate writelogger is currently unsupported") - // writeLoggerImpl = writeloggerclient.New(dax.Address(cfg.ComputerConfig.WriteLogger)) - } else if wlSvc != nil { - writeLoggerImpl = wlSvc - } else { - cfg.Logger.Warnf("No writelogger configured, dynamic scaling will not function properly.") - } - - var snapshotterImpl computer.SnapshotService - if cfg.ComputerConfig.Snapshotter != "" { - panic("running separate snapshotter is currently unsupported") - // snapshotterImpl = snapshotterclient.New(dax.Address(cfg.ComputerConfig.Snapshotter)) - } else if ssSvc != nil { - snapshotterImpl = ssSvc - } else { - cfg.Logger.Warnf("No snapshotter configured.") - } - fbcmd := fbserver.NewCommand(cfg.Stderr, fbserver.OptCommandSetConfig(&cfg.ComputerConfig), fbserver.OptCommandServerOptions( @@ -180,13 +169,13 @@ func newCommand(addr dax.Address, cfg CommandConfig) *fbserver.Command { featurebase.OptServerLogger(cfg.Logger), ), fbserver.OptCommandInjections(fbserver.Injections{ - WriteLogger: writeLoggerImpl, - Snapshotter: snapshotterImpl, + Writelogger: wlSvc, + Snapshotter: ssSvc, IsComputeNode: true, }), ) - return fbcmd + return fbcmd, nil } type nopListener struct{} diff --git a/dax/controller.go b/dax/controller.go new file mode 100644 index 000000000..8f3d842eb --- /dev/null +++ b/dax/controller.go @@ -0,0 +1,20 @@ +package dax + +type Controller interface { + Noder + Schemar +} + +// Ensure type implements interface. +var _ Noder = &nopController{} +var _ Schemar = &nopController{} + +// nopController is a no-op implementation of the Controller interface. +type nopController struct { + nopNoder + NopSchemar +} + +func NewNopController() *nopController { + return &nopController{} +} diff --git a/dax/dax.go b/dax/dax.go index 052969d87..301521b86 100644 --- a/dax/dax.go +++ b/dax/dax.go @@ -7,5 +7,5 @@ const ( ServicePrefixMDS = "mds" ServicePrefixQueryer = "queryer" ServicePrefixSnapshotter = "snapshotter" - ServicePrefixWriteLogger = "writelogger" + ServicePrefixWritelogger = "writelogger" ) diff --git a/dax/docker-compose.yml b/dax/docker-compose.yml index c6ca7f8f7..a22fa75fc 100644 --- a/dax/docker-compose.yml +++ b/dax/docker-compose.yml @@ -39,10 +39,8 @@ services: FEATUREBASE_BIND: 0.0.0.0:8080 FEATUREBASE_VERBOSE: "true" FEATUREBASE_STORAGE_METHOD: boltdb - FEATUREBASE_WRITELOGGER_RUN: "true" - FEATUREBASE_WRITELOGGER_CONFIG_DATA_DIR: "/dax-data/writelogger" - FEATUREBASE_SNAPSHOTTER_RUN: "true" - FEATUREBASE_SNAPSHOTTER_CONFIG_DATA_DIR: "/dax-data/snapshotter" + FEATUREBASE_COMPUTER_CONFIG_WRITELOGGER_DIR: "/dax-data/writelogger" + FEATUREBASE_COMPUTER_CONFIG_SNAPSHOTTER_DIR: "/dax-data/snapshotter" volumes: - "./dax-data/writelogger:/dax-data/writelogger" - "./dax-data/snapshotter:/dax-data/snapshotter" diff --git a/dax/errors.go b/dax/errors.go index 728b297ea..fb9416447 100644 --- a/dax/errors.go +++ b/dax/errors.go @@ -7,8 +7,9 @@ import ( ) const ( - ErrDatabaseIDExists errors.Code = "DatabaseIDExists" - ErrDatabaseIDDoesNotExist errors.Code = "DatabaseIDDoesNotExist" + ErrDatabaseIDExists errors.Code = "DatabaseIDExists" + ErrDatabaseIDDoesNotExist errors.Code = "DatabaseIDDoesNotExist" + ErrDatabaseNameDoesNotExist errors.Code = "DatabaseNameDoesNotExist" ErrTableIDExists errors.Code = "TableIDExists" ErrTableKeyExists errors.Code = "TableKeyExists" @@ -42,6 +43,13 @@ func NewErrDatabaseIDDoesNotExist(qdbid QualifiedDatabaseID) error { ) } +func NewErrDatabaseNameDoesNotExist(dbName DatabaseName) error { + return errors.New( + ErrDatabaseNameDoesNotExist, + fmt.Sprintf("database name '%s' does not exist", dbName), + ) +} + func NewErrTableIDDoesNotExist(qtid QualifiedTableID) error { return errors.New( ErrTableIDDoesNotExist, diff --git a/dax/http/handler.go b/dax/http/handler.go index 44588ea31..78211a8a4 100644 --- a/dax/http/handler.go +++ b/dax/http/handler.go @@ -30,7 +30,7 @@ type Handler struct { server *http.Server mds *mds.MDS - writeLogger *writelogger.WriteLogger + writeLogger *writelogger.Writelogger snapshotter *snapshotter.Snapshotter queryer *queryer.Queryer @@ -56,7 +56,7 @@ func OptHandlerMDS(m *mds.MDS) HandlerOption { } } -func OptHandlerWriteLogger(w *writelogger.WriteLogger) HandlerOption { +func OptHandlerWritelogger(w *writelogger.Writelogger) HandlerOption { return func(h *Handler) error { h.writeLogger = w return nil diff --git a/dax/mds/client/client.go b/dax/mds/client/client.go index 3f2c1c163..721e86b78 100644 --- a/dax/mds/client/client.go +++ b/dax/mds/client/client.go @@ -10,6 +10,7 @@ import ( "net/http" "github.com/featurebasedb/featurebase/v3/dax" + "github.com/featurebasedb/featurebase/v3/dax/computer" mdshttp "github.com/featurebasedb/featurebase/v3/dax/mds/http" "github.com/featurebasedb/featurebase/v3/errors" "github.com/featurebasedb/featurebase/v3/logger" @@ -19,6 +20,10 @@ const ( defaultScheme = "http" ) +// Ensure type implements interface. +var _ computer.Registrar = (*Client)(nil) +var _ dax.Schemar = (*Client)(nil) + // Client is an HTTP client that operates on the MDS endpoints exposed by the // main MDS service. type Client struct { @@ -48,6 +53,159 @@ func (c *Client) Health() bool { return true } +func (c *Client) CreateDatabase(ctx context.Context, qdb *dax.QualifiedDatabase) error { + url := fmt.Sprintf("%s/create-database", c.address.WithScheme(defaultScheme)) + + // Encode the request. + postBody, err := json.Marshal(qdb) + if err != nil { + return errors.Wrap(err, "marshalling post request") + } + responseBody := bytes.NewBuffer(postBody) + + // Post the request. + resp, err := http.Post(url, "application/json", responseBody) + if err != nil { + return errors.Wrap(err, "posting create database 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 +} + +func (c *Client) DropDatabase(ctx context.Context, qdbid dax.QualifiedDatabaseID) error { + url := fmt.Sprintf("%s/drop-database", c.address.WithScheme(defaultScheme)) + + // Encode the request. + postBody, err := json.Marshal(qdbid) + if err != nil { + return errors.Wrap(err, "marshalling post request") + } + responseBody := bytes.NewBuffer(postBody) + + // Post the request. + resp, err := http.Post(url, "application/json", responseBody) + if err != nil { + return errors.Wrap(err, "posting drop database 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 +} + +func (c *Client) DatabaseByID(ctx context.Context, qdbid dax.QualifiedDatabaseID) (*dax.QualifiedDatabase, error) { + url := fmt.Sprintf("%s/database-by-id", c.address.WithScheme(defaultScheme)) + + // Encode the request. + postBody, err := json.Marshal(qdbid) + if err != nil { + return nil, errors.Wrap(err, "marshalling post request") + } + responseBody := bytes.NewBuffer(postBody) + + // Post the request. + c.logger.Debugf("POST database-by-id request: url: %s", url) + resp, err := http.Post(url, "application/json", responseBody) + if err != nil { + return nil, errors.Wrap(err, "posting database-by-id 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 qdb *dax.QualifiedDatabase + if err := json.NewDecoder(resp.Body).Decode(&qdb); err != nil { + return nil, errors.Wrap(err, "reading response body") + } + + return qdb, nil +} + +func (c *Client) DatabaseByName(ctx context.Context, orgID dax.OrganizationID, name dax.DatabaseName) (*dax.QualifiedDatabase, error) { + url := fmt.Sprintf("%s/database-by-name", c.address.WithScheme(defaultScheme)) + + req := &mdshttp.DatabaseByNameRequest{ + OrganizationID: orgID, + Name: name, + } + + // Encode the request. + postBody, err := json.Marshal(req) + if err != nil { + return nil, errors.Wrap(err, "marshalling post request") + } + responseBody := bytes.NewBuffer(postBody) + + // Post the request. + c.logger.Debugf("POST database-by-name request: url: %s", url) + resp, err := http.Post(url, "application/json", responseBody) + if err != nil { + return nil, errors.Wrap(err, "posting database-by-name 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 qdb *dax.QualifiedDatabase + if err := json.NewDecoder(resp.Body).Decode(&qdb); err != nil { + return nil, errors.Wrap(err, "reading response body") + } + + return qdb, nil +} + +func (c *Client) Databases(ctx context.Context, orgID dax.OrganizationID, ids ...dax.DatabaseID) ([]*dax.QualifiedDatabase, error) { + url := fmt.Sprintf("%s/databases", c.address.WithScheme(defaultScheme)) + + req := &mdshttp.DatabasesRequest{ + OrganizationID: orgID, + DatabaseIDs: ids, + } + + // Encode the request. + postBody, err := json.Marshal(req) + if err != nil { + return nil, errors.Wrap(err, "marshalling post request") + } + responseBody := bytes.NewBuffer(postBody) + + // Post the request. + c.logger.Debugf("POST databases request: url: %s", url) + resp, err := http.Post(url, "application/json", responseBody) + if err != nil { + return nil, errors.Wrap(err, "posting databases 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 qdbs []*dax.QualifiedDatabase + if err := json.NewDecoder(resp.Body).Decode(&qdbs); err != nil { + return nil, errors.Wrap(err, "reading response body") + } + + return qdbs, nil +} + // TODO(tlt): collapse Table into this func (c *Client) TableByID(ctx context.Context, qtid dax.QualifiedTableID) (*dax.QualifiedTable, error) { return c.Table(ctx, qtid) @@ -166,62 +324,6 @@ func (c *Client) Tables(ctx context.Context, qdbid dax.QualifiedDatabaseID, ids return qtables, nil } -func (c *Client) CreateDatabase(ctx context.Context, qdb *dax.QualifiedDatabase) error { - url := fmt.Sprintf("%s/create-database", c.address.WithScheme(defaultScheme)) - - // Encode the request. - postBody, err := json.Marshal(qdb) - if err != nil { - return errors.Wrap(err, "marshalling post request") - } - responseBody := bytes.NewBuffer(postBody) - - // Post the request. - resp, err := http.Post(url, "application/json", responseBody) - if err != nil { - return errors.Wrap(err, "posting create database 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 -} - -func (c *Client) DatabaseByID(ctx context.Context, qdbid dax.QualifiedDatabaseID) (*dax.QualifiedDatabase, error) { - url := fmt.Sprintf("%s/database-by-id", c.address.WithScheme(defaultScheme)) - - // Encode the request. - postBody, err := json.Marshal(qdbid) - if err != nil { - return nil, errors.Wrap(err, "marshalling post request") - } - responseBody := bytes.NewBuffer(postBody) - - // Post the request. - c.logger.Debugf("POST database request: url: %s", url) - resp, err := http.Post(url, "application/json", responseBody) - if err != nil { - return nil, errors.Wrap(err, "posting table 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 qdb *dax.QualifiedDatabase - if err := json.NewDecoder(resp.Body).Decode(&qdb); err != nil { - return nil, errors.Wrap(err, "reading response body") - } - - return qdb, nil -} - func (c *Client) CreateTable(ctx context.Context, qtbl *dax.QualifiedTable) error { url := fmt.Sprintf("%s/create-table", c.address.WithScheme(defaultScheme)) diff --git a/dax/mds/controller/balancer.go b/dax/mds/controller/balancer.go index 2344af873..3884bd997 100644 --- a/dax/mds/controller/balancer.go +++ b/dax/mds/controller/balancer.go @@ -37,6 +37,12 @@ type Balancer interface { // WorkersForTable returns the workers responsible for any job related to // the given table. WorkersForTable(tx dax.Transaction, roleType dax.RoleType, qtid dax.QualifiedTableID) ([]dax.WorkerInfo, error) + + // ReadNode returns the node for the given address. + ReadNode(tx dax.Transaction, addr dax.Address) (*dax.Node, error) + + // Nodes returns all nodes known by the Balancer. + Nodes(tx dax.Transaction) ([]*dax.Node, error) } // Ensure type implements interface. @@ -76,3 +82,9 @@ func (b *NopBalancer) WorkersForJobs(tx dax.Transaction, roleType dax.RoleType, func (b *NopBalancer) WorkersForTable(tx dax.Transaction, roleType dax.RoleType, qtid dax.QualifiedTableID) ([]dax.WorkerInfo, error) { return []dax.WorkerInfo{}, nil } +func (b *NopBalancer) ReadNode(tx dax.Transaction, addr dax.Address) (*dax.Node, error) { + return nil, nil +} +func (b *NopBalancer) Nodes(tx dax.Transaction) ([]*dax.Node, error) { + return []*dax.Node{}, nil +} diff --git a/dax/mds/controller/balancer/balancer.go b/dax/mds/controller/balancer/balancer.go index f29813819..1d10f1420 100644 --- a/dax/mds/controller/balancer/balancer.go +++ b/dax/mds/controller/balancer/balancer.go @@ -28,6 +28,8 @@ type Balancer struct { // current represents the current state of worker/job assigments. current WorkerJobService + nodeService controller.NodeService + // freeJobs is the set of jobs which have yet to be assigned to a worker. // This could be because there are no available workers, or because a worker // has been removed and the jobs for which it was responsible have yet to be @@ -42,9 +44,10 @@ type Balancer struct { } // New returns a new instance of Balancer. -func New(fjs FreeJobService, wjs WorkerJobService, fws FreeWorkerService, schemar schemar.Schemar, logger logger.Logger) *Balancer { +func New(ns controller.NodeService, fjs FreeJobService, wjs WorkerJobService, fws FreeWorkerService, schemar schemar.Schemar, logger logger.Logger) *Balancer { return &Balancer{ current: wjs, + nodeService: ns, freeJobs: fjs, freeWorkers: fws, schemar: schemar, @@ -60,6 +63,10 @@ func (b *Balancer) AddWorker(tx dax.Transaction, node *dax.Node) ([]dax.WorkerDi addr := node.Address b.logger.Debugf("AddWorker(%s)", addr) + if err := b.nodeService.CreateNode(tx, addr, node); err != nil { + return nil, errors.Wrapf(err, "creating node on node service: %s", addr) + } + diffs := NewInternalDiffs() // This logic means that a node is used for ALL of the role types specified. @@ -235,19 +242,29 @@ func (b *Balancer) databaseHasJobs(tx dax.Transaction, roleType dax.RoleType, qd func (b *Balancer) RemoveWorker(tx dax.Transaction, addr dax.Address) ([]dax.WorkerDiff, error) { diffs := NewInternalDiffs() - // See if the worker is assigned to a database. + // Remove the worker from the free worker list (if it's there). + for _, rt := range []dax.RoleType{dax.RoleTypeCompute, dax.RoleTypeTranslate} { + if err := b.freeWorkers.RemoveWorker(tx, rt, addr); err != nil { + return nil, errors.Wrapf(err, "removing worker from free list: (%s) %s", rt, addr) + } + } + + // Remove the worker (i.e. Node) from the node service. + if err := b.nodeService.DeleteNode(tx, addr); err != nil { + return nil, errors.Wrapf(err, "deleting node from node service: %s", addr) + } + + ////// The rest is database specific. //////////// + + // See if the worker is assigned to a database. If it's not, return early. dbkey := b.current.DatabaseForWorker(tx, addr) if dbkey == "" { return diffs.Output(), nil } + qdbid := dbkey.QualifiedDatabaseID() for _, rt := range []dax.RoleType{dax.RoleTypeCompute, dax.RoleTypeTranslate} { - // Remove the worker form the free worker list (if it's there). - if err := b.freeWorkers.RemoveWorker(tx, rt, addr); err != nil { - return nil, errors.Wrapf(err, "removing worker from free list: (%s) %s", rt, addr) - } - if diff, err := b.removeDatabaseWorker(tx, rt, qdbid, addr); err != nil { return nil, errors.Wrapf(err, "removing worker: (%s) %s", rt, addr) } else { @@ -787,6 +804,14 @@ func (b *Balancer) workerForJob(tx dax.Transaction, roleType dax.RoleType, qdbid return "", false, nil } +func (b *Balancer) ReadNode(tx dax.Transaction, addr dax.Address) (*dax.Node, error) { + return b.nodeService.ReadNode(tx, addr) +} + +func (b *Balancer) Nodes(tx dax.Transaction) ([]*dax.Node, error) { + return b.nodeService.Nodes(tx) +} + type WorkerJobService interface { WorkersJobs(tx dax.Transaction, roleType dax.RoleType, qdbid dax.QualifiedDatabaseID) ([]dax.WorkerInfo, error) diff --git a/dax/mds/controller/balancer/boltdb/balancer.go b/dax/mds/controller/balancer/boltdb/balancer.go index bb2725718..950b6763b 100644 --- a/dax/mds/controller/balancer/boltdb/balancer.go +++ b/dax/mds/controller/balancer/boltdb/balancer.go @@ -31,8 +31,9 @@ func NewBalancer(db *boltdb.DB, schemar schemar.Schemar, logger logger.Logger) c fjs := newFreeJobService(db) wjs := newWorkerJobService(db, logger) fws := newFreeWorkerService(db) + ns := NewNodeService(db, logger) - return balancer.New(fjs, wjs, fws, schemar, logger) + return balancer.New(ns, fjs, wjs, fws, schemar, logger) } // Ensure type implements interface. diff --git a/dax/boltdb/node.go b/dax/mds/controller/balancer/boltdb/node.go similarity index 75% rename from dax/boltdb/node.go rename to dax/mds/controller/balancer/boltdb/node.go index b06b3912d..a6214b2f9 100644 --- a/dax/boltdb/node.go +++ b/dax/mds/controller/balancer/boltdb/node.go @@ -6,32 +6,24 @@ import ( "fmt" "github.com/featurebasedb/featurebase/v3/dax" + "github.com/featurebasedb/featurebase/v3/dax/boltdb" + "github.com/featurebasedb/featurebase/v3/dax/mds/controller" "github.com/featurebasedb/featurebase/v3/errors" "github.com/featurebasedb/featurebase/v3/logger" ) -var ( - bucketNodes = Bucket("nodeServiceNodes") -) - -// NodeServiceBuckets defines the buckets used by this package. It can be called -// during setup to create the buckets ahead of time. -var NodeServiceBuckets []Bucket = []Bucket{ - bucketNodes, -} - // Ensure type implements interface. -var _ dax.NodeService = (*NodeService)(nil) +var _ controller.NodeService = (*NodeService)(nil) // NodeService represents a service for managing nodes. type NodeService struct { - db *DB + db *boltdb.DB logger logger.Logger } // NewNodeService returns a new instance of NodeService with default values. -func NewNodeService(db *DB, logger logger.Logger) *NodeService { +func NewNodeService(db *boltdb.DB, logger logger.Logger) *NodeService { return &NodeService{ db: db, logger: logger, @@ -39,14 +31,14 @@ func NewNodeService(db *DB, logger logger.Logger) *NodeService { } func (s *NodeService) CreateNode(tx dax.Transaction, addr dax.Address, node *dax.Node) error { - txx, ok := tx.(*Tx) + txx, ok := tx.(*boltdb.Tx) if !ok { return dax.NewErrInvalidTransaction() } - bkt := txx.Bucket(bucketNodes) + bkt := txx.Bucket(bucketBalancer) if bkt == nil { - return errors.Errorf(ErrFmtBucketNotFound, bucketNodes) + return errors.Errorf(boltdb.ErrFmtBucketNotFound, bucketBalancer) } val, err := json.Marshal(node) @@ -62,14 +54,14 @@ func (s *NodeService) CreateNode(tx dax.Transaction, addr dax.Address, node *dax } func (s *NodeService) ReadNode(tx dax.Transaction, addr dax.Address) (*dax.Node, error) { - txx, ok := tx.(*Tx) + txx, ok := tx.(*boltdb.Tx) if !ok { return nil, dax.NewErrInvalidTransaction() } - bkt := txx.Bucket(bucketNodes) + bkt := txx.Bucket(bucketBalancer) if bkt == nil { - return nil, errors.Errorf(ErrFmtBucketNotFound, bucketNodes) + return nil, errors.Errorf(boltdb.ErrFmtBucketNotFound, bucketBalancer) } b := bkt.Get(addressKey(addr)) @@ -86,14 +78,14 @@ func (s *NodeService) ReadNode(tx dax.Transaction, addr dax.Address) (*dax.Node, } func (s *NodeService) DeleteNode(tx dax.Transaction, addr dax.Address) error { - txx, ok := tx.(*Tx) + txx, ok := tx.(*boltdb.Tx) if !ok { return dax.NewErrInvalidTransaction() } - bkt := txx.Bucket(bucketNodes) + bkt := txx.Bucket(bucketBalancer) if bkt == nil { - return errors.Errorf(ErrFmtBucketNotFound, bucketNodes) + return errors.Errorf(boltdb.ErrFmtBucketNotFound, bucketBalancer) } if err := bkt.Delete(addressKey(addr)); err != nil { @@ -104,7 +96,7 @@ func (s *NodeService) DeleteNode(tx dax.Transaction, addr dax.Address) error { } func (s *NodeService) Nodes(tx dax.Transaction) ([]*dax.Node, error) { - txx, ok := tx.(*Tx) + txx, ok := tx.(*boltdb.Tx) if !ok { return nil, dax.NewErrInvalidTransaction() } @@ -117,8 +109,8 @@ func (s *NodeService) Nodes(tx dax.Transaction) ([]*dax.Node, error) { return nodes, nil } -func (s *NodeService) getNodes(tx *Tx) ([]*dax.Node, error) { - c := tx.Bucket(bucketNodes).Cursor() +func (s *NodeService) getNodes(tx *boltdb.Tx) ([]*dax.Node, error) { + c := tx.Bucket(bucketBalancer).Cursor() // Deserialize rows into Node objects. nodes := make([]*dax.Node, 0) diff --git a/dax/boltdb/node_test.go b/dax/mds/controller/balancer/boltdb/node_test.go similarity index 89% rename from dax/boltdb/node_test.go rename to dax/mds/controller/balancer/boltdb/node_test.go index 591bb6968..8a332f7b1 100644 --- a/dax/boltdb/node_test.go +++ b/dax/mds/controller/balancer/boltdb/node_test.go @@ -5,7 +5,7 @@ import ( "testing" "github.com/featurebasedb/featurebase/v3/dax" - "github.com/featurebasedb/featurebase/v3/dax/boltdb" + "github.com/featurebasedb/featurebase/v3/dax/mds/controller/balancer/boltdb" testbolt "github.com/featurebasedb/featurebase/v3/dax/test/boltdb" "github.com/featurebasedb/featurebase/v3/errors" "github.com/featurebasedb/featurebase/v3/logger" @@ -23,7 +23,7 @@ func TestNodeService(t *testing.T) { ctx := context.Background() // Initialize the buckets. - assert.NoError(t, db.InitializeBuckets(boltdb.NodeServiceBuckets...)) + assert.NoError(t, db.InitializeBuckets(boltdb.BalancerBuckets...)) t.Run("Nodes", func(t *testing.T) { ns := boltdb.NewNodeService(db, logger.NopLogger) diff --git a/dax/mds/controller/config.go b/dax/mds/controller/config.go index e67d3ddb0..368090e3b 100644 --- a/dax/mds/controller/config.go +++ b/dax/mds/controller/config.go @@ -19,6 +19,9 @@ type Config struct { StorageMethod string BoltDB *boltdb.DB + SnapshotterDir string + WriteloggerDir string + // RegistrationBatchTimeout is the time that the controller will // wait after a node registers itself to see if any more nodes // will register before sending out directives to all nodes which diff --git a/dax/mds/controller/controller.go b/dax/mds/controller/controller.go index abb4332f3..72f983727 100644 --- a/dax/mds/controller/controller.go +++ b/dax/mds/controller/controller.go @@ -9,31 +9,35 @@ import ( "github.com/featurebasedb/featurebase/v3/dax" "github.com/featurebasedb/featurebase/v3/dax/boltdb" + "github.com/featurebasedb/featurebase/v3/dax/computer" "github.com/featurebasedb/featurebase/v3/dax/mds/schemar" + "github.com/featurebasedb/featurebase/v3/dax/snapshotter" + "github.com/featurebasedb/featurebase/v3/dax/writelogger" "github.com/featurebasedb/featurebase/v3/errors" "github.com/featurebasedb/featurebase/v3/logger" "golang.org/x/sync/errgroup" ) +// Ensure type implements interface. +var _ computer.Registrar = (*Controller)(nil) +var _ dax.Schemar = (*Controller)(nil) +var _ dax.NodeService = (*Controller)(nil) + type Controller struct { // Schemar used by the controller to get table information. The controller // should NOT call Schemar methods which modify data. Schema mutations are // made outside of the controller (at this point that happens in MDS). Schemar schemar.Schemar - // nodeService is the interface to working with nodes, by address, which - // have registered with the controller. - nodeService dax.NodeService - boltDB *boltdb.DB Balancer Balancer + Snapshotter *snapshotter.Snapshotter + Writelogger *writelogger.Writelogger + // Director is used to send directives to computer workers. Director Director - // poller is used to notify a Poller if nodes have been added or removed. - poller dax.AddressManager - directiveVersion dax.DirectiveVersion registrationBatchTimeout time.Duration @@ -60,8 +64,6 @@ func New(cfg Config) *Controller { Director: NewNopDirector(), - poller: dax.NewNopAddressManager(), - registrationBatchTimeout: cfg.RegistrationBatchTimeout, nodeChan: make(chan *dax.Node, 10), snappingTurtleTimeout: cfg.SnappingTurtleTimeout, @@ -75,13 +77,11 @@ func New(cfg Config) *Controller { c.logger = cfg.Logger } + c.Snapshotter = snapshotter.New(cfg.SnapshotterDir, c.logger) + c.Writelogger = writelogger.New(cfg.WriteloggerDir, c.logger) + switch cfg.StorageMethod { case "boltdb": - if err := cfg.BoltDB.InitializeBuckets(boltdb.NodeServiceBuckets...); err != nil { - c.logger.Panicf("initializing node service buckets: %v", err) - } - c.nodeService = boltdb.NewNodeService(cfg.BoltDB, c.logger) - if err := cfg.BoltDB.InitializeBuckets(boltdb.DirectiveBuckets...); err != nil { c.logger.Panicf("initializing directive buckets: %v", err) } @@ -143,22 +143,6 @@ func (c *Controller) RegisterNodes(ctx context.Context, nodes ...*dax.Node) erro // and therefore need to be sent an updated Directive. workerSet := NewAddressSet() - // Create node if we don't already have it - for _, n := range nodes { - if node, _ := c.nodeService.ReadNode(tx, n.Address); node == nil { - if err := c.nodeService.CreateNode(tx, n.Address, n); err != nil { - return errors.Wrapf(err, "creating node: %s", n.Address) - } - - // Add the node to the workerSet so that it receives a directive. - // Even if there is currently no data for this worker (i.e. it - // doesn't result in a diffByAddr entry below), we still want to - // send it a "reset" directive so that in the off chance it has some - // local data, that data gets removed. - workerSet.Add(n.Address) - } - } - // diffByAddr keeps track of the diffs that have been applied to each // specific address. // TODO(tlt): I don't understand why we're keeping track of the @@ -166,7 +150,20 @@ func (c *Controller) RegisterNodes(ctx context.Context, nodes ...*dax.Node) erro // doesn't ever seem to be used. diffByAddr := make(map[dax.Address]dax.WorkerDiff) + // Create node if we don't already have it for _, n := range nodes { + // If the node already exists, skip it. + if node, _ := c.Balancer.ReadNode(tx, n.Address); node != nil { + continue + } + + // Add the node to the workerSet so that it receives a directive. + // Even if there is currently no data for this worker (i.e. it + // doesn't result in a diffByAddr entry below), we still want to + // send it a "reset" directive so that in the off chance it has some + // local data, that data gets removed. + workerSet.Add(n.Address) + adiffs, err := c.Balancer.AddWorker(tx, n) if err != nil { return errors.Wrap(err, "adding worker") @@ -188,16 +185,6 @@ func (c *Controller) RegisterNodes(ctx context.Context, nodes ...*dax.Node) erro workerSet.Add(addr) } - addrs := []dax.Address{} - for _, n := range nodes { - addrs = append(addrs, n.Address) - } - - // Tell the poller about the new nodes. - if err := c.poller.AddAddresses(tx.Context(), addrs...); err != nil { - return NewErrInternal(err.Error()) - } - // No need to send directives if the workerSet is empty. if len(workerSet) == 0 { return nil @@ -248,7 +235,7 @@ func (c *Controller) RegisterNode(ctx context.Context, n *dax.Node) error { } defer tx.Rollback() - if node, _ := c.nodeService.ReadNode(tx, n.Address); node != nil { + if node, _ := c.Balancer.ReadNode(tx, n.Address); node != nil { return nil } @@ -275,7 +262,7 @@ func (c *Controller) CheckInNode(ctx context.Context, n *dax.Node) error { // Directive; then we could check that the compute node is actually doing // what we expect it to be doing. But for now, we're just checking that we // know about the compute node at all. - if node, _ := c.nodeService.ReadNode(tx, n.Address); node != nil { + if node, _ := c.Balancer.ReadNode(tx, n.Address); node != nil { return nil } @@ -340,16 +327,6 @@ func (c *Controller) DeregisterNodes(ctx context.Context, addresses ...dax.Addre workerSet.Remove(addr) } - for _, address := range addresses { - if err := c.nodeService.DeleteNode(tx, address); err != nil { - return errors.Wrapf(err, "deleting node at address: %s", address) - } - } - - if err := c.poller.RemoveAddresses(tx.Context(), addresses...); err != nil { - return NewErrInternal(err.Error()) - } - // No need to send Directives if nothing has ultimately changed. if len(workerSet) == 0 { return tx.Commit() @@ -603,6 +580,65 @@ func (c *Controller) CreateDatabase(ctx context.Context, qdb *dax.QualifiedDatab return tx.Commit() } +func (c *Controller) DropDatabase(ctx context.Context, qdbid dax.QualifiedDatabaseID) error { + tx, err := c.boltDB.BeginTx(ctx, true) + if err != nil { + return errors.Wrap(err, "beginning tx") + } + defer tx.Rollback() + + // Get all the tables for the database and call dropTable on each one. + qtbls, err := c.Schemar.Tables(tx, qdbid) + if err != nil { + return errors.Wrapf(err, "getting tables for database: %s", qdbid) + } + + // workerSet maintains the set of workers which have a job assignment change + // and therefore need to be sent an updated Directive. + workerSet := NewAddressSet() + + for _, qtbl := range qtbls { + qtid := qtbl.QualifiedID() + addrs, err := c.dropTable(tx, qtid) + if err != nil { + return errors.Wrapf(err, "dropping table: %s", qtid) + } + workerSet.Merge(addrs) + } + + // Drop the database record from the schema. + if err := c.Schemar.DropDatabase(tx, qdbid); err != nil { + return errors.Wrap(err, "dropping database from schemar") + } + + // Convert the slice of addresses into a slice of addressMethod containing + // the appropriate method. + addressMethods := applyAddressMethod(workerSet.SortedSlice(), dax.DirectiveMethodDiff) + + // Finally, send the directives based on the dropTable calls. + if err := c.sendDirectives(tx, addressMethods...); err != nil { + return NewErrDirectiveSendFailure(err.Error()) + } + + return tx.Commit() +} + +// DatabaseByName returns the database for the given name. +func (c *Controller) DatabaseByName(ctx context.Context, orgID dax.OrganizationID, dbname dax.DatabaseName) (*dax.QualifiedDatabase, error) { + tx, err := c.boltDB.BeginTx(ctx, false) + if err != nil { + return nil, errors.Wrap(err, "beginning tx") + } + defer tx.Rollback() + + qdb, err := c.Schemar.DatabaseByName(tx, orgID, dbname) + if err != nil { + return nil, errors.Wrap(err, "getting database by name from schemar") + } + + return qdb, nil +} + // DatabaseByID returns the database for the given id. func (c *Controller) DatabaseByID(ctx context.Context, qdbid dax.QualifiedDatabaseID) (*dax.QualifiedDatabase, error) { tx, err := c.boltDB.BeginTx(ctx, false) @@ -613,7 +649,7 @@ func (c *Controller) DatabaseByID(ctx context.Context, qdbid dax.QualifiedDataba qdb, err := c.Schemar.DatabaseByID(tx, qdbid) if err != nil { - return nil, errors.Wrap(err, "getting database from schemar") + return nil, errors.Wrap(err, "getting database by id from schemar") } return qdb, nil @@ -634,6 +670,17 @@ func (c *Controller) SetDatabaseOptions(ctx context.Context, qdbid dax.Qualified return tx.Commit() } +func (c *Controller) Databases(ctx context.Context, orgID dax.OrganizationID, ids ...dax.DatabaseID) ([]*dax.QualifiedDatabase, error) { + tx, err := c.boltDB.BeginTx(ctx, false) + if err != nil { + return nil, errors.Wrap(err, "beginning tx") + } + defer tx.Rollback() + + // Get the tables from the schemar. + return c.Schemar.Databases(tx, orgID, ids...) +} + // CreateTable adds a table to the schemar, and then sends directives to all // affected nodes based on the change. func (c *Controller) CreateTable(ctx context.Context, qtbl *dax.QualifiedTable) error { @@ -731,9 +778,28 @@ func (c *Controller) DropTable(ctx context.Context, qtid dax.QualifiedTableID) e } defer tx.Rollback() + addrs, err := c.dropTable(tx, qtid) + if err != nil { + return errors.Wrapf(err, "dropping table: %s", qtid) + } + + // Convert the slice of addresses into a slice of addressMethod containing + // the appropriate method. + addressMethods := applyAddressMethod(addrs.SortedSlice(), dax.DirectiveMethodDiff) + + if err := c.sendDirectives(tx, addressMethods...); err != nil { + return NewErrDirectiveSendFailure(err.Error()) + } + + return tx.Commit() +} + +// dropTable removes a table from the schema and sends directives to all affected +// nodes based on the change. +func (c *Controller) dropTable(tx dax.Transaction, qtid dax.QualifiedTableID) (AddressSet, error) { // Get the table from the schemar. if _, err := c.Schemar.Table(tx, qtid); err != nil { - return errors.Wrapf(err, "table not in schemar: %s", qtid) + return nil, errors.Wrapf(err, "table not in schemar: %s", qtid) } // workerSet maintains the set of workers which have a job assignment change @@ -742,7 +808,7 @@ func (c *Controller) DropTable(ctx context.Context, qtid dax.QualifiedTableID) e diffs, err := c.Balancer.RemoveJobs(tx, dax.RoleTypeCompute, qtid) if err != nil { - return errors.Wrapf(err, "removing compute jobs for table: %s", qtid) + return nil, errors.Wrapf(err, "removing compute jobs for table: %s", qtid) } for _, diff := range diffs { workerSet.Add(dax.Address(diff.Address)) @@ -750,30 +816,30 @@ func (c *Controller) DropTable(ctx context.Context, qtid dax.QualifiedTableID) e diffs, err = c.Balancer.RemoveJobs(tx, dax.RoleTypeTranslate, qtid) if err != nil { - return errors.Wrapf(err, "removing translate jobs for table: %s", qtid) + return nil, errors.Wrapf(err, "removing translate jobs for table: %s", qtid) } for _, diff := range diffs { workerSet.Add(dax.Address(diff.Address)) } - // Convert the slice of addresses into a slice of addressMethod containing - // the appropriate method. - addressMethods := applyAddressMethod(workerSet.SortedSlice(), dax.DirectiveMethodDiff) - - if err := c.sendDirectives(tx, addressMethods...); err != nil { - return NewErrDirectiveSendFailure(err.Error()) - } - // Remove table from schemar. if err := c.Schemar.DropTable(tx, qtid); err != nil { - return errors.Wrapf(err, "dropping table from schemar: %s", qtid) + return nil, errors.Wrapf(err, "dropping table from schemar: %s", qtid) } - return tx.Commit() + // Delete relavent table files from snapshotter and writelogger. + if err := c.Snapshotter.DeleteTable(qtid); err != nil { + return nil, errors.Wrap(err, "deleting from snapshotter") + } + if err := c.Writelogger.DeleteTable(qtid); err != nil { + return nil, errors.Wrap(err, "deleting from writelogger") + } + + return workerSet, nil } -// Table returns a table by quaified table id. -func (c *Controller) Table(ctx context.Context, qtid dax.QualifiedTableID) (*dax.QualifiedTable, error) { +// TableByID returns a table by quaified table id. +func (c *Controller) TableByID(ctx context.Context, qtid dax.QualifiedTableID) (*dax.QualifiedTable, error) { tx, err := c.boltDB.BeginTx(ctx, false) if err != nil { return nil, errors.Wrap(err, "beginning tx") @@ -1055,10 +1121,10 @@ func (c *Controller) buildDirectives(tx dax.Transaction, addrs []addressMethod) if len(tableSet) > 0 { dTables := make([]*dax.QualifiedTable, 0) - for qual, tblIDs := range tableSet.QualifiedSortedSlice() { - qtbls, err := c.Schemar.Tables(tx, qual, tblIDs...) + for qdbid, tblIDs := range tableSet.QualifiedSortedSlice() { + qtbls, err := c.Schemar.Tables(tx, qdbid, tblIDs...) if err != nil { - return nil, errors.Wrapf(err, "getting directive tables for qual: %s", qual) + return nil, errors.Wrapf(err, "getting directive tables for qdbid: %s", qdbid) } dTables = append(dTables, qtbls...) } @@ -1077,34 +1143,6 @@ func (c *Controller) buildDirectives(tx dax.Transaction, addrs []addressMethod) return directives, nil } -func (c *Controller) SetPoller(poller dax.AddressManager) { - c.poller = poller -} - -// InitializePoller sends the list of known nodes (to be polled) to the poller. -// This is useful in the case where MDS has restarted (or has been replaced) and -// its poller is emtpy (i.e. it doesn't know about any nodes). -func (c *Controller) InitializePoller(ctx context.Context) error { - tx, err := c.boltDB.BeginTx(ctx, false) - if err != nil { - return errors.Wrap(err, "beginning tx") - } - defer tx.Rollback() - - nodes, err := c.nodeService.Nodes(tx) - if err != nil { - return errors.Wrap(err, "initializing poller") - } - - for _, node := range nodes { - if err := c.poller.AddAddresses(ctx, node.Address); err != nil { - return errors.Wrapf(err, "adding address to poller: %s", node.Address) - } - } - - return nil -} - // SnapshotTable snapshots a table. It might also snapshot everything // else... no guarantees here, only used in tests as of this writing. func (c *Controller) SnapshotTable(ctx context.Context, qtid dax.QualifiedTableID) error { @@ -1114,7 +1152,7 @@ func (c *Controller) SnapshotTable(ctx context.Context, qtid dax.QualifiedTableI // SnapshotShardData forces the compute node responsible for the given shard to // snapshot that shard, then increment its shard version for logs written to the -// WriteLogger. +// Writelogger. func (c *Controller) SnapshotShardData(ctx context.Context, qtid dax.QualifiedTableID, shardNum dax.ShardNum) error { tx, err := c.boltDB.BeginTx(ctx, false) if err != nil { @@ -1157,7 +1195,7 @@ func (c *Controller) snapshotShardData(tx dax.Transaction, qtid dax.QualifiedTab // SnapshotTableKeys forces the translate node responsible for the given // partition to snapshot the table keys for that partition, then increment its -// version for logs written to the WriteLogger. +// version for logs written to the Writelogger. func (c *Controller) SnapshotTableKeys(ctx context.Context, qtid dax.QualifiedTableID, partitionNum dax.PartitionNum) error { tx, err := c.boltDB.BeginTx(ctx, false) if err != nil { @@ -1200,7 +1238,7 @@ func (c *Controller) snapshotTableKeys(tx dax.Transaction, qtid dax.QualifiedTab // SnapshotFieldKeys forces the translate node responsible for the given field // to snapshot the keys for that field, then increment its version for logs -// written to the WriteLogger. +// written to the Writelogger. func (c *Controller) SnapshotFieldKeys(ctx context.Context, qtid dax.QualifiedTableID, field dax.FieldName) error { tx, err := c.boltDB.BeginTx(ctx, false) if err != nil { @@ -1691,7 +1729,7 @@ func (c *Controller) DebugNodes(ctx context.Context) ([]*dax.Node, error) { } defer tx.Rollback() - return c.nodeService.Nodes(tx) + return c.Balancer.Nodes(tx) } // sanitizeQTID populates Table.ID (by looking up the table, by name, in @@ -1707,7 +1745,25 @@ func (c *Controller) sanitizeQTID(tx dax.Transaction, qtid *dax.QualifiedTableID return nil } -// TableID handles a table id (i.e. by name) request. +// TableByName gets the full table by name. +func (c *Controller) TableByName(ctx context.Context, qdbid dax.QualifiedDatabaseID, name dax.TableName) (*dax.QualifiedTable, error) { + tx, err := c.boltDB.BeginTx(ctx, false) + if err != nil { + return nil, errors.Wrap(err, "beginning tx") + } + defer tx.Rollback() + + qtid, err := c.Schemar.TableID(tx, qdbid, name) + if err != nil { + return nil, errors.Wrap(err, "getting table id") + } + + // Get the table from the schemar. + return c.Schemar.Table(tx, qtid) +} + +// TableID returns the table id by table name. +// TODO(tlt): try to phase this out in favor of TableByName(). func (c *Controller) TableID(ctx context.Context, qdbid dax.QualifiedDatabaseID, name dax.TableName) (dax.QualifiedTableID, error) { tx, err := c.boltDB.BeginTx(ctx, false) if err != nil { @@ -1717,3 +1773,24 @@ func (c *Controller) TableID(ctx context.Context, qdbid dax.QualifiedDatabaseID, return c.Schemar.TableID(tx, qdbid, name) } + +// NodeService + +func (c *Controller) CreateNode(context.Context, dax.Address, *dax.Node) error { + return errors.Errorf("Controller.CreateNode() not implemented") +} +func (c *Controller) ReadNode(context.Context, dax.Address) (*dax.Node, error) { + return nil, errors.Errorf("Controller.ReadNode() not implemented") +} +func (c *Controller) DeleteNode(context.Context, dax.Address) error { + return errors.Errorf("Controller.DeleteNode() not implemented") +} +func (c *Controller) Nodes(ctx context.Context) ([]*dax.Node, error) { + tx, err := c.boltDB.BeginTx(ctx, false) + if err != nil { + return nil, errors.Wrap(err, "beginning tx") + } + defer tx.Rollback() + + return c.Balancer.Nodes(tx) +} diff --git a/dax/mds/controller/node.go b/dax/mds/controller/node.go new file mode 100644 index 000000000..db64fe438 --- /dev/null +++ b/dax/mds/controller/node.go @@ -0,0 +1,36 @@ +package controller + +import ( + "github.com/featurebasedb/featurebase/v3/dax" +) + +// NodeService represents a service for managing Nodes. +type NodeService interface { + CreateNode(dax.Transaction, dax.Address, *dax.Node) error + ReadNode(dax.Transaction, dax.Address) (*dax.Node, error) + DeleteNode(dax.Transaction, dax.Address) error + Nodes(dax.Transaction) ([]*dax.Node, error) +} + +// Ensure type implements interface. +var _ NodeService = &nopNodeService{} + +// nopNoder is a no-op implementation of the Noder interface. +type nopNodeService struct{} + +func NewNopNodeService() *nopNodeService { + return &nopNodeService{} +} + +func (n *nopNodeService) CreateNode(dax.Transaction, dax.Address, *dax.Node) error { + return nil +} +func (n *nopNodeService) ReadNode(dax.Transaction, dax.Address) (*dax.Node, error) { + return nil, nil +} +func (n *nopNodeService) DeleteNode(dax.Transaction, dax.Address) error { + return nil +} +func (n *nopNodeService) Nodes(dax.Transaction) ([]*dax.Node, error) { + return []*dax.Node{}, nil +} diff --git a/dax/mds/controller/sets.go b/dax/mds/controller/sets.go index 1a7313ec6..b5e6185ab 100644 --- a/dax/mds/controller/sets.go +++ b/dax/mds/controller/sets.go @@ -130,6 +130,12 @@ func (s AddressSet) Add(p dax.Address) { s[p] = struct{}{} } +func (s AddressSet) Merge(o AddressSet) { + for k := range o { + s[k] = struct{}{} + } +} + func (s AddressSet) Remove(p dax.Address) { delete(s, p) } diff --git a/dax/mds/http/handler.go b/dax/mds/http/handler.go index 532010bf9..a371907b2 100644 --- a/dax/mds/http/handler.go +++ b/dax/mds/http/handler.go @@ -19,7 +19,10 @@ func Handler(mds *mds.MDS) http.Handler { // mds endpoints. router.HandleFunc("/create-database", server.postCreateDatabase).Methods("POST").Name("PostCreateDatabase") + router.HandleFunc("/drop-database", server.postDropDatabase).Methods("POST").Name("PostDropDatabase") router.HandleFunc("/database-by-id", server.postDatabaseByID).Methods("POST").Name("PostDatabaseByID") + router.HandleFunc("/database-by-name", server.postDatabaseByName).Methods("POST").Name("PostDatabaseByName") + router.HandleFunc("/databases", server.postDatabases).Methods("POST").Name("PostDatabases") router.HandleFunc("/create-table", server.postCreateTable).Methods("POST").Name("PostCreateTable") router.HandleFunc("/drop-table", server.postDropTable).Methods("POST").Name("PostDropTable") @@ -85,7 +88,27 @@ func (s *server) postCreateDatabase(w http.ResponseWriter, r *http.Request) { } } -// POST /database +// POST /drop-database +func (s *server) postDropDatabase(w http.ResponseWriter, r *http.Request) { + body := r.Body + defer body.Close() + + ctx := r.Context() + + req := dax.QualifiedDatabaseID{} + if err := json.NewDecoder(body).Decode(&req); err != nil { + http.Error(w, err.Error(), http.StatusBadRequest) + return + } + + err := s.mds.DropDatabase(ctx, req) + if err != nil { + http.Error(w, err.Error(), http.StatusBadRequest) + return + } +} + +// POST /database-by-id func (s *server) postDatabaseByID(w http.ResponseWriter, r *http.Request) { body := r.Body defer body.Close() @@ -109,6 +132,68 @@ func (s *server) postDatabaseByID(w http.ResponseWriter, r *http.Request) { } } +// POST /database-by-name +func (s *server) postDatabaseByName(w http.ResponseWriter, r *http.Request) { + body := r.Body + defer body.Close() + + ctx := r.Context() + + req := &DatabaseByNameRequest{} + if err := json.NewDecoder(body).Decode(req); err != nil { + http.Error(w, err.Error(), http.StatusBadRequest) + return + } + resp, err := s.mds.DatabaseByName(ctx, req.OrganizationID, req.Name) + if err != nil { + http.Error(w, err.Error(), http.StatusBadRequest) + return + } + + if err := json.NewEncoder(w).Encode(resp); err != nil { + http.Error(w, err.Error(), http.StatusBadRequest) + return + } +} + +type DatabaseByNameRequest struct { + OrganizationID dax.OrganizationID `json:"org-id"` + Name dax.DatabaseName `json:"name"` +} + +// POST /databases +func (s *server) postDatabases(w http.ResponseWriter, r *http.Request) { + body := r.Body + defer body.Close() + + ctx := r.Context() + + req := DatabasesRequest{} + if err := json.NewDecoder(body).Decode(&req); err != nil { + http.Error(w, err.Error(), http.StatusBadRequest) + return + } + + ids := req.DatabaseIDs + + resp, err := s.mds.Databases(ctx, req.OrganizationID, ids...) + if err != nil { + http.Error(w, err.Error(), http.StatusBadRequest) + return + } + + if err := json.NewEncoder(w).Encode(resp); err != nil { + http.Error(w, err.Error(), http.StatusBadRequest) + return + } +} + +type DatabasesRequest struct { + OrganizationID dax.OrganizationID `json:"org-id"` + DatabaseIDs dax.DatabaseIDs `json:"database-ids"` + // DatabaseNames dax.DatabaseNames `json:"database-names"` +} + // POST /create-table func (s *server) postCreateTable(w http.ResponseWriter, r *http.Request) { body := r.Body @@ -146,7 +231,7 @@ func (s *server) postTable(w http.ResponseWriter, r *http.Request) { http.Error(w, err.Error(), http.StatusBadRequest) return } - resp, err := s.mds.Table(ctx, qtid) + resp, err := s.mds.TableByID(ctx, qtid) if err != nil { http.Error(w, err.Error(), http.StatusBadRequest) return @@ -171,11 +256,12 @@ func (s *server) postTableID(w http.ResponseWriter, r *http.Request) { return } - qtid, err := s.mds.TableID(ctx, req.QualifiedDatabaseID, req.Name) + qtbl, err := s.mds.TableByName(ctx, req.QualifiedDatabaseID, req.Name) if err != nil { http.Error(w, err.Error(), http.StatusBadRequest) return } + qtid := qtbl.QualifiedID() if err := json.NewEncoder(w).Encode(qtid); err != nil { http.Error(w, err.Error(), http.StatusBadRequest) diff --git a/dax/mds/mds.go b/dax/mds/mds.go index 657ee6461..ed155d42f 100644 --- a/dax/mds/mds.go +++ b/dax/mds/mds.go @@ -20,7 +20,7 @@ import ( ) type Config struct { - // Controller + // TODO(jaffee) director on config is a bit weird? Shouldn't this be set up internally? Director controller.Director `toml:"-"` // RegistrationBatchTimeout is the time that the controller will // wait after a node registers itself to see if any more nodes @@ -37,12 +37,16 @@ type Config struct { StorageMethod string `toml:"-"` DataDir string `toml:"-"` + SnapshotterDir string `toml:"snapshotter-dir"` + WriteloggerDir string `toml:"writelogger-dir"` + // Logger Logger logger.Logger `toml:"-"` } // Ensure type implements interface. var _ computer.Registrar = (*MDS)(nil) +var _ dax.Schemar = (*MDS)(nil) // MDS provides public MDS methods for an MDS service. type MDS struct { @@ -103,6 +107,8 @@ func New(cfg Config) *MDS { RegistrationBatchTimeout: cfg.RegistrationBatchTimeout, SnappingTurtleTimeout: cfg.SnappingTurtleTimeout, + SnapshotterDir: cfg.SnapshotterDir, + WriteloggerDir: cfg.WriteloggerDir, StorageMethod: cfg.StorageMethod, // just reusing this bolt for internal controller svcs @@ -116,18 +122,13 @@ func New(cfg Config) *MDS { pollerCfg := poller.Config{ AddressManager: controller, + NodeService: controller, NodePoller: poller.NewHTTPNodePoller(logr), PollInterval: cfg.PollInterval, Logger: logr, } poller := poller.New(pollerCfg) - // The controller needs to tell the poller about nodes which have been - // added/removed. - // TODO: this feels hacky. We need an elegant way to register interface - // implementations across services without an explicit Set method like this. - controller.SetPoller(poller) - return &MDS{ controller: controller, poller: poller, @@ -145,13 +146,7 @@ func New(cfg Config) *MDS { // Start starts MDS services, such as the Poller. func (m *MDS) Start() error { - // Initialize the poller (in the case where this MDS instance has restarted - // or is a replacement). Then start the poller. - if err := m.controller.InitializePoller(context.Background()); err != nil { - return errors.Wrap(err, "initializing the poller") - } m.poller.Run() - return m.controller.Run() } @@ -173,10 +168,22 @@ func (m *MDS) CreateDatabase(ctx context.Context, qdb *dax.QualifiedDatabase) er return m.controller.CreateDatabase(ctx, qdb) } +func (m *MDS) DropDatabase(ctx context.Context, qdbid dax.QualifiedDatabaseID) error { + return m.controller.DropDatabase(ctx, qdbid) +} + +func (m *MDS) DatabaseByName(ctx context.Context, orgID dax.OrganizationID, dbname dax.DatabaseName) (*dax.QualifiedDatabase, error) { + return m.controller.DatabaseByName(ctx, orgID, dbname) +} + func (m *MDS) DatabaseByID(ctx context.Context, qdbid dax.QualifiedDatabaseID) (*dax.QualifiedDatabase, error) { return m.controller.DatabaseByID(ctx, qdbid) } +func (m *MDS) Databases(ctx context.Context, orgID dax.OrganizationID, ids ...dax.DatabaseID) ([]*dax.QualifiedDatabase, error) { + return m.controller.Databases(ctx, orgID, ids...) +} + // CreateTable handles a create table request. func (m *MDS) CreateTable(ctx context.Context, qtbl *dax.QualifiedTable) error { return m.controller.CreateTable(ctx, qtbl) @@ -199,9 +206,9 @@ func (m *MDS) DropField(ctx context.Context, qtid dax.QualifiedTableID, fldName return m.controller.DropField(ctx, qtid, fldName) } -// Table handles a table request. -func (m *MDS) Table(ctx context.Context, qtid dax.QualifiedTableID) (*dax.QualifiedTable, error) { - return m.controller.Table(ctx, qtid) +// TableByID handles a table request. +func (m *MDS) TableByID(ctx context.Context, qtid dax.QualifiedTableID) (*dax.QualifiedTable, error) { + return m.controller.TableByID(ctx, qtid) } // Tables handles a tables request. @@ -209,9 +216,9 @@ func (m *MDS) Tables(ctx context.Context, qdbid dax.QualifiedDatabaseID, ids ... return m.controller.Tables(ctx, qdbid, ids...) } -// TableID handles a table id (i.e. by name) request. -func (m *MDS) TableID(ctx context.Context, qdbid dax.QualifiedDatabaseID, name dax.TableName) (dax.QualifiedTableID, error) { - return m.controller.TableID(ctx, qdbid, name) +// TableByName handles a table id (i.e. by name) request. +func (m *MDS) TableByName(ctx context.Context, qdbid dax.QualifiedDatabaseID, name dax.TableName) (*dax.QualifiedTable, error) { + return m.controller.TableByName(ctx, qdbid, name) } // IngestPartition handles an ingest partition request. diff --git a/dax/mds/poller/config.go b/dax/mds/poller/config.go index 6c0bdd151..3afdc4cd9 100644 --- a/dax/mds/poller/config.go +++ b/dax/mds/poller/config.go @@ -9,6 +9,7 @@ import ( type Config struct { AddressManager dax.AddressManager + NodeService dax.NodeService NodePoller NodePoller PollInterval time.Duration Logger logger.Logger diff --git a/dax/mds/poller/poller.go b/dax/mds/poller/poller.go index 1676ff7b8..16fcb57e3 100644 --- a/dax/mds/poller/poller.go +++ b/dax/mds/poller/poller.go @@ -14,10 +14,10 @@ import ( type Poller struct { mu sync.RWMutex - addresses map[dax.Address]struct{} - addressManager dax.AddressManager + nodeService dax.NodeService + nodePoller NodePoller pollInterval time.Duration @@ -30,8 +30,8 @@ type Poller struct { // New returns a new instance of Poller with default values. func New(cfg Config) *Poller { p := &Poller{ - addresses: make(map[dax.Address]struct{}), addressManager: dax.NewNopAddressManager(), + nodeService: dax.NewNopNodeService(), nodePoller: NewNopNodePoller(), pollInterval: time.Second, stopping: make(chan struct{}), @@ -42,6 +42,9 @@ func New(cfg Config) *Poller { if cfg.AddressManager != nil { p.addressManager = cfg.AddressManager } + if cfg.NodeService != nil { + p.nodeService = cfg.NodeService + } if cfg.NodePoller != nil { p.nodePoller = cfg.NodePoller } @@ -55,35 +58,15 @@ func New(cfg Config) *Poller { return p } -func (p *Poller) AddAddresses(ctx context.Context, addrs ...dax.Address) error { - p.mu.Lock() - defer p.mu.Unlock() - - for _, addr := range addrs { - p.addresses[addr] = struct{}{} - } - - return nil -} - -func (p *Poller) RemoveAddresses(ctx context.Context, addrs ...dax.Address) error { - p.mu.Lock() - defer p.mu.Unlock() - - for _, addr := range addrs { - delete(p.addresses, addr) - } - - return nil -} - func (p *Poller) Addresses() []dax.Address { - p.mu.RLock() - defer p.mu.RUnlock() + nodes, err := p.nodeService.Nodes(context.Background()) + if err != nil { + p.logger.Printf("POLLER: unable to get nodes from node service") + } - addrs := make([]dax.Address, 0, len(p.addresses)) - for addr := range p.addresses { - addrs = append(addrs, addr) + addrs := make([]dax.Address, 0, len(nodes)) + for _, node := range nodes { + addrs = append(addrs, node.Address) } return addrs diff --git a/dax/mds/poller/poller_test.go b/dax/mds/poller/poller_test.go index a1d1cf8fb..9db836a01 100644 --- a/dax/mds/poller/poller_test.go +++ b/dax/mds/poller/poller_test.go @@ -6,12 +6,14 @@ import ( "log" "net/http" "net/http/httptest" + "sync" "testing" "time" "github.com/featurebasedb/featurebase/v3/dax" - mds_http "github.com/featurebasedb/featurebase/v3/dax/mds/http" + mdshttp "github.com/featurebasedb/featurebase/v3/dax/mds/http" "github.com/featurebasedb/featurebase/v3/dax/mds/poller" + "github.com/featurebasedb/featurebase/v3/errors" "github.com/featurebasedb/featurebase/v3/logger" "github.com/stretchr/testify/assert" ) @@ -23,46 +25,50 @@ import ( func TestPoller(t *testing.T) { ctx := context.Background() + nodeService := newMemNodeService() + // node 1 node1 := newMockNode(t, "health", 0) defer node1.Close() addr1 := dax.Address(node1.URL()) + daxNode1 := &dax.Node{ + Address: addr1, + } - // node 1 + // node 2 node2 := newMockNode(t, "health", 3*time.Second) defer node2.Close() addr2 := dax.Address(node2.URL()) + daxNode2 := &dax.Node{ + Address: addr2, + } // manager - manager := newMockManager(t, ctx, "deregister-nodes", []dax.Address{addr1, addr2}) + manager := newMockManager(t, ctx, "deregister-nodes", nodeService) defer manager.Close() managerAddr := dax.Address(manager.URL()) t.Run("Poller", func(t *testing.T) { cfg := poller.Config{ - AddressManager: mds_http.NewAddressManager(managerAddr), + AddressManager: mdshttp.NewAddressManager(managerAddr), NodePoller: poller.NewHTTPNodePoller(logger.NopLogger), + NodeService: nodeService, } p := poller.New(cfg) - // This is a little strange, but basically we need the manager to be - // able to call poller.RemoveAddresses, and since this test poller isn't - // running as an http server (unlike everything else in this test: - // manager, nodes), we give the manager a pointer to the Poller here so - // it can call the RemoveAddresses method directly. - manager.setPoller(p) - done := make(chan struct{}) go func() { time.Sleep(5 * time.Second) close(done) }() + // Add nodes to nodeService so they are available to the poller. + nodeService.CreateNode(ctx, addr1, daxNode1) + nodeService.CreateNode(ctx, addr2, daxNode2) + p.Run() defer p.Stop() - p.AddAddresses(ctx, addr1, addr2) - // wait for a done <-done @@ -77,19 +83,13 @@ type mockManager struct { t *testing.T server *httptest.Server - poller *poller.Poller - addresses map[dax.Address]struct{} + nodeService dax.NodeService } -func newMockManager(t *testing.T, ctx context.Context, deregisterPath string, addrs []dax.Address) *mockManager { - addresses := make(map[dax.Address]struct{}) - for _, addr := range addrs { - addresses[addr] = struct{}{} - } - +func newMockManager(t *testing.T, ctx context.Context, deregisterPath string, nodeService dax.NodeService) *mockManager { mm := &mockManager{ - t: t, - addresses: addresses, + t: t, + nodeService: nodeService, } // deregister is a function used in this mock to remove the address from the @@ -97,9 +97,8 @@ func newMockManager(t *testing.T, ctx context.Context, deregisterPath string, ad // the Poller. deregister := func(addrs ...dax.Address) { for _, addr := range addrs { - delete(mm.addresses, addr) + mm.nodeService.DeleteNode(context.Background(), addr) } - mm.poller.RemoveAddresses(ctx, addrs...) } server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { @@ -111,7 +110,7 @@ func newMockManager(t *testing.T, ctx context.Context, deregisterPath string, ad body := r.Body defer body.Close() - req := mds_http.DeregisterNodesRequest{} + req := mdshttp.DeregisterNodesRequest{} if err := json.NewDecoder(body).Decode(&req); err != nil { http.Error(w, err.Error(), http.StatusBadRequest) return @@ -128,10 +127,6 @@ func newMockManager(t *testing.T, ctx context.Context, deregisterPath string, ad return mm } -func (m *mockManager) setPoller(p *poller.Poller) { - m.poller = p -} - func (m *mockManager) URL() string { if m.server != nil { return m.server.URL @@ -180,3 +175,50 @@ func (m *mockNode) Close() { m.server.Close() } } + +type memNodeService struct { + mu sync.RWMutex + addresses map[dax.Address]*dax.Node +} + +func newMemNodeService() *memNodeService { + return &memNodeService{ + addresses: make(map[dax.Address]*dax.Node), + } +} + +func (m *memNodeService) CreateNode(ctx context.Context, addr dax.Address, node *dax.Node) error { + m.mu.Lock() + defer m.mu.Unlock() + m.addresses[addr] = node + return nil +} + +func (m *memNodeService) ReadNode(ctx context.Context, addr dax.Address) (*dax.Node, error) { + m.mu.RLock() + defer m.mu.RUnlock() + node, ok := m.addresses[addr] + if !ok { + return nil, errors.Errorf("node does not exist") + } + return node, nil +} + +func (m *memNodeService) DeleteNode(ctx context.Context, addr dax.Address) error { + m.mu.Lock() + defer m.mu.Unlock() + delete(m.addresses, addr) + return nil +} + +func (m *memNodeService) Nodes(ctx context.Context) ([]*dax.Node, error) { + m.mu.RLock() + defer m.mu.RUnlock() + + nodes := make([]*dax.Node, 0, len(m.addresses)) + for _, node := range m.addresses { + nodes = append(nodes, node) + } + + return nodes, nil +} diff --git a/dax/mds/schemar/boltdb/schemar.go b/dax/mds/schemar/boltdb/schemar.go index 47c9951fd..10d3c5710 100644 --- a/dax/mds/schemar/boltdb/schemar.go +++ b/dax/mds/schemar/boltdb/schemar.go @@ -117,6 +117,38 @@ func (s *Schemar) databaseByID(tx *boltdb.Tx, orgID dax.OrganizationID, id dax.D return database, nil } +func (s *Schemar) DatabaseByName(tx dax.Transaction, orgID dax.OrganizationID, dbname dax.DatabaseName) (*dax.QualifiedDatabase, error) { + txx, ok := tx.(*boltdb.Tx) + if !ok { + return nil, dax.NewErrInvalidTransaction() + } + + return s.databaseByName(txx, orgID, dbname) +} + +func (s *Schemar) databaseByName(tx *boltdb.Tx, orgID dax.OrganizationID, name dax.DatabaseName) (*dax.QualifiedDatabase, error) { + qdbid, err := s.databaseIDByName(tx, orgID, name) + if err != nil { + return nil, errors.Wrap(err, "getting database ID") + } + + return s.databaseByID(tx, orgID, qdbid.DatabaseID) +} + +func (s *Schemar) databaseIDByName(tx *boltdb.Tx, orgID dax.OrganizationID, name dax.DatabaseName) (dax.QualifiedDatabaseID, error) { + bkt := tx.Bucket(bucketSchemar) + if bkt == nil { + return dax.QualifiedDatabaseID{}, errors.Errorf(boltdb.ErrFmtBucketNotFound, bucketSchemar) + } + + b := bkt.Get(databaseNameKey(orgID, name)) + if b == nil { + return dax.QualifiedDatabaseID{}, dax.NewErrDatabaseNameDoesNotExist(name) + } + + return keyQualifiedDatabaseID(b) +} + func (s *Schemar) putDatabase(tx *boltdb.Tx, qdb *dax.QualifiedDatabase) error { bkt := tx.Bucket(bucketSchemar) if bkt == nil { @@ -646,6 +678,19 @@ func keyQualifiedTableID(key []byte) (dax.QualifiedTableID, error) { ), nil } +// keyQualifedDatabaseID gets the QualifiedDatabaseID out of the key. +func keyQualifiedDatabaseID(key []byte) (dax.QualifiedDatabaseID, error) { + parts := strings.Split(string(key), "/") + if len(parts) != 3 { + return dax.QualifiedDatabaseID{}, errors.New(errors.ErrUncoded, "table key format expected: `databases/orgID/dbID`") + } + + return dax.NewQualifiedDatabaseID( + dax.OrganizationID(parts[1]), + dax.DatabaseID(parts[2]), + ), nil +} + func timestamp() int64 { return time.Now().UnixNano() } diff --git a/dax/mds/schemar/schemar.go b/dax/mds/schemar/schemar.go index 7f54f35ee..24e0d82fc 100644 --- a/dax/mds/schemar/schemar.go +++ b/dax/mds/schemar/schemar.go @@ -8,14 +8,15 @@ import ( type Schemar interface { CreateDatabase(dax.Transaction, *dax.QualifiedDatabase) error DropDatabase(dax.Transaction, dax.QualifiedDatabaseID) error + DatabaseByName(tx dax.Transaction, orgID dax.OrganizationID, dbname dax.DatabaseName) (*dax.QualifiedDatabase, error) DatabaseByID(dax.Transaction, dax.QualifiedDatabaseID) (*dax.QualifiedDatabase, error) SetDatabaseOptions(dax.Transaction, dax.QualifiedDatabaseID, dax.DatabaseOptions) error - // Databases returns a list of databases. If the OrganizationID is empty, - // all databases will be returned. If greater than zero database IDs are - // passed in the second argument, only databases matching those IDs will be - // returned. + // Databases returns a list of databases. If the list of DatabaseIDs is + // empty, all databases will be returned. If greater than zero DatabaseIDs + // are passed in the second argument, only databases matching those IDs will + // be returned. Databases(dax.Transaction, dax.OrganizationID, ...dax.DatabaseID) ([]*dax.QualifiedDatabase, error) CreateTable(dax.Transaction, *dax.QualifiedTable) error @@ -56,6 +57,10 @@ func (s *NopSchemar) DropDatabase(tx dax.Transaction, qdbid dax.QualifiedDatabas return nil } +func (s *NopSchemar) DatabaseByName(dax.Transaction, dax.OrganizationID, dax.DatabaseName) (*dax.QualifiedDatabase, error) { + return nil, nil +} + func (s *NopSchemar) DatabaseByID(dax.Transaction, dax.QualifiedDatabaseID) (*dax.QualifiedDatabase, error) { return nil, nil } diff --git a/dax/node.go b/dax/node.go index d1ddb3a03..30934608b 100644 --- a/dax/node.go +++ b/dax/node.go @@ -40,10 +40,33 @@ type AssignedNode struct { // NodeService represents a service for managing Nodes. type NodeService interface { - CreateNode(Transaction, Address, *Node) error - ReadNode(Transaction, Address) (*Node, error) - DeleteNode(Transaction, Address) error - Nodes(Transaction) ([]*Node, error) + CreateNode(context.Context, Address, *Node) error + ReadNode(context.Context, Address) (*Node, error) + DeleteNode(context.Context, Address) error + Nodes(context.Context) ([]*Node, error) +} + +// Ensure type implements interface. +var _ NodeService = &nopNodeService{} + +// nopNoder is a no-op implementation of the Noder interface. +type nopNodeService struct{} + +func NewNopNodeService() *nopNodeService { + return &nopNodeService{} +} + +func (n *nopNodeService) CreateNode(context.Context, Address, *Node) error { + return nil +} +func (n *nopNodeService) ReadNode(context.Context, Address) (*Node, error) { + return nil, nil +} +func (n *nopNodeService) DeleteNode(context.Context, Address) error { + return nil +} +func (n *nopNodeService) Nodes(context.Context) ([]*Node, error) { + return []*Node{}, nil } // ComputeNode represents a compute node and the table/shards for which it is @@ -82,7 +105,7 @@ type Noder interface { // Ensure type implements interface. var _ Noder = &nopNoder{} -// NopMDS is a no-op implementation of the MDS interface. +// nopNoder is a no-op implementation of the Noder interface. type nopNoder struct{} func NewNopNoder() *nopNoder { diff --git a/dax/queryer/http/handler.go b/dax/queryer/http/handler.go index 24725115f..03f437cc0 100644 --- a/dax/queryer/http/handler.go +++ b/dax/queryer/http/handler.go @@ -64,6 +64,7 @@ func (s *server) postQuery(w http.ResponseWriter, r *http.Request) { } } + w.Header().Set("Content-Type", "application/json") if err := json.NewEncoder(w).Encode(resp); err != nil { http.Error(w, err.Error(), http.StatusBadRequest) return @@ -90,6 +91,7 @@ func (s *server) postSQL(w http.ResponseWriter, r *http.Request) { return } + w.Header().Set("Content-Type", "application/json") if err := json.NewEncoder(w).Encode(resp); err != nil { http.Error(w, err.Error(), http.StatusBadRequest) return diff --git a/dax/queryer/orchestrator.go b/dax/queryer/orchestrator.go index b51f153f3..23caf109e 100644 --- a/dax/queryer/orchestrator.go +++ b/dax/queryer/orchestrator.go @@ -45,7 +45,7 @@ type Topologer interface { } type MDSTopology struct { - noder dax.Noder + controller dax.Controller } func (m *MDSTopology) ComputeNodes(ctx context.Context, index string, shards []uint64) ([]dax.ComputeNode, error) { @@ -59,7 +59,7 @@ func (m *MDSTopology) ComputeNodes(ctx context.Context, index string, shards []u // interface altogether and replace it with dax.Noder. qtid := dax.TableKey(index).QualifiedTableID() - return m.noder.ComputeNodes(ctx, qtid, daxShards...) + return m.controller.ComputeNodes(ctx, qtid, daxShards...) } // TODO(jaffee) we need version info in here ASAP. whenever schema or topo diff --git a/dax/queryer/queryer.go b/dax/queryer/queryer.go index ad2e5cd4e..33fd93dc8 100644 --- a/dax/queryer/queryer.go +++ b/dax/queryer/queryer.go @@ -35,8 +35,7 @@ type Queryer struct { fbClient *featurebase.InternalClient - noder dax.Noder - schemar dax.Schemar + controller dax.Controller logger logger.Logger } @@ -44,8 +43,7 @@ type Queryer struct { // New returns a new instance of Queryer. func New(cfg Config) *Queryer { q := &Queryer{ - noder: dax.NewNopNoder(), - schemar: dax.NewNopSchemar(), + controller: dax.NewNopController(), orchestrators: make(map[dax.QualifiedDatabaseID]*qualifiedOrchestrator), logger: logger.NopLogger, } @@ -80,12 +78,12 @@ func (q *Queryer) Orchestrator(qdbid dax.QualifiedDatabaseID) *qualifiedOrchestr return orch } - sapi := newQualifiedSchemaAPI(qdbid, q.schemar) + sapi := newQualifiedSchemaAPI(qdbid, q.controller) orch := &orchestrator{ schema: sapi, - trans: NewMDSTranslator(q.noder, q.schemar), - topology: &MDSTopology{noder: q.noder}, + trans: NewMDSTranslator(q.controller), + topology: &MDSTopology{controller: q.controller}, // TODO(jaffee) using default http.Client probably bad... need to set some timeouts. client: q.fbClient, logger: q.logger, @@ -97,21 +95,14 @@ func (q *Queryer) Orchestrator(qdbid dax.QualifiedDatabaseID) *qualifiedOrchestr return qorch } -func (q *Queryer) SetNoder(noder dax.Noder) error { - q.noder = noder - return nil -} - -func (q *Queryer) SetSchemar(schemar dax.Schemar) error { - q.schemar = schemar +func (q *Queryer) SetController(controller dax.Controller) error { + q.controller = controller return nil } func (q *Queryer) Start() error { - if q.noder == nil { - return errors.New(errors.ErrUncoded, "queryer requires noder to be configured") - } else if q.schemar == nil { - return errors.New(errors.ErrUncoded, "queryer requires schemar to be configured") + if q.controller == nil { + return errors.New(errors.ErrUncoded, "queryer requires controller to be configured") } // fbClient is an instance of internal client. It's used in one place in the @@ -174,10 +165,10 @@ func (q *Queryer) QuerySQL(ctx context.Context, qdbid dax.QualifiedDatabaseID, s } // SchemaAPI - sapi := newQualifiedSchemaAPI(qdbid, q.schemar) + sapi := newQualifiedSchemaAPI(qdbid, q.controller) // Importer - imp := idkmds.NewImporter(q.noder, q.schemar, qdbid, nil) + imp := idkmds.NewImporter(q.controller, qdbid, nil) // TODO(tlt): We need a dax-compatible implementation of the SystemAPI. sysapi := &featurebase.NopSystemAPI{} @@ -259,7 +250,7 @@ func (q *Queryer) parseAndQueryPQL(ctx context.Context, qdbid dax.QualifiedDatab // pql package. func (q *Queryer) convertIndex(ctx context.Context, qdbid dax.QualifiedDatabaseID, call *featurebase_pql.Call) { if index := call.CallIndex(); index != "" { - qtbl, err := q.schemar.TableByName(ctx, qdbid, dax.TableName(index)) + qtbl, err := q.controller.TableByName(ctx, qdbid, dax.TableName(index)) if err != nil { return } @@ -311,7 +302,7 @@ func (q *Queryer) queryPQL(ctx context.Context, qdbid dax.QualifiedDatabaseID, t // Replace any "index" arguments within the PQL with a TableKey. q.convertIndex(ctx, qdbid, qry.Calls[0]) - qtbl, err := q.schemar.TableByName(ctx, qdbid, dax.TableName(table)) + qtbl, err := q.controller.TableByName(ctx, qdbid, dax.TableName(table)) if err != nil { return nil, errors.Wrap(err, "converting index to qualified table") } diff --git a/dax/queryer/service/queryer.go b/dax/queryer/service/queryer.go index 32ad2227e..cf1954620 100644 --- a/dax/queryer/service/queryer.go +++ b/dax/queryer/service/queryer.go @@ -51,7 +51,6 @@ func (q *queryerService) HTTPHandler() http.Handler { func (q *queryerService) SetMDS(addr dax.Address) error { mdscli := mdsclient.New(addr, q.logger) - q.queryer.SetNoder(mdscli) - q.queryer.SetSchemar(mdscli) + q.queryer.SetController(mdscli) return nil } diff --git a/dax/queryer/translator.go b/dax/queryer/translator.go index 024082d56..46b529f43 100644 --- a/dax/queryer/translator.go +++ b/dax/queryer/translator.go @@ -17,14 +17,12 @@ import ( var _ Translator = (*mdsTranslator)(nil) type mdsTranslator struct { - noder dax.Noder - schemar dax.Schemar + controller dax.Controller } -func NewMDSTranslator(noder dax.Noder, schemar dax.Schemar) *mdsTranslator { +func NewMDSTranslator(controller dax.Controller) *mdsTranslator { return &mdsTranslator{ - noder: noder, - schemar: schemar, + controller: controller, } } @@ -43,7 +41,7 @@ func (m *mdsTranslator) CreateIndexKeys(ctx context.Context, table string, keys tkey := dax.TableKey(table) qtid := tkey.QualifiedTableID() - qtbl, err := m.schemar.TableByID(ctx, qtid) + qtbl, err := m.controller.TableByID(ctx, qtid) if err != nil { return nil, errors.Wrap(err, "getting table") } @@ -55,7 +53,7 @@ func (m *mdsTranslator) CreateIndexKeys(ctx context.Context, table string, keys out := make(map[string]uint64) for pNum := range pMap { - address, err := m.noder.IngestPartition(ctx, qtid, pNum) + address, err := m.controller.IngestPartition(ctx, qtid, pNum) if err != nil { return nil, errors.Wrapf(err, "calling ingest-partition on table: %s, partition: %d", table, pNum) } @@ -82,7 +80,7 @@ func (m *mdsTranslator) CreateIndexKeys(ctx context.Context, table string, keys func (m *mdsTranslator) CreateFieldKeys(ctx context.Context, table string, field string, keys []string) (map[string]uint64, error) { qtid := dax.TableKey(table).QualifiedTableID() - address, err := m.noder.IngestPartition(ctx, qtid, dax.PartitionNum(0)) + address, err := m.controller.IngestPartition(ctx, qtid, dax.PartitionNum(0)) if err != nil { return nil, errors.Wrapf(err, "calling ingest-partition on table: %s, partition: %d", table, dax.PartitionNum(0)) } @@ -102,7 +100,7 @@ func (m *mdsTranslator) FindIndexKeys(ctx context.Context, table string, keys [] tkey := dax.TableKey(table) qtid := tkey.QualifiedTableID() - qtbl, err := m.schemar.TableByID(ctx, qtid) + qtbl, err := m.controller.TableByID(ctx, qtid) if err != nil { return nil, errors.Wrap(err, "getting table") } @@ -117,7 +115,7 @@ func (m *mdsTranslator) FindIndexKeys(ctx context.Context, table string, keys [] pNums = append(pNums, k) } - translateNodes, err := m.noder.TranslateNodes(ctx, qtid, pNums...) + translateNodes, err := m.controller.TranslateNodes(ctx, qtid, pNums...) if err != nil { return nil, errors.Wrapf(err, "getting translate nodes for partitions on table: %s", table) } @@ -153,7 +151,7 @@ func (m *mdsTranslator) FindIndexKeys(ctx context.Context, table string, keys [] func (m *mdsTranslator) FindFieldKeys(ctx context.Context, table, field string, keys []string) (map[string]uint64, error) { qtid := dax.TableKey(table).QualifiedTableID() - address, err := m.noder.IngestPartition(ctx, qtid, dax.PartitionNum(0)) + address, err := m.controller.IngestPartition(ctx, qtid, dax.PartitionNum(0)) if err != nil { return nil, errors.Wrapf(err, "calling ingest-partition on table: %s, partition: %d", table, dax.PartitionNum(0)) } @@ -178,7 +176,7 @@ func (m *mdsTranslator) TranslateIndexIDs(ctx context.Context, index string, ids qtid := dax.TableKey(index).QualifiedTableID() - nodes, err := m.noder.TranslateNodes(ctx, qtid, daxPartitions...) + nodes, err := m.controller.TranslateNodes(ctx, qtid, daxPartitions...) if err != nil { return nil, errors.Wrapf(err, "calling translate-nodes on table: %s, partitions: %v", index, daxPartitions) } @@ -252,7 +250,7 @@ func (m *mdsTranslator) TranslateFieldIDs(ctx context.Context, tableKeyer dax.Ta } func (m *mdsTranslator) TranslateFieldListIDs(ctx context.Context, index, field string, ids []uint64) ([]string, error) { qtid := dax.TableKey(index).QualifiedTableID() - address, err := m.noder.IngestPartition(ctx, qtid, dax.PartitionNum(0)) + address, err := m.controller.IngestPartition(ctx, qtid, dax.PartitionNum(0)) if err != nil { return nil, errors.Wrapf(err, "calling ingest-partition on table: %s, partition: %d", index, dax.PartitionNum(0)) } diff --git a/dax/schema.go b/dax/schema.go index 05d31c43e..205d325f9 100644 --- a/dax/schema.go +++ b/dax/schema.go @@ -14,16 +14,16 @@ type Schemar interface { ////////////////////////////////////////////////////////////////////////// CreateDatabase(context.Context, *QualifiedDatabase) error - // DropDatabase(context.Context, QualifiedDatabaseID) error + DropDatabase(context.Context, QualifiedDatabaseID) error - // DatabaseByName(ctx context.Context, orgID OrganizationID, dbname DatabaseName) (*QualifiedDatabase, error) + DatabaseByName(ctx context.Context, orgID OrganizationID, dbname DatabaseName) (*QualifiedDatabase, error) DatabaseByID(ctx context.Context, qdbid QualifiedDatabaseID) (*QualifiedDatabase, error) - // // Databases returns a list of databases. If the OrganizationID is empty, - // // all databases will be returned. If greater than zero database IDs are - // // passed in the second argument, only databases matching those IDs will be - // // returned. - // Databases(context.Context, OrganizationID, ...DatabaseID) ([]*QualifiedDatabase, error) + // Databases returns a list of databases. If the list of DatabaseIDs is + // empty, all databases will be returned. If greater than zero DatabaseIDs + // are passed in the second argument, only databases matching those IDs will + // be returned. + Databases(context.Context, OrganizationID, ...DatabaseID) ([]*QualifiedDatabase, error) // SetDatabaseOptions(context.Context, QualifiedDatabaseID, DatabaseOptions) error @@ -67,9 +67,18 @@ func NewNopSchemar() *NopSchemar { func (s *NopSchemar) CreateDatabase(context.Context, *QualifiedDatabase) error { return nil } +func (s *NopSchemar) DropDatabase(context.Context, QualifiedDatabaseID) error { + return nil +} +func (s *NopSchemar) DatabaseByName(ctx context.Context, orgID OrganizationID, dbname DatabaseName) (*QualifiedDatabase, error) { + return nil, nil +} func (s *NopSchemar) DatabaseByID(ctx context.Context, qdbid QualifiedDatabaseID) (*QualifiedDatabase, error) { return nil, nil } +func (s *NopSchemar) Databases(context.Context, OrganizationID, ...DatabaseID) ([]*QualifiedDatabase, error) { + return nil, nil +} func (s *NopSchemar) TableByName(context.Context, QualifiedDatabaseID, TableName) (*QualifiedTable, error) { return nil, nil } diff --git a/dax/server/config.go b/dax/server/config.go index 5d9fe80b5..e039e237f 100644 --- a/dax/server/config.go +++ b/dax/server/config.go @@ -12,8 +12,6 @@ import ( "github.com/featurebasedb/featurebase/v3/dax/mds" "github.com/featurebasedb/featurebase/v3/dax/queryer" - "github.com/featurebasedb/featurebase/v3/dax/snapshotter" - "github.com/featurebasedb/featurebase/v3/dax/writelogger" "github.com/featurebasedb/featurebase/v3/errors" fbserver "github.com/featurebasedb/featurebase/v3/server" ) @@ -45,11 +43,9 @@ type Config struct { // LogPath configures where Pilosa will write logs. LogPath string `toml:"log-path"` - MDS MDSOptions `toml:"mds"` - WriteLogger WriteLoggerOptions `toml:"writelogger"` - Snapshotter SnapshotterOptions `toml:"snapshotter"` - Queryer QueryerOptions `toml:"queryer"` - Computer ComputerOptions `toml:"computer"` + MDS MDSOptions `toml:"mds"` + Queryer QueryerOptions `toml:"queryer"` + Computer ComputerOptions `toml:"computer"` } type MDSOptions struct { @@ -57,16 +53,6 @@ type MDSOptions struct { Config mds.Config `toml:"config"` } -type WriteLoggerOptions struct { - Run bool `toml:"run"` - Config writelogger.Config `toml:"config"` -} - -type SnapshotterOptions struct { - Run bool `toml:"run"` - Config snapshotter.Config `toml:"config"` -} - type QueryerOptions struct { Run bool `toml:"run"` Config queryer.Config `toml:"config"` diff --git a/dax/server/server.go b/dax/server/server.go index 34fe15709..e4d28beb4 100644 --- a/dax/server/server.go +++ b/dax/server/server.go @@ -279,19 +279,14 @@ func (m *Command) setupServer() error { func (m *Command) setupServices() error { // Set up MDS. if m.Config.MDS.Run { - mdsCfg := mds.Config{ - RegistrationBatchTimeout: m.Config.MDS.Config.RegistrationBatchTimeout, - StorageMethod: m.Config.MDS.Config.StorageMethod, - DataDir: m.Config.MDS.Config.DataDir, - SnappingTurtleTimeout: m.Config.MDS.Config.SnappingTurtleTimeout, - Logger: m.logger, - Director: controllerhttp.NewDirector( - controllerhttp.DirectorConfig{ - DirectivePath: "directive", - SnapshotRequestPath: "snapshot", - Logger: m.logger, - }), - } + mdsCfg := m.Config.MDS.Config + mdsCfg.Logger = m.logger + mdsCfg.Director = controllerhttp.NewDirector( + controllerhttp.DirectorConfig{ + DirectivePath: "directive", + SnapshotRequestPath: "snapshot", + Logger: m.logger, + }) m.svcmgr.MDS = mdssvc.New(m.advertiseURI, mds.New(mdsCfg)) if err := m.svcmgr.MDSStart(); err != nil { @@ -344,11 +339,7 @@ func (m *Command) setupServices() error { for i := 0; i < n; i++ { m.logger.Printf("Set up computer (%d)", i) cfg := computersvc.CommandConfig{ - WriteLoggerRun: m.Config.WriteLogger.Run, - WriteLoggerConfig: m.Config.WriteLogger.Config, - SnapshotterRun: m.Config.Snapshotter.Run, - SnapshotterConfig: m.Config.Snapshotter.Config, - ComputerConfig: m.Config.Computer.Config, + ComputerConfig: m.Config.Computer.Config, Listener: m.ln, RootDataDir: rootDataDir, diff --git a/dax/server/test/managed.go b/dax/server/test/managed.go index c3b5ada02..7f6ef3131 100644 --- a/dax/server/test/managed.go +++ b/dax/server/test/managed.go @@ -96,11 +96,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 { cfg := computersvc.CommandConfig{ - WriteLoggerRun: mc.Config.WriteLogger.Run, - WriteLoggerConfig: mc.Config.WriteLogger.Config, - SnapshotterRun: mc.Config.Snapshotter.Run, - SnapshotterConfig: mc.Config.Snapshotter.Config, - ComputerConfig: mc.Config.Computer.Config, + ComputerConfig: mc.Config.Computer.Config, RootDataDir: mc.Config.Computer.Config.DataDir, @@ -192,8 +188,10 @@ func NewManagedCommand(tb fbtest.DirCleaner, opts ...server.CommandOption) *Mana mc.Config.Bind = "http://localhost:0" mc.Config.MDS.Config.DataDir = path + "/mds" mc.Config.Computer.Config.DataDir = path - mc.Config.WriteLogger.Config.DataDir = path + "/wl" - mc.Config.Snapshotter.Config.DataDir = path + "/sn" + mc.Config.Computer.Config.WriteloggerDir = path + "/wl" + mc.Config.MDS.Config.WriteloggerDir = path + "/wl" + mc.Config.Computer.Config.SnapshotterDir = path + "/sn" + mc.Config.MDS.Config.SnapshotterDir = path + "/sn" return mc } @@ -207,8 +205,6 @@ func DefaultConfig() *server.Config { cfg.Queryer.Run = true cfg.Computer.Run = true cfg.Computer.N = 1 - cfg.WriteLogger.Run = true - cfg.Snapshotter.Run = true return cfg } diff --git a/dax/snapshotter/config.go b/dax/snapshotter/config.go deleted file mode 100644 index 7ace20546..000000000 --- a/dax/snapshotter/config.go +++ /dev/null @@ -1,8 +0,0 @@ -package snapshotter - -import "github.com/featurebasedb/featurebase/v3/logger" - -type Config struct { - DataDir string `toml:"data-dir"` - Logger logger.Logger `toml:"-"` -} diff --git a/dax/snapshotter/snapshotter.go b/dax/snapshotter/snapshotter.go index b311b0e12..7f58f0a81 100644 --- a/dax/snapshotter/snapshotter.go +++ b/dax/snapshotter/snapshotter.go @@ -12,6 +12,7 @@ import ( "sync" "syscall" + "github.com/featurebasedb/featurebase/v3/dax" "github.com/featurebasedb/featurebase/v3/dax/computer" "github.com/featurebasedb/featurebase/v3/errors" "github.com/featurebasedb/featurebase/v3/logger" @@ -28,10 +29,10 @@ type Snapshotter struct { logger logger.Logger } -func New(cfg Config) *Snapshotter { +func New(dir string, log logger.Logger) *Snapshotter { return &Snapshotter{ - dataDir: cfg.DataDir, - logger: logger.NopLogger, + dataDir: dir, + logger: log, } } @@ -135,6 +136,15 @@ func (s *Snapshotter) snapshotFileByKey(key string) (*os.File, error) { return f, nil } +func (s *Snapshotter) DeleteTable(qtid dax.QualifiedTableID) error { + dir := path.Join(s.dataDir, string(qtid.Key())) + err := os.RemoveAll(dir) + if err != nil { + return errors.Wrapf(err, "dropping %s from snapshotter", dir) + } + return nil +} + // fullKey returns the full file key including the bucket and version. func fullKey(bucket string, key string, version int) string { return path.Join(bucket, key, fmt.Sprintf("%d", version)) diff --git a/dax/snapshotter/snapshotter_test.go b/dax/snapshotter/snapshotter_test.go index 7aa0f5193..d36ca50f4 100644 --- a/dax/snapshotter/snapshotter_test.go +++ b/dax/snapshotter/snapshotter_test.go @@ -10,7 +10,7 @@ import ( ) func TestSnapshotter(t *testing.T) { - tmpDir, err := os.MkdirTemp("", "testWriteLogger-*") + tmpDir, err := os.MkdirTemp("", "testWritelogger-*") assert.NoError(t, err) // Remove the temp directory. diff --git a/dax/storage/storage.go b/dax/storage/storage.go index 96dc17876..428a3cc82 100644 --- a/dax/storage/storage.go +++ b/dax/storage/storage.go @@ -18,7 +18,7 @@ import ( // all of which use the same underlying snapshotter and writelogger. type ResourceManager struct { Snapshotter computer.SnapshotService - WriteLogger computer.WriteLogService + Writelogger computer.WritelogService Logger logger.Logger mu sync.Mutex @@ -27,10 +27,10 @@ type ResourceManager struct { fieldKeyResources map[fieldKeyK]*Resource } -func NewResourceManager(s computer.SnapshotService, w computer.WriteLogService, l logger.Logger) *ResourceManager { +func NewResourceManager(s computer.SnapshotService, w computer.WritelogService, l logger.Logger) *ResourceManager { return &ResourceManager{ Snapshotter: s, - WriteLogger: w, + Writelogger: w, Logger: l, shardResources: make(map[shardK]*Resource), @@ -66,7 +66,7 @@ func (mm *ResourceManager) GetShardResource(qtid dax.QualifiedTableID, partition } mm.shardResources[key] = (&Resource{ snapshotter: mm.Snapshotter, - writeLogger: mm.WriteLogger, + writeLogger: mm.Writelogger, bucket: partitionBucket(qtid.Key(), partition), key: shardKey(shard), log: mm.Logger, @@ -97,7 +97,7 @@ func (mm *ResourceManager) GetTableKeyResource(qtid dax.QualifiedTableID, partit } mm.tableKeyResources[key] = (&Resource{ snapshotter: mm.Snapshotter, - writeLogger: mm.WriteLogger, + writeLogger: mm.Writelogger, bucket: partitionBucket(qtid.Key(), partition), key: keysFileName, log: mm.Logger, @@ -127,7 +127,7 @@ func (mm *ResourceManager) GetFieldKeyResource(qtid dax.QualifiedTableID, field } mm.fieldKeyResources[key] = (&Resource{ snapshotter: mm.Snapshotter, - writeLogger: mm.WriteLogger, + writeLogger: mm.Writelogger, bucket: fieldBucket(qtid.Key(), field), key: keysFileName, log: mm.Logger, @@ -149,7 +149,11 @@ func (mm *ResourceManager) RemoveFieldKeyResource(qtid dax.QualifiedTableID, fie } // RemoveAll unlocks and deletes all resources held within this -// ResourceManager. +// ResourceManager. To be clear, the underlying resources are *not* +// deleted, they are simply no longer under this resource manager's +// purview. If there's a problem removing any resource, we'll save the +// error, but keep going... this is only called on shutdown, so we do +// our best to unlock everything. func (mm *ResourceManager) RemoveAll() error { mm.mu.Lock() defer mm.mu.Unlock() @@ -182,6 +186,40 @@ func (mm *ResourceManager) RemoveAll() error { return nil } +// RemoveTable, unlocks and removes all resources related to the given +// table from this ResourceManager. The underlying files are not +// deleted. (If the table is being dropped, deleting the files is +// under the purview of metadata services). In this case (in contrast +// to RemoveAll), we'll stop on any error and return it. Not sure it +// actually matters in either of these cases. +func (mm *ResourceManager) RemoveTable(qtid dax.QualifiedTableID) error { + mm.mu.Lock() + defer mm.mu.Unlock() + + for k, resource := range mm.shardResources { + if k.qtid == qtid { + if err := resource.Unlock(); err != nil { + return errors.Wrap(err, "unlocking shard") + } + } + } + for k, resource := range mm.tableKeyResources { + if k.qtid == qtid { + if err := resource.Unlock(); err != nil { + return errors.Wrap(err, "unlocking shard") + } + } + } + for k, resource := range mm.fieldKeyResources { + if k.qtid == qtid { + if err := resource.Unlock(); err != nil { + return errors.Wrap(err, "deleting field key resource %s") + } + } + } + return nil +} + // Resource wraps the snapshotter and writelogger to maintain messy // state between calls. Resource is *not* threadsafe, care should be // taken that concurrent calls are not made to Resource methods. The @@ -189,7 +227,7 @@ func (mm *ResourceManager) RemoveAll() error { // concurrently. type Resource struct { snapshotter computer.SnapshotService - writeLogger computer.WriteLogService + writeLogger computer.WritelogService bucket string key string diff --git a/dax/storage/storage_test.go b/dax/storage/storage_test.go index 081133580..4cedbe5f3 100644 --- a/dax/storage/storage_test.go +++ b/dax/storage/storage_test.go @@ -24,14 +24,12 @@ func TestResourceManager(t *testing.T) { os.RemoveAll(wdd) }() - sn := snapshotter.New(snapshotter.Config{ - DataDir: sdd, - }) - wl := writelogger.New(writelogger.Config{ - DataDir: wdd, - }) + log := logger.NewStandardLogger(os.Stderr) - mm := NewResourceManager(sn, wl, logger.NewStandardLogger(os.Stderr)) + sn := snapshotter.New(sdd, log) + wl := writelogger.New(wdd, log) + + mm := NewResourceManager(sn, wl, log) qtid := dax.QualifiedTableID{ QualifiedDatabaseID: dax.NewQualifiedDatabaseID( diff --git a/dax/table.go b/dax/table.go index 577174872..13c505273 100644 --- a/dax/table.go +++ b/dax/table.go @@ -113,6 +113,13 @@ type OrganizationID string // value could be any string. type DatabaseID string +// DatabaseIDs is a sortable slice of DatabaseID. +type DatabaseIDs []DatabaseID + +func (s DatabaseIDs) Len() int { return len(s) } +func (s DatabaseIDs) Less(i, j int) bool { return s[i] < s[j] } +func (s DatabaseIDs) Swap(i, j int) { s[i], s[j] = s[j], s[i] } + // DatabaseKey is a globally unique identifier for a database; it is effectively the // compound key: (org, database). This is (hopefully) the value that will // be used when interfacing with services which are unaware of qualifiers. diff --git a/dax/test/dax/dax_test.go b/dax/test/dax/dax_test.go index 823f8c1ac..cf11765d0 100644 --- a/dax/test/dax/dax_test.go +++ b/dax/test/dax/dax_test.go @@ -2,9 +2,10 @@ package dax_test import ( "context" - "errors" "fmt" + "io" "log" + "os" "sort" "testing" "time" @@ -17,6 +18,7 @@ import ( "github.com/featurebasedb/featurebase/v3/dax/server/test" "github.com/featurebasedb/featurebase/v3/logger" "github.com/featurebasedb/featurebase/v3/sql3/test/defs" + goerrors "github.com/pkg/errors" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" ) @@ -26,12 +28,15 @@ func TestDAXIntegration(t *testing.T) { t.Skip("skipping integration test") } - qdbid := dax.NewQualifiedDatabaseID("acme", "db1") + orgID := dax.OrganizationID("acme") + dbID := dax.DatabaseID("db1") + qdbid := dax.NewQualifiedDatabaseID(orgID, dbID) + dbname := dax.DatabaseName("dbname1") qdb := &dax.QualifiedDatabase{ OrganizationID: qdbid.OrganizationID, Database: dax.Database{ ID: qdbid.DatabaseID, - Name: "dbname1", + Name: dbname, Options: dax.DatabaseOptions{ WorkersMin: 1, WorkersMax: 1, @@ -507,6 +512,126 @@ func TestDAXIntegration(t *testing.T) { ) }) }) + + t.Run("Delete_Database", func(t *testing.T) { + mc := test.MustRunManagedCommand(t) + defer mc.Close() + svcmgr := mc.Manage() + + ctx := context.Background() + + // Set up MDS client. + mdsClient := mdsclient.New(svcmgr.MDS.Address(), svcmgr.Logger) + + // Create database. + qdb.Options.WorkersMin = 1 + qdb.Options.WorkersMax = 1 + assert.NoError(t, mdsClient.CreateDatabase(context.Background(), qdb)) + + // Create two tables with data. Query the data to ensure it exists. + runTableTests(t, + svcmgr.Queryer.Address(), + basicTableTestConfig(qdbid, defs.Keyed, defs.Unkeyed)..., + ) + + // Make sure the database and tables exist. + db, err := mdsClient.DatabaseByName(ctx, orgID, dbname) + assert.NoError(t, err) + assert.NotNil(t, db) + + tbl1, err := mdsClient.TableByName(ctx, qdbid, dax.TableName(defs.Keyed.Name(0))) + assert.NoError(t, err) + assert.NotNil(t, tbl1) + + tbl2, err := mdsClient.TableByName(ctx, qdbid, dax.TableName(defs.Unkeyed.Name(0))) + assert.NoError(t, err) + assert.NotNil(t, tbl2) + + // Drop the database + assert.NoError(t, mdsClient.DropDatabase(ctx, qdbid)) + + // Make sure the database and tables no longer exist. + db, err = mdsClient.DatabaseByName(ctx, orgID, dbname) + if assert.Error(t, err) { + assert.Contains(t, err.Error(), "database name 'dbname1' does not exist") + // TODO(tlt): replace the previous line with the following once we + // have threaded error codes through the http calls. + // assert.True(t, errors.Is(err, dax.ErrDatabaseNameDoesNotExist)) + } + assert.Nil(t, db) + + tbl1, err = mdsClient.TableByName(ctx, qdbid, dax.TableName(defs.Keyed.Name(0))) + if assert.Error(t, err) { + assert.Contains(t, err.Error(), "table name 'keyed' does not exist") + // TODO(tlt): replace the previous line with the following once we + // have threaded error codes through the http calls. + // assert.True(t, errors.Is(err, dax.ErrTableNameDoesNotExist)) + } + assert.Nil(t, tbl1) + + tbl2, err = mdsClient.TableByName(ctx, qdbid, dax.TableName(defs.Unkeyed.Name(0))) + if assert.Error(t, err) { + assert.Contains(t, err.Error(), "table name 'unkeyed' does not exist") + // TODO(tlt): replace the previous line with the following once we + // have threaded error codes through the http calls. + // assert.True(t, errors.Is(err, dax.ErrTableNameDoesNotExist)) + } + assert.Nil(t, tbl2) + }) + + t.Run("Delete_Table", func(t *testing.T) { + mc := test.MustRunManagedCommand(t) + defer mc.Close() + svcmgr := mc.Manage() + + // Set up MDS client. + mdsClient := mdsclient.New(svcmgr.MDS.Address(), svcmgr.Logger) + + // Create database. + qdb.Options.WorkersMin = 1 + qdb.Options.WorkersMax = 1 + assert.NoError(t, mdsClient.CreateDatabase(context.Background(), qdb)) + + testconfigs := basicTableTestConfig(qdbid, defs.Keyed) + for i := range testconfigs { + testconfigs[i].skipQuery = true + } + runTableTests(t, + svcmgr.Queryer.Address(), + testconfigs..., + ) + + rootDir := mc.Config.Computer.Config.DataDir + + // Ensure the index and writelogger directories are empty. + assert.False(t, dirIsEmpty(t, rootDir+"/computer0")) + assert.False(t, dirIsEmpty(t, rootDir+"/computer0/indexes")) + assert.False(t, dirIsEmpty(t, rootDir+"/mds")) + assert.False(t, dirIsEmpty(t, rootDir+"/wl")) + + resp := runSQL(t, svcmgr.Queryer.Address(), testconfigs[0].qdbid, "drop table keyed") + assert.Empty(t, resp.Error) + + // Ensure the index and writelogger directories are empty. + assert.False(t, dirIsEmpty(t, rootDir+"/computer0")) + assert.True(t, dirIsEmpty(t, rootDir+"/computer0/indexes")) + assert.False(t, dirIsEmpty(t, rootDir+"/mds")) + assert.True(t, dirIsEmpty(t, rootDir+"/wl")) + }) +} + +func dirIsEmpty(t *testing.T, name string) bool { + f, err := os.Open(name) + assert.NoError(t, err) + defer f.Close() + + _, err = f.Readdirnames(1) // Or f.Readdir(1) + if err == io.EOF { + return true + } + assert.NoError(t, err) + + return false } /////////////////////////////////////////////////// @@ -589,7 +714,7 @@ func runTableTests(t *testing.T, queryerAddr dax.Address, cfgs ...tableTestConfi rows := resp.Data var err error if resp.Error != "" { - err = errors.New(resp.Error) + err = goerrors.New(resp.Error) } // Check expected error instead of results. @@ -666,7 +791,7 @@ func runTableTests(t *testing.T, queryerAddr dax.Address, cfgs ...tableTestConfi rows := resp.Data var err error if resp.Error != "" { - err = errors.New(resp.Error) + err = goerrors.New(resp.Error) } // Check expected error instead of results. diff --git a/dax/writelogger/config.go b/dax/writelogger/config.go deleted file mode 100644 index 207b78f86..000000000 --- a/dax/writelogger/config.go +++ /dev/null @@ -1,8 +0,0 @@ -package writelogger - -import "github.com/featurebasedb/featurebase/v3/logger" - -type Config struct { - DataDir string `toml:"data-dir"` - Logger logger.Logger `toml:"-"` -} diff --git a/dax/writelogger/writelogger.go b/dax/writelogger/writelogger.go index 2df249c48..b974f8a17 100644 --- a/dax/writelogger/writelogger.go +++ b/dax/writelogger/writelogger.go @@ -8,15 +8,17 @@ import ( "os" "path" "strconv" + "strings" "sync" "syscall" + "github.com/featurebasedb/featurebase/v3/dax" "github.com/featurebasedb/featurebase/v3/dax/computer" "github.com/featurebasedb/featurebase/v3/errors" "github.com/featurebasedb/featurebase/v3/logger" ) -type WriteLogger struct { +type Writelogger struct { dataDir string mu sync.RWMutex @@ -26,22 +28,22 @@ type WriteLogger struct { logger logger.Logger } -func New(cfg Config) *WriteLogger { - return &WriteLogger{ - dataDir: cfg.DataDir, +func New(dir string, log logger.Logger) *Writelogger { + return &Writelogger{ + dataDir: dir, logFiles: make(map[string]*os.File), lockFiles: make(map[string]*os.File), - logger: logger.NopLogger, + logger: log, } } // SetLogger sets the logger used for logging messages. Note, this is not the -// same "logger" that the WriteLogger represents, which logs data writes. -func (w *WriteLogger) SetLogger(l logger.Logger) { +// same "logger" that the Writelogger represents, which logs data writes. +func (w *Writelogger) SetLogger(l logger.Logger) { w.logger = l } -func (w *WriteLogger) AppendMessage(bucket string, key string, version int, message []byte) error { +func (w *Writelogger) AppendMessage(bucket string, key string, version int, message []byte) error { fKey := fullKey(bucket, key, version) logFile, err := w.logFileByKey(fKey) if err != nil { @@ -56,7 +58,7 @@ func (w *WriteLogger) AppendMessage(bucket string, key string, version int, mess return errors.Wrapf(err, "syncing log file %s", logFile.Name()) } -func (w *WriteLogger) List(bucket, key string) ([]computer.WriteLogInfo, error) { +func (w *Writelogger) List(bucket, key string) ([]computer.WriteLogInfo, error) { dirpath := path.Join(w.dataDir, bucket, key) entries, err := os.ReadDir(dirpath) @@ -80,11 +82,11 @@ func (w *WriteLogger) List(bucket, key string) ([]computer.WriteLogInfo, error) return wLogs, nil } -func (w *WriteLogger) LogReader(bucket, key string, version int) (io.ReadCloser, error) { +func (w *Writelogger) LogReader(bucket, key string, version int) (io.ReadCloser, error) { return w.LogReaderFrom(bucket, key, version, 0) } -func (w *WriteLogger) LogReaderFrom(bucket string, key string, version int, offset int) (io.ReadCloser, error) { +func (w *Writelogger) LogReaderFrom(bucket string, key string, version int, offset int) (io.ReadCloser, error) { _, filePath := w.paths(fullKey(bucket, key, version)) f, err := os.Open(filePath) @@ -97,12 +99,12 @@ func (w *WriteLogger) LogReaderFrom(bucket string, key string, version int, offs if offset > 0 { f.Seek(int64(offset), io.SeekStart) } - w.logger.Debugf("WriteLogger LogReader file: %s", f.Name()) + w.logger.Debugf("Writelogger LogReader file: %s", f.Name()) return f, nil } -func (w *WriteLogger) DeleteLog(bucket string, key string, version int) error { +func (w *Writelogger) DeleteLog(bucket string, key string, version int) error { w.mu.Lock() defer w.mu.Unlock() @@ -118,16 +120,18 @@ func (w *WriteLogger) DeleteLog(bucket string, key string, version int) error { return errors.Wrap(err, "closing log file") } + delete(w.logFiles, fullKey) + // Remove the log file. return os.Remove(f.Name()) } -func (w *WriteLogger) lockFile(bucket, key string) (string, string) { +func (w *Writelogger) lockFile(bucket, key string) (string, string) { lockFile := path.Join(w.dataDir, bucket, fmt.Sprintf("_lock_%s", key)) return path.Dir(lockFile), lockFile } -func (w *WriteLogger) Lock(bucket, key string) error { +func (w *Writelogger) Lock(bucket, key string) error { lockDir, lockFile := w.lockFile(bucket, key) if err := os.MkdirAll(lockDir, 0777); err != nil { @@ -153,20 +157,30 @@ func (w *WriteLogger) Lock(bucket, key string) error { } -func (w *WriteLogger) Unlock(bucket, key string) error { +func (w *Writelogger) Unlock(bucket, key string) error { w.mu.Lock() defer w.mu.Unlock() + + // remove all local state associated with this bucket/key + keyPrefix := path.Join(bucket, key) + for logKey, logFile := range w.logFiles { + if strings.HasPrefix(logKey, keyPrefix) { + _ = logFile.Close() + delete(w.logFiles, logKey) + } + } // TODO(jaffee) since the file isn't guaranteed to be removed if // the process is killed, we should actually use flock instead of // EXCL file creation. Problem with that is it makes testing // tricky because file handles from the same process are able to // acquire the flock simultaneously. Headache. _, lockFile := w.lockFile(bucket, key) - f, ok := w.lockFiles[lockFile] - if !ok { - return errors.New(errors.ErrUncoded, "couldn't find file to unlock") + + if f, ok := w.lockFiles[lockFile]; ok { + _ = f.Close() + } else { + w.logger.Warnf("unlocking %s not find cached file to unlock", lockFile) } - f.Close() err := os.Remove(lockFile) delete(w.lockFiles, lockFile) @@ -182,10 +196,19 @@ func (w *WriteLogger) Unlock(bucket, key string) error { return errors.Wrap(err, "removing lock file") } +func (w *Writelogger) DeleteTable(qtid dax.QualifiedTableID) error { + dir := path.Join(w.dataDir, string(qtid.Key())) + err := os.RemoveAll(dir) + if err != nil { + return errors.Wrapf(err, "dropping %s from writelogger", dir) + } + return nil +} + // paths takes a key and returns the full file path (including the root data // directory) as well as the full directory path (i.e. the file path without the // file portion). -func (w *WriteLogger) paths(key string) (string, string) { +func (w *Writelogger) paths(key string) (string, string) { filePath := path.Join(w.dataDir, key) dirPath, _ := path.Split(filePath) return dirPath, filePath @@ -194,7 +217,7 @@ func (w *WriteLogger) paths(key string) (string, string) { // logFileByKey returns a pointer to the file specified by key. If the file does // not exist, the file is created (along with any directories in which the file // is nested). -func (w *WriteLogger) logFileByKey(key string) (*os.File, error) { +func (w *Writelogger) logFileByKey(key string) (*os.File, error) { w.mu.Lock() defer w.mu.Unlock() diff --git a/dax/writelogger/writelogger_test.go b/dax/writelogger/writelogger_test.go index 62d1d4667..1025acfab 100644 --- a/dax/writelogger/writelogger_test.go +++ b/dax/writelogger/writelogger_test.go @@ -9,11 +9,12 @@ import ( "testing" "github.com/featurebasedb/featurebase/v3/dax/writelogger" + "github.com/featurebasedb/featurebase/v3/logger" "github.com/stretchr/testify/assert" ) -func TestWriteLogger(t *testing.T) { - tmpDir, err := os.MkdirTemp("", "testWriteLogger-*") +func TestWritelogger(t *testing.T) { + tmpDir, err := os.MkdirTemp("", "testWritelogger-*") assert.NoError(t, err) // Remove the temp directory. @@ -27,10 +28,7 @@ func TestWriteLogger(t *testing.T) { Bar int `json:"bar"` } - cfg := writelogger.Config{ - DataDir: tmpDir, - } - wl := writelogger.New(cfg) + wl := writelogger.New(tmpDir, logger.NopLogger) table := "tbl" partition := 1 diff --git a/idk/docker-compose.yml b/idk/docker-compose.yml index 9f0e3097b..5bcde33b3 100644 --- a/idk/docker-compose.yml +++ b/idk/docker-compose.yml @@ -143,9 +143,7 @@ services: FEATUREBASE_QUERYER_RUN: "true" FEATUREBASE_MDS_RUN: "true" FEATUREBASE_MDS_CONFIG_DATA_DIR: /dax-data/mds - FEATUREBASE_WRITELOGGER_RUN: "true" - FEATUREBASE_WRITELOGGER_CONFIG_DATA_DIR: /dax-data/wl - FEATUREBASE_SNAPSHOTTER_RUN: "true" - FEATUREBASE_SNAPSHOTTER_CONFIG_DATA_DIR: /dax-data/snaps FEATUREBASE_COMPUTER_RUN: "true" FEATUREBASE_COMPUTER_CONFIG_DATA_DIR: /dax-data/computer + FEATUREBASE_COMPUTER_CONFIG_WRITELOGGER_DIR: /dax-data/wl + FEATUREBASE_COMPUTER_CONFIG_SNAPSHOTTER_DIR: /dax-data/snaps diff --git a/idk/ingest.go b/idk/ingest.go index e35405583..1f514ecab 100644 --- a/idk/ingest.go +++ b/idk/ingest.go @@ -1024,7 +1024,7 @@ func (m *Main) setupClient() (*tls.Config, error) { m.SchemaManager = mds.NewSchemaManager(dax.Address(m.MDSAddress), qdbid, m.log) m.NewImporterFn = func() pilosacore.Importer { - return mds.NewImporter(mdsClient, mdsClient, qtbl.Qualifier(), &qtbl.Table) + return mds.NewImporter(mdsClient, qtbl.Qualifier(), &qtbl.Table) } } else { m.SchemaManager = m.client diff --git a/idk/ingest_test.go b/idk/ingest_test.go index 63f949b40..1cf096550 100644 --- a/idk/ingest_test.go +++ b/idk/ingest_test.go @@ -58,7 +58,7 @@ func configureTestFlagsMDS(main *Main, address dax.Address, qtbl *dax.QualifiedT mdsClient := mdsclient.New(dax.Address(address), logger.StderrLogger) main.NewImporterFn = func() pilosa.Importer { - return mds.NewImporter(mdsClient, mdsClient, qtbl.QualifiedDatabaseID, &qtbl.Table) + return mds.NewImporter(mdsClient, qtbl.QualifiedDatabaseID, &qtbl.Table) } } diff --git a/idk/mds/importer.go b/idk/mds/importer.go index 0d7a1a31c..27a2757c3 100644 --- a/idk/mds/importer.go +++ b/idk/mds/importer.go @@ -18,20 +18,18 @@ var _ featurebase.Importer = &importer{} // importer type importer struct { - noder dax.Noder - schemar dax.Schemar + controller dax.Controller mu sync.Mutex qdbid dax.QualifiedDatabaseID tbl *dax.Table } -func NewImporter(noder dax.Noder, schemar dax.Schemar, qdbid dax.QualifiedDatabaseID, tbl *dax.Table) *importer { +func NewImporter(controller dax.Controller, qdbid dax.QualifiedDatabaseID, tbl *dax.Table) *importer { return &importer{ - noder: noder, - schemar: schemar, - qdbid: qdbid, - tbl: tbl, + controller: controller, + qdbid: qdbid, + tbl: tbl, } } @@ -77,7 +75,7 @@ func (m *importer) CreateTableKeys(ctx context.Context, tid dax.TableID, keys .. // all the partitions at once, then getting the distinct list of addresses // and looping over that instead. for partition, ks := range partitions { - address, err := m.noder.IngestPartition(context.Background(), qtbl.QualifiedID(), partition) + address, err := m.controller.IngestPartition(context.Background(), qtbl.QualifiedID(), partition) if err != nil { return nil, errors.Wrapf(err, "calling ingest-partition on table: %s, partition: %d", qtbl, partition) } @@ -114,7 +112,7 @@ func (m *importer) CreateFieldKeys(ctx context.Context, tid dax.TableID, fname d // different partitionN for field translation. partition := dax.PartitionNum(0) - address, err := m.noder.IngestPartition(context.Background(), qtbl.QualifiedID(), partition) + address, err := m.controller.IngestPartition(context.Background(), qtbl.QualifiedID(), partition) if err != nil { return nil, errors.Wrapf(err, "calling ingest-partition on table: %s, partition: %d", qtbl, partition) } @@ -139,7 +137,7 @@ func (m *importer) ImportRoaringBitmap(ctx context.Context, tid dax.TableID, fld return errors.Wrapf(err, "getting qtbl") } - address, err := m.noder.IngestShard(context.Background(), qtbl.QualifiedID(), dax.ShardNum(shard)) + address, err := m.controller.IngestShard(context.Background(), qtbl.QualifiedID(), dax.ShardNum(shard)) if err != nil { return errors.Wrap(err, "calling ingest-shard") } @@ -164,7 +162,7 @@ func (m *importer) ImportRoaringShard(ctx context.Context, tid dax.TableID, shar return errors.Wrapf(err, "getting qtbl") } - address, err := m.noder.IngestShard(context.Background(), qtbl.QualifiedID(), dax.ShardNum(shard)) + address, err := m.controller.IngestShard(context.Background(), qtbl.QualifiedID(), dax.ShardNum(shard)) if err != nil { return errors.Wrap(err, "calling ingest-shard") } @@ -184,7 +182,7 @@ func (m *importer) EncodeImportValues(ctx context.Context, tid dax.TableID, fld return "", nil, errors.Wrapf(err, "getting qtbl") } - address, err := m.noder.IngestShard(context.Background(), qtbl.QualifiedID(), dax.ShardNum(shard)) + address, err := m.controller.IngestShard(context.Background(), qtbl.QualifiedID(), dax.ShardNum(shard)) if err != nil { return "", nil, errors.Wrap(err, "calling ingest-shard") } @@ -209,7 +207,7 @@ func (m *importer) EncodeImport(ctx context.Context, tid dax.TableID, fld *dax.F return "", nil, errors.Wrapf(err, "getting qtbl") } - address, err := m.noder.IngestShard(context.Background(), qtbl.QualifiedID(), dax.ShardNum(shard)) + address, err := m.controller.IngestShard(context.Background(), qtbl.QualifiedID(), dax.ShardNum(shard)) if err != nil { return "", nil, errors.Wrap(err, "calling ingest-shard") } @@ -234,7 +232,7 @@ func (m *importer) DoImport(ctx context.Context, tid dax.TableID, fld *dax.Field return errors.Wrapf(err, "getting qtbl") } - address, err := m.noder.IngestShard(context.Background(), qtbl.QualifiedID(), dax.ShardNum(shard)) + address, err := m.controller.IngestShard(context.Background(), qtbl.QualifiedID(), dax.ShardNum(shard)) if err != nil { return errors.Wrap(err, "calling ingest-shard") } @@ -266,7 +264,7 @@ func (m *importer) getQtbl(ctx context.Context, tid dax.TableID) (*dax.Qualified qtid := dax.NewQualifiedTableID(m.qdbid, tid) - qtbl, err := m.schemar.TableByID(ctx, qtid) + qtbl, err := m.controller.TableByID(ctx, qtid) if err != nil { return nil, errors.Wrap(err, "getting table") } diff --git a/server/config.go b/server/config.go index 282e3fe56..d1e72a38b 100644 --- a/server/config.go +++ b/server/config.go @@ -58,13 +58,15 @@ type Config struct { // particular index. MDSAddress string `toml:"mds-address"` - // WriteLogger is the location at which this node should read/write change - // logs. - WriteLogger string `toml:"write-logger"` + // WriteloggerDir is the location at which this node should + // read/write change logs. Typically a network mounted filesystem + // for availability/durability. + WriteloggerDir string `toml:"writelogger-dir"` - // Snapshotter is the location at which this node should read/write - // snapshots. - Snapshotter string `toml:"snapshotter"` + // SnapshotterDir is the location at which this node should + // read/write snapshots. Typically a network mounted filesystem + // for availability/durability. + SnapshotterDir string `toml:"snapshotter-dir"` // DataDir is the directory where Pilosa stores both indexed data and // running state such as cluster topology information. diff --git a/server/server.go b/server/server.go index b72eee2e9..a4188aa34 100644 --- a/server/server.go +++ b/server/server.go @@ -74,7 +74,7 @@ type Command struct { Registrar computer.Registrar serverlessStorage *storage.ResourceManager - writeLogService computer.WriteLogService + writeLogService computer.WritelogService snapshotService computer.SnapshotService Handler pilosa.HandlerI @@ -124,7 +124,6 @@ func OptCommandConfig(config *Config) CommandOption { c.Config.Auth = config.Auth c.Config.TLS = config.TLS c.Config.MDSAddress = config.MDSAddress - c.Config.WriteLogger = config.WriteLogger c.Config.SQL.EndpointEnabled = config.SQL.EndpointEnabled return nil } @@ -147,8 +146,8 @@ func OptCommandSetConfig(config *Config) CommandOption { // OptCommandInjections injects the interface implementations. func OptCommandInjections(inj Injections) CommandOption { return func(c *Command) error { - if inj.WriteLogger != nil { - c.writeLogService = inj.WriteLogger + if inj.Writelogger != nil { + c.writeLogService = inj.Writelogger } if inj.Snapshotter != nil { c.snapshotService = inj.Snapshotter @@ -159,7 +158,7 @@ func OptCommandInjections(inj Injections) CommandOption { } type Injections struct { - WriteLogger computer.WriteLogService + Writelogger computer.WritelogService Snapshotter computer.SnapshotService IsComputeNode bool } diff --git a/sql3/test/defs/defs_unkeyed.go b/sql3/test/defs/defs_unkeyed.go index 038870480..ce54f79f3 100644 --- a/sql3/test/defs/defs_unkeyed.go +++ b/sql3/test/defs/defs_unkeyed.go @@ -2,6 +2,8 @@ package defs import "github.com/featurebasedb/featurebase/v3/pql" +var Unkeyed TableTest = unkeyed + var unkeyed = TableTest{ name: "unkeyed", Table: tbl(