realloc/frostfs-node
1
0
Fork 0
forked from TrueCloudLab/frostfs-node
Code Releases Packages Activity

[#135] get-object: Add tracing spans

Browse source 
Signed-off-by: Dmitrii Stepanov <d.stepanov@yadro.com>
This commit is contained in:
Dmitrii Stepanov 2023-03-13 14:37:35 +03:00 • committed by fyrchik
parent 5af9f58469
commit 0920d848d0
80 changed files with 523 additions and 231 deletions
Download patch file Download diff file Expand all files Collapse all files

16
pkg/local_object_storage/blobstor/blobovniczatree/exists.go
View file

@ -1,15 +1,27 @@
package blobovniczatree
import (
"context"
"encoding/hex"
"path/filepath"
"git.frostfs.info/TrueCloudLab/frostfs-api-go/v2/pkg/tracing"
"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/local_object_storage/blobovnicza"
"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/local_object_storage/blobstor/common"
"go.opentelemetry.io/otel/attribute"
"go.opentelemetry.io/otel/trace"
"go.uber.org/zap"
)
// Exists implements common.Storage.
func (b *Blobovniczas) Exists(prm common.ExistsPrm) (common.ExistsRes, error) {
func (b *Blobovniczas) Exists(ctx context.Context, prm common.ExistsPrm) (common.ExistsRes, error) {
ctx, span := tracing.StartSpanFromContext(ctx, "Blobovniczas.Exists",
trace.WithAttributes(
attribute.String("address", prm.Address.EncodeToString()),
attribute.String("storage_id", hex.EncodeToString(prm.StorageID)),
))
defer span.End()
if prm.StorageID != nil {
id := blobovnicza.NewIDFromBytes(prm.StorageID)
blz, err := b.openBlobovnicza(id.String())
@ -32,7 +44,7 @@ func (b *Blobovniczas) Exists(prm common.ExistsPrm) (common.ExistsRes, error) {
_, ok := activeCache[dirPath]
_, err := b.getObjectFromLevel(gPrm, p, !ok)
_, err := b.getObjectFromLevel(ctx, gPrm, p, !ok)
if err != nil {
if !blobovnicza.IsErrNotFound(err) {
b.log.Debug("could not get object from level",

5
pkg/local_object_storage/blobstor/blobovniczatree/exists_test.go
View file

@ -1,6 +1,7 @@
package blobovniczatree
import (
"context"
"os"
"path/filepath"
"testing"
@ -44,7 +45,7 @@ func TestExistsInvalidStorageID(t *testing.T) {
storageID[0]--
}
res, err := b.Exists(common.ExistsPrm{Address: addr, StorageID: storageID})
res, err := b.Exists(context.Background(), common.ExistsPrm{Address: addr, StorageID: storageID})
require.NoError(t, err)
require.False(t, res.Exists)
})
@ -57,7 +58,7 @@ func TestExistsInvalidStorageID(t *testing.T) {
require.NoError(t, os.Chmod(badDir, 0))
t.Cleanup(func() { _ = os.Chmod(filepath.Join(dir, "9"), os.ModePerm) })
res, err := b.Exists(common.ExistsPrm{Address: addr, StorageID: storageID})
res, err := b.Exists(context.Background(), common.ExistsPrm{Address: addr, StorageID: storageID})
require.Error(t, err)
require.False(t, res.Exists)
})

31
pkg/local_object_storage/blobstor/blobovniczatree/get.go
View file

@ -1,14 +1,19 @@
package blobovniczatree
import (
"context"
"encoding/hex"
"fmt"
"path/filepath"
"git.frostfs.info/TrueCloudLab/frostfs-api-go/v2/pkg/tracing"
"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/local_object_storage/blobovnicza"
"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/local_object_storage/blobstor/common"
"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/local_object_storage/util/logicerr"
apistatus "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/client/status"
objectSDK "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/object"
"go.opentelemetry.io/otel/attribute"
"go.opentelemetry.io/otel/trace"
"go.uber.org/zap"
)
@ -16,7 +21,15 @@ import (
//
// If blobocvnicza ID is specified, only this blobovnicza is processed.
// Otherwise, all Blobovniczas are processed descending weight.
func (b *Blobovniczas) Get(prm common.GetPrm) (res common.GetRes, err error) {
func (b *Blobovniczas) Get(ctx context.Context, prm common.GetPrm) (res common.GetRes, err error) {
ctx, span := tracing.StartSpanFromContext(ctx, "Blobovniczas.Get",
trace.WithAttributes(
attribute.String("address", prm.Address.EncodeToString()),
attribute.String("storage_id", hex.EncodeToString(prm.StorageID)),
attribute.Bool("raw", prm.Raw),
))
defer span.End()
var bPrm blobovnicza.GetPrm
bPrm.SetAddress(prm.Address)
@ -27,7 +40,7 @@ func (b *Blobovniczas) Get(prm common.GetPrm) (res common.GetRes, err error) {
return res, err
}
return b.getObject(blz, bPrm)
return b.getObject(ctx, blz, bPrm)
}
activeCache := make(map[string]struct{})
@ -37,7 +50,7 @@ func (b *Blobovniczas) Get(prm common.GetPrm) (res common.GetRes, err error) {
_, ok := activeCache[dirPath]
res, err = b.getObjectFromLevel(bPrm, p, !ok)
res, err = b.getObjectFromLevel(ctx, bPrm, p, !ok)
if err != nil {
if !blobovnicza.IsErrNotFound(err) {
b.log.Debug("could not get object from level",
@ -64,7 +77,7 @@ func (b *Blobovniczas) Get(prm common.GetPrm) (res common.GetRes, err error) {
// tries to read object from particular blobovnicza.
//
// returns error if object could not be read from any blobovnicza of the same level.
func (b *Blobovniczas) getObjectFromLevel(prm blobovnicza.GetPrm, blzPath string, tryActive bool) (common.GetRes, error) {
func (b *Blobovniczas) getObjectFromLevel(ctx context.Context, prm blobovnicza.GetPrm, blzPath string, tryActive bool) (common.GetRes, error) {
lvlPath := filepath.Dir(blzPath)
// try to read from blobovnicza if it is opened
@ -72,7 +85,7 @@ func (b *Blobovniczas) getObjectFromLevel(prm blobovnicza.GetPrm, blzPath string
v, ok := b.opened.Get(blzPath)
b.lruMtx.Unlock()
if ok {
if res, err := b.getObject(v, prm); err == nil {
if res, err := b.getObject(ctx, v, prm); err == nil {
return res, err
} else if !blobovnicza.IsErrNotFound(err) {
b.log.Debug("could not read object from opened blobovnicza",
@ -92,7 +105,7 @@ func (b *Blobovniczas) getObjectFromLevel(prm blobovnicza.GetPrm, blzPath string
b.activeMtx.RUnlock()
if ok && tryActive {
if res, err := b.getObject(active.blz, prm); err == nil {
if res, err := b.getObject(ctx, active.blz, prm); err == nil {
return res, err
} else if !blobovnicza.IsErrNotFound(err) {
b.log.Debug("could not get object from active blobovnicza",
@ -117,12 +130,12 @@ func (b *Blobovniczas) getObjectFromLevel(prm blobovnicza.GetPrm, blzPath string
return common.GetRes{}, err
}
return b.getObject(blz, prm)
return b.getObject(ctx, blz, prm)
}
// reads object from blobovnicza and returns GetSmallRes.
func (b *Blobovniczas) getObject(blz *blobovnicza.Blobovnicza, prm blobovnicza.GetPrm) (common.GetRes, error) {
res, err := blz.Get(prm)
func (b *Blobovniczas) getObject(ctx context.Context, blz *blobovnicza.Blobovnicza, prm blobovnicza.GetPrm) (common.GetRes, error) {
res, err := blz.Get(ctx, prm)
if err != nil {
return common.GetRes{}, err
}

33
pkg/local_object_storage/blobstor/blobovniczatree/get_range.go
View file

@ -1,14 +1,20 @@
package blobovniczatree
import (
"context"
"encoding/hex"
"fmt"
"path/filepath"
"strconv"
"git.frostfs.info/TrueCloudLab/frostfs-api-go/v2/pkg/tracing"
"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/local_object_storage/blobovnicza"
"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/local_object_storage/blobstor/common"
"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/local_object_storage/util/logicerr"
apistatus "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/client/status"
objectSDK "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/object"
"go.opentelemetry.io/otel/attribute"
"go.opentelemetry.io/otel/trace"
"go.uber.org/zap"
)
@ -16,7 +22,16 @@ import (
//
// If blobocvnicza ID is specified, only this blobovnicza is processed.
// Otherwise, all Blobovniczas are processed descending weight.
func (b *Blobovniczas) GetRange(prm common.GetRangePrm) (res common.GetRangeRes, err error) {
func (b *Blobovniczas) GetRange(ctx context.Context, prm common.GetRangePrm) (res common.GetRangeRes, err error) {
ctx, span := tracing.StartSpanFromContext(ctx, "Blobovniczas.GetRange",
trace.WithAttributes(
attribute.String("address", prm.Address.EncodeToString()),
attribute.String("storage_id", hex.EncodeToString(prm.StorageID)),
attribute.String("offset", strconv.FormatUint(prm.Range.GetOffset(), 10)),
attribute.String("length", strconv.FormatUint(prm.Range.GetLength(), 10)),
))
defer span.End()
if prm.StorageID != nil {
id := blobovnicza.NewIDFromBytes(prm.StorageID)
blz, err := b.openBlobovnicza(id.String())
@ -24,7 +39,7 @@ func (b *Blobovniczas) GetRange(prm common.GetRangePrm) (res common.GetRangeRes,
return common.GetRangeRes{}, err
}
return b.getObjectRange(blz, prm)
return b.getObjectRange(ctx, blz, prm)
}
activeCache := make(map[string]struct{})
@ -35,7 +50,7 @@ func (b *Blobovniczas) GetRange(prm common.GetRangePrm) (res common.GetRangeRes,
_, ok := activeCache[dirPath]
res, err = b.getRangeFromLevel(prm, p, !ok)
res, err = b.getRangeFromLevel(ctx, prm, p, !ok)
if err != nil {
outOfBounds := isErrOutOfRange(err)
if !outOfBounds && !blobovnicza.IsErrNotFound(err) {
@ -68,7 +83,7 @@ func (b *Blobovniczas) GetRange(prm common.GetRangePrm) (res common.GetRangeRes,
// tries to read range of object payload data from particular blobovnicza.
//
// returns error if object could not be read from any blobovnicza of the same level.
func (b *Blobovniczas) getRangeFromLevel(prm common.GetRangePrm, blzPath string, tryActive bool) (common.GetRangeRes, error) {
func (b *Blobovniczas) getRangeFromLevel(ctx context.Context, prm common.GetRangePrm, blzPath string, tryActive bool) (common.GetRangeRes, error) {
lvlPath := filepath.Dir(blzPath)
// try to read from blobovnicza if it is opened
@ -76,7 +91,7 @@ func (b *Blobovniczas) getRangeFromLevel(prm common.GetRangePrm, blzPath string,
v, ok := b.opened.Get(blzPath)
b.lruMtx.Unlock()
if ok {
res, err := b.getObjectRange(v, prm)
res, err := b.getObjectRange(ctx, v, prm)
switch {
case err == nil,
isErrOutOfRange(err):
@ -101,7 +116,7 @@ func (b *Blobovniczas) getRangeFromLevel(prm common.GetRangePrm, blzPath string,
b.activeMtx.RUnlock()
if ok && tryActive {
res, err := b.getObjectRange(active.blz, prm)
res, err := b.getObjectRange(ctx, active.blz, prm)
switch {
case err == nil,
isErrOutOfRange(err):
@ -131,11 +146,11 @@ func (b *Blobovniczas) getRangeFromLevel(prm common.GetRangePrm, blzPath string,
return common.GetRangeRes{}, err
}
return b.getObjectRange(blz, prm)
return b.getObjectRange(ctx, blz, prm)
}
// reads range of object payload data from blobovnicza and returns GetRangeSmallRes.
func (b *Blobovniczas) getObjectRange(blz *blobovnicza.Blobovnicza, prm common.GetRangePrm) (common.GetRangeRes, error) {
func (b *Blobovniczas) getObjectRange(ctx context.Context, blz *blobovnicza.Blobovnicza, prm common.GetRangePrm) (common.GetRangeRes, error) {
var gPrm blobovnicza.GetPrm
gPrm.SetAddress(prm.Address)
@ -143,7 +158,7 @@ func (b *Blobovniczas) getObjectRange(blz *blobovnicza.Blobovnicza, prm common.G
// stores data that is compressed on BlobStor side.
// If blobovnicza learns to do the compression itself,
// we can start using GetRange.
res, err := blz.Get(gPrm)
res, err := blz.Get(ctx, gPrm)
if err != nil {
return common.GetRangeRes{}, err
}

5
pkg/local_object_storage/blobstor/blobstor_test.go
View file

@ -1,6 +1,7 @@
package blobstor
import (
"context"
"path/filepath"
"testing"
@ -62,11 +63,11 @@ func TestCompression(t *testing.T) {
}
testGet := func(t *testing.T, b *BlobStor, i int) {
res1, err := b.Get(common.GetPrm{Address: object.AddressOf(smallObj[i])})
res1, err := b.Get(context.Background(), common.GetPrm{Address: object.AddressOf(smallObj[i])})
require.NoError(t, err)
require.Equal(t, smallObj[i], res1.Object)
res2, err := b.Get(common.GetPrm{Address: object.AddressOf(bigObj[i])})
res2, err := b.Get(context.Background(), common.GetPrm{Address: object.AddressOf(bigObj[i])})
require.NoError(t, err)
require.Equal(t, bigObj[i], res2.Object)
}

12
pkg/local_object_storage/blobstor/common/storage.go
View file

@ -1,6 +1,10 @@
package common
import "git.frostfs.info/TrueCloudLab/frostfs-node/pkg/local_object_storage/blobstor/compression"
import (
"context"
"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/local_object_storage/blobstor/compression"
)
// Storage represents key-value object storage.
// It is used as a building block for a blobstor of a shard.
@ -16,9 +20,9 @@ type Storage interface {
// This function MUST be called before Open.
SetReportErrorFunc(f func(string, error))
Get(GetPrm) (GetRes, error)
GetRange(GetRangePrm) (GetRangeRes, error)
Exists(ExistsPrm) (ExistsRes, error)
Get(context.Context, GetPrm) (GetRes, error)
GetRange(context.Context, GetRangePrm) (GetRangeRes, error)
Exists(context.Context, ExistsPrm) (ExistsRes, error)
Put(PutPrm) (PutRes, error)
Delete(DeletePrm) (DeleteRes, error)
Iterate(IteratePrm) (IterateRes, error)

21
pkg/local_object_storage/blobstor/exists.go
View file

@ -1,7 +1,13 @@
package blobstor
import (
"context"
"encoding/hex"
"git.frostfs.info/TrueCloudLab/frostfs-api-go/v2/pkg/tracing"
"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/local_object_storage/blobstor/common"
"go.opentelemetry.io/otel/attribute"
"go.opentelemetry.io/otel/trace"
"go.uber.org/zap"
)
@ -9,15 +15,22 @@ import (
//
// Returns any error encountered that did not allow
// to completely check object existence.
func (b *BlobStor) Exists(prm common.ExistsPrm) (common.ExistsRes, error) {
func (b *BlobStor) Exists(ctx context.Context, prm common.ExistsPrm) (common.ExistsRes, error) {
ctx, span := tracing.StartSpanFromContext(ctx, "BlobStor.Exists",
trace.WithAttributes(
attribute.String("address", prm.Address.EncodeToString()),
attribute.String("storage_id", hex.EncodeToString(prm.StorageID)),
))
defer span.End()
b.modeMtx.RLock()
defer b.modeMtx.RUnlock()
if prm.StorageID != nil {
if len(prm.StorageID) == 0 {
return b.storage[len(b.storage)-1].Storage.Exists(prm)
return b.storage[len(b.storage)-1].Storage.Exists(ctx, prm)
}
return b.storage[0].Storage.Exists(prm)
return b.storage[0].Storage.Exists(ctx, prm)
}
// If there was an error during existence check below,
@ -31,7 +44,7 @@ func (b *BlobStor) Exists(prm common.ExistsPrm) (common.ExistsRes, error) {
// error | error | log the first error, return the second
var errors []error
for i := range b.storage {
res, err := b.storage[i].Storage.Exists(prm)
res, err := b.storage[i].Storage.Exists(ctx, prm)
if err == nil && res.Exists {
return res, nil
} else if err != nil {

9
pkg/local_object_storage/blobstor/exists_test.go
View file

@ -1,6 +1,7 @@
package blobstor
import (
"context"
"os"
"testing"
@ -43,13 +44,13 @@ func TestExists(t *testing.T) {
for i := range objects {
prm.Address = objectCore.AddressOf(objects[i])
res, err := b.Exists(prm)
res, err := b.Exists(context.Background(), prm)
require.NoError(t, err)
require.True(t, res.Exists)
}
prm.Address = oidtest.Address()
res, err := b.Exists(prm)
res, err := b.Exists(context.Background(), prm)
require.NoError(t, err)
require.False(t, res.Exists)
@ -60,13 +61,13 @@ func TestExists(t *testing.T) {
// Object exists, first error is logged.
prm.Address = objectCore.AddressOf(objects[0])
res, err := b.Exists(prm)
res, err := b.Exists(context.Background(), prm)
require.NoError(t, err)
require.True(t, res.Exists)
// Object doesn't exist, first error is returned.
prm.Address = objectCore.AddressOf(objects[1])
_, err = b.Exists(prm)
_, err = b.Exists(context.Background(), prm)
require.Error(t, err)
require.ErrorIs(t, err, teststore.ErrDiskExploded)
})

46
pkg/local_object_storage/blobstor/fstree/fstree.go
View file

@ -1,6 +1,7 @@
package fstree
import (
"context"
"crypto/sha256"
"errors"
"fmt"
@ -11,6 +12,7 @@ import (
"strings"
"syscall"
"git.frostfs.info/TrueCloudLab/frostfs-api-go/v2/pkg/tracing"
"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/local_object_storage/blobstor/common"
"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/local_object_storage/blobstor/compression"
"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/local_object_storage/util/logicerr"
@ -19,6 +21,8 @@ import (
cid "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/container/id"
objectSDK "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/object"
oid "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/object/id"
"go.opentelemetry.io/otel/attribute"
"go.opentelemetry.io/otel/trace"
)
// FSTree represents an object storage as a filesystem tree.
@ -208,7 +212,13 @@ func (t *FSTree) Delete(prm common.DeletePrm) (common.DeleteRes, error) {
// Exists returns the path to the file with object contents if it exists in the storage
// and an error otherwise.
func (t *FSTree) Exists(prm common.ExistsPrm) (common.ExistsRes, error) {
func (t *FSTree) Exists(ctx context.Context, prm common.ExistsPrm) (common.ExistsRes, error) {
_, span := tracing.StartSpanFromContext(ctx, "FSTree.Exists",
trace.WithAttributes(
attribute.String("address", prm.Address.EncodeToString()),
))
defer span.End()
p := t.treePath(prm.Address)
_, err := os.Stat(p)
@ -336,16 +346,30 @@ func (t *FSTree) PutStream(addr oid.Address, handler func(*os.File) error) error
}
// Get returns an object from the storage by address.
func (t *FSTree) Get(prm common.GetPrm) (common.GetRes, error) {
func (t *FSTree) Get(ctx context.Context, prm common.GetPrm) (common.GetRes, error) {
ctx, span := tracing.StartSpanFromContext(ctx, "FSTree.Get",
trace.WithAttributes(
attribute.Bool("raw", prm.Raw),
attribute.String("address", prm.Address.EncodeToString()),
))
defer span.End()
p := t.treePath(prm.Address)
if _, err := os.Stat(p); os.IsNotExist(err) {
return common.GetRes{}, logicerr.Wrap(apistatus.ObjectNotFound{})
}
data, err := os.ReadFile(p)
if err != nil {
return common.GetRes{}, err
var data []byte
var err error
{
_, span := tracing.StartSpanFromContext(ctx, "FSTree.Get.ReadFile")
defer span.End()
data, err = os.ReadFile(p)
if err != nil {
return common.GetRes{}, err
}
}
data, err = t.Decompress(data)
@ -362,8 +386,16 @@ func (t *FSTree) Get(prm common.GetPrm) (common.GetRes, error) {
}
// GetRange implements common.Storage.
func (t *FSTree) GetRange(prm common.GetRangePrm) (common.GetRangeRes, error) {
res, err := t.Get(common.GetPrm{Address: prm.Address})
func (t *FSTree) GetRange(ctx context.Context, prm common.GetRangePrm) (common.GetRangeRes, error) {
ctx, span := tracing.StartSpanFromContext(ctx, "FSTree.GetRange",
trace.WithAttributes(
attribute.String("address", prm.Address.EncodeToString()),
attribute.String("offset", strconv.FormatUint(prm.Range.GetOffset(), 10)),
attribute.String("length", strconv.FormatUint(prm.Range.GetLength(), 10)),
))
defer span.End()
res, err := t.Get(ctx, common.GetPrm{Address: prm.Address})
if err != nil {
return common.GetRangeRes{}, err
}

21
pkg/local_object_storage/blobstor/get.go
View file

@ -1,23 +1,36 @@
package blobstor
import (
"context"
"encoding/hex"
"errors"
"git.frostfs.info/TrueCloudLab/frostfs-api-go/v2/pkg/tracing"
"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/local_object_storage/blobstor/common"
"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/local_object_storage/util/logicerr"
apistatus "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/client/status"
"go.opentelemetry.io/otel/attribute"
"go.opentelemetry.io/otel/trace"
)
// Get reads the object from b.
// If the descriptor is present, only one sub-storage is tried,
// Otherwise, each sub-storage is tried in order.
func (b *BlobStor) Get(prm common.GetPrm) (common.GetRes, error) {
func (b *BlobStor) Get(ctx context.Context, prm common.GetPrm) (common.GetRes, error) {
ctx, span := tracing.StartSpanFromContext(ctx, "BlobStor.Get",
trace.WithAttributes(
attribute.String("address", prm.Address.EncodeToString()),
attribute.Bool("raw", prm.Raw),
attribute.String("storage_id", hex.EncodeToString(prm.StorageID)),
))
defer span.End()
b.modeMtx.RLock()
defer b.modeMtx.RUnlock()
if prm.StorageID == nil {
for i := range b.storage {
res, err := b.storage[i].Storage.Get(prm)
res, err := b.storage[i].Storage.Get(ctx, prm)
if err == nil || !errors.As(err, new(apistatus.ObjectNotFound)) {
return res, err
}
@ -26,7 +39,7 @@ func (b *BlobStor) Get(prm common.GetPrm) (common.GetRes, error) {
return common.GetRes{}, logicerr.Wrap(apistatus.ObjectNotFound{})
}
if len(prm.StorageID) == 0 {
return b.storage[len(b.storage)-1].Storage.Get(prm)
return b.storage[len(b.storage)-1].Storage.Get(ctx, prm)
}
return b.storage[0].Storage.Get(prm)
return b.storage[0].Storage.Get(ctx, prm)
}

23
pkg/local_object_storage/blobstor/get_range.go
View file

@ -1,23 +1,38 @@
package blobstor
import (
"context"
"encoding/hex"
"errors"
"strconv"
"git.frostfs.info/TrueCloudLab/frostfs-api-go/v2/pkg/tracing"
"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/local_object_storage/blobstor/common"
"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/local_object_storage/util/logicerr"
apistatus "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/client/status"
"go.opentelemetry.io/otel/attribute"
"go.opentelemetry.io/otel/trace"
)
// GetRange reads object payload data from b.
// If the descriptor is present, only one sub-storage is tried,
// Otherwise, each sub-storage is tried in order.
func (b *BlobStor) GetRange(prm common.GetRangePrm) (common.GetRangeRes, error) {
func (b *BlobStor) GetRange(ctx context.Context, prm common.GetRangePrm) (common.GetRangeRes, error) {
ctx, span := tracing.StartSpanFromContext(ctx, "BlobStor.GetRange",
trace.WithAttributes(
attribute.String("address", prm.Address.EncodeToString()),
attribute.String("storage_id", hex.EncodeToString(prm.StorageID)),
attribute.String("offset", strconv.FormatUint(prm.Range.GetOffset(), 10)),
attribute.String("length", strconv.FormatUint(prm.Range.GetLength(), 10)),
))
defer span.End()
b.modeMtx.RLock()
defer b.modeMtx.RUnlock()
if prm.StorageID == nil {
for i := range b.storage {
res, err := b.storage[i].Storage.GetRange(prm)
res, err := b.storage[i].Storage.GetRange(ctx, prm)
if err == nil || !errors.As(err, new(apistatus.ObjectNotFound)) {
return res, err
}
@ -26,7 +41,7 @@ func (b *BlobStor) GetRange(prm common.GetRangePrm) (common.GetRangeRes, error)
return common.GetRangeRes{}, logicerr.Wrap(apistatus.ObjectNotFound{})
}
if len(prm.StorageID) == 0 {
return b.storage[len(b.storage)-1].Storage.GetRange(prm)
return b.storage[len(b.storage)-1].Storage.GetRange(ctx, prm)
}
return b.storage[0].Storage.GetRange(prm)
return b.storage[0].Storage.GetRange(ctx, prm)
}

3
pkg/local_object_storage/blobstor/internal/blobstortest/control.go
View file

@ -1,6 +1,7 @@
package blobstortest
import (
"context"
"math/rand"
"testing"
@ -26,7 +27,7 @@ func TestControl(t *testing.T, cons Constructor, min, max uint64) {
prm.StorageID = objects[i].storageID
prm.Raw = true
_, err := s.Get(prm)
_, err := s.Get(context.Background(), prm)
require.NoError(t, err)
}

9
pkg/local_object_storage/blobstor/internal/blobstortest/delete.go
View file

@ -1,6 +1,7 @@
package blobstortest
import (
"context"
"testing"
"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/local_object_storage/blobstor/common"
@ -35,18 +36,18 @@ func TestDelete(t *testing.T, cons Constructor, min, max uint64) {
t.Run("exists fail", func(t *testing.T) {
prm := common.ExistsPrm{Address: oidtest.Address()}
res, err := s.Exists(prm)
res, err := s.Exists(context.Background(), prm)
require.NoError(t, err)
require.False(t, res.Exists)
})
t.Run("get fail", func(t *testing.T) {
prm := common.GetPrm{Address: oidtest.Address()}
_, err := s.Get(prm)
_, err := s.Get(context.Background(), prm)
require.ErrorAs(t, err, new(apistatus.ObjectNotFound))
})
t.Run("getrange fail", func(t *testing.T) {
prm := common.GetRangePrm{Address: oidtest.Address()}
_, err := s.GetRange(prm)
_, err := s.GetRange(context.Background(), prm)
require.ErrorAs(t, err, new(apistatus.ObjectNotFound))
})
})
@ -75,7 +76,7 @@ func TestDelete(t *testing.T, cons Constructor, min, max uint64) {
prm.Address = objects[3].addr
prm.Raw = true
res, err := s.Get(prm)
res, err := s.Get(context.Background(), prm)
require.NoError(t, err)
require.Equal(t, objects[3].raw, res.RawData)
})

7
pkg/local_object_storage/blobstor/internal/blobstortest/exists.go
View file

@ -1,6 +1,7 @@
package blobstortest
import (
"context"
"testing"
"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/local_object_storage/blobstor/common"
@ -18,7 +19,7 @@ func TestExists(t *testing.T, cons Constructor, min, max uint64) {
t.Run("missing object", func(t *testing.T) {
prm := common.ExistsPrm{Address: oidtest.Address()}
res, err := s.Exists(prm)
res, err := s.Exists(context.Background(), prm)
require.NoError(t, err)
require.False(t, res.Exists)
})
@ -29,7 +30,7 @@ func TestExists(t *testing.T, cons Constructor, min, max uint64) {
t.Run("without storage ID", func(t *testing.T) {
prm.StorageID = nil
res, err := s.Exists(prm)
res, err := s.Exists(context.Background(), prm)
require.NoError(t, err)
require.True(t, res.Exists)
})
@ -37,7 +38,7 @@ func TestExists(t *testing.T, cons Constructor, min, max uint64) {
t.Run("with storage ID", func(t *testing.T) {
prm.StorageID = objects[0].storageID
res, err := s.Exists(prm)
res, err := s.Exists(context.Background(), prm)
require.NoError(t, err)
require.True(t, res.Exists)
})

9
pkg/local_object_storage/blobstor/internal/blobstortest/get.go
View file

@ -1,6 +1,7 @@
package blobstortest
import (
"context"
"testing"
"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/local_object_storage/blobstor/common"
@ -19,7 +20,7 @@ func TestGet(t *testing.T, cons Constructor, min, max uint64) {
t.Run("missing object", func(t *testing.T) {
gPrm := common.GetPrm{Address: oidtest.Address()}
_, err := s.Get(gPrm)
_, err := s.Get(context.Background(), gPrm)
require.ErrorAs(t, err, new(apistatus.ObjectNotFound))
})
@ -29,13 +30,13 @@ func TestGet(t *testing.T, cons Constructor, min, max uint64) {
// With storage ID.
gPrm.StorageID = objects[i].storageID
res, err := s.Get(gPrm)
res, err := s.Get(context.Background(), gPrm)
require.NoError(t, err)
require.Equal(t, objects[i].obj, res.Object)
// Without storage ID.
gPrm.StorageID = nil
res, err = s.Get(gPrm)
res, err = s.Get(context.Background(), gPrm)
require.NoError(t, err)
require.Equal(t, objects[i].obj, res.Object)
@ -43,7 +44,7 @@ func TestGet(t *testing.T, cons Constructor, min, max uint64) {
gPrm.StorageID = objects[i].storageID
gPrm.Raw = true
res, err = s.Get(gPrm)
res, err = s.Get(context.Background(), gPrm)
require.NoError(t, err)
require.Equal(t, objects[i].raw, res.RawData)
}

15
pkg/local_object_storage/blobstor/internal/blobstortest/get_range.go
View file

@ -1,6 +1,7 @@
package blobstortest
import (
"context"
"math"
"testing"
@ -20,7 +21,7 @@ func TestGetRange(t *testing.T, cons Constructor, min, max uint64) {
t.Run("missing object", func(t *testing.T) {
gPrm := common.GetRangePrm{Address: oidtest.Address()}
_, err := s.GetRange(gPrm)
_, err := s.GetRange(context.Background(), gPrm)
require.ErrorAs(t, err, new(apistatus.ObjectNotFound))
})
@ -38,14 +39,14 @@ func TestGetRange(t *testing.T, cons Constructor, min, max uint64) {
t.Run("without storage ID", func(t *testing.T) {
// Without storage ID.
res, err := s.GetRange(gPrm)
res, err := s.GetRange(context.Background(), gPrm)
require.NoError(t, err)
require.Equal(t, payload[start:stop], res.Data)
})
t.Run("with storage ID", func(t *testing.T) {
gPrm.StorageID = objects[0].storageID
res, err := s.GetRange(gPrm)
res, err := s.GetRange(context.Background(), gPrm)
require.NoError(t, err)
require.Equal(t, payload[start:stop], res.Data)
})
@ -54,7 +55,7 @@ func TestGetRange(t *testing.T, cons Constructor, min, max uint64) {
gPrm.Range.SetOffset(uint64(len(payload) + 10))
gPrm.Range.SetLength(10)
_, err := s.GetRange(gPrm)
_, err := s.GetRange(context.Background(), gPrm)
require.ErrorAs(t, err, new(apistatus.ObjectOutOfRange))
})
@ -62,7 +63,7 @@ func TestGetRange(t *testing.T, cons Constructor, min, max uint64) {
gPrm.Range.SetOffset(10)
gPrm.Range.SetLength(uint64(len(payload)))
_, err := s.GetRange(gPrm)
_, err := s.GetRange(context.Background(), gPrm)
require.ErrorAs(t, err, new(apistatus.ObjectOutOfRange))
})
@ -70,7 +71,7 @@ func TestGetRange(t *testing.T, cons Constructor, min, max uint64) {
gPrm.Range.SetOffset(0)
gPrm.Range.SetLength(1 << 63)
_, err := s.GetRange(gPrm)
_, err := s.GetRange(context.Background(), gPrm)
require.ErrorAs(t, err, new(apistatus.ObjectOutOfRange))
})
@ -78,7 +79,7 @@ func TestGetRange(t *testing.T, cons Constructor, min, max uint64) {
gPrm.Range.SetOffset(10)
gPrm.Range.SetLength(math.MaxUint64 - 2)
_, err := s.GetRange(gPrm)
_, err := s.GetRange(context.Background(), gPrm)
require.ErrorAs(t, err, new(apistatus.ObjectOutOfRange))
})
}

9
pkg/local_object_storage/blobstor/memstore/memstore.go
View file

@ -2,6 +2,7 @@
package memstore
import (
"context"
"fmt"
"sync"
@ -32,7 +33,7 @@ func New(opts ...Option) common.Storage {
return st
}
func (s *memstoreImpl) Get(req common.GetPrm) (common.GetRes, error) {
func (s *memstoreImpl) Get(_ context.Context, req common.GetPrm) (common.GetRes, error) {
key := req.Address.EncodeToString()
s.mu.RLock()
@ -58,8 +59,8 @@ func (s *memstoreImpl) Get(req common.GetPrm) (common.GetRes, error) {
return common.GetRes{Object: obj, RawData: data}, nil
}
func (s *memstoreImpl) GetRange(req common.GetRangePrm) (common.GetRangeRes, error) {
getResp, err := s.Get(common.GetPrm{
func (s *memstoreImpl) GetRange(ctx context.Context, req common.GetRangePrm) (common.GetRangeRes, error) {
getResp, err := s.Get(ctx, common.GetPrm{
Address: req.Address,
StorageID: req.StorageID,
})
@ -80,7 +81,7 @@ func (s *memstoreImpl) GetRange(req common.GetRangePrm) (common.GetRangeRes, err
}, nil
}
func (s *memstoreImpl) Exists(req common.ExistsPrm) (common.ExistsRes, error) {
func (s *memstoreImpl) Exists(_ context.Context, req common.ExistsPrm) (common.ExistsRes, error) {
key := req.Address.EncodeToString()
s.mu.RLock()

9
pkg/local_object_storage/blobstor/memstore/memstore_test.go
View file

@ -1,6 +1,7 @@
package memstore
import (
"context"
"testing"
"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/core/object"
@ -32,13 +33,13 @@ func TestSimpleLifecycle(t *testing.T) {
}
{
resp, err := s.Exists(common.ExistsPrm{Address: addr})
resp, err := s.Exists(context.Background(), common.ExistsPrm{Address: addr})
require.NoError(t, err)
require.True(t, resp.Exists)
}
{
resp, err := s.Get(common.GetPrm{Address: addr})
resp, err := s.Get(context.Background(), common.GetPrm{Address: addr})
require.NoError(t, err)
require.Equal(t, obj.Payload(), resp.Object.Payload())
}
@ -47,7 +48,7 @@ func TestSimpleLifecycle(t *testing.T) {
var objRange objectSDK.Range
objRange.SetOffset(256)
objRange.SetLength(512)
resp, err := s.GetRange(common.GetRangePrm{
resp, err := s.GetRange(context.Background(), common.GetRangePrm{
Address: addr,
Range: objRange,
})
@ -61,7 +62,7 @@ func TestSimpleLifecycle(t *testing.T) {
}
{
resp, err := s.Exists(common.ExistsPrm{Address: addr})
resp, err := s.Exists(context.Background(), common.ExistsPrm{Address: addr})
require.NoError(t, err)
require.False(t, resp.Exists)
}

3
pkg/local_object_storage/blobstor/perf_test.go
View file

@ -1,6 +1,7 @@
package blobstor
import (
"context"
"fmt"
"os"
"testing"
@ -127,7 +128,7 @@ func BenchmarkSubstorageReadPerf(b *testing.B) {
b.ResetTimer()
b.RunParallel(func(pb *testing.PB) {
for pb.Next() {
_, err := st.Get(common.GetPrm{Address: addrGen.Next()})
_, err := st.Get(context.Background(), common.GetPrm{Address: addrGen.Next()})
require.NoError(b, err)
}
})

13
pkg/local_object_storage/blobstor/teststore/teststore.go
View file

@ -13,6 +13,7 @@
package teststore
import (
"context"
"errors"
"fmt"
"sync"
@ -140,36 +141,36 @@ func (s *TestStore) SetReportErrorFunc(f func(string, error)) {
}
}
func (s *TestStore) Get(req common.GetPrm) (common.GetRes, error) {
func (s *TestStore) Get(ctx context.Context, req common.GetPrm) (common.GetRes, error) {
switch {
case s.overrides.Get != nil:
return s.overrides.Get(req)
case s.st != nil:
return s.st.Get(req)
return s.st.Get(ctx, req)
default:
panic(fmt.Sprintf("unexpected storage call: Get(%+v)", req))
}
}
func (s *TestStore) GetRange(req common.GetRangePrm) (common.GetRangeRes, error) {
func (s *TestStore) GetRange(ctx context.Context, req common.GetRangePrm) (common.GetRangeRes, error) {
s.mu.RLock()
defer s.mu.RUnlock()
switch {
case s.overrides.GetRange != nil:
return s.overrides.GetRange(req)
case s.st != nil:
return s.st.GetRange(req)
return s.st.GetRange(ctx, req)
default:
panic(fmt.Sprintf("unexpected storage call: GetRange(%+v)", req))
}
}
func (s *TestStore) Exists(req common.ExistsPrm) (common.ExistsRes, error) {
func (s *TestStore) Exists(ctx context.Context, req common.ExistsPrm) (common.ExistsRes, error) {
switch {
case s.overrides.Exists != nil:
return s.overrides.Exists(req)
case s.st != nil:
return s.st.Exists(req)
return s.st.Exists(ctx, req)
default:
panic(fmt.Sprintf("unexpected storage call: Exists(%+v)", req))
}