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(