dstepanov-yadro/frostfs-node
1
0
Fork 0
forked from TrueCloudLab/frostfs-node
Code Issues Pull requests Projects Releases Packages Wiki Activity

[#242] treesvc: Add tracing spans

Browse source 
Signed-off-by: Dmitrii Stepanov <d.stepanov@yadro.com>
This commit is contained in:
Dmitrii Stepanov 2023-04-13 15:36:20 +03:00
parent d5c2c3a244
commit 941e8cde03
13 changed files with 601 additions and 195 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

175
pkg/local_object_storage/engine/tree.go
View file

@ -1,24 +1,39 @@
package engine
import (
"context"
"errors"
"fmt"
"git.frostfs.info/TrueCloudLab/frostfs-api-go/v2/pkg/tracing"
"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/local_object_storage/pilorama"
"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/local_object_storage/shard"
cidSDK "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/container/id"
"go.opentelemetry.io/otel/attribute"
"go.opentelemetry.io/otel/trace"
"go.uber.org/zap"
)
var _ pilorama.Forest = (*StorageEngine)(nil)
// TreeMove implements the pilorama.Forest interface.
func (e *StorageEngine) TreeMove(d pilorama.CIDDescriptor, treeID string, m *pilorama.Move) (*pilorama.Move, error) {
index, lst, err := e.getTreeShard(d.CID, treeID)
func (e *StorageEngine) TreeMove(ctx context.Context, d pilorama.CIDDescriptor, treeID string, m *pilorama.Move) (*pilorama.Move, error) {
ctx, span := tracing.StartSpanFromContext(ctx, "StorageEngine.TreeMove",
trace.WithAttributes(
attribute.String("container_id", d.CID.EncodeToString()),
attribute.Int("position", d.Position),
attribute.Int("size", d.Size),
attribute.String("tree_id", treeID),
),
)
defer span.End()
index, lst, err := e.getTreeShard(ctx, d.CID, treeID)
if err != nil && !errors.Is(err, pilorama.ErrTreeNotFound) {
return nil, err
}
lm, err := lst[index].TreeMove(d, treeID, m)
lm, err := lst[index].TreeMove(ctx, d, treeID, m)
if err != nil {
if !errors.Is(err, shard.ErrReadOnlyMode) && err != shard.ErrPiloramaDisabled {
e.reportShardError(lst[index], "can't perform `TreeMove`", err,
@ -32,13 +47,26 @@ func (e *StorageEngine) TreeMove(d pilorama.CIDDescriptor, treeID string, m *pil
}
// TreeAddByPath implements the pilorama.Forest interface.
func (e *StorageEngine) TreeAddByPath(d pilorama.CIDDescriptor, treeID string, attr string, path []string, m []pilorama.KeyValue) ([]pilorama.Move, error) {
index, lst, err := e.getTreeShard(d.CID, treeID)
func (e *StorageEngine) TreeAddByPath(ctx context.Context, d pilorama.CIDDescriptor, treeID string, attr string, path []string, m []pilorama.KeyValue) ([]pilorama.Move, error) {
ctx, span := tracing.StartSpanFromContext(ctx, "StorageEngine.TreeAddByPath",
trace.WithAttributes(
attribute.String("container_id", d.CID.EncodeToString()),
attribute.Int("position", d.Position),
attribute.Int("size", d.Size),
attribute.String("tree_id", treeID),
attribute.String("attr", attr),
attribute.Int("path_count", len(path)),
attribute.Int("meta_count", len(m)),
),
)
defer span.End()
index, lst, err := e.getTreeShard(ctx, d.CID, treeID)
if err != nil && !errors.Is(err, pilorama.ErrTreeNotFound) {
return nil, err
}
lm, err := lst[index].TreeAddByPath(d, treeID, attr, path, m)
lm, err := lst[index].TreeAddByPath(ctx, d, treeID, attr, path, m)
if err != nil {
if !errors.Is(err, shard.ErrReadOnlyMode) && err != shard.ErrPiloramaDisabled {
e.reportShardError(lst[index], "can't perform `TreeAddByPath`", err,
@ -51,13 +79,22 @@ func (e *StorageEngine) TreeAddByPath(d pilorama.CIDDescriptor, treeID string, a
}
// TreeApply implements the pilorama.Forest interface.
func (e *StorageEngine) TreeApply(cnr cidSDK.ID, treeID string, m *pilorama.Move, backgroundSync bool) error {
index, lst, err := e.getTreeShard(cnr, treeID)
func (e *StorageEngine) TreeApply(ctx context.Context, cnr cidSDK.ID, treeID string, m *pilorama.Move, backgroundSync bool) error {
ctx, span := tracing.StartSpanFromContext(ctx, "StorageEngine.TreeApply",
trace.WithAttributes(
attribute.String("container_id", cnr.EncodeToString()),
attribute.String("tree_id", treeID),
attribute.Bool("background", backgroundSync),
),
)
defer span.End()
index, lst, err := e.getTreeShard(ctx, cnr, treeID)
if err != nil && !errors.Is(err, pilorama.ErrTreeNotFound) {
return err
}
err = lst[index].TreeApply(cnr, treeID, m, backgroundSync)
err = lst[index].TreeApply(ctx, cnr, treeID, m, backgroundSync)
if err != nil {
if !errors.Is(err, shard.ErrReadOnlyMode) && err != shard.ErrPiloramaDisabled {
e.reportShardError(lst[index], "can't perform `TreeApply`", err,
@ -70,11 +107,22 @@ func (e *StorageEngine) TreeApply(cnr cidSDK.ID, treeID string, m *pilorama.Move
}
// TreeGetByPath implements the pilorama.Forest interface.
func (e *StorageEngine) TreeGetByPath(cid cidSDK.ID, treeID string, attr string, path []string, latest bool) ([]pilorama.Node, error) {
func (e *StorageEngine) TreeGetByPath(ctx context.Context, cid cidSDK.ID, treeID string, attr string, path []string, latest bool) ([]pilorama.Node, error) {
ctx, span := tracing.StartSpanFromContext(ctx, "StorageEngine.TreeGetByPath",
trace.WithAttributes(
attribute.String("container_id", cid.EncodeToString()),
attribute.String("tree_id", treeID),
attribute.String("attr", attr),
attribute.Int("path_count", len(path)),
attribute.Bool("latest", latest),
),
)
defer span.End()
var err error
var nodes []pilorama.Node
for _, sh := range e.sortShardsByWeight(cid) {
nodes, err = sh.TreeGetByPath(cid, treeID, attr, path, latest)
nodes, err = sh.TreeGetByPath(ctx, cid, treeID, attr, path, latest)
if err != nil {
if err == shard.ErrPiloramaDisabled {
break
@ -92,12 +140,21 @@ func (e *StorageEngine) TreeGetByPath(cid cidSDK.ID, treeID string, attr string,
}
// TreeGetMeta implements the pilorama.Forest interface.
func (e *StorageEngine) TreeGetMeta(cid cidSDK.ID, treeID string, nodeID pilorama.Node) (pilorama.Meta, uint64, error) {
func (e *StorageEngine) TreeGetMeta(ctx context.Context, cid cidSDK.ID, treeID string, nodeID pilorama.Node) (pilorama.Meta, uint64, error) {
ctx, span := tracing.StartSpanFromContext(ctx, "StorageEngine.TreeGetMeta",
trace.WithAttributes(
attribute.String("container_id", cid.EncodeToString()),
attribute.String("tree_id", treeID),
attribute.String("node_id", fmt.Sprintf("%d", nodeID)),
),
)
defer span.End()
var err error
var m pilorama.Meta
var p uint64
for _, sh := range e.sortShardsByWeight(cid) {
m, p, err = sh.TreeGetMeta(cid, treeID, nodeID)
m, p, err = sh.TreeGetMeta(ctx, cid, treeID, nodeID)
if err != nil {
if err == shard.ErrPiloramaDisabled {
break
@ -115,11 +172,20 @@ func (e *StorageEngine) TreeGetMeta(cid cidSDK.ID, treeID string, nodeID piloram
}
// TreeGetChildren implements the pilorama.Forest interface.
func (e *StorageEngine) TreeGetChildren(cid cidSDK.ID, treeID string, nodeID pilorama.Node) ([]uint64, error) {
func (e *StorageEngine) TreeGetChildren(ctx context.Context, cid cidSDK.ID, treeID string, nodeID pilorama.Node) ([]uint64, error) {
ctx, span := tracing.StartSpanFromContext(ctx, "StorageEngine.TreeGetChildren",
trace.WithAttributes(
attribute.String("container_id", cid.EncodeToString()),
attribute.String("tree_id", treeID),
attribute.String("node_id", fmt.Sprintf("%d", nodeID)),
),
)
defer span.End()
var err error
var nodes []uint64
for _, sh := range e.sortShardsByWeight(cid) {
nodes, err = sh.TreeGetChildren(cid, treeID, nodeID)
nodes, err = sh.TreeGetChildren(ctx, cid, treeID, nodeID)
if err != nil {
if err == shard.ErrPiloramaDisabled {
break
@ -137,11 +203,20 @@ func (e *StorageEngine) TreeGetChildren(cid cidSDK.ID, treeID string, nodeID pil
}
// TreeGetOpLog implements the pilorama.Forest interface.
func (e *StorageEngine) TreeGetOpLog(cid cidSDK.ID, treeID string, height uint64) (pilorama.Move, error) {
func (e *StorageEngine) TreeGetOpLog(ctx context.Context, cid cidSDK.ID, treeID string, height uint64) (pilorama.Move, error) {
ctx, span := tracing.StartSpanFromContext(ctx, "StorageEngine.TreeGetOpLog",
trace.WithAttributes(
attribute.String("container_id", cid.EncodeToString()),
attribute.String("tree_id", treeID),
attribute.String("height", fmt.Sprintf("%d", height)),
),
)
defer span.End()
var err error
var lm pilorama.Move
for _, sh := range e.sortShardsByWeight(cid) {
lm, err = sh.TreeGetOpLog(cid, treeID, height)
lm, err = sh.TreeGetOpLog(ctx, cid, treeID, height)
if err != nil {
if err == shard.ErrPiloramaDisabled {
break
@ -159,10 +234,18 @@ func (e *StorageEngine) TreeGetOpLog(cid cidSDK.ID, treeID string, height uint64
}
// TreeDrop implements the pilorama.Forest interface.
func (e *StorageEngine) TreeDrop(cid cidSDK.ID, treeID string) error {
func (e *StorageEngine) TreeDrop(ctx context.Context, cid cidSDK.ID, treeID string) error {
ctx, span := tracing.StartSpanFromContext(ctx, "StorageEngine.TreeDrop",
trace.WithAttributes(
attribute.String("container_id", cid.EncodeToString()),
attribute.String("tree_id", treeID),
),
)
defer span.End()
var err error
for _, sh := range e.sortShardsByWeight(cid) {
err = sh.TreeDrop(cid, treeID)
err = sh.TreeDrop(ctx, cid, treeID)
if err != nil {
if err == shard.ErrPiloramaDisabled {
break
@ -180,11 +263,18 @@ func (e *StorageEngine) TreeDrop(cid cidSDK.ID, treeID string) error {
}
// TreeList implements the pilorama.Forest interface.
func (e *StorageEngine) TreeList(cid cidSDK.ID) ([]string, error) {
func (e *StorageEngine) TreeList(ctx context.Context, cid cidSDK.ID) ([]string, error) {
ctx, span := tracing.StartSpanFromContext(ctx, "StorageEngine.TreeList",
trace.WithAttributes(
attribute.String("container_id", cid.EncodeToString()),
),
)
defer span.End()
var resIDs []string
for _, sh := range e.unsortedShards() {
ids, err := sh.TreeList(cid)
ids, err := sh.TreeList(ctx, cid)
if err != nil {
if errors.Is(err, shard.ErrPiloramaDisabled) || errors.Is(err, shard.ErrReadOnlyMode) {
return nil, err
@ -205,8 +295,16 @@ func (e *StorageEngine) TreeList(cid cidSDK.ID) ([]string, error) {
}
// TreeExists implements the pilorama.Forest interface.
func (e *StorageEngine) TreeExists(cid cidSDK.ID, treeID string) (bool, error) {
_, _, err := e.getTreeShard(cid, treeID)
func (e *StorageEngine) TreeExists(ctx context.Context, cid cidSDK.ID, treeID string) (bool, error) {
ctx, span := tracing.StartSpanFromContext(ctx, "StorageEngine.TreeExists",
trace.WithAttributes(
attribute.String("container_id", cid.EncodeToString()),
attribute.String("tree_id", treeID),
),
)
defer span.End()
_, _, err := e.getTreeShard(ctx, cid, treeID)
if errors.Is(err, pilorama.ErrTreeNotFound) {
return false, nil
}
@ -214,13 +312,22 @@ func (e *StorageEngine) TreeExists(cid cidSDK.ID, treeID string) (bool, error) {
}
// TreeUpdateLastSyncHeight implements the pilorama.Forest interface.
func (e *StorageEngine) TreeUpdateLastSyncHeight(cid cidSDK.ID, treeID string, height uint64) error {
index, lst, err := e.getTreeShard(cid, treeID)
func (e *StorageEngine) TreeUpdateLastSyncHeight(ctx context.Context, cid cidSDK.ID, treeID string, height uint64) error {
ctx, span := tracing.StartSpanFromContext(ctx, "StorageEngine.TreeUpdateLastSyncHeight",
trace.WithAttributes(
attribute.String("container_id", cid.EncodeToString()),
attribute.String("tree_id", treeID),
attribute.String("height", fmt.Sprintf("%d", height)),
),
)
defer span.End()
index, lst, err := e.getTreeShard(ctx, cid, treeID)
if err != nil && !errors.Is(err, pilorama.ErrTreeNotFound) {
return err
}
err = lst[index].TreeUpdateLastSyncHeight(cid, treeID, height)
err = lst[index].TreeUpdateLastSyncHeight(ctx, cid, treeID, height)
if err != nil && !errors.Is(err, shard.ErrReadOnlyMode) && err != shard.ErrPiloramaDisabled {
e.reportShardError(lst[index], "can't update tree synchronization height", err,
zap.Stringer("cid", cid),
@ -230,11 +337,19 @@ func (e *StorageEngine) TreeUpdateLastSyncHeight(cid cidSDK.ID, treeID string, h
}
// TreeLastSyncHeight implements the pilorama.Forest interface.
func (e *StorageEngine) TreeLastSyncHeight(cid cidSDK.ID, treeID string) (uint64, error) {
func (e *StorageEngine) TreeLastSyncHeight(ctx context.Context, cid cidSDK.ID, treeID string) (uint64, error) {
ctx, span := tracing.StartSpanFromContext(ctx, "StorageEngine.TreeLastSyncHeight",
trace.WithAttributes(
attribute.String("container_id", cid.EncodeToString()),
attribute.String("tree_id", treeID),
),
)
defer span.End()
var err error
var height uint64
for _, sh := range e.sortShardsByWeight(cid) {
height, err = sh.TreeLastSyncHeight(cid, treeID)
height, err = sh.TreeLastSyncHeight(ctx, cid, treeID)
if err != nil {
if err == shard.ErrPiloramaDisabled {
break
@ -251,10 +366,10 @@ func (e *StorageEngine) TreeLastSyncHeight(cid cidSDK.ID, treeID string) (uint64
return height, err
}
func (e *StorageEngine) getTreeShard(cid cidSDK.ID, treeID string) (int, []hashedShard, error) {
func (e *StorageEngine) getTreeShard(ctx context.Context, cid cidSDK.ID, treeID string) (int, []hashedShard, error) {
lst := e.sortShardsByWeight(cid)
for i, sh := range lst {
exists, err := sh.TreeExists(cid, treeID)
exists, err := sh.TreeExists(ctx, cid, treeID)
if err != nil {
return 0, nil, err
}

4
pkg/local_object_storage/engine/tree_test.go
View file

@ -36,7 +36,7 @@ func benchmarkTreeVsSearch(b *testing.B, objCount int) {
if err != nil {
b.Fatal(err)
}
_, err = te.ng.TreeAddByPath(d, treeID, pilorama.AttributeFilename, nil,
_, err = te.ng.TreeAddByPath(context.Background(), d, treeID, pilorama.AttributeFilename, nil,
[]pilorama.KeyValue{{pilorama.AttributeFilename, []byte(strconv.Itoa(i))}})
if err != nil {
b.Fatal(err)
@ -63,7 +63,7 @@ func benchmarkTreeVsSearch(b *testing.B, objCount int) {
})
b.Run("TreeGetByPath", func(b *testing.B) {
for i := 0; i < b.N; i++ {
nodes, err := te.ng.TreeGetByPath(cid, treeID, pilorama.AttributeFilename, []string{strconv.Itoa(objCount / 2)}, true)
nodes, err := te.ng.TreeGetByPath(context.Background(), cid, treeID, pilorama.AttributeFilename, []string{strconv.Itoa(objCount / 2)}, true)
if err != nil {
b.Fatal(err)
}

138
pkg/local_object_storage/pilorama/boltdb.go
View file

@ -2,6 +2,7 @@ package pilorama
import (
"bytes"
"context"
"encoding/binary"
"errors"
"fmt"
@ -11,12 +12,15 @@ import (
"sync"
"time"
"git.frostfs.info/TrueCloudLab/frostfs-api-go/v2/pkg/tracing"
"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/local_object_storage/shard/mode"
"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/local_object_storage/util/logicerr"
"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/util"
cidSDK "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/container/id"
"github.com/nspcc-dev/neo-go/pkg/io"
"go.etcd.io/bbolt"
"go.opentelemetry.io/otel/attribute"
"go.opentelemetry.io/otel/trace"
)
type boltForest struct {
@ -144,7 +148,17 @@ func (t *boltForest) Close() error {
}
// TreeMove implements the Forest interface.
func (t *boltForest) TreeMove(d CIDDescriptor, treeID string, m *Move) (*Move, error) {
func (t *boltForest) TreeMove(ctx context.Context, d CIDDescriptor, treeID string, m *Move) (*Move, error) {
_, span := tracing.StartSpanFromContext(ctx, "boltForest.TreeMove",
trace.WithAttributes(
attribute.String("container_id", d.CID.EncodeToString()),
attribute.Int("position", d.Position),
attribute.Int("size", d.Size),
attribute.String("tree_id", treeID),
),
)
defer span.End()
if !d.checkValid() {
return nil, ErrInvalidCIDDescriptor
}
@ -175,7 +189,15 @@ func (t *boltForest) TreeMove(d CIDDescriptor, treeID string, m *Move) (*Move, e
}
// TreeExists implements the Forest interface.
func (t *boltForest) TreeExists(cid cidSDK.ID, treeID string) (bool, error) {
func (t *boltForest) TreeExists(ctx context.Context, cid cidSDK.ID, treeID string) (bool, error) {
_, span := tracing.StartSpanFromContext(ctx, "boltForest.TreeExists",
trace.WithAttributes(
attribute.String("container_id", cid.EncodeToString()),
attribute.String("tree_id", treeID),
),
)
defer span.End()
t.modeMtx.RLock()
defer t.modeMtx.RUnlock()
@ -197,7 +219,16 @@ func (t *boltForest) TreeExists(cid cidSDK.ID, treeID string) (bool, error) {
var syncHeightKey = []byte{'h'}
// TreeUpdateLastSyncHeight implements the pilorama.Forest interface.
func (t *boltForest) TreeUpdateLastSyncHeight(cid cidSDK.ID, treeID string, height uint64) error {
func (t *boltForest) TreeUpdateLastSyncHeight(ctx context.Context, cid cidSDK.ID, treeID string, height uint64) error {
_, span := tracing.StartSpanFromContext(ctx, "boltForest.TreeUpdateLastSyncHeight",
trace.WithAttributes(
attribute.String("container_id", cid.EncodeToString()),
attribute.String("tree_id", treeID),
attribute.String("height", fmt.Sprintf("%d", height)),
),
)
defer span.End()
rawHeight := make([]byte, 8)
binary.LittleEndian.PutUint64(rawHeight, height)
@ -214,7 +245,15 @@ func (t *boltForest) TreeUpdateLastSyncHeight(cid cidSDK.ID, treeID string, heig
}
// TreeLastSyncHeight implements the pilorama.Forest interface.
func (t *boltForest) TreeLastSyncHeight(cid cidSDK.ID, treeID string) (uint64, error) {
func (t *boltForest) TreeLastSyncHeight(ctx context.Context, cid cidSDK.ID, treeID string) (uint64, error) {
_, span := tracing.StartSpanFromContext(ctx, "boltForest.TreeLastSyncHeight",
trace.WithAttributes(
attribute.String("container_id", cid.EncodeToString()),
attribute.String("tree_id", treeID),
),
)
defer span.End()
var height uint64
buck := bucketName(cid, treeID)
@ -235,7 +274,20 @@ func (t *boltForest) TreeLastSyncHeight(cid cidSDK.ID, treeID string) (uint64, e
}
// TreeAddByPath implements the Forest interface.
func (t *boltForest) TreeAddByPath(d CIDDescriptor, treeID string, attr string, path []string, meta []KeyValue) ([]Move, error) {
func (t *boltForest) TreeAddByPath(ctx context.Context, d CIDDescriptor, treeID string, attr string, path []string, meta []KeyValue) ([]Move, error) {
_, span := tracing.StartSpanFromContext(ctx, "boltForest.TreeAddByPath",
trace.WithAttributes(
attribute.String("container_id", d.CID.EncodeToString()),
attribute.Int("position", d.Position),
attribute.Int("size", d.Size),
attribute.String("tree_id", treeID),
attribute.String("attr", attr),
attribute.Int("path_count", len(path)),
attribute.Int("meta_count", len(meta)),
),
)
defer span.End()
if !d.checkValid() {
return nil, ErrInvalidCIDDescriptor
}
@ -329,7 +381,16 @@ func (t *boltForest) findSpareID(bTree *bbolt.Bucket) uint64 {
}
// TreeApply implements the Forest interface.
func (t *boltForest) TreeApply(cnr cidSDK.ID, treeID string, m *Move, backgroundSync bool) error {
func (t *boltForest) TreeApply(ctx context.Context, cnr cidSDK.ID, treeID string, m *Move, backgroundSync bool) error {
_, span := tracing.StartSpanFromContext(ctx, "boltForest.TreeApply",
trace.WithAttributes(
attribute.String("container_id", cnr.EncodeToString()),
attribute.String("tree_id", treeID),
attribute.Bool("background", backgroundSync),
),
)
defer span.End()
t.modeMtx.RLock()
defer t.modeMtx.RUnlock()
@ -627,7 +688,18 @@ func (t *boltForest) isAncestor(b *bbolt.Bucket, parent, child Node) bool {
}
// TreeGetByPath implements the Forest interface.
func (t *boltForest) TreeGetByPath(cid cidSDK.ID, treeID string, attr string, path []string, latest bool) ([]Node, error) {
func (t *boltForest) TreeGetByPath(ctx context.Context, cid cidSDK.ID, treeID string, attr string, path []string, latest bool) ([]Node, error) {
_, span := tracing.StartSpanFromContext(ctx, "boltForest.TreeGetByPath",
trace.WithAttributes(
attribute.String("container_id", cid.EncodeToString()),
attribute.String("tree_id", treeID),
attribute.String("attr", attr),
attribute.Int("path_count", len(path)),
attribute.Bool("latest", latest),
),
)
defer span.End()
if !isAttributeInternal(attr) {
return nil, ErrNotPathAttribute
}
@ -686,7 +758,16 @@ func (t *boltForest) TreeGetByPath(cid cidSDK.ID, treeID string, attr string, pa
}
// TreeGetMeta implements the forest interface.
func (t *boltForest) TreeGetMeta(cid cidSDK.ID, treeID string, nodeID Node) (Meta, Node, error) {
func (t *boltForest) TreeGetMeta(ctx context.Context, cid cidSDK.ID, treeID string, nodeID Node) (Meta, Node, error) {
_, span := tracing.StartSpanFromContext(ctx, "boltForest.TreeGetMeta",
trace.WithAttributes(
attribute.String("container_id", cid.EncodeToString()),
attribute.String("tree_id", treeID),
attribute.String("node_id", fmt.Sprintf("%d", nodeID)),
),
)
defer span.End()
t.modeMtx.RLock()
defer t.modeMtx.RUnlock()
@ -717,7 +798,16 @@ func (t *boltForest) TreeGetMeta(cid cidSDK.ID, treeID string, nodeID Node) (Met
}
// TreeGetChildren implements the Forest interface.
func (t *boltForest) TreeGetChildren(cid cidSDK.ID, treeID string, nodeID Node) ([]uint64, error) {
func (t *boltForest) TreeGetChildren(ctx context.Context, cid cidSDK.ID, treeID string, nodeID Node) ([]uint64, error) {
_, span := tracing.StartSpanFromContext(ctx, "boltForest.TreeGetChildren",
trace.WithAttributes(
attribute.String("container_id", cid.EncodeToString()),
attribute.String("tree_id", treeID),
attribute.String("node_id", fmt.Sprintf("%d", nodeID)),
),
)
defer span.End()
t.modeMtx.RLock()
defer t.modeMtx.RUnlock()
@ -749,7 +839,14 @@ func (t *boltForest) TreeGetChildren(cid cidSDK.ID, treeID string, nodeID Node)
}
// TreeList implements the Forest interface.
func (t *boltForest) TreeList(cid cidSDK.ID) ([]string, error) {
func (t *boltForest) TreeList(ctx context.Context, cid cidSDK.ID) ([]string, error) {
_, span := tracing.StartSpanFromContext(ctx, "boltForest.TreeList",
trace.WithAttributes(
attribute.String("container_id", cid.EncodeToString()),
),
)
defer span.End()
t.modeMtx.RLock()
defer t.modeMtx.RUnlock()
@ -783,7 +880,16 @@ func (t *boltForest) TreeList(cid cidSDK.ID) ([]string, error) {
}
// TreeGetOpLog implements the pilorama.Forest interface.
func (t *boltForest) TreeGetOpLog(cid cidSDK.ID, treeID string, height uint64) (Move, error) {
func (t *boltForest) TreeGetOpLog(ctx context.Context, cid cidSDK.ID, treeID string, height uint64) (Move, error) {
_, span := tracing.StartSpanFromContext(ctx, "boltForest.TreeGetOpLog",
trace.WithAttributes(
attribute.String("container_id", cid.EncodeToString()),
attribute.String("tree_id", treeID),
attribute.String("height", fmt.Sprintf("%d", height)),
),
)
defer span.End()
t.modeMtx.RLock()
defer t.modeMtx.RUnlock()
@ -813,7 +919,15 @@ func (t *boltForest) TreeGetOpLog(cid cidSDK.ID, treeID string, height uint64) (
}
// TreeDrop implements the pilorama.Forest interface.
func (t *boltForest) TreeDrop(cid cidSDK.ID, treeID string) error {
func (t *boltForest) TreeDrop(ctx context.Context, cid cidSDK.ID, treeID string) error {
_, span := tracing.StartSpanFromContext(ctx, "boltForest.TreeDrop",
trace.WithAttributes(
attribute.String("container_id", cid.EncodeToString()),
attribute.String("tree_id", treeID),
),
)
defer span.End()
t.modeMtx.RLock()
defer t.modeMtx.RUnlock()

25
pkg/local_object_storage/pilorama/forest.go
View file

@ -1,6 +1,7 @@
package pilorama
import (
"context"
"sort"
"strings"
@ -25,7 +26,7 @@ func NewMemoryForest() ForestStorage {
}
// TreeMove implements the Forest interface.
func (f *memoryForest) TreeMove(d CIDDescriptor, treeID string, op *Move) (*Move, error) {
func (f *memoryForest) TreeMove(_ context.Context, d CIDDescriptor, treeID string, op *Move) (*Move, error) {
if !d.checkValid() {
return nil, ErrInvalidCIDDescriptor
}
@ -48,7 +49,7 @@ func (f *memoryForest) TreeMove(d CIDDescriptor, treeID string, op *Move) (*Move
}
// TreeAddByPath implements the Forest interface.
func (f *memoryForest) TreeAddByPath(d CIDDescriptor, treeID string, attr string, path []string, m []KeyValue) ([]Move, error) {
func (f *memoryForest) TreeAddByPath(_ context.Context, d CIDDescriptor, treeID string, attr string, path []string, m []KeyValue) ([]Move, error) {
if !d.checkValid() {
return nil, ErrInvalidCIDDescriptor
}
@ -93,7 +94,7 @@ func (f *memoryForest) TreeAddByPath(d CIDDescriptor, treeID string, attr string
}
// TreeApply implements the Forest interface.
func (f *memoryForest) TreeApply(cnr cid.ID, treeID string, op *Move, _ bool) error {
func (f *memoryForest) TreeApply(_ context.Context, cnr cid.ID, treeID string, op *Move, _ bool) error {
fullID := cnr.String() + "/" + treeID
s, ok := f.treeMap[fullID]
if !ok {
@ -119,7 +120,7 @@ func (f *memoryForest) Close() error {
}
// TreeGetByPath implements the Forest interface.
func (f *memoryForest) TreeGetByPath(cid cid.ID, treeID string, attr string, path []string, latest bool) ([]Node, error) {
func (f *memoryForest) TreeGetByPath(_ context.Context, cid cid.ID, treeID string, attr string, path []string, latest bool) ([]Node, error) {
if !isAttributeInternal(attr) {
return nil, ErrNotPathAttribute
}
@ -134,7 +135,7 @@ func (f *memoryForest) TreeGetByPath(cid cid.ID, treeID string, attr string, pat
}
// TreeGetMeta implements the Forest interface.
func (f *memoryForest) TreeGetMeta(cid cid.ID, treeID string, nodeID Node) (Meta, Node, error) {
func (f *memoryForest) TreeGetMeta(_ context.Context, cid cid.ID, treeID string, nodeID Node) (Meta, Node, error) {
fullID := cid.String() + "/" + treeID
s, ok := f.treeMap[fullID]
if !ok {
@ -145,7 +146,7 @@ func (f *memoryForest) TreeGetMeta(cid cid.ID, treeID string, nodeID Node) (Meta
}
// TreeGetChildren implements the Forest interface.
func (f *memoryForest) TreeGetChildren(cid cid.ID, treeID string, nodeID Node) ([]uint64, error) {
func (f *memoryForest) TreeGetChildren(_ context.Context, cid cid.ID, treeID string, nodeID Node) ([]uint64, error) {
fullID := cid.String() + "/" + treeID
s, ok := f.treeMap[fullID]
if !ok {
@ -163,7 +164,7 @@ func (f *memoryForest) TreeGetChildren(cid cid.ID, treeID string, nodeID Node) (
}
// TreeGetOpLog implements the pilorama.Forest interface.
func (f *memoryForest) TreeGetOpLog(cid cid.ID, treeID string, height uint64) (Move, error) {
func (f *memoryForest) TreeGetOpLog(_ context.Context, cid cid.ID, treeID string, height uint64) (Move, error) {
fullID := cid.String() + "/" + treeID
s, ok := f.treeMap[fullID]
if !ok {
@ -180,7 +181,7 @@ func (f *memoryForest) TreeGetOpLog(cid cid.ID, treeID string, height uint64) (M
}
// TreeDrop implements the pilorama.Forest interface.
func (f *memoryForest) TreeDrop(cid cid.ID, treeID string) error {
func (f *memoryForest) TreeDrop(_ context.Context, cid cid.ID, treeID string) error {
cidStr := cid.String()
if treeID == "" {
for k := range f.treeMap {
@ -200,7 +201,7 @@ func (f *memoryForest) TreeDrop(cid cid.ID, treeID string) error {
}
// TreeList implements the pilorama.Forest interface.
func (f *memoryForest) TreeList(cid cid.ID) ([]string, error) {
func (f *memoryForest) TreeList(_ context.Context, cid cid.ID) ([]string, error) {
var res []string
cidStr := cid.EncodeToString()
@ -217,14 +218,14 @@ func (f *memoryForest) TreeList(cid cid.ID) ([]string, error) {
}
// TreeExists implements the pilorama.Forest interface.
func (f *memoryForest) TreeExists(cid cid.ID, treeID string) (bool, error) {
func (f *memoryForest) TreeExists(_ context.Context, cid cid.ID, treeID string) (bool, error) {
fullID := cid.EncodeToString() + "/" + treeID
_, ok := f.treeMap[fullID]
return ok, nil
}
// TreeUpdateLastSyncHeight implements the pilorama.Forest interface.
func (f *memoryForest) TreeUpdateLastSyncHeight(cid cid.ID, treeID string, height uint64) error {
func (f *memoryForest) TreeUpdateLastSyncHeight(_ context.Context, cid cid.ID, treeID string, height uint64) error {
fullID := cid.EncodeToString() + "/" + treeID
t, ok := f.treeMap[fullID]
if !ok {
@ -235,7 +236,7 @@ func (f *memoryForest) TreeUpdateLastSyncHeight(cid cid.ID, treeID string, heigh
}
// TreeLastSyncHeight implements the pilorama.Forest interface.
func (f *memoryForest) TreeLastSyncHeight(cid cid.ID, treeID string) (uint64, error) {
func (f *memoryForest) TreeLastSyncHeight(_ context.Context, cid cid.ID, treeID string) (uint64, error) {
fullID := cid.EncodeToString() + "/" + treeID
t, ok := f.treeMap[fullID]
if !ok {

155
pkg/local_object_storage/pilorama/forest_test.go
View file

@ -1,6 +1,7 @@
package pilorama
import (
"context"
"fmt"
"math/rand"
"os"
@ -49,7 +50,7 @@ var providers = []struct {
}
func testMeta(t *testing.T, f Forest, cid cidSDK.ID, treeID string, nodeID, parentID Node, expected Meta) {
actualMeta, actualParent, err := f.TreeGetMeta(cid, treeID, nodeID)
actualMeta, actualParent, err := f.TreeGetMeta(context.Background(), cid, treeID, nodeID)
require.NoError(t, err)
require.Equal(t, parentID, actualParent)
require.Equal(t, expected, actualMeta)
@ -71,13 +72,13 @@ func testForestTreeMove(t *testing.T, s Forest) {
meta := []KeyValue{
{Key: AttributeVersion, Value: []byte("XXX")},
{Key: AttributeFilename, Value: []byte("file.txt")}}
lm, err := s.TreeAddByPath(d, treeID, AttributeFilename, []string{"path", "to"}, meta)
lm, err := s.TreeAddByPath(context.Background(), d, treeID, AttributeFilename, []string{"path", "to"}, meta)
require.NoError(t, err)
require.Equal(t, 3, len(lm))
nodeID := lm[2].Child
t.Run("invalid descriptor", func(t *testing.T) {
_, err = s.TreeMove(CIDDescriptor{cid, 0, 0}, treeID, &Move{
_, err = s.TreeMove(context.Background(), CIDDescriptor{cid, 0, 0}, treeID, &Move{
Parent: lm[1].Child,
Meta: Meta{Items: append(meta, KeyValue{Key: "NewKey", Value: []byte("NewValue")})},
Child: nodeID,
@ -85,7 +86,7 @@ func testForestTreeMove(t *testing.T, s Forest) {
require.ErrorIs(t, err, ErrInvalidCIDDescriptor)
})
t.Run("same parent, update meta", func(t *testing.T) {
res, err := s.TreeMove(d, treeID, &Move{
res, err := s.TreeMove(context.Background(), d, treeID, &Move{
Parent: lm[1].Child,
Meta: Meta{Items: append(meta, KeyValue{Key: "NewKey", Value: []byte("NewValue")})},
Child: nodeID,
@ -93,12 +94,12 @@ func testForestTreeMove(t *testing.T, s Forest) {
require.NoError(t, err)
require.Equal(t, res.Child, nodeID)
nodes, err := s.TreeGetByPath(cid, treeID, AttributeFilename, []string{"path", "to", "file.txt"}, false)
nodes, err := s.TreeGetByPath(context.Background(), cid, treeID, AttributeFilename, []string{"path", "to", "file.txt"}, false)
require.NoError(t, err)
require.ElementsMatch(t, []Node{nodeID}, nodes)
})
t.Run("different parent", func(t *testing.T) {
res, err := s.TreeMove(d, treeID, &Move{
res, err := s.TreeMove(context.Background(), d, treeID, &Move{
Parent: RootID,
Meta: Meta{Items: append(meta, KeyValue{Key: "NewKey", Value: []byte("NewValue")})},
Child: nodeID,
@ -106,11 +107,11 @@ func testForestTreeMove(t *testing.T, s Forest) {
require.NoError(t, err)
require.Equal(t, res.Child, nodeID)
nodes, err := s.TreeGetByPath(cid, treeID, AttributeFilename, []string{"path", "to", "file.txt"}, false)
nodes, err := s.TreeGetByPath(context.Background(), cid, treeID, AttributeFilename, []string{"path", "to", "file.txt"}, false)
require.NoError(t, err)
require.True(t, len(nodes) == 0)
nodes, err = s.TreeGetByPath(cid, treeID, AttributeFilename, []string{"file.txt"}, false)
nodes, err = s.TreeGetByPath(context.Background(), cid, treeID, AttributeFilename, []string{"file.txt"}, false)
require.NoError(t, err)
require.ElementsMatch(t, []Node{nodeID}, nodes)
})
@ -130,7 +131,7 @@ func testForestTreeGetChildren(t *testing.T, s Forest) {
treeID := "version"
treeAdd := func(t *testing.T, child, parent Node) {
_, err := s.TreeMove(d, treeID, &Move{
_, err := s.TreeMove(context.Background(), d, treeID, &Move{
Parent: parent,
Child: child,
})
@ -152,7 +153,7 @@ func testForestTreeGetChildren(t *testing.T, s Forest) {
treeAdd(t, 7, 0)
testGetChildren := func(t *testing.T, nodeID Node, expected []Node) {
actual, err := s.TreeGetChildren(cid, treeID, nodeID)
actual, err := s.TreeGetChildren(context.Background(), cid, treeID, nodeID)
require.NoError(t, err)
require.ElementsMatch(t, expected, actual)
}
@ -168,7 +169,7 @@ func testForestTreeGetChildren(t *testing.T, s Forest) {
testGetChildren(t, 42, nil)
})
t.Run("missing tree", func(t *testing.T) {
_, err := s.TreeGetChildren(cid, treeID+"123", 0)
_, err := s.TreeGetChildren(context.Background(), cid, treeID+"123", 0)
require.ErrorIs(t, err, ErrTreeNotFound)
})
}
@ -191,10 +192,10 @@ func testForestTreeDrop(t *testing.T, s Forest) {
cid := cids[0]
t.Run("return nil if not found", func(t *testing.T) {
require.ErrorIs(t, s.TreeDrop(cid, "123"), ErrTreeNotFound)
require.ErrorIs(t, s.TreeDrop(context.Background(), cid, "123"), ErrTreeNotFound)
})
require.NoError(t, s.TreeDrop(cid, ""))
require.NoError(t, s.TreeDrop(context.Background(), cid, ""))
trees := []string{"tree1", "tree2"}
var descs [cidsSize]CIDDescriptor
@ -203,39 +204,39 @@ func testForestTreeDrop(t *testing.T, s Forest) {
}
d := descs[0]
for i := range trees {
_, err := s.TreeAddByPath(d, trees[i], AttributeFilename, []string{"path"},
_, err := s.TreeAddByPath(context.Background(), d, trees[i], AttributeFilename, []string{"path"},
[]KeyValue{{Key: "TreeName", Value: []byte(trees[i])}})
require.NoError(t, err)
}
err := s.TreeDrop(cid, trees[0])
err := s.TreeDrop(context.Background(), cid, trees[0])
require.NoError(t, err)
_, err = s.TreeGetByPath(cid, trees[0], AttributeFilename, []string{"path"}, true)
_, err = s.TreeGetByPath(context.Background(), cid, trees[0], AttributeFilename, []string{"path"}, true)
require.ErrorIs(t, err, ErrTreeNotFound)
_, err = s.TreeGetByPath(cid, trees[1], AttributeFilename, []string{"path"}, true)
_, err = s.TreeGetByPath(context.Background(), cid, trees[1], AttributeFilename, []string{"path"}, true)
require.NoError(t, err)
for j := range descs {
for i := range trees {
_, err := s.TreeAddByPath(descs[j], trees[i], AttributeFilename, []string{"path"},
_, err := s.TreeAddByPath(context.Background(), descs[j], trees[i], AttributeFilename, []string{"path"},
[]KeyValue{{Key: "TreeName", Value: []byte(trees[i])}})
require.NoError(t, err)
}
}
list, err := s.TreeList(cid)
list, err := s.TreeList(context.Background(), cid)
require.NoError(t, err)
require.NotEmpty(t, list)
require.NoError(t, s.TreeDrop(cid, ""))
require.NoError(t, s.TreeDrop(context.Background(), cid, ""))
list, err = s.TreeList(cid)
list, err = s.TreeList(context.Background(), cid)
require.NoError(t, err)
require.Empty(t, list)
for j := 1; j < len(cids); j++ {
list, err = s.TreeList(cids[j])
list, err = s.TreeList(context.Background(), cids[j])
require.NoError(t, err)
require.Equal(t, len(list), len(trees))
}
@ -264,24 +265,24 @@ func testForestTreeAdd(t *testing.T, s Forest) {
}
t.Run("invalid descriptor", func(t *testing.T) {
_, err := s.TreeMove(CIDDescriptor{cid, 0, 0}, treeID, m)
_, err := s.TreeMove(context.Background(), CIDDescriptor{cid, 0, 0}, treeID, m)
require.ErrorIs(t, err, ErrInvalidCIDDescriptor)
})
lm, err := s.TreeMove(d, treeID, m)
lm, err := s.TreeMove(context.Background(), d, treeID, m)
require.NoError(t, err)
testMeta(t, s, cid, treeID, lm.Child, lm.Parent, Meta{Time: lm.Time, Items: meta})
nodes, err := s.TreeGetByPath(cid, treeID, AttributeFilename, []string{"file.txt"}, false)
nodes, err := s.TreeGetByPath(context.Background(), cid, treeID, AttributeFilename, []string{"file.txt"}, false)
require.NoError(t, err)
require.ElementsMatch(t, []Node{lm.Child}, nodes)
t.Run("other trees are unaffected", func(t *testing.T) {
_, err := s.TreeGetByPath(cid, treeID+"123", AttributeFilename, []string{"file.txt"}, false)
_, err := s.TreeGetByPath(context.Background(), cid, treeID+"123", AttributeFilename, []string{"file.txt"}, false)
require.ErrorIs(t, err, ErrTreeNotFound)
_, _, err = s.TreeGetMeta(cid, treeID+"123", 0)
_, _, err = s.TreeGetMeta(context.Background(), cid, treeID+"123", 0)
require.ErrorIs(t, err, ErrTreeNotFound)
})
}
@ -304,15 +305,15 @@ func testForestTreeAddByPath(t *testing.T, s Forest) {
{Key: AttributeFilename, Value: []byte("file.txt")}}
t.Run("invalid descriptor", func(t *testing.T) {
_, err := s.TreeAddByPath(CIDDescriptor{cid, 0, 0}, treeID, AttributeFilename, []string{"yyy"}, meta)
_, err := s.TreeAddByPath(context.Background(), CIDDescriptor{cid, 0, 0}, treeID, AttributeFilename, []string{"yyy"}, meta)
require.ErrorIs(t, err, ErrInvalidCIDDescriptor)
})
t.Run("invalid attribute", func(t *testing.T) {
_, err := s.TreeAddByPath(d, treeID, AttributeVersion, []string{"yyy"}, meta)
_, err := s.TreeAddByPath(context.Background(), d, treeID, AttributeVersion, []string{"yyy"}, meta)
require.ErrorIs(t, err, ErrNotPathAttribute)
})
lm, err := s.TreeAddByPath(d, treeID, AttributeFilename, []string{"path", "to"}, meta)
lm, err := s.TreeAddByPath(context.Background(), d, treeID, AttributeFilename, []string{"path", "to"}, meta)
require.NoError(t, err)
require.Equal(t, 3, len(lm))
testMeta(t, s, cid, treeID, lm[0].Child, lm[0].Parent, Meta{Time: lm[0].Time, Items: []KeyValue{{AttributeFilename, []byte("path")}}})
@ -322,7 +323,7 @@ func testForestTreeAddByPath(t *testing.T, s Forest) {
testMeta(t, s, cid, treeID, firstID, lm[2].Parent, Meta{Time: lm[2].Time, Items: meta})
meta[0].Value = []byte("YYY")
lm, err = s.TreeAddByPath(d, treeID, AttributeFilename, []string{"path", "to"}, meta)
lm, err = s.TreeAddByPath(context.Background(), d, treeID, AttributeFilename, []string{"path", "to"}, meta)
require.NoError(t, err)
require.Equal(t, 1, len(lm))
@ -331,19 +332,19 @@ func testForestTreeAddByPath(t *testing.T, s Forest) {
t.Run("get versions", func(t *testing.T) {
// All versions.
nodes, err := s.TreeGetByPath(cid, treeID, AttributeFilename, []string{"path", "to", "file.txt"}, false)
nodes, err := s.TreeGetByPath(context.Background(), cid, treeID, AttributeFilename, []string{"path", "to", "file.txt"}, false)
require.NoError(t, err)
require.ElementsMatch(t, []Node{firstID, secondID}, nodes)
// Latest version.
nodes, err = s.TreeGetByPath(cid, treeID, AttributeFilename, []string{"path", "to", "file.txt"}, true)
nodes, err = s.TreeGetByPath(context.Background(), cid, treeID, AttributeFilename, []string{"path", "to", "file.txt"}, true)
require.NoError(t, err)
require.Equal(t, []Node{secondID}, nodes)
})
meta[0].Value = []byte("ZZZ")
meta[1].Value = []byte("cat.jpg")
lm, err = s.TreeAddByPath(d, treeID, AttributeFilename, []string{"path", "dir"}, meta)
lm, err = s.TreeAddByPath(context.Background(), d, treeID, AttributeFilename, []string{"path", "dir"}, meta)
require.NoError(t, err)
require.Equal(t, 2, len(lm))
testMeta(t, s, cid, treeID, lm[0].Child, lm[0].Parent, Meta{Time: lm[0].Time, Items: []KeyValue{{AttributeFilename, []byte("dir")}}})
@ -352,7 +353,7 @@ func testForestTreeAddByPath(t *testing.T, s Forest) {
t.Run("create internal nodes", func(t *testing.T) {
meta[0].Value = []byte("SomeValue")
meta[1].Value = []byte("another")
lm, err = s.TreeAddByPath(d, treeID, AttributeFilename, []string{"path"}, meta)
lm, err = s.TreeAddByPath(context.Background(), d, treeID, AttributeFilename, []string{"path"}, meta)
require.NoError(t, err)
require.Equal(t, 1, len(lm))
@ -360,7 +361,7 @@ func testForestTreeAddByPath(t *testing.T, s Forest) {
meta[0].Value = []byte("Leaf")
meta[1].Value = []byte("file.txt")
lm, err = s.TreeAddByPath(d, treeID, AttributeFilename, []string{"path", "another"}, meta)
lm, err = s.TreeAddByPath(context.Background(), d, treeID, AttributeFilename, []string{"path", "another"}, meta)
require.NoError(t, err)
require.Equal(t, 2, len(lm))
@ -375,12 +376,12 @@ func testForestTreeAddByPath(t *testing.T, s Forest) {
{AttributeFilename, []byte("another")}}})
t.Run("get by path", func(t *testing.T) {
nodes, err := s.TreeGetByPath(cid, treeID, AttributeFilename, []string{"path", "another"}, false)
nodes, err := s.TreeGetByPath(context.Background(), cid, treeID, AttributeFilename, []string{"path", "another"}, false)
require.NoError(t, err)
require.Equal(t, 2, len(nodes))
require.ElementsMatch(t, []Node{lm[0].Child, oldMove.Child}, nodes)
nodes, err = s.TreeGetByPath(cid, treeID, AttributeFilename, []string{"path", "another", "file.txt"}, false)
nodes, err = s.TreeGetByPath(context.Background(), cid, treeID, AttributeFilename, []string{"path", "another", "file.txt"}, false)
require.NoError(t, err)
require.Equal(t, 1, len(nodes))
require.Equal(t, lm[1].Child, nodes[0])
@ -391,11 +392,11 @@ func testForestTreeAddByPath(t *testing.T, s Forest) {
meta := []KeyValue{
{Key: AttributeVersion, Value: []byte("XXX")},
{Key: AttributeFilename, Value: []byte{}}}
lm, err := s.TreeAddByPath(d, treeID, AttributeFilename, []string{"path", "to"}, meta)
lm, err := s.TreeAddByPath(context.Background(), d, treeID, AttributeFilename, []string{"path", "to"}, meta)
require.NoError(t, err)
require.Equal(t, 1, len(lm))
nodes, err := s.TreeGetByPath(d.CID, treeID, AttributeFilename, []string{"path", "to", ""}, false)
nodes, err := s.TreeGetByPath(context.Background(), d.CID, treeID, AttributeFilename, []string{"path", "to", ""}, false)
require.NoError(t, err)
require.Equal(t, 1, len(nodes))
require.Equal(t, lm[0].Child, nodes[0])
@ -415,7 +416,7 @@ func testForestTreeApply(t *testing.T, constructor func(t testing.TB, _ ...Optio
treeID := "version"
testApply := func(t *testing.T, s Forest, child, parent Node, meta Meta) {
require.NoError(t, s.TreeApply(cid, treeID, &Move{
require.NoError(t, s.TreeApply(context.Background(), cid, treeID, &Move{
Child: child,
Parent: parent,
Meta: meta,
@ -475,16 +476,16 @@ func testForestTreeGetOpLog(t *testing.T, constructor func(t testing.TB, _ ...Op
s := constructor(t)
t.Run("empty log, no panic", func(t *testing.T) {
_, err := s.TreeGetOpLog(cid, treeID, 0)
_, err := s.TreeGetOpLog(context.Background(), cid, treeID, 0)
require.ErrorIs(t, err, ErrTreeNotFound)
})
for i := range logs {
require.NoError(t, s.TreeApply(cid, treeID, &logs[i], false))
require.NoError(t, s.TreeApply(context.Background(), cid, treeID, &logs[i], false))
}
testGetOpLog := func(t *testing.T, height uint64, m Move) {
lm, err := s.TreeGetOpLog(cid, treeID, height)
lm, err := s.TreeGetOpLog(context.Background(), cid, treeID, height)
require.NoError(t, err)
require.Equal(t, m, lm)
}
@ -498,7 +499,7 @@ func testForestTreeGetOpLog(t *testing.T, constructor func(t testing.TB, _ ...Op
testGetOpLog(t, 261, Move{})
})
t.Run("missing tree", func(t *testing.T) {
_, err := s.TreeGetOpLog(cid, treeID+"123", 4)
_, err := s.TreeGetOpLog(context.Background(), cid, treeID+"123", 4)
require.ErrorIs(t, err, ErrTreeNotFound)
})
}
@ -515,7 +516,7 @@ func testForestTreeExists(t *testing.T, constructor func(t testing.TB, opts ...O
s := constructor(t)
checkExists := func(t *testing.T, expected bool, cid cidSDK.ID, treeID string) {
actual, err := s.TreeExists(cid, treeID)
actual, err := s.TreeExists(context.Background(), cid, treeID)
require.NoError(t, err)
require.Equal(t, expected, actual)
}
@ -527,13 +528,13 @@ func testForestTreeExists(t *testing.T, constructor func(t testing.TB, opts ...O
checkExists(t, false, cid, treeID)
})
require.NoError(t, s.TreeApply(cid, treeID, &Move{Parent: 0, Child: 1}, false))
require.NoError(t, s.TreeApply(context.Background(), cid, treeID, &Move{Parent: 0, Child: 1}, false))
checkExists(t, true, cid, treeID)
checkExists(t, false, cidtest.ID(), treeID) // different CID, same tree
checkExists(t, false, cid, "another tree") // same CID, different tree
t.Run("can be removed", func(t *testing.T) {
require.NoError(t, s.TreeDrop(cid, treeID))
require.NoError(t, s.TreeDrop(context.Background(), cid, treeID))
checkExists(t, false, cid, treeID)
})
}
@ -563,11 +564,11 @@ func TestApplyTricky1(t *testing.T) {
t.Run(providers[i].name, func(t *testing.T) {
s := providers[i].construct(t)
for i := range ops {
require.NoError(t, s.TreeApply(cid, treeID, &ops[i], false))
require.NoError(t, s.TreeApply(context.Background(), cid, treeID, &ops[i], false))
}
for i := range expected {
_, parent, err := s.TreeGetMeta(cid, treeID, expected[i].child)
_, parent, err := s.TreeGetMeta(context.Background(), cid, treeID, expected[i].child)
require.NoError(t, err)
require.Equal(t, expected[i].parent, parent)
}
@ -624,11 +625,11 @@ func TestApplyTricky2(t *testing.T) {
t.Run(providers[i].name, func(t *testing.T) {
s := providers[i].construct(t)
for i := range ops {
require.NoError(t, s.TreeApply(cid, treeID, &ops[i], false))
require.NoError(t, s.TreeApply(context.Background(), cid, treeID, &ops[i], false))
}
for i := range expected {
_, parent, err := s.TreeGetMeta(cid, treeID, expected[i].child)
_, parent, err := s.TreeGetMeta(context.Background(), cid, treeID, expected[i].child)
require.NoError(t, err)
require.Equal(t, expected[i].parent, parent)
}
@ -697,9 +698,9 @@ func prepareRandomTree(nodeCount, opCount int) []Move {
func compareForests(t *testing.T, expected, actual Forest, cid cidSDK.ID, treeID string, nodeCount int) {
for i := uint64(0); i < uint64(nodeCount); i++ {
expectedMeta, expectedParent, err := expected.TreeGetMeta(cid, treeID, i)
expectedMeta, expectedParent, err := expected.TreeGetMeta(context.Background(), cid, treeID, i)
require.NoError(t, err)
actualMeta, actualParent, err := actual.TreeGetMeta(cid, treeID, i)
actualMeta, actualParent, err := actual.TreeGetMeta(context.Background(), cid, treeID, i)
require.NoError(t, err)
require.Equal(t, expectedParent, actualParent, "node id: %d", i)
require.Equal(t, expectedMeta, actualMeta, "node id: %d", i)
@ -738,7 +739,7 @@ func testForestTreeParallelApply(t *testing.T, constructor func(t testing.TB, _
expected := constructor(t)
for i := range ops {
require.NoError(t, expected.TreeApply(cid, treeID, &ops[i], false))
require.NoError(t, expected.TreeApply(context.Background(), cid, treeID, &ops[i], false))
}
for i := 0; i < iterCount; i++ {
@ -753,7 +754,7 @@ func testForestTreeParallelApply(t *testing.T, constructor func(t testing.TB, _
go func() {
defer wg.Done()
for op := range ch {
require.NoError(t, actual.TreeApply(cid, treeID, op, false))
require.NoError(t, actual.TreeApply(context.Background(), cid, treeID, op, false))
}
}()
}
@ -783,7 +784,7 @@ func testForestTreeApplyRandom(t *testing.T, constructor func(t testing.TB, _ ..
expected := constructor(t)
for i := range ops {
require.NoError(t, expected.TreeApply(cid, treeID, &ops[i], false))
require.NoError(t, expected.TreeApply(context.Background(), cid, treeID, &ops[i], false))
}
const iterCount = 200
@ -793,7 +794,7 @@ func testForestTreeApplyRandom(t *testing.T, constructor func(t testing.TB, _ ..
actual := constructor(t)
for i := range ops {
require.NoError(t, actual.TreeApply(cid, treeID, &ops[i], false))
require.NoError(t, actual.TreeApply(context.Background(), cid, treeID, &ops[i], false))
}
compareForests(t, expected, actual, cid, treeID, nodeCount)
}
@ -886,7 +887,7 @@ func benchmarkApply(b *testing.B, s Forest, genFunc func(int) []Move) {
b.SetParallelism(10)
b.RunParallel(func(pb *testing.PB) {
for pb.Next() {
if err := s.TreeApply(cid, treeID, &ops[<-ch], false); err != nil {
if err := s.TreeApply(context.Background(), cid, treeID, &ops[<-ch], false); err != nil {
b.Fatalf("error in `Apply`: %v", err)
}
}
@ -929,27 +930,27 @@ func testTreeGetByPath(t *testing.T, s Forest) {
}
t.Run("invalid attribute", func(t *testing.T) {
_, err := s.TreeGetByPath(cid, treeID, AttributeVersion, []string{"", "TTT"}, false)
_, err := s.TreeGetByPath(context.Background(), cid, treeID, AttributeVersion, []string{"", "TTT"}, false)
require.ErrorIs(t, err, ErrNotPathAttribute)
})
nodes, err := s.TreeGetByPath(cid, treeID, AttributeFilename, []string{"b", "cat1.jpg"}, false)
nodes, err := s.TreeGetByPath(context.Background(), cid, treeID, AttributeFilename, []string{"b", "cat1.jpg"}, false)
require.NoError(t, err)
require.Equal(t, []Node{4, 5}, nodes)
nodes, err = s.TreeGetByPath(cid, treeID, AttributeFilename, []string{"a", "cat1.jpg"}, false)
nodes, err = s.TreeGetByPath(context.Background(), cid, treeID, AttributeFilename, []string{"a", "cat1.jpg"}, false)
require.Equal(t, []Node{3}, nodes)
t.Run("missing child", func(t *testing.T) {
nodes, err = s.TreeGetByPath(cid, treeID, AttributeFilename, []string{"a", "cat3.jpg"}, false)
nodes, err = s.TreeGetByPath(context.Background(), cid, treeID, AttributeFilename, []string{"a", "cat3.jpg"}, false)
require.True(t, len(nodes) == 0)
})
t.Run("missing parent", func(t *testing.T) {
nodes, err = s.TreeGetByPath(cid, treeID, AttributeFilename, []string{"xyz", "cat1.jpg"}, false)
nodes, err = s.TreeGetByPath(context.Background(), cid, treeID, AttributeFilename, []string{"xyz", "cat1.jpg"}, false)
require.True(t, len(nodes) == 0)
})
t.Run("empty path", func(t *testing.T) {
nodes, err = s.TreeGetByPath(cid, treeID, AttributeFilename, nil, false)
nodes, err = s.TreeGetByPath(context.Background(), cid, treeID, AttributeFilename, nil, false)
require.True(t, len(nodes) == 0)
})
}
@ -961,7 +962,7 @@ func testMove(t *testing.T, s Forest, ts int, node, parent Node, cid cidSDK.ID,
items = append(items, KeyValue{AttributeVersion, []byte(version)})
}
require.NoError(t, s.TreeApply(cid, treeID, &Move{
require.NoError(t, s.TreeApply(context.Background(), cid, treeID, &Move{
Parent: parent,
Child: node,
Meta: Meta{
@ -1000,7 +1001,7 @@ func testTreeGetTrees(t *testing.T, s Forest) {
d.CID = cid
for _, treeID := range treeIDs[cid] {
_, err := s.TreeAddByPath(d, treeID, objectSDK.AttributeFileName, []string{"path"}, nil)
_, err := s.TreeAddByPath(context.Background(), d, treeID, objectSDK.AttributeFileName, []string{"path"}, nil)
require.NoError(t, err)
}
}
@ -1008,7 +1009,7 @@ func testTreeGetTrees(t *testing.T, s Forest) {
for _, cid := range cids {
d.CID = cid
trees, err := s.TreeList(cid)
trees, err := s.TreeList(context.Background(), cid)
require.NoError(t, err)
require.ElementsMatch(t, treeIDs[cid], trees)
@ -1028,38 +1029,38 @@ func testTreeLastSyncHeight(t *testing.T, f Forest) {
treeID := "someTree"
t.Run("ErrNotFound if no log operations are stored for a tree", func(t *testing.T) {
_, err := f.TreeLastSyncHeight(cnr, treeID)
_, err := f.TreeLastSyncHeight(context.Background(), cnr, treeID)
require.ErrorIs(t, err, ErrTreeNotFound)
err = f.TreeUpdateLastSyncHeight(cnr, treeID, 1)
err = f.TreeUpdateLastSyncHeight(context.Background(), cnr, treeID, 1)
require.ErrorIs(t, err, ErrTreeNotFound)
})
_, err := f.TreeMove(CIDDescriptor{CID: cnr, Size: 1}, treeID, &Move{
_, err := f.TreeMove(context.Background(), CIDDescriptor{CID: cnr, Size: 1}, treeID, &Move{
Parent: RootID,
Child: 1,
})
require.NoError(t, err)
h, err := f.TreeLastSyncHeight(cnr, treeID)
h, err := f.TreeLastSyncHeight(context.Background(), cnr, treeID)
require.NoError(t, err)
require.EqualValues(t, 0, h)
t.Run("separate storages for separate containers", func(t *testing.T) {
_, err := f.TreeLastSyncHeight(cidtest.ID(), treeID)
_, err := f.TreeLastSyncHeight(context.Background(), cidtest.ID(), treeID)
require.ErrorIs(t, err, ErrTreeNotFound)
})
require.NoError(t, f.TreeUpdateLastSyncHeight(cnr, treeID, 10))
require.NoError(t, f.TreeUpdateLastSyncHeight(context.Background(), cnr, treeID, 10))
h, err = f.TreeLastSyncHeight(cnr, treeID)
h, err = f.TreeLastSyncHeight(context.Background(), cnr, treeID)
require.NoError(t, err)
require.EqualValues(t, 10, h)
t.Run("removed correctly", func(t *testing.T) {
require.NoError(t, f.TreeDrop(cnr, treeID))
require.NoError(t, f.TreeDrop(context.Background(), cnr, treeID))
_, err := f.TreeLastSyncHeight(cnr, treeID)
_, err := f.TreeLastSyncHeight(context.Background(), cnr, treeID)
require.ErrorIs(t, err, ErrTreeNotFound)
})
}

26
pkg/local_object_storage/pilorama/interface.go
View file

@ -1,6 +1,8 @@
package pilorama
import (
"context"
"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/local_object_storage/shard/mode"
"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/local_object_storage/util/logicerr"
cidSDK "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/container/id"
@ -11,43 +13,43 @@ type Forest interface {
// TreeMove moves node in the tree.
// If the parent of the move operation is TrashID, the node is removed.
// If the child of the move operation is RootID, new ID is generated and added to a tree.
TreeMove(d CIDDescriptor, treeID string, m *Move) (*Move, error)
TreeMove(ctx context.Context, d CIDDescriptor, treeID string, m *Move) (*Move, error)
// TreeAddByPath adds new node in the tree using provided path.
// The path is constructed by descending from the root using the values of the attr in meta.
// Internal nodes in path should have exactly one attribute, otherwise a new node is created.
TreeAddByPath(d CIDDescriptor, treeID string, attr string, path []string, meta []KeyValue) ([]Move, error)
TreeAddByPath(ctx context.Context, d CIDDescriptor, treeID string, attr string, path []string, meta []KeyValue) ([]Move, error)
// TreeApply applies replicated operation from another node.
// If background is true, TreeApply will first check whether an operation exists.
TreeApply(cnr cidSDK.ID, treeID string, m *Move, backgroundSync bool) error
TreeApply(ctx context.Context, cnr cidSDK.ID, treeID string, m *Move, backgroundSync bool) error
// TreeGetByPath returns all nodes corresponding to the path.
// The path is constructed by descending from the root using the values of the
// AttributeFilename in meta.
// The last argument determines whether only the node with the latest timestamp is returned.
// Should return ErrTreeNotFound if the tree is not found, and empty result if the path is not in the tree.
TreeGetByPath(cid cidSDK.ID, treeID string, attr string, path []string, latest bool) ([]Node, error)
TreeGetByPath(ctx context.Context, cid cidSDK.ID, treeID string, attr string, path []string, latest bool) ([]Node, error)
// TreeGetMeta returns meta information of the node with the specified ID.
// Should return ErrTreeNotFound if the tree is not found, and empty result if the node is not in the tree.
TreeGetMeta(cid cidSDK.ID, treeID string, nodeID Node) (Meta, Node, error)
TreeGetMeta(ctx context.Context, cid cidSDK.ID, treeID string, nodeID Node) (Meta, Node, error)
// 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)
TreeGetChildren(ctx context.Context, 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)
TreeGetOpLog(ctx context.Context, cid cidSDK.ID, treeID string, height uint64) (Move, error)
// TreeDrop drops a tree from the database.
// If the tree is not found, ErrTreeNotFound should be returned.
// In case of empty treeID drops all trees related to container.
TreeDrop(cid cidSDK.ID, treeID string) error
TreeDrop(ctx context.Context, cid cidSDK.ID, treeID string) error
// TreeList returns all the tree IDs that have been added to the
// passed container ID. Nil slice should be returned if no tree found.
TreeList(cid cidSDK.ID) ([]string, error)
TreeList(ctx context.Context, cid cidSDK.ID) ([]string, error)
// TreeExists checks if a tree exists locally.
// If the tree is not found, false and a nil error should be returned.
TreeExists(cid cidSDK.ID, treeID string) (bool, error)
TreeExists(ctx context.Context, cid cidSDK.ID, treeID string) (bool, error)
// TreeUpdateLastSyncHeight updates last log height synchronized with _all_ container nodes.
TreeUpdateLastSyncHeight(cid cidSDK.ID, treeID string, height uint64) error
TreeUpdateLastSyncHeight(ctx context.Context, cid cidSDK.ID, treeID string, height uint64) error
// TreeLastSyncHeight returns last log height synchronized with _all_ container nodes.
TreeLastSyncHeight(cid cidSDK.ID, treeID string) (uint64, error)
TreeLastSyncHeight(ctx context.Context, cid cidSDK.ID, treeID string) (uint64, error)
}
type ForestStorage interface {

176
pkg/local_object_storage/shard/tree.go
View file

@ -1,9 +1,15 @@
package shard
import (
"context"
"fmt"
"git.frostfs.info/TrueCloudLab/frostfs-api-go/v2/pkg/tracing"
"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/local_object_storage/pilorama"
"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/local_object_storage/util/logicerr"
cidSDK "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/container/id"
"go.opentelemetry.io/otel/attribute"
"go.opentelemetry.io/otel/trace"
)
var _ pilorama.Forest = (*Shard)(nil)
@ -12,7 +18,18 @@ var _ pilorama.Forest = (*Shard)(nil)
var ErrPiloramaDisabled = logicerr.New("pilorama is disabled")
// TreeMove implements the pilorama.Forest interface.
func (s *Shard) TreeMove(d pilorama.CIDDescriptor, treeID string, m *pilorama.Move) (*pilorama.Move, error) {
func (s *Shard) TreeMove(ctx context.Context, d pilorama.CIDDescriptor, treeID string, m *pilorama.Move) (*pilorama.Move, error) {
ctx, span := tracing.StartSpanFromContext(ctx, "Shard.TreeMove",
trace.WithAttributes(
attribute.String("shard_id", s.ID().String()),
attribute.String("container_id", d.CID.EncodeToString()),
attribute.Int("position", d.Position),
attribute.Int("size", d.Size),
attribute.String("tree_id", treeID),
),
)
defer span.End()
if s.pilorama == nil {
return nil, ErrPiloramaDisabled
}
@ -26,11 +43,25 @@ func (s *Shard) TreeMove(d pilorama.CIDDescriptor, treeID string, m *pilorama.Mo
if s.info.Mode.NoMetabase() {
return nil, ErrDegradedMode
}
return s.pilorama.TreeMove(d, treeID, m)
return s.pilorama.TreeMove(ctx, d, treeID, m)
}
// TreeAddByPath implements the pilorama.Forest interface.
func (s *Shard) TreeAddByPath(d pilorama.CIDDescriptor, treeID string, attr string, path []string, meta []pilorama.KeyValue) ([]pilorama.Move, error) {
func (s *Shard) TreeAddByPath(ctx context.Context, d pilorama.CIDDescriptor, treeID string, attr string, path []string, meta []pilorama.KeyValue) ([]pilorama.Move, error) {
ctx, span := tracing.StartSpanFromContext(ctx, "Shard.TreeAddByPath",
trace.WithAttributes(
attribute.String("shard_id", s.ID().String()),
attribute.String("container_id", d.CID.EncodeToString()),
attribute.Int("position", d.Position),
attribute.Int("size", d.Size),
attribute.String("tree_id", treeID),
attribute.String("attr", attr),
attribute.Int("path_count", len(path)),
attribute.Int("meta_count", len(meta)),
),
)
defer span.End()
if s.pilorama == nil {
return nil, ErrPiloramaDisabled
}
@ -44,11 +75,21 @@ func (s *Shard) TreeAddByPath(d pilorama.CIDDescriptor, treeID string, attr stri
if s.info.Mode.NoMetabase() {
return nil, ErrDegradedMode
}
return s.pilorama.TreeAddByPath(d, treeID, attr, path, meta)
return s.pilorama.TreeAddByPath(ctx, d, treeID, attr, path, meta)
}
// TreeApply implements the pilorama.Forest interface.
func (s *Shard) TreeApply(cnr cidSDK.ID, treeID string, m *pilorama.Move, backgroundSync bool) error {
func (s *Shard) TreeApply(ctx context.Context, cnr cidSDK.ID, treeID string, m *pilorama.Move, backgroundSync bool) error {
ctx, span := tracing.StartSpanFromContext(ctx, "Shard.TreeApply",
trace.WithAttributes(
attribute.String("shard_id", s.ID().String()),
attribute.String("container_id", cnr.EncodeToString()),
attribute.String("tree_id", treeID),
attribute.Bool("background", backgroundSync),
),
)
defer span.End()
if s.pilorama == nil {
return ErrPiloramaDisabled
}
@ -62,11 +103,23 @@ func (s *Shard) TreeApply(cnr cidSDK.ID, treeID string, m *pilorama.Move, backgr
if s.info.Mode.NoMetabase() {
return ErrDegradedMode
}
return s.pilorama.TreeApply(cnr, treeID, m, backgroundSync)
return s.pilorama.TreeApply(ctx, cnr, treeID, m, backgroundSync)
}
// TreeGetByPath implements the pilorama.Forest interface.
func (s *Shard) TreeGetByPath(cid cidSDK.ID, treeID string, attr string, path []string, latest bool) ([]pilorama.Node, error) {
func (s *Shard) TreeGetByPath(ctx context.Context, cid cidSDK.ID, treeID string, attr string, path []string, latest bool) ([]pilorama.Node, error) {
ctx, span := tracing.StartSpanFromContext(ctx, "Shard.TreeGetByPath",
trace.WithAttributes(
attribute.String("shard_id", s.ID().String()),
attribute.String("container_id", cid.EncodeToString()),
attribute.String("tree_id", treeID),
attribute.String("attr", attr),
attribute.Int("path_count", len(path)),
attribute.Bool("latest", latest),
),
)
defer span.End()
if s.pilorama == nil {
return nil, ErrPiloramaDisabled
}
@ -77,11 +130,21 @@ func (s *Shard) TreeGetByPath(cid cidSDK.ID, treeID string, attr string, path []
if s.info.Mode.NoMetabase() {
return nil, ErrDegradedMode
}
return s.pilorama.TreeGetByPath(cid, treeID, attr, path, latest)
return s.pilorama.TreeGetByPath(ctx, cid, treeID, attr, path, latest)
}
// TreeGetMeta implements the pilorama.Forest interface.
func (s *Shard) TreeGetMeta(cid cidSDK.ID, treeID string, nodeID pilorama.Node) (pilorama.Meta, uint64, error) {
func (s *Shard) TreeGetMeta(ctx context.Context, cid cidSDK.ID, treeID string, nodeID pilorama.Node) (pilorama.Meta, uint64, error) {
ctx, span := tracing.StartSpanFromContext(ctx, "Shard.TreeGetMeta",
trace.WithAttributes(
attribute.String("shard_id", s.ID().String()),
attribute.String("container_id", cid.EncodeToString()),
attribute.String("tree_id", treeID),
attribute.String("node_id", fmt.Sprintf("%d", nodeID)),
),
)
defer span.End()
if s.pilorama == nil {
return pilorama.Meta{}, 0, ErrPiloramaDisabled
}
@ -92,11 +155,21 @@ func (s *Shard) TreeGetMeta(cid cidSDK.ID, treeID string, nodeID pilorama.Node)
if s.info.Mode.NoMetabase() {
return pilorama.Meta{}, 0, ErrDegradedMode
}
return s.pilorama.TreeGetMeta(cid, treeID, nodeID)
return s.pilorama.TreeGetMeta(ctx, cid, treeID, nodeID)
}
// TreeGetChildren implements the pilorama.Forest interface.
func (s *Shard) TreeGetChildren(cid cidSDK.ID, treeID string, nodeID pilorama.Node) ([]uint64, error) {
func (s *Shard) TreeGetChildren(ctx context.Context, cid cidSDK.ID, treeID string, nodeID pilorama.Node) ([]uint64, error) {
ctx, span := tracing.StartSpanFromContext(ctx, "Shard.TreeGetChildren",
trace.WithAttributes(
attribute.String("shard_id", s.ID().String()),
attribute.String("container_id", cid.EncodeToString()),
attribute.String("tree_id", treeID),
attribute.String("node_id", fmt.Sprintf("%d", nodeID)),
),
)
defer span.End()
if s.pilorama == nil {
return nil, ErrPiloramaDisabled
}
@ -107,11 +180,21 @@ func (s *Shard) TreeGetChildren(cid cidSDK.ID, treeID string, nodeID pilorama.No
if s.info.Mode.NoMetabase() {
return nil, ErrDegradedMode
}
return s.pilorama.TreeGetChildren(cid, treeID, nodeID)
return s.pilorama.TreeGetChildren(ctx, cid, treeID, nodeID)
}
// TreeGetOpLog implements the pilorama.Forest interface.
func (s *Shard) TreeGetOpLog(cid cidSDK.ID, treeID string, height uint64) (pilorama.Move, error) {
func (s *Shard) TreeGetOpLog(ctx context.Context, cid cidSDK.ID, treeID string, height uint64) (pilorama.Move, error) {
ctx, span := tracing.StartSpanFromContext(ctx, "Shard.TreeGetOpLog",
trace.WithAttributes(
attribute.String("shard_id", s.ID().String()),
attribute.String("container_id", cid.EncodeToString()),
attribute.String("tree_id", treeID),
attribute.String("height", fmt.Sprintf("%d", height)),
),
)
defer span.End()
if s.pilorama == nil {
return pilorama.Move{}, ErrPiloramaDisabled
}
@ -122,11 +205,20 @@ func (s *Shard) TreeGetOpLog(cid cidSDK.ID, treeID string, height uint64) (pilor
if s.info.Mode.NoMetabase() {
return pilorama.Move{}, ErrDegradedMode
}
return s.pilorama.TreeGetOpLog(cid, treeID, height)
return s.pilorama.TreeGetOpLog(ctx, cid, treeID, height)
}
// TreeDrop implements the pilorama.Forest interface.
func (s *Shard) TreeDrop(cid cidSDK.ID, treeID string) error {
func (s *Shard) TreeDrop(ctx context.Context, cid cidSDK.ID, treeID string) error {
ctx, span := tracing.StartSpanFromContext(ctx, "Shard.TreeDrop",
trace.WithAttributes(
attribute.String("shard_id", s.ID().String()),
attribute.String("container_id", cid.EncodeToString()),
attribute.String("tree_id", treeID),
),
)
defer span.End()
if s.pilorama == nil {
return ErrPiloramaDisabled
}
@ -137,11 +229,19 @@ func (s *Shard) TreeDrop(cid cidSDK.ID, treeID string) error {
if s.info.Mode.NoMetabase() {
return ErrDegradedMode
}
return s.pilorama.TreeDrop(cid, treeID)
return s.pilorama.TreeDrop(ctx, cid, treeID)
}
// TreeList implements the pilorama.Forest interface.
func (s *Shard) TreeList(cid cidSDK.ID) ([]string, error) {
func (s *Shard) TreeList(ctx context.Context, cid cidSDK.ID) ([]string, error) {
ctx, span := tracing.StartSpanFromContext(ctx, "Shard.TreeList",
trace.WithAttributes(
attribute.String("shard_id", s.ID().String()),
attribute.String("container_id", cid.EncodeToString()),
),
)
defer span.End()
if s.pilorama == nil {
return nil, ErrPiloramaDisabled
}
@ -152,11 +252,20 @@ func (s *Shard) TreeList(cid cidSDK.ID) ([]string, error) {
if s.info.Mode.NoMetabase() {
return nil, ErrDegradedMode
}
return s.pilorama.TreeList(cid)
return s.pilorama.TreeList(ctx, cid)
}
// TreeExists implements the pilorama.Forest interface.
func (s *Shard) TreeExists(cid cidSDK.ID, treeID string) (bool, error) {
func (s *Shard) TreeExists(ctx context.Context, cid cidSDK.ID, treeID string) (bool, error) {
ctx, span := tracing.StartSpanFromContext(ctx, "Shard.TreeExists",
trace.WithAttributes(
attribute.String("shard_id", s.ID().String()),
attribute.String("container_id", cid.EncodeToString()),
attribute.String("tree_id", treeID),
),
)
defer span.End()
if s.pilorama == nil {
return false, ErrPiloramaDisabled
}
@ -167,11 +276,21 @@ func (s *Shard) TreeExists(cid cidSDK.ID, treeID string) (bool, error) {
if s.info.Mode.NoMetabase() {
return false, ErrDegradedMode
}
return s.pilorama.TreeExists(cid, treeID)
return s.pilorama.TreeExists(ctx, cid, treeID)
}
// TreeUpdateLastSyncHeight implements the pilorama.Forest interface.
func (s *Shard) TreeUpdateLastSyncHeight(cid cidSDK.ID, treeID string, height uint64) error {
func (s *Shard) TreeUpdateLastSyncHeight(ctx context.Context, cid cidSDK.ID, treeID string, height uint64) error {
ctx, span := tracing.StartSpanFromContext(ctx, "Shard.TreeUpdateLastSyncHeight",
trace.WithAttributes(
attribute.String("shard_id", s.ID().String()),
attribute.String("container_id", cid.EncodeToString()),
attribute.String("tree_id", treeID),
attribute.String("height", fmt.Sprintf("%d", height)),
),
)
defer span.End()
if s.pilorama == nil {
return ErrPiloramaDisabled
}
@ -185,11 +304,20 @@ func (s *Shard) TreeUpdateLastSyncHeight(cid cidSDK.ID, treeID string, height ui
if s.info.Mode.NoMetabase() {
return ErrDegradedMode
}
return s.pilorama.TreeUpdateLastSyncHeight(cid, treeID, height)
return s.pilorama.TreeUpdateLastSyncHeight(ctx, cid, treeID, height)
}
// TreeLastSyncHeight implements the pilorama.Forest interface.
func (s *Shard) TreeLastSyncHeight(cid cidSDK.ID, treeID string) (uint64, error) {
func (s *Shard) TreeLastSyncHeight(ctx context.Context, cid cidSDK.ID, treeID string) (uint64, error) {
ctx, span := tracing.StartSpanFromContext(ctx, "Shard.TreeLastSyncHeight",
trace.WithAttributes(
attribute.String("shard_id", s.ID().String()),
attribute.String("container_id", cid.EncodeToString()),
attribute.String("tree_id", treeID),
),
)
defer span.End()
if s.pilorama == nil {
return 0, ErrPiloramaDisabled
}
@ -200,5 +328,5 @@ func (s *Shard) TreeLastSyncHeight(cid cidSDK.ID, treeID string) (uint64, error)
if s.info.Mode.NoMetabase() {
return 0, ErrDegradedMode
}
return s.pilorama.TreeLastSyncHeight(cid, treeID)
return s.pilorama.TreeLastSyncHeight(ctx, cid, treeID)
}

4
pkg/services/tree/drop.go
View file

@ -7,8 +7,8 @@ import (
)
// DropTree drops a tree from the database. If treeID is empty, all the trees are dropped.
func (s *Service) DropTree(_ context.Context, cid cid.ID, treeID string) error {
func (s *Service) DropTree(ctx context.Context, cid cid.ID, treeID string) error {
// The only current use-case is a container removal, where all trees should be removed.
// Thus there is no need to replicate the operation on other node.
return s.forest.TreeDrop(cid, treeID)
return s.forest.TreeDrop(ctx, cid, treeID)
}

7
pkg/services/tree/getsubtree_test.go
View file

@ -1,6 +1,7 @@
package tree
import (
"context"
"errors"
"testing"
@ -32,7 +33,7 @@ func TestGetSubTree(t *testing.T) {
meta := []pilorama.KeyValue{
{Key: pilorama.AttributeFilename, Value: []byte(path[len(path)-1])}}
lm, err := p.TreeAddByPath(d, treeID, pilorama.AttributeFilename, path[:len(path)-1], meta)
lm, err := p.TreeAddByPath(context.Background(), d, treeID, pilorama.AttributeFilename, path[:len(path)-1], meta)
require.NoError(t, err)
require.Equal(t, 1, len(lm))
@ -41,7 +42,7 @@ func TestGetSubTree(t *testing.T) {
testGetSubTree := func(t *testing.T, rootID uint64, depth uint32, errIndex int) []uint64 {
acc := subTreeAcc{errIndex: errIndex}
err := getSubTree(&acc, d.CID, &GetSubTreeRequest_Body{
err := getSubTree(context.Background(), &acc, d.CID, &GetSubTreeRequest_Body{
TreeId: treeID,
RootId: rootID,
Depth: depth,
@ -68,7 +69,7 @@ func TestGetSubTree(t *testing.T) {
// GetSubTree must return valid meta.
for i := range acc.seen {
b := acc.seen[i].Body
meta, node, err := p.TreeGetMeta(d.CID, treeID, b.NodeId)
meta, node, err := p.TreeGetMeta(context.Background(), d.CID, treeID, b.NodeId)
require.NoError(t, err)
require.Equal(t, node, b.ParentId)
require.Equal(t, meta.Time, b.Timestamp)

9
pkg/services/tree/redirect.go
View file

@ -5,8 +5,11 @@ import (
"context"
"errors"
"git.frostfs.info/TrueCloudLab/frostfs-api-go/v2/pkg/tracing"
"git.frostfs.info/TrueCloudLab/frostfs-node/internal/logs"
netmapSDK "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/netmap"
"go.opentelemetry.io/otel/attribute"
"go.opentelemetry.io/otel/trace"
"go.uber.org/zap"
)
@ -25,6 +28,12 @@ func (s *Service) forEachNode(ctx context.Context, cntNodes []netmapSDK.NodeInfo
for _, n := range cntNodes {
var stop bool
n.IterateNetworkEndpoints(func(endpoint string) bool {
ctx, span := tracing.StartSpanFromContext(ctx, "TreeService.IterateNetworkEndpoints",
trace.WithAttributes(
attribute.String("endpoint", endpoint),
))
defer span.End()
c, err := s.cache.get(ctx, endpoint)
if err != nil {
return false

32
pkg/services/tree/replicator.go
View file

@ -8,10 +8,13 @@ import (
"fmt"
"time"
"git.frostfs.info/TrueCloudLab/frostfs-api-go/v2/pkg/tracing"
"git.frostfs.info/TrueCloudLab/frostfs-node/internal/logs"
"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/local_object_storage/pilorama"
cidSDK "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/container/id"
netmapSDK "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/netmap"
"go.opentelemetry.io/otel/attribute"
"go.opentelemetry.io/otel/trace"
"go.uber.org/zap"
)
@ -38,17 +41,25 @@ const (
defaultReplicatorSendTimeout = time.Second * 5
)
func (s *Service) localReplicationWorker() {
func (s *Service) localReplicationWorker(ctx context.Context) {
for {
select {
case <-s.closeCh:
return
case op := <-s.replicateLocalCh:
err := s.forest.TreeApply(op.cid, op.treeID, &op.Move, false)
ctx, span := tracing.StartSpanFromContext(ctx, "TreeService.HandleReplicationOperation",
trace.WithAttributes(
attribute.String("tree_id", op.treeID),
attribute.String("container_id", op.cid.EncodeToString()),
),
)
err := s.forest.TreeApply(ctx, op.cid, op.treeID, &op.Move, false)
if err != nil {
s.log.Error(logs.TreeFailedToApplyReplicatedOperation,
zap.String("err", err.Error()))
}
span.End()
}
}
}
@ -59,10 +70,24 @@ func (s *Service) replicationWorker(ctx context.Context) {
case <-s.closeCh:
return
case task := <-s.replicationTasks:
ctx, span := tracing.StartSpanFromContext(ctx, "TreeService.HandleReplicationTask",
trace.WithAttributes(
attribute.String("public_key", hex.EncodeToString(task.n.PublicKey())),
),
)
var lastErr error
var lastAddr string
task.n.IterateNetworkEndpoints(func(addr string) bool {
ctx, span := tracing.StartSpanFromContext(ctx, "TreeService.HandleReplicationTaskOnEndpoint",
trace.WithAttributes(
attribute.String("public_key", hex.EncodeToString(task.n.PublicKey())),
attribute.String("address", addr),
),
)
defer span.End()
lastAddr = addr
c, err := s.cache.get(ctx, addr)
@ -89,6 +114,7 @@ func (s *Service) replicationWorker(ctx context.Context) {
zap.String("key", hex.EncodeToString(task.n.PublicKey())))
}
}
span.End()
}
}
}
@ -96,7 +122,7 @@ func (s *Service) replicationWorker(ctx context.Context) {
func (s *Service) replicateLoop(ctx context.Context) {
for i := 0; i < s.replicatorWorkerCount; i++ {
go s.replicationWorker(ctx)
go s.localReplicationWorker()
go s.localReplicationWorker(ctx)
}
defer func() {
for len(s.replicationTasks) != 0 {

30
pkg/services/tree/service.go
View file

@ -119,7 +119,7 @@ func (s *Service) Add(ctx context.Context, req *AddRequest) (*AddResponse, error
}
d := pilorama.CIDDescriptor{CID: cid, Position: pos, Size: len(ns)}
log, err := s.forest.TreeMove(d, b.GetTreeId(), &pilorama.Move{
log, err := s.forest.TreeMove(ctx, d, b.GetTreeId(), &pilorama.Move{
Parent: b.GetParentId(),
Child: pilorama.RootID,
Meta: pilorama.Meta{Items: protoToMeta(b.GetMeta())},
@ -174,7 +174,7 @@ func (s *Service) AddByPath(ctx context.Context, req *AddByPathRequest) (*AddByP
}
d := pilorama.CIDDescriptor{CID: cid, Position: pos, Size: len(ns)}
logs, err := s.forest.TreeAddByPath(d, b.GetTreeId(), attr, b.GetPath(), meta)
logs, err := s.forest.TreeAddByPath(ctx, d, b.GetTreeId(), attr, b.GetPath(), meta)
if err != nil {
return nil, err
}
@ -231,7 +231,7 @@ func (s *Service) Remove(ctx context.Context, req *RemoveRequest) (*RemoveRespon
}
d := pilorama.CIDDescriptor{CID: cid, Position: pos, Size: len(ns)}
log, err := s.forest.TreeMove(d, b.GetTreeId(), &pilorama.Move{
log, err := s.forest.TreeMove(ctx, d, b.GetTreeId(), &pilorama.Move{
Parent: pilorama.TrashID,
Child: b.GetNodeId(),
})
@ -280,7 +280,7 @@ func (s *Service) Move(ctx context.Context, req *MoveRequest) (*MoveResponse, er
}
d := pilorama.CIDDescriptor{CID: cid, Position: pos, Size: len(ns)}
log, err := s.forest.TreeMove(d, b.GetTreeId(), &pilorama.Move{
log, err := s.forest.TreeMove(ctx, d, b.GetTreeId(), &pilorama.Move{
Parent: b.GetParentId(),
Child: b.GetNodeId(),
Meta: pilorama.Meta{Items: protoToMeta(b.GetMeta())},
@ -328,14 +328,14 @@ func (s *Service) GetNodeByPath(ctx context.Context, req *GetNodeByPathRequest)
attr = pilorama.AttributeFilename
}
nodes, err := s.forest.TreeGetByPath(cid, b.GetTreeId(), attr, b.GetPath(), b.GetLatestOnly())
nodes, err := s.forest.TreeGetByPath(ctx, cid, b.GetTreeId(), attr, b.GetPath(), b.GetLatestOnly())
if err != nil {
return nil, err
}
info := make([]*GetNodeByPathResponse_Info, 0, len(nodes))
for _, node := range nodes {
m, parent, err := s.forest.TreeGetMeta(cid, b.GetTreeId(), node)
m, parent, err := s.forest.TreeGetMeta(ctx, cid, b.GetTreeId(), node)
if err != nil {
return nil, err
}
@ -406,10 +406,10 @@ func (s *Service) GetSubTree(req *GetSubTreeRequest, srv TreeService_GetSubTreeS
return nil
}
return getSubTree(srv, cid, b, s.forest)
return getSubTree(srv.Context(), srv, cid, b, s.forest)
}
func getSubTree(srv TreeService_GetSubTreeServer, cid cidSDK.ID, b *GetSubTreeRequest_Body, forest pilorama.Forest) error {
func getSubTree(ctx context.Context, srv TreeService_GetSubTreeServer, cid cidSDK.ID, b *GetSubTreeRequest_Body, forest pilorama.Forest) error {
// Traverse the tree in a DFS manner. Because we need to support arbitrary depth,
// recursive implementation is not suitable here, so we maintain explicit stack.
stack := [][]uint64{{b.GetRootId()}}
@ -425,7 +425,7 @@ func getSubTree(srv TreeService_GetSubTreeServer, cid cidSDK.ID, b *GetSubTreeRe
nodeID := stack[len(stack)-1][0]
stack[len(stack)-1] = stack[len(stack)-1][1:]
m, p, err := forest.TreeGetMeta(cid, b.GetTreeId(), nodeID)
m, p, err := forest.TreeGetMeta(ctx, cid, b.GetTreeId(), nodeID)
if err != nil {
return err
}
@ -442,7 +442,7 @@ func getSubTree(srv TreeService_GetSubTreeServer, cid cidSDK.ID, b *GetSubTreeRe
}
if b.GetDepth() == 0 || uint32(len(stack)) < b.GetDepth() {
children, err := forest.TreeGetChildren(cid, b.GetTreeId(), nodeID)
children, err := forest.TreeGetChildren(ctx, cid, b.GetTreeId(), nodeID)
if err != nil {
return err
}
@ -455,7 +455,7 @@ func getSubTree(srv TreeService_GetSubTreeServer, cid cidSDK.ID, b *GetSubTreeRe
}
// Apply locally applies operation from the remote node to the tree.
func (s *Service) Apply(_ context.Context, req *ApplyRequest) (*ApplyResponse, error) {
func (s *Service) Apply(ctx context.Context, req *ApplyRequest) (*ApplyResponse, error) {
err := verifyMessage(req)
if err != nil {
return nil, err
@ -468,7 +468,7 @@ func (s *Service) Apply(_ context.Context, req *ApplyRequest) (*ApplyResponse, e
key := req.GetSignature().GetKey()
_, pos, _, err := s.getContainerInfo(cid, key)
_, pos, _, err := s.getContainerInfo(ctx, cid, key)
if err != nil {
return nil, err
}
@ -532,7 +532,7 @@ func (s *Service) GetOpLog(req *GetOpLogRequest, srv TreeService_GetOpLogServer)
h := b.GetHeight()
for {
lm, err := s.forest.TreeGetOpLog(cid, b.GetTreeId(), h)
lm, err := s.forest.TreeGetOpLog(srv.Context(), cid, b.GetTreeId(), h)
if err != nil || lm.Time == 0 {
return err
}
@ -587,7 +587,7 @@ func (s *Service) TreeList(ctx context.Context, req *TreeListRequest) (*TreeList
return resp, outErr
}
ids, err := s.forest.TreeList(cid)
ids, err := s.forest.TreeList(ctx, cid)
if err != nil {
return nil, err
}
@ -623,7 +623,7 @@ func metaToProto(arr []pilorama.KeyValue) []*KeyValue {
// getContainerInfo returns the list of container nodes, position in the container for the node
// with pub key and total amount of nodes in all replicas.
func (s *Service) getContainerInfo(cid cidSDK.ID, pub []byte) ([]netmapSDK.NodeInfo, int, int, error) {
func (s *Service) getContainerInfo(ctx context.Context, cid cidSDK.ID, pub []byte) ([]netmapSDK.NodeInfo, int, int, error) {
cntNodes, _, err := s.getContainerNodes(cid)
if err != nil {
return nil, 0, 0, err

15
pkg/services/tree/sync.go
View file

@ -85,7 +85,7 @@ func (s *Service) synchronizeAllTrees(ctx context.Context, cid cid.ID) error {
}
for _, tid := range treesToSync {
h, err := s.forest.TreeLastSyncHeight(cid, tid)
h, err := s.forest.TreeLastSyncHeight(ctx, cid, tid)
if err != nil && !errors.Is(err, pilorama.ErrTreeNotFound) {
s.log.Warn(logs.TreeCouldNotGetLastSynchronizedHeightForATree,
zap.Stringer("cid", cid),
@ -94,7 +94,7 @@ func (s *Service) synchronizeAllTrees(ctx context.Context, cid cid.ID) error {
}
newHeight := s.synchronizeTree(ctx, cid, h, tid, nodes)
if h < newHeight {
if err := s.forest.TreeUpdateLastSyncHeight(cid, tid, newHeight); err != nil {
if err := s.forest.TreeUpdateLastSyncHeight(ctx, cid, tid, newHeight); err != nil {
s.log.Warn(logs.TreeCouldNotUpdateLastSynchronizedHeightForATree,
zap.Stringer("cid", cid),
zap.String("tree", tid))
@ -232,7 +232,7 @@ func (s *Service) synchronizeSingle(ctx context.Context, cid cid.ID, treeID stri
if err := m.Meta.FromBytes(lm.Meta); err != nil {
return newHeight, err
}
if err := s.forest.TreeApply(cid, treeID, m, true); err != nil {
if err := s.forest.TreeApply(ctx, cid, treeID, m, true); err != nil {
return newHeight, err
}
if m.Time > newHeight {
@ -284,11 +284,13 @@ func (s *Service) syncLoop(ctx context.Context) {
case <-ctx.Done():
return
case <-s.syncChan:
ctx, span := tracing.StartSpanFromContext(ctx, "TreeService.sync")
s.log.Debug(logs.TreeSyncingTrees)
cnrs, err := s.cfg.cnrSource.List()
if err != nil {
s.log.Error(logs.TreeCouldNotFetchContainers, zap.Error(err))
span.End()
continue
}
@ -299,11 +301,15 @@ func (s *Service) syncLoop(ctx context.Context) {
s.removeContainers(ctx, newMap)
s.log.Debug(logs.TreeTreesHaveBeenSynchronized)
span.End()
}
}
}
func (s *Service) syncContainers(ctx context.Context, cnrs []cid.ID) {
ctx, span := tracing.StartSpanFromContext(ctx, "TreeService.syncContainers")
defer span.End()
// sync new containers
var wg sync.WaitGroup
for _, cnr := range cnrs {
@ -335,6 +341,9 @@ func (s *Service) syncContainers(ctx context.Context, cnrs []cid.ID) {
}
func (s *Service) removeContainers(ctx context.Context, newContainers map[cid.ID]struct{}) {
ctx, span := tracing.StartSpanFromContext(ctx, "TreeService.removeContainers")
defer span.End()
s.cnrMapMtx.Lock()
defer s.cnrMapMtx.Unlock()