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

extremely WIP

Browse source 
(cherry picked from commit 35c472a54f)
This commit is contained in:
Matthew Jaffee 2022-12-12 21:49:00 -06:00 • committed by Joe Friedrich
parent 71e3c00b46
commit ed7c6d419e
18 changed files with 907 additions and 326 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

3
Makefile
View file

@ -228,7 +228,8 @@ build-for-quick:
docker-image-featurebase-quick: build-for-quick
docker build \
--build-arg GO_VERSION=$(GO_VERSION) \
--file Dockerfile-dax-quick ./.quick/
--file Dockerfile-dax-quick \
--tag dax/featurebase ./.quick/
docker-image-datagen: vendor

238
api.go
View file

@ -55,9 +55,7 @@ type API struct {
Serializer Serializer
writeLogReader computer.WriteLogReader
writeLogWriter computer.WriteLogWriter
snapshotReadWriter computer.SnapshotReadWriter
serverlessStorage *storage.ManagerManager
directiveWorkerPoolSize int
@ -83,6 +81,13 @@ func OptAPIServer(s *Server) apiOption {
}
}
func OptAPIServerlessStorage(mm *storage.ManagerManager) apiOption {
return func(a *API) error {
a.serverlessStorage = mm
return nil
}
}
func OptAPIImportWorkerPoolSize(size int) apiOption {
return func(a *API) error {
a.importWorkerPoolSize = size
@ -90,27 +95,6 @@ func OptAPIImportWorkerPoolSize(size int) apiOption {
}
}
func OptAPIWriteLogReader(wlr computer.WriteLogReader) apiOption {
return func(a *API) error {
a.writeLogReader = wlr
return nil
}
}
func OptAPIWriteLogWriter(wlw computer.WriteLogWriter) apiOption {
return func(a *API) error {
a.writeLogWriter = wlw
return nil
}
}
func OptAPISnapshotter(snap computer.SnapshotReadWriter) apiOption {
return func(a *API) error {
a.snapshotReadWriter = snap
return nil
}
}
func OptAPIDirectiveWorkerPoolSize(size int) apiOption {
return func(a *API) error {
a.directiveWorkerPoolSize = size
@ -129,9 +113,6 @@ func OptAPIIsComputeNode(is bool) apiOption {
func NewAPI(opts ...apiOption) (*API, error) {
api := &API{
importWorkerPoolSize: 2,
writeLogReader: computer.NewNopWriteLogReader(),
writeLogWriter: computer.NewNopWriteLogWriter(),
snapshotReadWriter: computer.NewNopSnapshotReadWriter(),
directiveWorkerPoolSize: 2,
}
@ -709,20 +690,20 @@ func (api *API) ImportRoaring(ctx context.Context, indexName, fieldName string,
Views: req.Views,
}
// Get the current version for shard.
version, err := api.getOrCreateShardVersion(ctx, indexName, shard)
if err != nil {
return errors.Wrap(err, "get or creating shard version")
}
tkey := dax.TableKey(indexName)
qtid := tkey.QualifiedTableID()
partitionNum := dax.PartitionNum(partition)
shardNum := dax.ShardNum(shard)
api.server.logger.Debugf("importroaring writing to writelogger: %+v, %[1]T len(msg.Views): %d, table: %s", api.writeLogWriter, len(msg.Views), msg.Table)
if err := api.writeLogWriter.WriteShard(ctx, qtid, partitionNum, shardNum, version, msg); err != nil {
return err
b, err := computer.MarshalLogMessage(msg, computer.EncodeTypeJSON)
if err != nil {
return errors.Wrap(err, "marshalling log message")
}
mgr := api.serverlessStorage.GetShardManager(qtid, partitionNum, shardNum)
err = mgr.Append(b)
if err != nil {
return errors.Wrap(err, "appending shard data") // TODO do we need to set err0 or something?
}
}
@ -1286,8 +1267,13 @@ func (api *API) DeleteView(ctx context.Context, indexName string, fieldName stri
return errors.Wrap(err, "sending DeleteView message")
}
// IndexShardSnapshot returns a reader that contains the contents of an RBF snapshot for an index/shard.
func (api *API) IndexShardSnapshot(ctx context.Context, indexName string, shard uint64) (io.ReadCloser, error) {
// IndexShardSnapshot returns a reader that contains the contents of
// an RBF snapshot for an index/shard. When snapshotting for
// serverless, we need to be able to transactionally move the write
// log to the new version, so we expose writeTx to allow the caller to
// request a write transaction for the snapshot even though we'll just
// be reading inside RBF.
func (api *API) IndexShardSnapshot(ctx context.Context, indexName string, shard uint64, writeTx bool) (io.ReadCloser, error) {
span, _ := tracing.StartSpanFromContext(ctx, "API.IndexShardSnapshot")
defer span.Finish()
@ -1298,7 +1284,7 @@ func (api *API) IndexShardSnapshot(ctx context.Context, indexName string, shard
}
// Start transaction.
tx := index.holder.txf.NewTx(Txo{Index: index, Shard: shard})
tx := index.holder.txf.NewTx(Txo{Index: index, Shard: shard, Write: writeTx})
// Ensure transaction is an RBF transaction.
rtx, ok := tx.(*RBFTx)
@ -1517,20 +1503,20 @@ func (api *API) Import(ctx context.Context, qcx *Qcx, req *ImportRequest, opts .
}
if api.isComputeNode && !options.suppressLog {
// Get the current version for shard.
version, err := api.getOrCreateShardVersion(ctx, req.Index, req.Shard)
if err != nil {
return errors.Wrap(err, "get or creating shard version")
}
tkey := dax.TableKey(req.Index)
qtid := tkey.QualifiedTableID()
partitionNum := dax.PartitionNum(partition)
shardNum := dax.ShardNum(req.Shard)
// Write the request to the write logger.
if err := api.writeLogWriter.WriteShard(ctx, qtid, partitionNum, shardNum, version, msg); err != nil {
return err
b, err := computer.MarshalLogMessage(msg, computer.EncodeTypeJSON)
if err != nil {
return errors.Wrap(err, "marshalling log message")
}
mgr := api.serverlessStorage.GetShardManager(qtid, partitionNum, shardNum)
err = mgr.Append(b)
if err != nil {
return errors.Wrap(err, "appending shard data") // TODO do we need to set err0 or something?
}
}
@ -1765,21 +1751,21 @@ func (api *API) ImportRoaringShard(ctx context.Context, indexName string, shard
ClearRecords: view.ClearRecords,
}
}
// Get the current version for shard.
version, err := api.getOrCreateShardVersion(ctx, indexName, shard)
if err != nil {
err1 = errors.Wrap(err, "get or creating shard version")
return err1
}
tkey := dax.TableKey(indexName)
qtid := tkey.QualifiedTableID()
partitionNum := dax.PartitionNum(partition)
shardNum := dax.ShardNum(shard)
api.server.logger.Debugf("importroaringshard writing shard to writelogger: %+v, len(msg.Views): %d, table: %s", api.writeLogWriter, len(msg.Views), msg.Table)
b, err := computer.MarshalLogMessage(msg, computer.EncodeTypeJSON)
if err != nil {
err1 = errors.Wrap(err, "marshalling log message")
return err1
}
if err := api.writeLogWriter.WriteShard(ctx, qtid, partitionNum, shardNum, version, msg); err != nil {
err1 = errors.Wrap(err, "writing import-roaring-shard to writelogger")
mgr := api.serverlessStorage.GetShardManager(qtid, partitionNum, shardNum)
err1 = errors.Wrap(mgr.Append(b), "appending shard data")
if err1 != nil {
return err1
}
}
@ -1860,20 +1846,21 @@ func (api *API) ImportValue(ctx context.Context, qcx *Qcx, req *ImportValueReque
if api.isComputeNode && !options.suppressLog {
// Get the current version for shard.
version, err := api.getOrCreateShardVersion(ctx, req.Index, req.Shard)
if err != nil {
return errors.Wrap(err, "get or creating shard version")
}
tkey := dax.TableKey(req.Index)
qtid := tkey.QualifiedTableID()
partitionNum := dax.PartitionNum(partition)
shardNum := dax.ShardNum(req.Shard)
// Write the request to the write logger.
if err := api.writeLogWriter.WriteShard(ctx, qtid, partitionNum, shardNum, version, msg); err != nil {
return errors.Wrap(err, "writing shard to write logger")
b, err := computer.MarshalLogMessage(msg, computer.EncodeTypeJSON)
if err != nil {
return errors.Wrap(err, "marshalling log message")
}
mgr := api.serverlessStorage.GetShardManager(qtid, partitionNum, shardNum)
err = mgr.Append(b)
if err != nil {
return errors.Wrap(err, "appending shard data") // TODO do we need to set err0 or something?
}
}
return nil
@ -3107,47 +3094,33 @@ func (api *API) DirectiveApplied(ctx context.Context) (bool, error) {
// SnapshotShardData triggers the node to perform a shard snapshot based on the
// provided SnapshotShardDataRequest.
func (api *API) SnapshotShardData(ctx context.Context, req *dax.SnapshotShardDataRequest) error {
qtid := req.TableKey.QualifiedTableID()
// TODO(jaffee) confirm this node is actually responsible for the given
// shard? Not sure we need to given that this request comes from
// MDS, but might be a belt&suspenders situation.
// Confirm that this node is currently responsible for table/shard/fromVersion.
var version int
if v, ok, err := api.holder.versionStore.ShardVersion(ctx, qtid, req.ShardNum); err != nil {
return err
} else if !ok {
return errors.Errorf("shard not managed by this node: %s, %d", req.TableKey, req.ShardNum)
} else if v != req.FromVersion {
return errors.Errorf("shard managed by this node is at version: %d, not: %d", v, req.FromVersion)
} else {
version = v
}
qtid := req.TableKey.QualifiedTableID()
partition := disco.ShardToShardPartition(string(req.TableKey), uint64(req.ShardNum), disco.DefaultPartitionN)
partitionNum := dax.PartitionNum(partition)
// Create the snapshot for the current version.
rc, err := api.IndexShardSnapshot(ctx, string(req.TableKey), uint64(req.ShardNum))
// Create the snapshot for the current version. How do we ensure
// here that any new writes go to the new write log since we are
// now reading a version of the shard which exists only at this
// exact point in time? Ans: we'll cut over to the new storage
// manager and call IncrementWriteLogVersion while a write Tx is
// held on RBF.
rc, err := api.IndexShardSnapshot(ctx, string(req.TableKey), uint64(req.ShardNum), true)
if err != nil {
return errors.Wrap(err, "getting index/shard readcloser")
}
// The following closes rc, the ReadCloser.
if err := api.snapshotReadWriter.WriteShardData(ctx, qtid, partitionNum, req.ShardNum, version, rc); err != nil {
return errors.Wrap(err, "snapshotting shard data")
mgr := api.serverlessStorage.GetShardManager(qtid, partitionNum, req.ShardNum)
if err := mgr.IncrementWLVersion(); err != nil {
return errors.Wrap(err, "incrementing write log version")
}
// Increment the version of the shard managed by this node.
if err := api.holder.versionStore.AddShards(ctx, qtid,
dax.NewVersionedShard(req.ShardNum, req.ToVersion),
); err != nil {
return errors.Wrap(err, "incrementing shard version locally")
}
// Update the cached directive on the holder.
api.holder.SetDirective(&req.Directive)
api.holder.SetDirectiveApplied(true)
// Finally, delete the log file for the previous version.
return api.writeLogWriter.DeleteShard(ctx, qtid, partitionNum, req.ShardNum, req.FromVersion)
// TODO(jaffee) look into downgrading Tx on RBF to read lock here now that WL version is incremented.
err = mgr.Snapshot(rc)
return errors.Wrap(err, "snapshotting shard data")
}
// SnapshotTableKeys triggers the node to perform a table keys snapshot based on
@ -3162,41 +3135,21 @@ func (api *API) SnapshotTableKeys(ctx context.Context, req *dax.SnapshotTableKey
qtid := req.TableKey.QualifiedTableID()
// Confirm that this node is currently responsible for table/partition/fromVersion.
var version int
if v, ok, err := api.holder.versionStore.PartitionVersion(ctx, qtid, req.PartitionNum); err != nil {
return err
} else if !ok {
return errors.Errorf("partition not managed by this node: %s, %d", req.TableKey, req.PartitionNum)
} else if v != req.FromVersion {
return errors.Errorf("partition managed by this node is at version: %d, not: %d", v, req.FromVersion)
} else {
version = v
}
// Create the snapshot for the current version.
wrTo, err := api.TranslateData(ctx, string(req.TableKey), int(req.PartitionNum))
if err != nil {
return errors.Wrapf(err, "getting index/partition writeto: %s/%d", req.TableKey, req.PartitionNum)
}
if err := api.snapshotReadWriter.WriteTableKeys(ctx, qtid, req.PartitionNum, version, wrTo); err != nil {
return errors.Wrap(err, "snapshotting table keys")
// TODO(jaffee) need to ensure writes to translation data can't
// occur while this is happening.
mgr := api.serverlessStorage.GetTableKeyManager(qtid, req.PartitionNum)
if err := mgr.IncrementWLVersion(); err != nil {
return errors.Wrap(err, "incrementing write log version")
}
// Increment the version of the partition managed by this node.
if err := api.holder.versionStore.AddPartitions(ctx, qtid,
dax.NewVersionedPartition(req.PartitionNum, req.ToVersion),
); err != nil {
return errors.Wrap(err, "incrementing partition version locally")
}
// Update the cached directive on the holder.
api.holder.SetDirective(&req.Directive)
api.holder.SetDirectiveApplied(true)
// Finally, delete the log file for the previous version.
return api.writeLogWriter.DeleteTableKeys(ctx, qtid, req.PartitionNum, req.FromVersion)
// TODO(jaffee) downgrade (currently non-existent) lock to read-only
err = mgr.SnapshotTo(wrTo)
return errors.Wrap(err, "snapshotting table keys")
}
// SnapshotFieldKeys triggers the node to perform a field keys snapshot based on
@ -3204,41 +3157,20 @@ func (api *API) SnapshotTableKeys(ctx context.Context, req *dax.SnapshotTableKey
func (api *API) SnapshotFieldKeys(ctx context.Context, req *dax.SnapshotFieldKeysRequest) error {
qtid := req.TableKey.QualifiedTableID()
// Confirm that this node is currently responsible for table/field/fromVersion.
var version int
if v, ok, err := api.holder.versionStore.FieldVersion(ctx, qtid, req.Field); err != nil {
return err
} else if !ok {
return errors.Errorf("field not managed by this node: %s, %s", req.TableKey, req.Field)
} else if v != req.FromVersion {
return errors.Errorf("field managed by this node is at version: %d, not: %d", v, req.FromVersion)
} else {
version = v
}
// Create the snapshot for the current version.
// TODO(jaffee) change this to get write lock
wrTo, err := api.FieldTranslateData(ctx, string(req.TableKey), string(req.Field))
if err != nil {
return errors.Wrap(err, "getting index/field writeto")
}
if err := api.snapshotReadWriter.WriteFieldKeys(ctx, qtid, req.Field, version, wrTo); err != nil {
return errors.Wrap(err, "snapshotting field keys")
mgr := api.serverlessStorage.GetFieldKeyManager(qtid, req.Field)
if err := mgr.IncrementWLVersion(); err != nil {
return errors.Wrap(err, "incrementing writelog version")
}
// Increment the version of the field managed by this node.
if err := api.holder.versionStore.AddFields(ctx, qtid,
dax.NewVersionedField(req.Field, req.ToVersion),
); err != nil {
return errors.Wrap(err, "incrementing field version locally")
}
// Update the cached directive on the holder.
api.holder.SetDirective(&req.Directive)
api.holder.SetDirectiveApplied(true)
// Finally, delete the log file for the previous version.
return api.writeLogWriter.DeleteFieldKeys(ctx, qtid, req.Field, req.FromVersion)
// TODO(jaffee) downgrade to read lock
err = mgr.SnapshotTo(wrTo)
return errors.Wrap(err, "snapshotTo in FieldKeys")
}
type serverInfo struct {

33
api_directive.go
View file

@ -348,29 +348,26 @@ func (api *API) pushJobsTableKeys(ctx context.Context, jobs chan<- directiveJobT
}
}
func (api *API) loadTableKeys(ctx context.Context, idx *Index, tkey dax.TableKey, partition dax.VersionedPartition) error {
func (api *API) loadTableKeys(ctx context.Context, idx *Index, tkey dax.TableKey, partition dax.PartitionNum) error {
qtid := tkey.QualifiedTableID()
// Load the previous snapshot. Version 0 doesn't have a snapshot
// file; it only has log entries.
if partition.Version > 0 {
// Load partition snapshot: version - 1
previousVersion := partition.Version - 1
rc, err := api.snapshotReadWriter.ReadTableKeys(ctx, qtid, partition.Num, previousVersion)
if err != nil {
return errors.Wrap(err, "reading table keys snapshot")
}
defer rc.Close()
mgr := api.serverlessStorage.GetTableKeyManager(qtid, partition)
rc, err := mgr.LoadLatestSnapshot()
if err != nil {
return errors.Wrap(err, "loading table key snapshot")
}
defer rc.Close()
if err := api.TranslateIndexDB(ctx, string(tkey), int(partition), rc); err != nil {
return errors.Wrap(err, "restoring table keys")
if err := api.TranslateIndexDB(ctx, string(tkey), int(partition.Num), rc); err != nil {
return errors.Wrap(err, "restoring table keys")
}
}
if err := func() error {
store := idx.TranslateStore(int(partition.Num))
store := idx.TranslateStore(int(partition))
reader, err := mgr.LoadWriteLog()
if err != nil {
return errors.Wrap(err, "loading write log table keys")
}
reader := api.writeLogReader.TableKeyReader(ctx, qtid, partition.Num, partition.Version)
if err := reader.Open(); err != nil {
// TODO: this log can be confusing because on a create
@ -506,7 +503,7 @@ func (api *API) pushJobsShards(ctx context.Context, jobs chan<- directiveJobType
}
}
func (api *API) loadShard(ctx context.Context, tkey dax.TableKey, shard dax.VersionedShard) error {
func (api *API) loadShard(ctx context.Context, tkey dax.TableKey, shard dax.ShardNum) error {
qtid := tkey.QualifiedTableID()
partition := disco.ShardToShardPartition(string(tkey), uint64(shard.Num), disco.DefaultPartitionN)

68
cluster.go
View file

@ -4,6 +4,7 @@ package pilosa
import (
"context"
"encoding/json"
"fmt"
"log"
"sync"
@ -72,8 +73,8 @@ type cluster struct { // nolint: maligned
partitionAssigner string
writeLogWriter computer.WriteLogWriter
versionStore dax.VersionStore
serverlessStorage *storage.ManagerManager
versionStore dax.VersionStore
// isComputeNode is set to true if this node is running as a DAX compute
// node.
@ -100,8 +101,6 @@ func newCluster() *cluster {
disCo: disco.NopDisCo,
noder: disco.NewEmptyLocalNoder(),
writeLogWriter: computer.NewNopWriteLogWriter(),
}
}
@ -318,6 +317,44 @@ func (c *cluster) findFieldKeys(ctx context.Context, field *Field, keys ...strin
return translations, nil
}
func (c *cluster) appendFieldKeysWriteLog(ctx context.Context, qtid dax.QualifiedTableID, fieldName dax.FieldName, translations map[string]uint64) error {
// TODO move marshaling somewhere more centralized and less... explicitly json-y
msg := computer.FieldKeyMap{
TableKey: qtid.Key(),
Field: fieldName,
StringToID: translations,
}
b, err := json.Marshal(msg)
if err != nil {
return errors.Wrap(err, "marshalling field key map to json")
}
mgr := c.serverlessStorage.GetFieldKeyManager(qtid, fieldName)
err = mgr.Append(b)
if err != nil {
return errors.Wrap(err, "appending field keys")
}
return nil
}
func (c *cluster) appendTableKeysWriteLog(ctx context.Context, qtid dax.QualifiedTableID, partition dax.PartitionNum, translations map[string]uint64) error {
msg := computer.PartitionKeyMap{
TableKey: qtid.Key(),
Partition: partition,
StringToID: translations,
}
b, err := json.Marshal(msg)
if err != nil {
return errors.Wrap(err, "marshalling partition key map to json")
}
mgr := c.serverlessStorage.GetTableKeyManager(qtid, partition)
return errors.Wrap(mgr.Append(b), "appending table keys")
}
func (c *cluster) createFieldKeys(ctx context.Context, field *Field, keys ...string) (map[string]uint64, error) {
if idx := field.ForeignIndex(); idx != "" {
// The field uses foreign index keys.
@ -350,17 +387,9 @@ func (c *cluster) createFieldKeys(ctx context.Context, field *Field, keys ...str
tkey := dax.TableKey(field.Index())
qtid := tkey.QualifiedTableID()
fieldName := dax.FieldName(field.Name())
// Get the current version for field.
version, found, err := c.versionStore.FieldVersion(ctx, qtid, fieldName)
err = c.appendFieldKeysWriteLog(ctx, qtid, fieldName, translations)
if err != nil {
return nil, errors.Wrap(err, "getting field version")
} else if !found {
return nil, errors.Errorf("no version found for table(%s) field(%s)", qtid, fieldName)
}
if err := c.writeLogWriter.CreateFieldKeys(ctx, qtid, fieldName, version, translations); err != nil {
return nil, errors.Errorf("logging field(%s/%s) keys(%v)", field.Index(), field.Name(), keys)
return nil, errors.Wrap(err, "appending to write log")
}
return translations, nil
@ -754,16 +783,7 @@ func (c *cluster) createIndexKeys(ctx context.Context, indexName string, keys ..
tkey := dax.TableKey(idx.Name())
qtid := tkey.QualifiedTableID()
partitionNum := dax.PartitionNum(partitionID)
// Get the current version for partition.
version, found, err := c.versionStore.PartitionVersion(ctx, qtid, partitionNum)
if err != nil {
return errors.Wrap(err, "getting partition version")
} else if !found {
return errors.Errorf("no version found for table(%s) partition(%d)", qtid, partitionNum)
}
return c.writeLogWriter.CreateTableKeys(ctx, qtid, partitionNum, version, translations)
return c.appendTableKeysWriteLog(ctx, qtid, partitionNum, translations)
})
}

3
dax/Makefile
View file

@ -78,6 +78,9 @@ dc-logs-%:
dc-prereqs:
mkdir -p ../.quick
dc-cli:
featurebase cli --host localhost --port 8080 --org-id=testorg --db-id=testdb
# This is just an example. For it to work, you'll first need to:
# featurebase cli --host localhost --port 8080 --org-id=testorg --db-id=testdb
# create table keysidstbl2 (_id string, slice idset);

87
dax/computer/interfaces.go
View file

@ -18,8 +18,12 @@ type Registrar interface {
// 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.Reader, io.Closer, error)
LogReader(bucket string, key string, version int) (io.ReadCloser, error)
LogReaderFrom(bucket string, key string, version int, offset int) (io.ReadCloser, error)
DeleteLog(bucket string, key string, version int) error
List(bucket, key string) ([]WriteLogInfo, error)
Lock(bucket, key string) error
Unlock(bucket, key string) error
}
// SnapshotService represents the SnapshotService methods which Computer uses.
@ -28,8 +32,89 @@ type SnapshotService interface {
Read(bucket string, key string, version int) (io.ReadCloser, error)
Write(bucket string, key string, version int, rc io.ReadCloser) error
WriteTo(bucket string, key string, version int, wrTo io.WriterTo) error
List(bucket, key string) ([]SnapInfo, error)
}
// ServerlessStorage is the interface to a particular shard or
// translation store that contains both writelogger and
// snapshotter. The interface is designed such that implementations
// are expected to be stateful.
//
// One must not call LoadWriteLog until after calling
// LoadLatestSnapshot. One must not call Append, IncrementWLVersion,
// or Snapshot until after successfully calling Lock.
type ServerlessStorage interface {
// LoadLatestSnapshot loads the latest available snapshot in the snapshot store.
LoadLatestSnapshot() (data io.ReadCloser, err error)
// // Potential future methods to support getting older versions. SnapInfo would have timestamp information as well.
//
// ListSnapshots() []SnapInfo
// LoadSnapshot(version int) (data io.ReadCloser, err error)
// LoadWriteLog can be called after LoadLatestSnapshot. It loads
// any writelog data which has been written since the latest
// snapshot. Subsequent calls to LoadWriteLog will only return new
// data that hasn't previously been returned from LoadWriteLog.
LoadWriteLog() (data io.ReadCloser, err error)
// Lock acquires an advisory lock for this resource which grants
// us exclusive access to write to it. The normal pattern is to
// call:
//
// 1. LoadLatestSnapshot
// 2. LoadWriteLog
// 3. Lock
// 4. LoadWriteLog
//
// The second call to LoadWriteLog is necessary in case any writes
// occurred between the last load and acquiring the lock. Once the
// lock is acquired it should not be possible for any more writes
// to occur. Lock will error if (a) we fail to acquire the lock or
// (b) the state of the snapshot store for this resource is not
// identical to what is was before the lock was acquired. Case (b)
// means that quite a lot has happened in between LoadWriteLog and
// Lock, and we should probably just die and start over.
Lock() error
// Append appends the msg to the write log. It will fail if we
// haven't properly loaded and gotten a lock for the resource
// we're writing to.
Append(msg []byte) error
// IncrementWLVersion should be called during snapshotting with a
// write Tx held on the local resource. This ensures that any
// writes which completed prior to the snapshot are in the prior
// WL and any that complete after the snapshot are in the
// incremented WL.
IncrementWLVersion() error
// Snapshot takes a ReadCloser which has the contents of the
// resource being tracked at a particular point in time and writes
// them to the Snapshot Store. Upon a successful write it will
// truncate any write logs which are now incorporated into the
// snapshot.
Snapshot(rc io.ReadCloser) error
SnapshotTo(wt io.WriterTo) error
// Unlock releases the lock. This should be called if control of
// the underlying resource is being transitioned to another
// node. Ideally it's also called if the process crashes (e.g. via
// a defer), but an implementation based on filesystem locks
// should have those removed by the operating system when the
// process exits anyway.
Unlock() error
}
// SnapInfo holds metadata about a snapshot.
type SnapInfo struct {
Version int
// Date time.Time
}
// WriteLogInfo holds metadata about a write log.
type WriteLogInfo SnapInfo
// SnapshotReadWriter provides the interface for all snapshot read and writes in
// FeatureBase.
type SnapshotReadWriter interface {

18
dax/computer/writelog.go
View file

@ -132,13 +132,13 @@ func newTableKeyReader(wl WriteLogService, qtid dax.QualifiedTableID, partition
func (r *tableKeyReader) Open() error {
bucket := partitionBucket(r.table, r.partition)
reader, closer, err := r.wl.LogReader(bucket, keysFileName, r.version)
readcloser, err := r.wl.LogReader(bucket, keysFileName, r.version)
if err != nil {
return errors.Wrapf(err, "getting log reader: %s, %s, %d", bucket, keysFileName, r.version)
}
r.closer = closer
r.scanner = bufio.NewScanner(reader)
r.closer = readcloser
r.scanner = bufio.NewScanner(readcloser)
return nil
}
@ -201,13 +201,13 @@ func newFieldKeyReader(wl WriteLogService, qtid dax.QualifiedTableID, field dax.
func (r *fieldKeyReader) Open() error {
bucket := fieldBucket(r.table, r.field)
reader, closer, err := r.wl.LogReader(bucket, keysFileName, r.version)
readcloser, err := r.wl.LogReader(bucket, keysFileName, r.version)
if err != nil {
return errors.Wrapf(err, "getting log reader: %s, %s, %d", bucket, keysFileName, r.version)
}
r.closer = closer
r.scanner = bufio.NewScanner(reader)
r.closer = readcloser
r.scanner = bufio.NewScanner(readcloser)
return nil
}
@ -273,13 +273,13 @@ func (r *shardReader) Open() error {
bucket := partitionBucket(r.table, r.partition)
shardKey := shardKey(r.shard)
reader, closer, err := r.wl.LogReader(bucket, shardKey, r.version)
readcloser, err := r.wl.LogReader(bucket, shardKey, r.version)
if err != nil {
return errors.Wrapf(err, "getting log reader: %s, %s, %d", bucket, shardKey, r.version)
}
r.closer = closer
r.scanner = bufio.NewScanner(reader)
r.closer = readcloser
r.scanner = bufio.NewScanner(readcloser)
return nil
}

2
dax/mds/client/client.go
View file

@ -451,7 +451,7 @@ func (c *Client) RegisterNode(ctx context.Context, node *dax.Node) error {
if resp.StatusCode != http.StatusOK {
b, _ := io.ReadAll(resp.Body)
return errors.Errorf("status code: %d: %s", resp.StatusCode, b)
return errors.Errorf("registration request to %s status code: %d: %s", url, resp.StatusCode, b)
}
return nil

54
dax/mds/controller/controller.go
View file

@ -1401,17 +1401,6 @@ func (c *Controller) SnapshotTable(ctx context.Context, qtid dax.QualifiedTableI
// snapshot that shard, then increment its shard version for logs written to the
// WriteLogger.
func (c *Controller) SnapshotShardData(ctx context.Context, qtid dax.QualifiedTableID, shardNum dax.ShardNum) error {
// Confirm table/shard is being tracked; get the current shard.
fromShardVersion, ok, err := c.versionStore.ShardVersion(ctx, qtid, shardNum)
if err != nil {
return errors.Wrapf(err, "getting shard version: %s, %d", qtid, shardNum)
} else if !ok {
return NewErrInternal(
fmt.Sprintf("shard to snapshot not found: %s, %d", qtid, shardNum),
)
}
toShardVersion := fromShardVersion + 1
// Get the node responsible for the shard.
bal := c.ComputeBalancer
@ -1428,54 +1417,17 @@ func (c *Controller) SnapshotShardData(ctx context.Context, qtid dax.QualifiedTa
addr := dax.Address(workers[0].ID)
// Make a copy of the controller's versionStore, and update the current
// shard so that the directive sent along with the SnapshotRequest reflects
// the state that we want after a successful snapshot.
versionStoreCopy, err := c.versionStore.Copy(ctx)
if err != nil {
return errors.Wrap(err, "copying version store")
}
if err := versionStoreCopy.AddShards(ctx, qtid,
dax.NewVersionedShard(shardNum, toShardVersion),
); err != nil {
return NewErrInternal(err.Error())
}
// Convert the address into a slice of addressMethod containing the
// appropriate method.
addressMethods := applyAddressMethod([]dax.Address{addr}, dax.DirectiveMethodSnapshot)
var toDirective dax.Directive
if directives, err := c.buildDirectives(ctx, addressMethods, versionStoreCopy); err != nil {
return NewErrInternal(err.Error())
} else if ld := len(directives); ld != 1 {
msg := fmt.Sprintf("buildDirectives returned invalid number of directives: %d", ld)
return NewErrInternal(msg)
} else {
toDirective = *directives[0]
}
// Send the node a snapshot request.
req := &dax.SnapshotShardDataRequest{
Address: addr,
TableKey: qtid.Key(),
ShardNum: shardNum,
FromVersion: fromShardVersion,
ToVersion: toShardVersion,
Directive: toDirective,
Address: addr,
TableKey: qtid.Key(),
ShardNum: shardNum,
}
if err := c.Director.SendSnapshotShardDataRequest(ctx, req); err != nil {
return NewErrInternal(err.Error())
}
// A successful request means the shard version can be incremented.
if err := c.versionStore.AddShards(ctx, qtid,
dax.NewVersionedShard(shardNum, toShardVersion),
); err != nil {
return NewErrInternal(err.Error())
}
return nil
}

8
dax/snapshot.go
View file

@ -3,12 +3,8 @@ package dax
type SnapshotShardDataRequest struct {
Address Address `json:"address"`
TableKey TableKey `json:"table-key"`
ShardNum ShardNum `json:"shard"`
FromVersion int `json:"from-version"`
ToVersion int `json:"to-version"`
Directive Directive `json:"directive"`
TableKey TableKey `json:"table-key"`
ShardNum ShardNum `json:"shard"`
}
type SnapshotTableKeysRequest struct {

28
dax/snapshotter/snapshotter.go
View file

@ -8,12 +8,17 @@ import (
"io/fs"
"os"
"path"
"strconv"
"sync"
"syscall"
"github.com/featurebasedb/featurebase/v3/errors"
"github.com/featurebasedb/featurebase/v3/logger"
)
// bucket = table + partition or table + field
// key = "shard/num" or "keys"
type Snapshotter struct {
mu sync.RWMutex
@ -50,6 +55,29 @@ func (s *Snapshotter) Write(bucket string, key string, version int, rc io.ReadCl
return snapshotFile.Sync()
}
func (s *Snapshotter) List(bucket, key string) ([]computer.SnapInfo, error) {
dirpath := path.Join(s.dataDir, bucket, key)
entries, err := os.ReadDir(dirpath)
if err != nil {
if pe, ok := err.(*os.PathError); ok && pe.Err == syscall.ENOENT {
return nil, nil
}
return nil, errors.Wrap(err, "reading directory")
}
snaps := make([]computer.SnapInfo, len(entries))
for i, entry := range entries {
version, err := strconv.ParseInt(entry.Name(), 10, 64)
if err != nil {
return nil, errors.Wrapf(err, "filename '%s' could not be parsed to version number", entry.Name())
}
snaps[i] = computer.SnapInfo{
Version: int(version),
}
}
return snaps, nil
}
func (s *Snapshotter) Read(bucket string, key string, version int) (io.ReadCloser, error) {
_, filePath := s.paths(fullKey(bucket, key, version))
f, err := os.Open(filePath)

320
dax/storage/storage.go Normal file
View file

@ -0,0 +1,320 @@
package storage
import (
"bytes"
"fmt"
"io"
"path"
"sync"
"github.com/molecula/featurebase/v3/dax"
"github.com/molecula/featurebase/v3/dax/computer"
"github.com/molecula/featurebase/v3/errors"
"github.com/molecula/featurebase/v3/logger"
)
// ManagerManager holds all the various Managers each of which is
// specific to a particular shard, table key partition or field, but
// all of which use the same underlying snapshotter and writelogger.
type ManagerManager struct {
Snapshotter computer.SnapshotService
WriteLogger computer.WriteLogService
Logger logger.Logger
mu sync.Mutex
shardManagers map[shardK]*Manager
tableKeyManagers map[tableKeyK]*Manager
fieldKeyManagers map[fieldKeyK]*Manager
}
func NewManagerManager(s computer.SnapshotService, w computer.WriteLogService, l logger.Logger) *ManagerManager {
return &ManagerManager{
Snapshotter: s,
WriteLogger: w,
Logger: l,
shardManagers: make(map[shardK]*Manager),
tableKeyManagers: make(map[tableKeyK]*Manager),
fieldKeyManagers: make(map[fieldKeyK]*Manager),
}
}
// compound map keys
type shardK struct {
qtid dax.QualifiedTableID
partition dax.PartitionNum
shard dax.ShardNum
}
type tableKeyK struct {
qtid dax.QualifiedTableID
partition dax.PartitionNum
}
type fieldKeyK struct {
qtid dax.QualifiedTableID
field dax.FieldName
}
func (mm *ManagerManager) GetShardManager(qtid dax.QualifiedTableID, partition dax.PartitionNum, shard dax.ShardNum) *Manager {
mm.mu.Lock()
defer mm.mu.Unlock()
key := shardK{qtid: qtid, partition: partition, shard: shard}
if m, ok := mm.shardManagers[key]; ok {
return m
}
mm.shardManagers[key] = (&Manager{
snapshotter: mm.Snapshotter,
writeLogger: mm.WriteLogger,
bucket: partitionBucket(qtid.Key(), partition),
key: shardKey(shard),
log: mm.Logger,
}).initialize()
return mm.shardManagers[key]
}
func (mm *ManagerManager) GetTableKeyManager(qtid dax.QualifiedTableID, partition dax.PartitionNum) *Manager {
mm.mu.Lock()
defer mm.mu.Unlock()
key := tableKeyK{qtid: qtid, partition: partition}
if m, ok := mm.tableKeyManagers[key]; ok {
return m
}
mm.tableKeyManagers[key] = (&Manager{
snapshotter: mm.Snapshotter,
writeLogger: mm.WriteLogger,
bucket: partitionBucket(qtid.Key(), partition),
key: keysFileName,
log: mm.Logger,
}).initialize()
return mm.tableKeyManagers[key]
}
func (mm *ManagerManager) GetFieldKeyManager(qtid dax.QualifiedTableID, field dax.FieldName) *Manager {
mm.mu.Lock()
defer mm.mu.Unlock()
key := fieldKeyK{qtid: qtid, field: field}
if m, ok := mm.fieldKeyManagers[key]; ok {
return m
}
mm.fieldKeyManagers[key] = (&Manager{
snapshotter: mm.Snapshotter,
writeLogger: mm.WriteLogger,
bucket: fieldBucket(qtid.Key(), field),
key: keysFileName,
log: mm.Logger,
}).initialize()
return mm.fieldKeyManagers[key]
}
// Manager wraps the snapshotter and writelogger to implement the
// ServerlessStorage interface and maintain messy state between
// calls. Manager is *not* threadsafe, care should be taken that
// concurrent calls are not made to Manager methods. The exception
// being that Snapshot and Append are safe to call concurrently.
type Manager struct {
snapshotter computer.SnapshotService
writeLogger computer.WriteLogService
bucket string
key string
log logger.Logger
loadWLsPastVersion int
latestWLVersion int
lastWLPos int
locked bool
}
func (m *Manager) initialize() *Manager {
m.loadWLsPastVersion = -2
m.latestWLVersion = -1
m.lastWLPos = -1
return m
}
func (m *Manager) LoadLatestSnapshot() (data io.ReadCloser, err error) {
snaps, err := m.snapshotter.List(m.bucket, m.key)
if err != nil {
return nil, errors.Wrap(err, "listing snapshots")
}
m.lastWLPos = 0
if len(snaps) == 0 {
m.loadWLsPastVersion = -1
return io.NopCloser(bytes.NewReader([]byte{})), nil
}
// assuming snapshots come back in sorted order
latest := snaps[len(snaps)-1]
m.loadWLsPastVersion = latest.Version
// TODO(jaffee): whatever is using the snapshot may discover that
// it is corrupted/incomplete. We don't want to separately check
// the checksum in here because then we'd have to read the whole
// snapshot twice. Need a way to catch the checksum error and tell
// Manager to mark that version as bad and remove it, then try
// LoadLatestSnapshot again.
return m.snapshotter.Read(m.bucket, m.key, latest.Version)
}
// // Potential future methods to support getting older versions. SnapInfo would have timestamp information as well.
//
// ListSnapshots() []SnapInfo
// LoadSnapshot(version int) (data io.ReadCloser, err error)
// LoadWriteLog can be called after LoadLatestSnapshot. It loads
// any writelog data which has been written since the latest
// snapshot. Subsequent calls to LoadWriteLog will only return new
// data that hasn't previously been returned from LoadWriteLog.
func (m *Manager) LoadWriteLog() (data io.ReadCloser, err error) {
if m.loadWLsPastVersion == -2 {
return nil, errors.New(errors.ErrUncoded, "LoadWriteLog called in inconsistent state, can't tell what version to load from")
}
wLogs, err := m.writeLogger.List(m.bucket, m.key)
versions := make([]int, 0, len(wLogs))
for _, log := range wLogs {
if log.Version > m.loadWLsPastVersion {
versions = append(versions, log.Version)
}
}
if len(versions) > 1 {
// TODO(jaffee) This can happen if there's a failure writing a
// snapshot. Need to implement a MultiReadCloser or similar
// that wraps all the latest write logs into one ReadCloser.
// It should only wrap the last one in a trackingReader.
return nil, errors.New(dax.ErrUnimplemented, "UNIMPLEMENTED: multiple write log versions ahead of latest snapshot.")
}
if len(versions) == 0 {
m.log.Debugf("LoadWriteLog: no logs after snapshot: %d on %s", m.loadWLsPastVersion, path.Join(m.bucket, m.key))
m.latestWLVersion = m.loadWLsPastVersion + 1
return io.NopCloser(bytes.NewReader([]byte{})), nil
}
if m.locked && m.latestWLVersion != versions[0] {
return nil, errors.New(errors.ErrUncoded, "write log version gone since locking")
}
m.latestWLVersion = versions[0]
r, err := m.writeLogger.LogReaderFrom(m.bucket, m.key, versions[0], m.lastWLPos)
if err != nil {
return nil, errors.Wrap(err, "getting writelog")
}
return &trackingReader{
r: r,
update: func(n int, err error) {
m.lastWLPos += n
},
}, nil
}
// Lock acquires an advisory lock for this resource which grants
// us exclusive access to write to it. The normal pattern is to
// call:
//
// 1. LoadLatestSnapshot
// 2. LoadWriteLog
// 3. Lock
// 4. LoadWriteLog
//
// The second call to LoadWriteLog is necessary in case any writes
// occurred between the last load and acquiring the lock. Once the
// lock is acquired it should not be possible for any more writes
// to occur. Lock will error if (a) we fail to acquire the lock or
// (b) the state of the snapshot store for this resource is not
// identical to what is was before the lock was acquired. Case (b)
// means that quite a lot has happened in between LoadWriteLog and
// Lock, and we should probably just die and start over.
func (m *Manager) Lock() error {
// lock is sort of arbitrarily on the write log interface
if err := m.writeLogger.Lock(m.bucket, m.key); err != nil {
return errors.Wrap(err, "acquiring lock")
}
m.locked = true
return nil
}
// Append appends the msg to the write log. It will fail if we
// haven't properly loaded and gotten a lock for the resource
// we're writing to.
func (m *Manager) Append(msg []byte) error {
if m.latestWLVersion < 0 {
return errors.New(errors.ErrUncoded, "can't call append before loading and locking write log")
}
return m.writeLogger.AppendMessage(m.bucket, m.key, m.latestWLVersion, msg)
}
// IncrementWLVersion should be called during snapshotting with a
// write Tx held on the local resource. This ensures that any
// writes which completed prior to the snapshot are in the prior
// WL and any that complete after the snapshot are in the
// incremented WL.
func (m *Manager) IncrementWLVersion() error {
m.latestWLVersion++
m.lastWLPos = -1
m.loadWLsPastVersion = -1
return nil
}
// Snapshot takes a ReadCloser which has the contents of the
// resource being tracked at a particular point in time and writes
// them to the Snapshot Store. Upon a successful write it will
// truncate any write logs which are now incorporated into the
// snapshot.
func (m *Manager) Snapshot(rc io.ReadCloser) error {
// latestWLVersion has already been incremented at this point, so
// we write that version minus 1.
err := m.snapshotter.Write(m.bucket, m.key, m.latestWLVersion-1, rc)
if err != nil {
return errors.Wrap(err, "writing snapshot")
}
err = m.writeLogger.DeleteLog(m.bucket, m.key, m.latestWLVersion-1)
return errors.Wrap(err, "deleting old write log")
}
// SnapshotTo is Snapshot's ugly stepsister supporting the weirdness
// of reading from translate stores who we're hoping to off in the
// next season.
func (m *Manager) SnapshotTo(wt io.WriterTo) error {
err := m.snapshotter.WriteTo(m.bucket, m.key, m.latestWLVersion-1, wt)
if err != nil {
return errors.Wrap(err, "writing snapshot SnapshotTo")
}
err = m.writeLogger.DeleteLog(m.bucket, m.key, m.latestWLVersion-1)
return errors.Wrap(err, "deleting old write log snapshotTo")
}
// Unlock releases the lock. This should be called if control of
// the underlying resource is being transitioned to another
// node. Ideally it's also called if the process crashes (e.g. via
// a defer), but an implementation based on filesystem locks
// should have those removed by the operating system when the
// process exits anyway.
func (m *Manager) Unlock() error {
if err := m.writeLogger.Unlock(m.bucket, m.key); err != nil {
return errors.Wrap(err, "unlocking")
}
m.locked = false
return nil
}
const (
keysFileName = "keys"
)
func partitionBucket(table dax.TableKey, partition dax.PartitionNum) string {
return path.Join(string(table), "partition", fmt.Sprintf("%d", partition))
}
func shardKey(shard dax.ShardNum) string {
return path.Join("shard", fmt.Sprintf("%d", shard))
}
func fieldBucket(table dax.TableKey, field dax.FieldName) string {
return path.Join(string(table), "field", string(field))
}

165
dax/storage/storage_test.go Normal file
View file

@ -0,0 +1,165 @@
package storage
import (
"bytes"
"io"
"os"
"testing"
"github.com/molecula/featurebase/v3/dax"
"github.com/molecula/featurebase/v3/dax/snapshotter"
"github.com/molecula/featurebase/v3/dax/writelogger"
"github.com/molecula/featurebase/v3/logger"
"github.com/stretchr/testify/assert"
)
func TestManagerManager(t *testing.T) {
sdd, err := os.MkdirTemp("", "snaptest*")
assert.NoError(t, err)
wdd, err := os.MkdirTemp("", "wltest*")
assert.NoError(t, err)
defer func() {
os.RemoveAll(sdd)
os.RemoveAll(wdd)
}()
sn := snapshotter.New(snapshotter.Config{
DataDir: sdd,
})
wl := writelogger.New(writelogger.Config{
DataDir: wdd,
})
mm := NewManagerManager(sn, wl, logger.NewStandardLogger(os.Stderr))
qtid := dax.QualifiedTableID{
TableQualifier: dax.TableQualifier{
OrganizationID: dax.OrganizationID("org1"),
DatabaseID: dax.DatabaseID("db1"),
},
ID: dax.TableID("blah"),
Name: "blah",
}
// get a manager and perform normal startup routine on empty data
mgr := mm.GetShardManager(qtid, dax.PartitionNum(1), dax.ShardNum(1))
d, err := mgr.LoadLatestSnapshot()
assert.NoError(t, err)
n, err := d.Read(make([]byte, 8))
assert.Equal(t, 0, n)
assert.Equal(t, err, io.EOF)
wld, err := mgr.LoadWriteLog()
assert.NoError(t, err)
n, err = wld.Read(make([]byte, 8))
assert.Equal(t, 0, n)
assert.Equal(t, io.EOF, err)
err = mgr.Lock()
assert.NoError(t, err)
wld, err = mgr.LoadWriteLog()
assert.NoError(t, err)
n, err = wld.Read(make([]byte, 8))
assert.Equal(t, 0, n)
assert.Equal(t, io.EOF, err)
// append some data
err = mgr.Append([]byte("blahblah"))
assert.NoError(t, err)
mm2 := NewManagerManager(sn, wl, logger.NewStandardLogger(os.Stderr))
// get second manager for same stuff
mgr2 := mm2.GetShardManager(qtid, dax.PartitionNum(1), dax.ShardNum(1))
// load snapshot on 2nd manager (empty)
d, err = mgr2.LoadLatestSnapshot()
assert.NoError(t, err)
n, err = d.Read(make([]byte, 8))
assert.Equal(t, 0, n)
assert.Equal(t, io.EOF, err)
// load WL on 2nd manager (blahblah)
wld, err = mgr2.LoadWriteLog()
assert.NoError(t, err)
buf := make([]byte, 16)
n, err = wld.Read(buf)
assert.Equal(t, 9, n)
assert.Equal(t, "blahblah\n", string(buf[:9]))
n, err = wld.Read(buf)
assert.Equal(t, 0, n)
assert.Equal(t, io.EOF, err)
// begin snapshot procedure on 1st manager
err = mgr.IncrementWLVersion()
assert.NoError(t, err)
// do append on 1st manager mid-snapshot
err = mgr.Append([]byte("blahbla2"))
assert.NoError(t, err)
// snapshot 1st manager
rc := io.NopCloser(bytes.NewBufferString("hahaha"))
err = mgr.Snapshot(rc)
assert.NoError(t, err)
// append again on 1st manager
err = mgr.Append([]byte("blahbla3"))
assert.NoError(t, err)
// locking 2nd manager should fail
err = mgr2.Lock()
assert.NotNil(t, err)
// exit 1st manager
err = mgr.Unlock()
assert.NoError(t, err)
// locking 2nd manager should succeed
err = mgr2.Lock()
assert.NoError(t, err)
// loading write log should fail since there's been a snapshot
// between the last load and locking.
wld, err = mgr2.LoadWriteLog()
assert.NotNil(t, err)
// mgr2 dies due to error loading write lock
err = mgr2.Unlock()
assert.NoError(t, err)
// get third manager for same stuff
mm3 := NewManagerManager(sn, wl, logger.NewStandardLogger(os.Stderr))
mgr3 := mm3.GetShardManager(qtid, dax.PartitionNum(1), dax.ShardNum(1))
// load snapshot on 3nd manager
d, err = mgr3.LoadLatestSnapshot()
assert.NoError(t, err)
buf = make([]byte, 6)
n, err = d.Read(buf)
assert.Equal(t, 6, n)
assert.Equal(t, "hahaha", string(buf))
assert.Equal(t, nil, err)
// load write log on 3rd manager, get previous 2 writes
wld, err = mgr3.LoadWriteLog()
assert.NoError(t, err)
buf = make([]byte, 20)
n, _ = wld.Read(buf)
assert.Equal(t, 18, n)
assert.Equal(t, "blahbla2\nblahbla3\n", string(buf[:18]))
n, err = wld.Read(buf)
assert.Equal(t, 0, n)
assert.Equal(t, io.EOF, err)
// lock 3rd manager
err = mgr3.Lock()
assert.NoError(t, err)
// reload write log (should be empty)
wld, err = mgr3.LoadWriteLog()
assert.NoError(t, err)
n, err = wld.Read(make([]byte, 8))
assert.Equal(t, 0, n)
assert.Equal(t, io.EOF, err)
}

24
dax/storage/util.go Normal file
View file

@ -0,0 +1,24 @@
package storage
import "io"
// trackingReader wraps a Reader and calls an custom "update" function
// whenever Read is called. Used by the storage layer to keep track of
// how much of the writelog has been read.
type trackingReader struct {
r io.Reader
update func(int, error)
}
func (tr *trackingReader) Read(p []byte) (n int, err error) {
n, err = tr.r.Read(p)
tr.update(n, err)
return n, err
}
func (tr *trackingReader) Close() error {
if closer, ok := tr.r.(io.Closer); ok {
return closer.Close()
}
return nil
}

120
dax/writelogger/writelogger.go
View file

@ -7,7 +7,9 @@ import (
"io/fs"
"os"
"path"
"strconv"
"sync"
"syscall"
"github.com/featurebasedb/featurebase/v3/errors"
"github.com/featurebasedb/featurebase/v3/logger"
@ -16,17 +18,19 @@ import (
type WriteLogger struct {
mu sync.RWMutex
dataDir string
logFiles map[string]*os.File
dataDir string
logFiles map[string]*os.File
lockFiles map[string]*os.File
logger logger.Logger
}
func New(cfg Config) *WriteLogger {
return &WriteLogger{
dataDir: cfg.DataDir,
logFiles: make(map[string]*os.File),
logger: logger.NopLogger,
dataDir: cfg.DataDir,
logFiles: make(map[string]*os.File),
lockFiles: make(map[string]*os.File),
logger: logger.NopLogger,
}
}
@ -43,25 +47,58 @@ func (w *WriteLogger) AppendMessage(bucket string, key string, version int, mess
return errors.Wrapf(err, "getting log file by key: %s", fKey)
}
logFile.Write(append(message, "\n"...))
logFile.Sync()
return nil
_, err = logFile.Write(append(message, "\n"...))
if err != nil {
return errors.Wrapf(err, "writing to log file %s", logFile.Name())
}
err = logFile.Sync()
return errors.Wrapf(err, "syncing log file %s", logFile.Name())
}
func (w *WriteLogger) LogReader(bucket string, key string, version int) (io.Reader, io.Closer, error) {
func (w *WriteLogger) List(bucket, key string) ([]computer.WriteLogInfo, error) {
dirpath := path.Join(w.dataDir, bucket, key)
entries, err := os.ReadDir(dirpath)
if err != nil {
if pe, ok := err.(*os.PathError); ok && pe.Err == syscall.ENOENT {
return nil, nil
}
return nil, errors.Wrap(err, "reading directory")
}
wLogs := make([]computer.WriteLogInfo, len(entries))
for i, entry := range entries {
version, err := strconv.ParseInt(entry.Name(), 10, 64)
if err != nil {
return nil, errors.Wrapf(err, "writelog filename '%s' could not be parsed to version number", entry.Name())
}
wLogs[i] = computer.WriteLogInfo{
Version: int(version),
}
}
return wLogs, nil
}
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) {
_, filePath := w.paths(fullKey(bucket, key, version))
f, err := os.Open(filePath)
if err != nil {
if e, ok := err.(*fs.PathError); ok {
return nil, nil, e
return nil, e
}
return nil, nil, err
return nil, err
}
if offset > 0 {
f.Seek(int64(offset), io.SeekStart)
}
w.logger.Debugf("WriteLogger LogReader file: %s", f.Name())
return f, f, nil
return f, nil
}
func (w *WriteLogger) DeleteLog(bucket string, key string, version int) error {
@ -84,6 +121,62 @@ func (w *WriteLogger) DeleteLog(bucket string, key string, version int) error {
return os.Remove(f.Name())
}
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 {
lockDir, lockFile := w.lockFile(bucket, key)
fmt.Println("lock dir:", lockDir)
fmt.Println("lock fil:", lockFile)
if err := os.MkdirAll(lockDir, 0777); err != nil {
return errors.Wrapf(err, "lock dir %s", lockDir)
}
f, err := os.OpenFile(lockFile, os.O_CREATE|os.O_EXCL, 0644)
if err != nil {
return errors.Wrapf(err, "opening lock file: %s", lockFile)
}
w.lockFiles[lockFile] = f
// fd, err = syscall.Open(lockFile, syscall.O_RDWR|syscall.O_CREAT, 0644)
// if err != nil {
// return 0, errors.Wrapf(err, "syscall opening %s", lockFile)
// }
// err = syscall.FcntlFlock(uintptr(fd), syscall.F_SETLK, &syscall.Flock_t{
// Type: syscall.F_WRLCK,
// })
return errors.Wrap(err, "locking")
}
func (w *WriteLogger) Unlock(bucket, key string) error {
// 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")
}
f.Close()
err := os.Remove(lockFile)
delete(w.lockFiles, lockFile)
// defer func() {
// err := syscall.Close(fd)
// if err != nil {
// w.logger.Printf("error closing lockfile %s", lockFile)
// }
// }()
// err := syscall.FcntlFlock(uintptr(fd), syscall.F_SETLK, &syscall.Flock_t{
// Type: syscall.F_UNLCK,
// })
return errors.Wrap(err, "closing lock file")
}
// 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).
@ -116,6 +209,7 @@ func (w *WriteLogger) logFileByKey(key string) (*os.File, error) {
if err != nil {
return nil, errors.Wrapf(err, "opening file: %s", filePath)
}
fmt.Printf("opened %s for key %s\n", filePath, key)
w.logFiles[key] = f
return f, nil

2
http_handler.go
View file

@ -2447,7 +2447,7 @@ func (h *Handler) handleGetIndexShardSnapshot(w http.ResponseWriter, r *http.Req
return
}
rc, err := h.api.IndexShardSnapshot(r.Context(), indexName, shard)
rc, err := h.api.IndexShardSnapshot(r.Context(), indexName, shard, false)
if err != nil {
switch errors.Cause(err) {
case ErrIndexNotFound:

30
server.go
View file

@ -97,9 +97,7 @@ type Server struct { // nolint: maligned
executionPlannerFn ExecutionPlannerFn
writeLogReader computer.WriteLogReader
writeLogWriter computer.WriteLogWriter
snapshotReadWriter computer.SnapshotReadWriter
serverlessStorage *daxstorage.ManagerManager
dataframeEnabled bool
}
@ -429,15 +427,6 @@ func OptServerPartitionAssigner(p string) ServerOption {
}
}
// OptServerWriteLogReader provides an implemenation of the WriteLogReader
// interface.
func OptServerWriteLogReader(wlr computer.WriteLogReader) ServerOption {
return func(s *Server) error {
s.writeLogReader = wlr
return nil
}
}
func OptServerExecutionPlannerFn(fn ExecutionPlannerFn) ServerOption {
return func(s *Server) error {
s.executionPlannerFn = fn
@ -445,20 +434,9 @@ func OptServerExecutionPlannerFn(fn ExecutionPlannerFn) ServerOption {
}
}
// OptServerWriteLogWriter provides an implemenation of the WriteLogWriter
// interface.
func OptServerWriteLogWriter(wlw computer.WriteLogWriter) ServerOption {
func OptServerServerlessStorage(mm *daxstorage.ManagerManager) ServerOption {
return func(s *Server) error {
s.writeLogWriter = wlw
return nil
}
}
// OptServerSnapshotReadWriter provides an implemenation of the
// SnapshotReadWriter interface.
func OptServerSnapshotReadWriter(snap computer.SnapshotReadWriter) ServerOption {
return func(s *Server) error {
s.snapshotReadWriter = snap
s.serverlessStorage = mm
return nil
}
}
@ -579,7 +557,7 @@ func NewServer(opts ...ServerOption) (*Server, error) {
s.cluster.disCo = s.disCo
s.cluster.noder = s.noder
s.cluster.sharder = s.sharder
s.cluster.writeLogWriter = s.writeLogWriter
s.cluster.serverlessStorage = s.serverlessStorage
s.cluster.versionStore = versionStore
// Append the NodeID tag to stats.

30
server/server.go
View file

@ -75,9 +75,10 @@ type Command struct {
logger loggerLogger
queryLogger loggerLogger
Registrar computer.Registrar
writeLogService computer.WriteLogService
snapshotService computer.SnapshotService
Registrar computer.Registrar
serverlessStorage *storage.ManagerManager
writeLogService computer.WriteLogService
snapshotService computer.SnapshotService
Handler pilosa.HandlerI
httpHandler http.Handler
@ -550,19 +551,8 @@ func (m *Command) setupServer() error {
m.Config.Etcd.Dir = filepath.Join(path, pilosa.DiscoDir)
}
// WriteLogger setup.
var wlw computer.WriteLogWriter = computer.NewNopWriteLogWriter()
var wlr computer.WriteLogReader = computer.NewNopWriteLogReader()
if m.writeLogService != nil {
wlrw := computer.NewWriteLogReadWriter(m.writeLogService)
wlr = wlrw
wlw = wlrw
}
// Snapshotter setup.
var snap computer.SnapshotReadWriter = computer.NewNopSnapshotReadWriter()
if m.snapshotService != nil {
snap = computer.NewSnapshotReadWriter(m.snapshotService)
if m.writeLogService != nil && m.snapshotService != nil {
m.serverlessStorage = storage.NewManagerManager(m.snapshotService, m.writeLogService, m.logger)
}
executionPlannerFn := func(e pilosa.Executor, api *pilosa.API, sql string) sql3.CompilePlanner {
@ -599,9 +589,7 @@ func (m *Command) setupServer() error {
pilosa.OptServerQueryHistoryLength(m.Config.QueryHistoryLength),
pilosa.OptServerPartitionAssigner(m.Config.Cluster.PartitionToNodeAssignment),
pilosa.OptServerExecutionPlannerFn(executionPlannerFn),
pilosa.OptServerWriteLogReader(wlr),
pilosa.OptServerWriteLogWriter(wlw),
pilosa.OptServerSnapshotReadWriter(snap),
pilosa.OptServerServerlessStorage(m.serverlessStorage),
pilosa.OptServerIsDataframeEnabled(m.Config.Dataframe.Enable),
}
@ -646,9 +634,7 @@ func (m *Command) setupServer() error {
m.API, err = pilosa.NewAPI(
pilosa.OptAPIServer(m.Server),
pilosa.OptAPIImportWorkerPoolSize(m.Config.ImportWorkerPoolSize),
pilosa.OptAPIWriteLogReader(wlr),
pilosa.OptAPIWriteLogWriter(wlw),
pilosa.OptAPISnapshotter(snap),
pilosa.OptAPIServerlessStorage(m.serverlessStorage),
pilosa.OptAPIDirectiveWorkerPoolSize(m.Config.DirectiveWorkerPoolSize),
pilosa.OptAPIIsComputeNode(m.isComputeNode),
)