[#1329] services/tree: Implement GetOpLog RPC

Signed-off-by: Evgenii Stratonikov <evgeniy@nspcc.ru>
This commit is contained in:
Evgenii Stratonikov 2022-05-11 16:29:04 +03:00
parent b19de6116f
commit 9cbd4271f1
11 changed files with 195 additions and 3 deletions

View file

@ -116,3 +116,22 @@ func (e *StorageEngine) TreeGetChildren(cid cidSDK.ID, treeID string, nodeID pil
}
return nil, err
}
// TreeGetOpLog implements the pilorama.Forest interface.
func (e *StorageEngine) TreeGetOpLog(cid cidSDK.ID, treeID string, height uint64) (pilorama.Move, error) {
var err error
var lm pilorama.Move
for _, sh := range e.sortShardsByWeight(cid) {
lm, err = sh.TreeGetOpLog(cid, treeID, height)
if err != nil {
e.log.Debug("can't perform `GetOpLog`",
zap.Stringer("cid", cid),
zap.String("tree", treeID),
zap.Uint64("height", height),
zap.String("err", err.Error()))
continue
}
return lm, nil
}
return lm, err
}

View file

@ -211,7 +211,7 @@ func (t *boltForest) applyOperation(logBucket, treeBucket *bbolt.Bucket, m *Move
// 1. Undo up until the desired timestamp is here.
for len(key) == 8 && binary.BigEndian.Uint64(key) > m.Time {
if err := t.logFromBytes(&tmp, key, value); err != nil {
if err := t.logFromBytes(&tmp, value); err != nil {
return nil, err
}
if err := t.undo(&tmp.Move, &tmp, treeBucket, cKey[:]); err != nil {
@ -231,7 +231,7 @@ func (t *boltForest) applyOperation(logBucket, treeBucket *bbolt.Bucket, m *Move
// 3. Re-apply all other operations.
for len(key) == 8 {
if err := t.logFromBytes(&tmp, key, value); err != nil {
if err := t.logFromBytes(&tmp, value); err != nil {
return nil, err
}
if err := t.do(logBucket, treeBucket, cKey[:], &tmp); err != nil {
@ -446,6 +446,29 @@ func (t *boltForest) TreeGetChildren(cid cidSDK.ID, treeID string, nodeID Node)
return children, err
}
// TreeGetOpLog implements the pilorama.Forest interface.
func (t *boltForest) TreeGetOpLog(cid cidSDK.ID, treeID string, height uint64) (Move, error) {
key := make([]byte, 8)
binary.BigEndian.PutUint64(key, height)
var lm Move
err := t.db.View(func(tx *bbolt.Tx) error {
treeRoot := tx.Bucket(bucketName(cid, treeID))
if treeRoot == nil {
return ErrTreeNotFound
}
c := treeRoot.Bucket(logBucket).Cursor()
if _, data := c.Seek(key); data != nil {
return t.moveFromBytes(&lm, data)
}
return nil
})
return lm, err
}
func (t *boltForest) getPathPrefix(bTree *bbolt.Bucket, attr string, path []string) (int, Node, error) {
var key [9]byte
@ -483,7 +506,17 @@ loop:
return len(path), curNode, nil
}
func (t *boltForest) logFromBytes(lm *LogMove, key []byte, data []byte) error {
func (t *boltForest) moveFromBytes(m *Move, data []byte) error {
r := io.NewBinReaderFromBuf(data)
m.Child = r.ReadU64LE()
m.Parent = r.ReadU64LE()
if err := m.Meta.FromBytes(r.ReadVarBytes()); err != nil {
return err
}
return r.Err
}
func (t *boltForest) logFromBytes(lm *LogMove, data []byte) error {
r := io.NewBinReaderFromBuf(data)
lm.Child = r.ReadU64LE()
lm.Parent = r.ReadU64LE()

View file

@ -1,6 +1,8 @@
package pilorama
import (
"sort"
cidSDK "github.com/nspcc-dev/neofs-sdk-go/container/id"
)
@ -133,3 +135,20 @@ func (f *memoryForest) TreeGetChildren(cid cidSDK.ID, treeID string, nodeID Node
copy(res, children)
return res, nil
}
// TreeGetOpLog implements the pilorama.Forest interface.
func (f *memoryForest) TreeGetOpLog(cid cidSDK.ID, treeID string, height uint64) (Move, error) {
fullID := cid.String() + "/" + treeID
s, ok := f.treeMap[fullID]
if !ok {
return Move{}, ErrTreeNotFound
}
n := sort.Search(len(s.operations), func(i int) bool {
return s.operations[i].Time >= height
})
if n == len(s.operations) {
return Move{}, nil
}
return s.operations[n].Move, nil
}

View file

@ -324,6 +324,64 @@ func testForestTreeApply(t *testing.T, constructor func(t *testing.T) Forest) {
})
}
func TestForest_GetOpLog(t *testing.T) {
for i := range providers {
t.Run(providers[i].name, func(t *testing.T) {
testForestTreeGetOpLog(t, providers[i].construct)
})
}
}
func testForestTreeGetOpLog(t *testing.T, constructor func(t *testing.T) Forest) {
cid := cidtest.ID()
treeID := "version"
logs := []Move{
{
Meta: Meta{Time: 4, Items: []KeyValue{{"grand", []byte{1}}}},
Child: 1,
},
{
Meta: Meta{Time: 5, Items: []KeyValue{{"second", []byte{1, 2, 3}}}},
Child: 4,
},
{
Parent: 10,
Meta: Meta{Time: 256 + 4, Items: []KeyValue{}}, // make sure keys are big-endian
Child: 11,
},
}
s := constructor(t)
t.Run("empty log, no panic", func(t *testing.T) {
_, err := s.TreeGetOpLog(cid, treeID, 0)
require.ErrorIs(t, err, ErrTreeNotFound)
})
for i := range logs {
require.NoError(t, s.TreeApply(cid, treeID, &logs[i]))
}
testGetOpLog := func(t *testing.T, height uint64, m Move) {
lm, err := s.TreeGetOpLog(cid, treeID, height)
require.NoError(t, err)
require.Equal(t, m, lm)
}
testGetOpLog(t, 0, logs[0])
testGetOpLog(t, 4, logs[0])
testGetOpLog(t, 5, logs[1])
testGetOpLog(t, 6, logs[2])
testGetOpLog(t, 260, logs[2])
t.Run("missing entry", func(t *testing.T) {
testGetOpLog(t, 261, Move{})
})
t.Run("missing tree", func(t *testing.T) {
_, err := s.TreeGetOpLog(cid, treeID+"123", 4)
require.ErrorIs(t, err, ErrTreeNotFound)
})
}
func TestForest_ApplyRandom(t *testing.T) {
for i := range providers {
t.Run(providers[i].name, func(t *testing.T) {

View file

@ -26,6 +26,9 @@ type Forest interface {
// TreeGetChildren returns children of the node with the specified ID. The order is arbitrary.
// Should return ErrTreeNotFound if the tree is not found, and empty result if the node is not in the tree.
TreeGetChildren(cid cidSDK.ID, treeID string, nodeID Node) ([]uint64, error)
// TreeGetOpLog returns first log operation stored at or above the height.
// In case no such operation is found, empty Move and nil error should be returned.
TreeGetOpLog(cid cidSDK.ID, treeID string, height uint64) (Move, error)
}
type ForestStorage interface {

View file

@ -36,3 +36,8 @@ func (s *Shard) TreeGetMeta(cid cidSDK.ID, treeID string, nodeID pilorama.Node)
func (s *Shard) TreeGetChildren(cid cidSDK.ID, treeID string, nodeID pilorama.Node) ([]uint64, error) {
return s.pilorama.TreeGetChildren(cid, treeID, nodeID)
}
// TreeGetOpLog implements the pilorama.Forest interface.
func (s *Shard) TreeGetOpLog(cid cidSDK.ID, treeID string, height uint64) (pilorama.Move, error) {
return s.pilorama.TreeGetOpLog(cid, treeID, height)
}

View file

@ -342,6 +342,38 @@ loop:
})
}
func (s *Service) GetOpLog(req *GetOpLogRequest, srv TreeService_GetOpLogServer) error {
b := req.GetBody()
var cid cidSDK.ID
if err := cid.Decode(req.GetBody().GetContainerId()); err != nil {
return err
}
h := b.GetHeight()
for {
lm, err := s.forest.TreeGetOpLog(cid, b.GetTreeId(), h)
if err != nil || lm.Time == 0 {
return err
}
err = srv.Send(&GetOpLogResponse{
Body: &GetOpLogResponse_Body{
Operation: &LogMove{
ParentId: lm.Parent,
Meta: lm.Meta.Bytes(),
ChildId: lm.Child,
},
},
})
if err != nil {
return err
}
h = lm.Time + 1
}
}
func protoToMeta(arr []*KeyValue) []pilorama.KeyValue {
meta := make([]pilorama.KeyValue, len(arr))
for i, kv := range arr {

Binary file not shown.

View file

@ -28,6 +28,7 @@ service TreeService {
// Apply pushes log operation from another node to the current.
// The request must be signed by a container node.
rpc Apply (ApplyRequest) returns (ApplyResponse);
rpc GetOpLog(GetOpLogRequest) returns (stream GetOpLogResponse);
}
message AddRequest {
@ -200,3 +201,25 @@ message ApplyResponse {
Body body = 1;
Signature signature = 2;
};
message GetOpLogRequest {
message Body {
bytes container_id = 1;
string tree_id = 2;
uint64 height = 3;
uint64 count = 4;
}
Body body = 1;
Signature signature = 2;
}
message GetOpLogResponse {
message Body {
LogMove operation = 1;
}
Body body = 1;
Signature signature = 2;
};

Binary file not shown.

Binary file not shown.