diff --git a/api.go b/api.go index e3ed39af8..75f749454 100644 --- a/api.go +++ b/api.go @@ -3108,18 +3108,14 @@ func (api *API) SnapshotShardData(ctx context.Context, req *dax.SnapshotShardDat partition := disco.ShardToShardPartition(string(req.TableKey), uint64(req.ShardNum), disco.DefaultPartitionN) partitionNum := dax.PartitionNum(partition) - // 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. + // Open a write Tx snapshotting current version. rc, err := api.IndexShardSnapshot(ctx, string(req.TableKey), uint64(req.ShardNum), true) if err != nil { return errors.Wrap(err, "getting index/shard readcloser") } mgr := api.serverlessStorage.GetShardManager(qtid, partitionNum, req.ShardNum) + // Bump writelog version while write Tx is held. if err := mgr.IncrementWLVersion(); err != nil { return errors.Wrap(err, "incrementing write log version") } diff --git a/dax/computer/interfaces.go b/dax/computer/interfaces.go index 61ac6ef9c..d0e87af54 100644 --- a/dax/computer/interfaces.go +++ b/dax/computer/interfaces.go @@ -35,78 +35,6 @@ type SnapshotService interface { 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. -// TODO(jaffee), this doesn't need to be an interface. Remove. -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 @@ -116,70 +44,6 @@ type SnapInfo struct { // 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 { - WriteShardData(ctx context.Context, qtid dax.QualifiedTableID, partition dax.PartitionNum, shard dax.ShardNum, version int, rc io.ReadCloser) error - ReadShardData(ctx context.Context, qtid dax.QualifiedTableID, partition dax.PartitionNum, shard dax.ShardNum, version int) (io.ReadCloser, error) - - WriteTableKeys(ctx context.Context, qtid dax.QualifiedTableID, partition dax.PartitionNum, version int, wrTo io.WriterTo) error - ReadTableKeys(ctx context.Context, qtid dax.QualifiedTableID, partition dax.PartitionNum, version int) (io.ReadCloser, error) - - WriteFieldKeys(ctx context.Context, qtid dax.QualifiedTableID, field dax.FieldName, version int, wrTo io.WriterTo) error - ReadFieldKeys(ctx context.Context, qtid dax.QualifiedTableID, field dax.FieldName, version int) (io.ReadCloser, error) -} - -// WriteLogWriter provides the interface for all data writes to FeatureBase. After -// data has been written to the local FeatureBase node, the respective interface -// method(s) will be called. -type WriteLogWriter interface { - // CreateTableKeys sends a map of string key to uint64 ID for the table and - // partition provided. - CreateTableKeys(ctx context.Context, qtid dax.QualifiedTableID, partition dax.PartitionNum, version int, _ map[string]uint64) error - - // DeleteTableKeys deletes all table keys for the table and partition - // provided. - DeleteTableKeys(ctx context.Context, qtid dax.QualifiedTableID, partition dax.PartitionNum, version int) error - - // CreateFieldKeys sends a map of string key to uint64 ID for the table and - // field provided. - CreateFieldKeys(ctx context.Context, qtid dax.QualifiedTableID, field dax.FieldName, version int, _ map[string]uint64) error - - // DeleteTableKeys deletes all field keys for the table and field provided. - DeleteFieldKeys(ctx context.Context, qtid dax.QualifiedTableID, field dax.FieldName, version int) error - - // WriteShard sends shard data for the table and shard provided. - WriteShard(ctx context.Context, qtid dax.QualifiedTableID, partition dax.PartitionNum, shard dax.ShardNum, version int, msg LogMessage) error - - // DeleteShard deletes all data for the table and shard provided. - DeleteShard(ctx context.Context, qtid dax.QualifiedTableID, partition dax.PartitionNum, shard dax.ShardNum, version int) error -} - -// WriteLogReader provides the interface for all reads from the write log. -type WriteLogReader interface { - ShardReader(ctx context.Context, qtid dax.QualifiedTableID, partition dax.PartitionNum, shard dax.ShardNum, version int) ShardReader - TableKeyReader(ctx context.Context, qtid dax.QualifiedTableID, partition dax.PartitionNum, version int) TableKeyReader - FieldKeyReader(ctx context.Context, qtid dax.QualifiedTableID, field dax.FieldName, version int) FieldKeyReader -} - -type TableKeyReader interface { - Open() error - Read() (PartitionKeyMap, error) - Close() error -} - -type FieldKeyReader interface { - Open() error - Read() (FieldKeyMap, error) - Close() error -} - -type ShardReader interface { - Open() error - Read() (LogMessage, error) - Close() error -} - // LogMessage is implemented by a variety of types which can be serialized as // messages to the WriteLogger. type LogMessage interface{} diff --git a/dax/computer/noop.go b/dax/computer/noop.go deleted file mode 100644 index 01e6cc41c..000000000 --- a/dax/computer/noop.go +++ /dev/null @@ -1,162 +0,0 @@ -package computer - -import ( - "context" - "io" - - "github.com/molecula/featurebase/v3/dax" -) - -// Ensure type implements interface. -var _ WriteLogWriter = (*NopWriteLogWriter)(nil) - -// NopWriteLogWriter is a no-op implementation of the WriteLogWriter interface. -type NopWriteLogWriter struct{} - -func NewNopWriteLogWriter() *NopWriteLogWriter { - return &NopWriteLogWriter{} -} - -func (w *NopWriteLogWriter) CreateTableKeys(ctx context.Context, qtid dax.QualifiedTableID, partition dax.PartitionNum, version int, m map[string]uint64) error { - return nil -} - -func (w *NopWriteLogWriter) DeleteTableKeys(ctx context.Context, qtid dax.QualifiedTableID, partition dax.PartitionNum, version int) error { - return nil -} - -func (w *NopWriteLogWriter) CreateFieldKeys(ctx context.Context, qtid dax.QualifiedTableID, field dax.FieldName, version int, m map[string]uint64) error { - return nil -} - -func (w *NopWriteLogWriter) DeleteFieldKeys(ctx context.Context, qtid dax.QualifiedTableID, field dax.FieldName, version int) error { - return nil -} - -func (w *NopWriteLogWriter) WriteShard(ctx context.Context, qtid dax.QualifiedTableID, partition dax.PartitionNum, shard dax.ShardNum, version int, msg LogMessage) error { - return nil -} - -func (w *NopWriteLogWriter) DeleteShard(ctx context.Context, qtid dax.QualifiedTableID, partition dax.PartitionNum, shard dax.ShardNum, version int) error { - return nil -} - -// Ensure type implements interface. -var _ WriteLogReader = (*NopWriteLogReader)(nil) - -// NopWriteLogReader is a no-op implementation of the WriteLogReader interface. -type NopWriteLogReader struct{} - -func NewNopWriteLogReader() *NopWriteLogReader { - return &NopWriteLogReader{} -} - -func (w *NopWriteLogReader) TableKeyReader(ctx context.Context, qtid dax.QualifiedTableID, partition dax.PartitionNum, version int) TableKeyReader { - return NewNopTableKeyReader() -} - -func (w *NopWriteLogReader) FieldKeyReader(ctx context.Context, qtid dax.QualifiedTableID, field dax.FieldName, version int) FieldKeyReader { - return NewNopFieldKeyReader() -} - -func (w *NopWriteLogReader) ShardReader(ctx context.Context, qtid dax.QualifiedTableID, partition dax.PartitionNum, shard dax.ShardNum, version int) ShardReader { - return NewNopShardReader() -} - -//////////////////////////////////////////////// - -// Ensure type implements interface. -var _ TableKeyReader = &NopTableKeyReader{} - -// NopTableKeyReader is a no-op implementation of the TableKeyReader -// interface. -type NopTableKeyReader struct{} - -func NewNopTableKeyReader() *NopTableKeyReader { - return &NopTableKeyReader{} -} - -func (r *NopTableKeyReader) Open() error { return nil } -func (r *NopTableKeyReader) Read() (PartitionKeyMap, error) { - return PartitionKeyMap{}, io.EOF -} -func (r *NopTableKeyReader) Close() error { return nil } - -//////////////////////////////////////////////// - -// Ensure type implements interface. -var _ FieldKeyReader = &NopFieldKeyReader{} - -// NopFieldKeyReader is a no-op implementation of the FieldKeyReader -// interface. -type NopFieldKeyReader struct{} - -func NewNopFieldKeyReader() *NopFieldKeyReader { - return &NopFieldKeyReader{} -} - -func (r *NopFieldKeyReader) Open() error { return nil } -func (r *NopFieldKeyReader) Read() (FieldKeyMap, error) { - return FieldKeyMap{}, io.EOF -} -func (r *NopFieldKeyReader) Close() error { return nil } - -//////////////////////////////////////////////// - -// Ensure type implements interface. -var _ ShardReader = &NopShardReader{} - -// NopShardReader is a no-op implementation of the ShardReader interface. -type NopShardReader struct{} - -func NewNopShardReader() *NopShardReader { - return &NopShardReader{} -} - -func (r *NopShardReader) Open() error { return nil } -func (r *NopShardReader) Read() (LogMessage, error) { - return nil, io.EOF -} -func (r *NopShardReader) Close() error { return nil } - -////////////// SNAPSHOT //////////////////////// - -// Ensure type implements interface. -var _ SnapshotReadWriter = &NopSnapshotReadWriter{} - -// NopSnapshotReadWriter is a no-op implementation of the SnapshotReadWriter -// interface. -type NopSnapshotReadWriter struct{} - -func NewNopSnapshotReadWriter() *NopSnapshotReadWriter { - return &NopSnapshotReadWriter{} -} - -func (w *NopSnapshotReadWriter) WriteShardData(ctx context.Context, qtid dax.QualifiedTableID, partition dax.PartitionNum, shard dax.ShardNum, version int, rc io.ReadCloser) error { - return nil -} - -func (w *NopSnapshotReadWriter) ReadShardData(ctx context.Context, qtid dax.QualifiedTableID, partition dax.PartitionNum, shard dax.ShardNum, version int) (io.ReadCloser, error) { - return &nopReadCloser{}, nil -} - -func (w *NopSnapshotReadWriter) WriteTableKeys(ctx context.Context, qtid dax.QualifiedTableID, partition dax.PartitionNum, version int, wrTo io.WriterTo) error { - return nil -} - -func (w *NopSnapshotReadWriter) ReadTableKeys(ctx context.Context, qtid dax.QualifiedTableID, partition dax.PartitionNum, version int) (io.ReadCloser, error) { - return &nopReadCloser{}, nil -} - -func (w *NopSnapshotReadWriter) WriteFieldKeys(ctx context.Context, qtid dax.QualifiedTableID, field dax.FieldName, version int, wrTo io.WriterTo) error { - return nil -} - -func (w *NopSnapshotReadWriter) ReadFieldKeys(ctx context.Context, qtid dax.QualifiedTableID, field dax.FieldName, version int) (io.ReadCloser, error) { - return &nopReadCloser{}, nil -} - -type nopReadCloser struct{} - -func (n *nopReadCloser) Read([]byte) (int, error) { return 0, nil } -func (n *nopReadCloser) Close() error { return nil } diff --git a/dax/computer/snapshot.go b/dax/computer/snapshot.go deleted file mode 100644 index ac30f8b72..000000000 --- a/dax/computer/snapshot.go +++ /dev/null @@ -1,94 +0,0 @@ -package computer - -import ( - "context" - "io" - - "github.com/molecula/featurebase/v3/dax" - "github.com/molecula/featurebase/v3/errors" -) - -// Ensure type implements interface. -var _ SnapshotReadWriter = &snapshotReadWriter{} - -// snapshotReadWriter uses a SnapshotService implementation (which could be, for -// example, an http client or a locally running sub-service) to store its -// snapshots. -type snapshotReadWriter struct { - ss SnapshotService -} - -func NewSnapshotReadWriter(ss SnapshotService) *snapshotReadWriter { - return &snapshotReadWriter{ - ss: ss, - } -} - -func (s *snapshotReadWriter) WriteShardData(ctx context.Context, qtid dax.QualifiedTableID, partition dax.PartitionNum, shard dax.ShardNum, version int, rc io.ReadCloser) error { - bucket := partitionBucket(qtid.Key(), partition) - key := shardKey(shard) - - if err := s.ss.Write(bucket, key, version, rc); err != nil { - return errors.Wrapf(err, "writing shard data: %s, %d", key, version) - } - - return nil -} - -func (s *snapshotReadWriter) ReadShardData(ctx context.Context, qtid dax.QualifiedTableID, partition dax.PartitionNum, shard dax.ShardNum, version int) (io.ReadCloser, error) { - bucket := partitionBucket(qtid.Key(), partition) - key := shardKey(shard) - - rc, err := s.ss.Read(bucket, key, version) - if err != nil { - return nil, errors.Wrapf(err, "reading shard data: %s, %s, %d", bucket, key, version) - } - - return rc, nil -} - -func (s *snapshotReadWriter) WriteTableKeys(ctx context.Context, qtid dax.QualifiedTableID, partition dax.PartitionNum, version int, wrTo io.WriterTo) error { - bucket := partitionBucket(qtid.Key(), partition) - key := keysFileName - - if err := s.ss.WriteTo(bucket, key, version, wrTo); err != nil { - return errors.Wrapf(err, "writing table keys: %s, %d", key, version) - } - - return nil -} - -func (s *snapshotReadWriter) ReadTableKeys(ctx context.Context, qtid dax.QualifiedTableID, partition dax.PartitionNum, version int) (io.ReadCloser, error) { - bucket := partitionBucket(qtid.Key(), partition) - key := keysFileName - - rc, err := s.ss.Read(bucket, key, version) - if err != nil { - return nil, errors.Wrapf(err, "reading table keys: %s, %s, %d", bucket, key, version) - } - - return rc, nil -} - -func (s *snapshotReadWriter) WriteFieldKeys(ctx context.Context, qtid dax.QualifiedTableID, field dax.FieldName, version int, wrTo io.WriterTo) error { - bucket := fieldBucket(qtid.Key(), field) - key := keysFileName - - if err := s.ss.WriteTo(bucket, key, version, wrTo); err != nil { - return errors.Wrapf(err, "writing field keys: %s, %d", key, version) - } - - return nil -} - -func (s *snapshotReadWriter) ReadFieldKeys(ctx context.Context, qtid dax.QualifiedTableID, field dax.FieldName, version int) (io.ReadCloser, error) { - bucket := fieldBucket(qtid.Key(), field) - key := keysFileName - - rc, err := s.ss.Read(bucket, key, version) - if err != nil { - return nil, errors.Wrapf(err, "reading field keys: %s, %s, %d", bucket, key, version) - } - - return rc, nil -} diff --git a/dax/computer/writelog.go b/dax/computer/writelog.go deleted file mode 100644 index 9c01a2e72..000000000 --- a/dax/computer/writelog.go +++ /dev/null @@ -1,307 +0,0 @@ -package computer - -import ( - "bufio" - "context" - "encoding/json" - "io" - - "github.com/molecula/featurebase/v3/dax" - "github.com/molecula/featurebase/v3/errors" -) - -// Ensure type implements interface. -var _ WriteLogReader = &writeLogReadWriter{} -var _ WriteLogWriter = &writeLogReadWriter{} - -// writeLogReadWriter is an implementation of the WriteLogReader and WriteLogWriter -// interfaces. It uses a WriteLogService implementation (which could be, for -// example, an http client or a locally running sub-service) to store its log -// messages. -type writeLogReadWriter struct { - wls WriteLogService -} - -func NewWriteLogReadWriter(wls WriteLogService) *writeLogReadWriter { - return &writeLogReadWriter{ - wls: wls, - } -} - -func (w *writeLogReadWriter) CreateTableKeys(ctx context.Context, qtid dax.QualifiedTableID, partition dax.PartitionNum, version int, m map[string]uint64) error { - msg := PartitionKeyMap{ - TableKey: qtid.Key(), - Partition: partition, - StringToID: m, - } - - b, err := json.Marshal(msg) - if err != nil { - return errors.Wrap(err, "marshalling partition key map to json") - } - - bucket := partitionBucket(qtid.Key(), partition) - - if err := w.wls.AppendMessage(bucket, keysFileName, version, b); err != nil { - return errors.Wrapf(err, "appending partition key message: %s, %d", keysFileName, version) - } - - return nil -} - -func (w *writeLogReadWriter) DeleteTableKeys(ctx context.Context, qtid dax.QualifiedTableID, partition dax.PartitionNum, version int) error { - bucket := partitionBucket(qtid.Key(), partition) - return w.wls.DeleteLog(bucket, keysFileName, version) -} - -func (w *writeLogReadWriter) CreateFieldKeys(ctx context.Context, qtid dax.QualifiedTableID, field dax.FieldName, version int, m map[string]uint64) error { - msg := FieldKeyMap{ - TableKey: qtid.Key(), - Field: field, - StringToID: m, - } - - b, err := json.Marshal(msg) - if err != nil { - return errors.Wrap(err, "marshalling field key map to json") - } - - bucket := fieldBucket(qtid.Key(), field) - - if err := w.wls.AppendMessage(bucket, keysFileName, version, b); err != nil { - return errors.Wrapf(err, "appending field key message: %s, %d", keysFileName, version) - } - - return nil -} - -func (w *writeLogReadWriter) DeleteFieldKeys(ctx context.Context, qtid dax.QualifiedTableID, field dax.FieldName, version int) error { - bucket := fieldBucket(qtid.Key(), field) - return w.wls.DeleteLog(bucket, keysFileName, version) -} - -func (w *writeLogReadWriter) WriteShard(ctx context.Context, qtid dax.QualifiedTableID, partition dax.PartitionNum, shard dax.ShardNum, version int, msg LogMessage) error { - b, err := MarshalLogMessage(msg, EncodeTypeJSON) - if err != nil { - return errors.Wrap(err, "marshalling log message") - } - - bucket := partitionBucket(qtid.Key(), partition) - shardKey := shardKey(shard) - - if err := w.wls.AppendMessage(bucket, shardKey, version, b); err != nil { - return errors.Wrapf(err, "appending shard key message: %s, %d", shardKey, version) - } - - return nil -} - -func (w *writeLogReadWriter) DeleteShard(ctx context.Context, qtid dax.QualifiedTableID, partition dax.PartitionNum, shard dax.ShardNum, version int) error { - bucket := partitionBucket(qtid.Key(), partition) - shardKey := shardKey(shard) - - return w.wls.DeleteLog(bucket, shardKey, version) -} - -//////////////////////////////////////////////// - -func (w *writeLogReadWriter) TableKeyReader(ctx context.Context, qtid dax.QualifiedTableID, partition dax.PartitionNum, version int) TableKeyReader { - return newTableKeyReader(w.wls, qtid, partition, version) -} - -type tableKeyReader struct { - wl WriteLogService - table dax.TableKey - partition dax.PartitionNum - version int - scanner *bufio.Scanner - closer io.Closer -} - -func newTableKeyReader(wl WriteLogService, qtid dax.QualifiedTableID, partition dax.PartitionNum, version int) *tableKeyReader { - r := &tableKeyReader{ - wl: wl, - table: qtid.Key(), - partition: partition, - version: version, - } - - return r -} - -func (r *tableKeyReader) Open() error { - bucket := partitionBucket(r.table, r.partition) - - 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 = readcloser - r.scanner = bufio.NewScanner(readcloser) - - return nil -} - -func (r *tableKeyReader) Read() (PartitionKeyMap, error) { - if r.scanner == nil { - return PartitionKeyMap{}, io.EOF - } - - var b []byte - var out PartitionKeyMap - - if r.scanner.Scan() { - b = r.scanner.Bytes() - if err := json.Unmarshal(b, &out); err != nil { - return out, err - } - return out, nil - } - if err := r.scanner.Err(); err != nil { - return out, err - } - - return out, io.EOF -} - -func (r *tableKeyReader) Close() error { - if r.closer != nil { - return r.closer.Close() - } - return nil -} - -//////////////////////////////////////////////// - -func (w *writeLogReadWriter) FieldKeyReader(ctx context.Context, qtid dax.QualifiedTableID, field dax.FieldName, version int) FieldKeyReader { - return newFieldKeyReader(w.wls, qtid, field, version) -} - -type fieldKeyReader struct { - wl WriteLogService - table dax.TableKey - field dax.FieldName - version int - scanner *bufio.Scanner - closer io.Closer -} - -func newFieldKeyReader(wl WriteLogService, qtid dax.QualifiedTableID, field dax.FieldName, version int) *fieldKeyReader { - r := &fieldKeyReader{ - wl: wl, - table: qtid.Key(), - field: field, - version: version, - } - - return r -} - -func (r *fieldKeyReader) Open() error { - bucket := fieldBucket(r.table, r.field) - - 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 = readcloser - r.scanner = bufio.NewScanner(readcloser) - - return nil -} - -func (r *fieldKeyReader) Read() (FieldKeyMap, error) { - if r.scanner == nil { - return FieldKeyMap{}, io.EOF - } - - var b []byte - var out FieldKeyMap - - if r.scanner.Scan() { - b = r.scanner.Bytes() - if err := json.Unmarshal(b, &out); err != nil { - return out, err - } - return out, nil - } - if err := r.scanner.Err(); err != nil { - return out, err - } - - return out, io.EOF -} - -func (r *fieldKeyReader) Close() error { - if r.closer != nil { - return r.closer.Close() - } - return nil -} - -//////////////////////////////////////////////// - -func (w *writeLogReadWriter) ShardReader(ctx context.Context, qtid dax.QualifiedTableID, partition dax.PartitionNum, shard dax.ShardNum, version int) ShardReader { - return newShardReader(w.wls, qtid, partition, shard, version) -} - -type shardReader struct { - wl WriteLogService - table dax.TableKey - partition dax.PartitionNum - shard dax.ShardNum - version int - scanner *bufio.Scanner - closer io.Closer -} - -func newShardReader(wl WriteLogService, qtid dax.QualifiedTableID, partition dax.PartitionNum, shard dax.ShardNum, version int) *shardReader { - r := &shardReader{ - wl: wl, - table: qtid.Key(), - partition: partition, - shard: shard, - version: version, - } - - return r -} - -func (r *shardReader) Open() error { - bucket := partitionBucket(r.table, r.partition) - shardKey := shardKey(r.shard) - - 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 = readcloser - r.scanner = bufio.NewScanner(readcloser) - - return nil -} - -func (r *shardReader) Read() (LogMessage, error) { - if r.scanner == nil { - return nil, io.EOF - } - - if r.scanner.Scan() { - return UnmarshalLogMessage(r.scanner.Bytes()) - } - if err := r.scanner.Err(); err != nil { - return nil, err - } - - return nil, io.EOF -} - -func (r *shardReader) Close() error { - if r.closer != nil { - return r.closer.Close() - } - return nil -} diff --git a/dax/snapshotter/api/openapi.yaml b/dax/snapshotter/api/openapi.yaml deleted file mode 100644 index bfb5072ed..000000000 --- a/dax/snapshotter/api/openapi.yaml +++ /dev/null @@ -1,101 +0,0 @@ -openapi: 3.0.3 - -info: - title: Snapshotter - description: The alpha implementation of the Snapshotter interface. - version: 0.0.0 - -paths: - /snapshotter/health: - get: - summary: Health check endpoint. - description: Provides an endpoint to check the overall health of the Snapshotter service. - operationId: GetHealth - responses: - 200: - description: Service is healthy. - - - /snapshotter/write-snapshot: - post: - summary: Write snapshot. - description: Write snapshot based on bucket/key. - operationId: PostWriteSnapshot - parameters: - - name: bucket - in: query - description: bucket containing snapshot key - required: true - schema: - type: string - - name: key - in: query - description: key identifying snapshot - required: true - schema: - type: string - - name: version - in: query - description: bucket/key version - required: true - schema: - type: integer - format: int64 - requestBody: - content: - text/plain: - schema: - type: string - format: byte - responses: - 200: - $ref: '#/components/responses/WriteSnapshotResponse' - - /snapshotter/read-snapshot: - get: - summary: Read snapshot. - description: Read snapshot based on bucket/key. - operationId: GetReadSnapshot - parameters: - - name: bucket - in: query - description: bucket containing snapshot key - required: true - schema: - type: string - - name: key - in: query - description: key identifying snapshot - required: true - schema: - type: string - - name: version - in: query - description: bucket/key version - required: true - schema: - type: integer - format: int64 - requestBody: - content: - text/plain: - schema: - type: string - format: byte - responses: - 200: - description: Bytes making up the contents of the snapshot. - content: - text/plain: - schema: - type: string - format: byte - -components: - responses: - WriteSnapshotResponse: - description: Placeholder response. - content: - application/json: - schema: - type: object \ No newline at end of file diff --git a/dax/snapshotter/client/client.go b/dax/snapshotter/client/client.go deleted file mode 100644 index 863bb240d..000000000 --- a/dax/snapshotter/client/client.go +++ /dev/null @@ -1,111 +0,0 @@ -// Package client contains an http implementation of the WriteLogger client. -package client - -// import ( -// "bytes" -// "encoding/json" -// "fmt" -// "io" -// "net/http" -// "net/url" - -// "github.com/molecula/featurebase/v3/dax" -// snapshotterhttp "github.com/molecula/featurebase/v3/dax/snapshotter/http" -// "github.com/molecula/featurebase/v3/errors" -// ) - -// const defaultScheme = "http" - -// // TODO(jaffee): remove this? - -// // Snapshotter is a client for the Snapshotter API methods. -// type Snapshotter struct { -// address dax.Address -// } - -// func New(address dax.Address) *Snapshotter { -// return &Snapshotter{ -// address: address, -// } -// } - -// func (s *Snapshotter) Write(bucket string, key string, version int, rc io.ReadCloser) error { -// url := fmt.Sprintf("%s/snapshotter/write-snapshot?bucket=%s&key=%s&version=%d", -// s.address.WithScheme(defaultScheme), -// url.QueryEscape(bucket), -// url.QueryEscape(key), -// version, -// ) - -// // Post the request. -// resp, err := http.Post(url, "", rc) -// if err != nil { -// return errors.Wrap(err, "posting write-snapshot") -// } -// defer resp.Body.Close() - -// if resp.StatusCode != http.StatusOK { -// b, _ := io.ReadAll(resp.Body) -// return errors.Errorf("status code: %d: %s", resp.StatusCode, b) -// } - -// var wsr snapshotterhttp.WriteSnapshotResponse -// if err := json.NewDecoder(resp.Body).Decode(&wsr); err != nil { -// return errors.Wrap(err, "reading response body") -// } - -// return nil -// } - -// // WriteTo is exactly the same as Write, except that it takes an io.WriteTo -// // instead of an io.ReadCloser. This needs to be cleaned up so that we're only -// // using one or the other. -// func (s *Snapshotter) WriteTo(bucket string, key string, version int, wrTo io.WriterTo) error { -// url := fmt.Sprintf("%s/snapshotter/write-snapshot?bucket=%s&key=%s&version=%d", -// s.address.WithScheme(defaultScheme), -// url.QueryEscape(bucket), -// url.QueryEscape(key), -// version, -// ) - -// buf := &bytes.Buffer{} -// if _, err := wrTo.WriteTo(buf); err != nil { -// return errors.Wrap(err, "writing to buffer") -// } - -// // Post the request. -// resp, err := http.Post(url, "", buf) -// if err != nil { -// return errors.Wrap(err, "posting write-snapshot") -// } -// defer resp.Body.Close() - -// if resp.StatusCode != http.StatusOK { -// b, _ := io.ReadAll(resp.Body) -// return errors.Errorf("status code: %d: %s", resp.StatusCode, b) -// } - -// var wsr snapshotterhttp.WriteSnapshotResponse -// if err := json.NewDecoder(resp.Body).Decode(&wsr); err != nil { -// return errors.Wrap(err, "reading response body") -// } - -// return nil -// } - -// func (s *Snapshotter) Read(bucket string, key string, version int) (io.ReadCloser, error) { -// url := fmt.Sprintf("%s/snapshotter/read-snapshot?bucket=%s&key=%s&version=%d", -// s.address.WithScheme(defaultScheme), -// url.QueryEscape(bucket), -// url.QueryEscape(key), -// version, -// ) - -// // Get the request. -// resp, err := http.Get(url) -// if err != nil { -// return nil, errors.Wrap(err, "getting read-snapshot") -// } - -// return resp.Body, nil -// } diff --git a/dax/snapshotter/http/handler.go b/dax/snapshotter/http/handler.go deleted file mode 100644 index 84a93f17b..000000000 --- a/dax/snapshotter/http/handler.go +++ /dev/null @@ -1,120 +0,0 @@ -package http - -// import ( -// "encoding/json" -// "io" -// "net/http" -// "strconv" - -// "github.com/gorilla/mux" -// "github.com/molecula/featurebase/v3/dax/snapshotter" -// "github.com/molecula/featurebase/v3/rbf" -// ) - -// func Handler(s *snapshotter.Snapshotter) http.Handler { -// svr := &server{ -// snapshotter: s, -// } - -// router := mux.NewRouter() -// router.HandleFunc("/health", svr.getHealth).Methods("GET").Name("GetHealth") -// router.HandleFunc("/write-snapshot", svr.postWriteSnapshot).Methods("POST").Name("PostWriteSnapshot") -// router.HandleFunc("/read-snapshot", svr.getReadSnapshot).Methods("GET").Name("GetReadSnapshot") -// return router -// } - -// type server struct { -// snapshotter *snapshotter.Snapshotter -// } - -// // GET /health -// func (s *server) getHealth(w http.ResponseWriter, r *http.Request) { -// w.WriteHeader(http.StatusOK) -// } - -// // POST /write-snapshot -// func (s *server) postWriteSnapshot(w http.ResponseWriter, r *http.Request) { -// bucket := r.URL.Query().Get("bucket") -// if bucket == "" { -// http.Error(w, "bucket required", http.StatusBadRequest) -// return -// } - -// key := r.URL.Query().Get("key") -// if key == "" { -// http.Error(w, "key required", http.StatusBadRequest) -// return -// } - -// versionArg := r.URL.Query().Get("version") -// versionInt64, err := strconv.ParseInt(versionArg, 10, 64) -// if err != nil { -// http.Error(w, "bad shard", http.StatusBadRequest) -// return -// } -// version := int(versionInt64) - -// body := r.Body -// defer body.Close() - -// if err := s.snapshotter.Write(bucket, key, version, body); err != nil { -// http.Error(w, err.Error(), http.StatusBadRequest) -// return -// } - -// resp := &WriteSnapshotResponse{} - -// if err := json.NewEncoder(w).Encode(resp); err != nil { -// http.Error(w, err.Error(), http.StatusBadRequest) -// return -// } -// } - -// type WriteSnapshotResponse struct{} - -// // GET /read-snapshot -// func (s *server) getReadSnapshot(w http.ResponseWriter, r *http.Request) { -// bucket := r.URL.Query().Get("bucket") -// if bucket == "" { -// http.Error(w, "bucket required", http.StatusBadRequest) -// return -// } - -// key := r.URL.Query().Get("key") -// if key == "" { -// http.Error(w, "key required", http.StatusBadRequest) -// return -// } - -// versionArg := r.URL.Query().Get("version") -// versionInt64, err := strconv.ParseInt(versionArg, 10, 64) -// if err != nil { -// http.Error(w, "bad shard", http.StatusBadRequest) -// return -// } -// version := int(versionInt64) - -// rc, err := s.snapshotter.Read(bucket, key, version) -// if err != nil { -// http.Error(w, err.Error(), http.StatusBadRequest) -// return -// } -// defer rc.Close() - -// // TODO: is rbf.PageSize a problem here for non-RBF snapshots (i.e. keys)? -// // Copy data to response body. -// if _, err := io.CopyBuffer(&passthroughWriter{w}, rc, make([]byte, rbf.PageSize)); err != nil { -// http.Error(w, err.Error(), http.StatusInternalServerError) -// return -// } -// } - -// // passthroughWriter is used to remove non-Writer interfaces from an io.Writer. -// // For example, a writer that implements io.ReaderFrom can change io.Copy() behavior. -// type passthroughWriter struct { -// w io.Writer -// } - -// func (w *passthroughWriter) Write(p []byte) (int, error) { -// return w.w.Write(p) -// } diff --git a/dax/writelogger/api/openapi.yaml b/dax/writelogger/api/openapi.yaml deleted file mode 100644 index d82fe6820..000000000 --- a/dax/writelogger/api/openapi.yaml +++ /dev/null @@ -1,113 +0,0 @@ -openapi: 3.0.3 - -info: - title: WriteLogger - description: The alpha implementation of the WriteLogger interface. - version: 0.0.0 - -paths: - /writelogger/health: - get: - summary: Health check endpoint. - description: Provides an endpoint to check the overall health of the WriteLogger service. - operationId: GetHealth - responses: - 200: - description: Service is healthy. - - - /writelogger/append-message: - post: - summary: Append message to WriteLogger. - description: Appends a message to a versioned bucket/key. - operationId: PostAppendMessage - requestBody: - content: - application/json: - example: - bucket: example-bucket - key: unique-key - version: 4 - message: SGVsbG8gV29ybGQ= - schema: - type: object - properties: - bucket: - type: string - key: - type: string - version: - type: integer - format: int64 - message: - type: string - format: byte - responses: - 200: - $ref: '#/components/responses/AppendMessageResponse' - - /writelogger/log-reader: - post: - summary: Read log. - description: Reads an entire log (collection of messages) at bucket/key for the given version. - operationId: PostLogReader - requestBody: - content: - application/json: - example: - bucket: example-bucket - key: unique-key - version: 4 - schema: - type: object - properties: - bucket: - type: string - key: - type: string - version: - type: integer - format: int64 - responses: - 200: - description: Bytes making up the contents of the log. - content: - text/plain: - schema: - type: string - format: byte - - /writelogger/delete-log: - post: - summary: Delete log. - description: Deletes the log at bucket/key for the given version. - operationId: PostDeleteLog - requestBody: - content: - application/json: - example: - bucket: example-bucket - key: unique-key - version: 4 - schema: - type: object - properties: - bucket: - type: string - key: - type: string - version: - type: integer - format: int64 - responses: - 200: - description: Log was deleted. - -components: - responses: - AppendMessageResponse: - description: Placeholder response. - content: - application/json: - schema: - type: object \ No newline at end of file diff --git a/dax/writelogger/client/client.go b/dax/writelogger/client/client.go deleted file mode 100644 index 25a2a3469..000000000 --- a/dax/writelogger/client/client.go +++ /dev/null @@ -1,147 +0,0 @@ -// Package client contains an http implementation of the WriteLogger client. -package client - -// import ( -// "bytes" -// "encoding/json" -// "fmt" -// "io" -// "net/http" - -// "github.com/molecula/featurebase/v3/dax" -// "github.com/molecula/featurebase/v3/errors" -// ) - -// // TODO(jaffee): remove this? - -// const defaultScheme = "http" - -// // WriteLogger is a client for the WriteLogger API methods. -// type WriteLogger struct { -// address dax.Address -// } - -// func New(address dax.Address) *WriteLogger { -// return &WriteLogger{ -// address: address, -// } -// } - -// func (w *WriteLogger) AppendMessage(bucket string, key string, version int, msg []byte) error { -// url := fmt.Sprintf("%s/writelogger/append-message", w.address.WithScheme(defaultScheme)) - -// req := &AppendMessageRequest{ -// Bucket: bucket, -// Key: key, -// Version: version, -// Message: msg, -// } - -// // Encode the request. -// postBody, err := json.Marshal(req) -// if err != nil { -// return errors.Wrap(err, "marshalling post request") -// } -// requestBody := bytes.NewBuffer(postBody) - -// // Post the request. -// resp, err := http.Post(url, "application/json", requestBody) -// if err != nil { -// return errors.Wrap(err, "posting append-message 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) -// } - -// var isr *AppendMessageResponse -// if err := json.NewDecoder(resp.Body).Decode(&isr); err != nil { -// return errors.Wrap(err, "reading response body") -// } - -// return nil -// } - -// type AppendMessageRequest struct { -// Bucket string `json:"bucket"` -// Key string `json:"key"` -// Version int `json:"version"` -// Message []byte `json:"message"` -// } -// type AppendMessageResponse struct{} - -// func (w *WriteLogger) LogReader(bucket string, key string, version int) (io.Reader, io.Closer, error) { -// url := fmt.Sprintf("%s/writelogger/log-reader", w.address.WithScheme(defaultScheme)) - -// req := &LogReaderRequest{ -// Bucket: bucket, -// Version: version, -// Key: key, -// } - -// // Encode the request. -// postBody, err := json.Marshal(req) -// if err != nil { -// return nil, nil, errors.Wrap(err, "marshalling post request") -// } -// requestBody := bytes.NewBuffer(postBody) - -// // Post the request. -// resp, err := http.Post(url, "application/json", requestBody) -// if err != nil { -// return nil, nil, errors.Wrap(err, "posting log-reader request") -// } - -// if resp.StatusCode != http.StatusOK { -// b, _ := io.ReadAll(resp.Body) -// defer resp.Body.Close() -// return nil, nil, errors.Errorf("status code: %d: %s", resp.StatusCode, b) -// } - -// return resp.Body, resp.Body, nil -// } - -// type LogReaderRequest struct { -// Bucket string `json:"bucket"` -// Version int `json:"version"` -// Key string `json:"key"` -// } - -// func (w *WriteLogger) DeleteLog(bucket string, key string, version int) error { -// url := fmt.Sprintf("%s/writelogger/delete-log", w.address.WithScheme(defaultScheme)) - -// req := &DeleteLogRequest{ -// Bucket: bucket, -// Version: version, -// Key: key, -// } - -// // Encode the request. -// postBody, err := json.Marshal(req) -// if err != nil { -// return errors.Wrap(err, "marshalling post request") -// } -// requestBody := bytes.NewBuffer(postBody) - -// // Post the request. -// resp, err := http.Post(url, "application/json", requestBody) -// if err != nil { -// return errors.Wrap(err, "posting log-reader request") -// } - -// if resp.StatusCode != http.StatusOK { -// b, _ := io.ReadAll(resp.Body) -// defer resp.Body.Close() -// return errors.Errorf("status code: %d: %s", resp.StatusCode, b) -// } - -// return nil -// } - -// type DeleteLogRequest struct { -// Bucket string `json:"bucket"` -// Version int `json:"version"` -// Key string `json:"key"` -// } diff --git a/dax/writelogger/http/handler.go b/dax/writelogger/http/handler.go deleted file mode 100644 index 7c2fe7fc1..000000000 --- a/dax/writelogger/http/handler.go +++ /dev/null @@ -1,121 +0,0 @@ -package http - -// import ( -// "encoding/json" -// "io" -// "net/http" - -// "github.com/gorilla/mux" -// "github.com/molecula/featurebase/v3/dax/writelogger" -// "github.com/molecula/featurebase/v3/logger" -// ) - -// func Handler(w *writelogger.WriteLogger, logger logger.Logger) http.Handler { -// svr := &server{ -// writeLogger: w, -// logger: logger, -// } - -// router := mux.NewRouter() -// router.HandleFunc("/health", svr.getHealth).Methods("GET").Name("GetHealth") -// router.HandleFunc("/append-message", svr.postAppendMessage).Methods("POST").Name("PostAppendMessage") -// router.HandleFunc("/log-reader", svr.postLogReader).Methods("POST").Name("PostLogReader") -// router.HandleFunc("/delete-log", svr.postDeleteLog).Methods("POST").Name("PostDeleteLog") -// return router -// } - -// type server struct { -// writeLogger *writelogger.WriteLogger -// logger logger.Logger -// } - -// // GET /health -// func (s *server) getHealth(w http.ResponseWriter, r *http.Request) { -// w.WriteHeader(http.StatusOK) -// } - -// // POST /append-message -// func (s *server) postAppendMessage(w http.ResponseWriter, r *http.Request) { -// body := r.Body -// defer body.Close() - -// req := AppendMessageRequest{} -// if err := json.NewDecoder(body).Decode(&req); err != nil { -// http.Error(w, err.Error(), http.StatusBadRequest) -// return -// } - -// err := s.writeLogger.AppendMessage(req.Bucket, req.Key, req.Version, req.Message) -// if err != nil { -// http.Error(w, err.Error(), http.StatusBadRequest) -// return -// } - -// resp := AppendMessageResponse{} -// if err := json.NewEncoder(w).Encode(resp); err != nil { -// http.Error(w, err.Error(), http.StatusBadRequest) -// return -// } -// } - -// type AppendMessageRequest struct { -// Bucket string `json:"bucket"` -// Key string `json:"key"` -// Version int `json:"version"` -// Message []byte `json:"message"` -// } - -// type AppendMessageResponse struct{} - -// // POST /log-reader -// func (s *server) postLogReader(w http.ResponseWriter, r *http.Request) { -// body := r.Body -// defer body.Close() - -// req := LogReaderRequest{} -// if err := json.NewDecoder(body).Decode(&req); err != nil { -// http.Error(w, err.Error(), http.StatusBadRequest) -// return -// } - -// reader, closer, err := s.writeLogger.LogReader(req.Bucket, req.Key, req.Version) -// if err != nil { -// http.Error(w, err.Error(), http.StatusBadRequest) -// return -// } - -// defer closer.Close() - -// if _, err := io.Copy(w, reader); err != nil { -// s.logger.Printf("error streaming log data: %s", err) -// } -// } - -// type LogReaderRequest struct { -// Bucket string `json:"bucket"` -// Version int `json:"version"` -// Key string `json:"key"` -// } - -// // POST /delete-log -// func (s *server) postDeleteLog(w http.ResponseWriter, r *http.Request) { -// body := r.Body -// defer body.Close() - -// req := DeleteLogRequest{} -// if err := json.NewDecoder(body).Decode(&req); err != nil { -// http.Error(w, err.Error(), http.StatusBadRequest) -// return -// } - -// if err := s.writeLogger.DeleteLog(req.Bucket, req.Key, req.Version); err != nil { -// http.Error(w, err.Error(), http.StatusBadRequest) -// return -// } -// } - -// type DeleteLogRequest struct { -// Bucket string `json:"bucket"` -// Version int `json:"version"` -// Key string `json:"key"` -// }