forked from TrueCloudLab/frostfs-node
Compare commits
9 commits
a4e10975fc
...
ff1e7b531f
Author | SHA1 | Date | |
---|---|---|---|
ff1e7b531f | |||
5ffd1541a8 | |||
56f723e4fe | |||
4440eeb0e1 | |||
efa22b6ded | |||
3c50022aba | |||
0cf5d70177 | |||
1bb4eb1ec0 | |||
8a57c78f5f |
9 changed files with 25 additions and 26 deletions
|
@ -45,6 +45,8 @@ func (r ListContainersRes) Containers() []cid.ID {
|
|||
//
|
||||
// Returns an error if executions are blocked (see BlockExecution).
|
||||
func (e *StorageEngine) ContainerSize(ctx context.Context, prm ContainerSizePrm) (res ContainerSizeRes, err error) {
|
||||
defer elapsed("ContainerSize", e.metrics.AddMethodDuration)()
|
||||
|
||||
err = e.execIfNotBlocked(func() error {
|
||||
res, err = e.containerSize(ctx, prm)
|
||||
return err
|
||||
|
@ -68,8 +70,6 @@ func ContainerSize(ctx context.Context, e *StorageEngine, id cid.ID) (uint64, er
|
|||
}
|
||||
|
||||
func (e *StorageEngine) containerSize(ctx context.Context, prm ContainerSizePrm) (res ContainerSizeRes, err error) {
|
||||
defer elapsed("EstimateContainerSize", e.metrics.AddMethodDuration)()
|
||||
|
||||
e.iterateOverUnsortedShards(func(sh hashedShard) (stop bool) {
|
||||
var csPrm shard.ContainerSizePrm
|
||||
csPrm.SetContainerID(prm.cnr)
|
||||
|
@ -93,6 +93,8 @@ func (e *StorageEngine) containerSize(ctx context.Context, prm ContainerSizePrm)
|
|||
//
|
||||
// Returns an error if executions are blocked (see BlockExecution).
|
||||
func (e *StorageEngine) ListContainers(ctx context.Context, _ ListContainersPrm) (res ListContainersRes, err error) {
|
||||
defer elapsed("ListContainers", e.metrics.AddMethodDuration)()
|
||||
|
||||
err = e.execIfNotBlocked(func() error {
|
||||
res, err = e.listContainers(ctx)
|
||||
return err
|
||||
|
@ -114,8 +116,6 @@ func ListContainers(ctx context.Context, e *StorageEngine) ([]cid.ID, error) {
|
|||
}
|
||||
|
||||
func (e *StorageEngine) listContainers(ctx context.Context) (ListContainersRes, error) {
|
||||
defer elapsed("ListContainers", e.metrics.AddMethodDuration)()
|
||||
|
||||
uniqueIDs := make(map[string]cid.ID)
|
||||
|
||||
e.iterateOverUnsortedShards(func(sh hashedShard) (stop bool) {
|
||||
|
|
|
@ -58,6 +58,7 @@ func (e *StorageEngine) Delete(ctx context.Context, prm DeletePrm) (res DeleteRe
|
|||
attribute.Bool("force_removal", prm.forceRemoval),
|
||||
))
|
||||
defer span.End()
|
||||
defer elapsed("Delete", e.metrics.AddMethodDuration)()
|
||||
|
||||
err = e.execIfNotBlocked(func() error {
|
||||
res, err = e.delete(ctx, prm)
|
||||
|
@ -68,8 +69,6 @@ func (e *StorageEngine) Delete(ctx context.Context, prm DeletePrm) (res DeleteRe
|
|||
}
|
||||
|
||||
func (e *StorageEngine) delete(ctx context.Context, prm DeletePrm) (DeleteRes, error) {
|
||||
defer elapsed("Delete", e.metrics.AddMethodDuration)()
|
||||
|
||||
var locked struct {
|
||||
is bool
|
||||
}
|
||||
|
|
|
@ -56,6 +56,7 @@ func (e *StorageEngine) Get(ctx context.Context, prm GetPrm) (res GetRes, err er
|
|||
attribute.String("address", prm.addr.EncodeToString()),
|
||||
))
|
||||
defer span.End()
|
||||
defer elapsed("Get", e.metrics.AddMethodDuration)()
|
||||
|
||||
err = e.execIfNotBlocked(func() error {
|
||||
res, err = e.get(ctx, prm)
|
||||
|
@ -66,8 +67,6 @@ func (e *StorageEngine) Get(ctx context.Context, prm GetPrm) (res GetRes, err er
|
|||
}
|
||||
|
||||
func (e *StorageEngine) get(ctx context.Context, prm GetPrm) (GetRes, error) {
|
||||
defer elapsed("Get", e.metrics.AddMethodDuration)()
|
||||
|
||||
errNotFound := new(apistatus.ObjectNotFound)
|
||||
|
||||
var shPrm shard.GetPrm
|
||||
|
|
|
@ -70,6 +70,7 @@ var errInhumeFailure = errors.New("inhume operation failed")
|
|||
func (e *StorageEngine) Inhume(ctx context.Context, prm InhumePrm) (res InhumeRes, err error) {
|
||||
ctx, span := tracing.StartSpanFromContext(ctx, "StorageEngine.Inhume")
|
||||
defer span.End()
|
||||
defer elapsed("Inhume", e.metrics.AddMethodDuration)()
|
||||
|
||||
err = e.execIfNotBlocked(func() error {
|
||||
res, err = e.inhume(ctx, prm)
|
||||
|
@ -80,8 +81,6 @@ func (e *StorageEngine) Inhume(ctx context.Context, prm InhumePrm) (res InhumeRe
|
|||
}
|
||||
|
||||
func (e *StorageEngine) inhume(ctx context.Context, prm InhumePrm) (InhumeRes, error) {
|
||||
defer elapsed("Inhume", e.metrics.AddMethodDuration)()
|
||||
|
||||
var shPrm shard.InhumePrm
|
||||
if prm.forceRemoval {
|
||||
shPrm.ForceRemoval()
|
||||
|
|
|
@ -7,6 +7,7 @@ import (
|
|||
|
||||
objectcore "git.frostfs.info/TrueCloudLab/frostfs-node/pkg/core/object"
|
||||
"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/local_object_storage/shard"
|
||||
"git.frostfs.info/TrueCloudLab/frostfs-observability/tracing"
|
||||
)
|
||||
|
||||
// ErrEndOfListing is returned from an object listing with cursor
|
||||
|
@ -98,6 +99,10 @@ func (l ListWithCursorRes) Cursor() *Cursor {
|
|||
// Returns ErrEndOfListing if there are no more objects to return or count
|
||||
// parameter set to zero.
|
||||
func (e *StorageEngine) ListWithCursor(ctx context.Context, prm ListWithCursorPrm) (ListWithCursorRes, error) {
|
||||
ctx, span := tracing.StartSpanFromContext(ctx, "StorageEngine.ListWithCursor")
|
||||
defer span.End()
|
||||
defer elapsed("ListWithCursor", e.metrics.AddMethodDuration)()
|
||||
|
||||
result := make([]objectcore.Info, 0, prm.count)
|
||||
|
||||
// Set initial cursors
|
||||
|
|
|
@ -32,6 +32,7 @@ func (e *StorageEngine) Lock(ctx context.Context, idCnr cid.ID, locker oid.ID, l
|
|||
attribute.Int("locked_count", len(locked)),
|
||||
))
|
||||
defer span.End()
|
||||
defer elapsed("Lock", e.metrics.AddMethodDuration)()
|
||||
|
||||
return e.execIfNotBlocked(func() error {
|
||||
return e.lock(ctx, idCnr, locker, locked)
|
||||
|
|
|
@ -56,6 +56,7 @@ func (e *StorageEngine) Put(ctx context.Context, prm PutPrm) (err error) {
|
|||
attribute.String("address", object.AddressOf(prm.Object).EncodeToString()),
|
||||
))
|
||||
defer span.End()
|
||||
defer elapsed("Put", e.metrics.AddMethodDuration)()
|
||||
|
||||
err = e.execIfNotBlocked(func() error {
|
||||
err = e.put(ctx, prm)
|
||||
|
@ -66,8 +67,6 @@ func (e *StorageEngine) Put(ctx context.Context, prm PutPrm) (err error) {
|
|||
}
|
||||
|
||||
func (e *StorageEngine) put(ctx context.Context, prm PutPrm) error {
|
||||
defer elapsed("Put", e.metrics.AddMethodDuration)()
|
||||
|
||||
addr := object.AddressOf(prm.Object)
|
||||
|
||||
// In #1146 this check was parallelized, however, it became
|
||||
|
|
|
@ -65,6 +65,15 @@ func (r RngRes) Object() *objectSDK.Object {
|
|||
//
|
||||
// Returns an error if executions are blocked (see BlockExecution).
|
||||
func (e *StorageEngine) GetRange(ctx context.Context, prm RngPrm) (res RngRes, err error) {
|
||||
ctx, span := tracing.StartSpanFromContext(ctx, "StorageEngine.getRange",
|
||||
trace.WithAttributes(
|
||||
attribute.String("address", prm.addr.EncodeToString()),
|
||||
attribute.String("offset", strconv.FormatUint(prm.off, 10)),
|
||||
attribute.String("length", strconv.FormatUint(prm.ln, 10)),
|
||||
))
|
||||
defer span.End()
|
||||
defer elapsed("GetRange", e.metrics.AddMethodDuration)()
|
||||
|
||||
err = e.execIfNotBlocked(func() error {
|
||||
res, err = e.getRange(ctx, prm)
|
||||
return err
|
||||
|
@ -74,16 +83,6 @@ func (e *StorageEngine) GetRange(ctx context.Context, prm RngPrm) (res RngRes, e
|
|||
}
|
||||
|
||||
func (e *StorageEngine) getRange(ctx context.Context, prm RngPrm) (RngRes, error) {
|
||||
ctx, span := tracing.StartSpanFromContext(ctx, "StorageEngine.getRange",
|
||||
trace.WithAttributes(
|
||||
attribute.String("address", prm.addr.EncodeToString()),
|
||||
attribute.String("offset", strconv.FormatUint(prm.off, 10)),
|
||||
attribute.String("length", strconv.FormatUint(prm.ln, 10)),
|
||||
))
|
||||
defer span.End()
|
||||
|
||||
defer elapsed("GetRange", e.metrics.AddMethodDuration)()
|
||||
|
||||
var shPrm shard.RngPrm
|
||||
shPrm.SetAddress(prm.addr)
|
||||
shPrm.SetRange(prm.off, prm.ln)
|
||||
|
|
|
@ -51,6 +51,7 @@ func (e *StorageEngine) Select(ctx context.Context, prm SelectPrm) (res SelectRe
|
|||
attribute.String("container_id", prm.cnr.EncodeToString()),
|
||||
))
|
||||
defer span.End()
|
||||
defer elapsed("Select", e.metrics.AddMethodDuration)()
|
||||
|
||||
err = e.execIfNotBlocked(func() error {
|
||||
res, err = e._select(ctx, prm)
|
||||
|
@ -61,8 +62,6 @@ func (e *StorageEngine) Select(ctx context.Context, prm SelectPrm) (res SelectRe
|
|||
}
|
||||
|
||||
func (e *StorageEngine) _select(ctx context.Context, prm SelectPrm) (SelectRes, error) {
|
||||
defer elapsed("Search", e.metrics.AddMethodDuration)()
|
||||
|
||||
addrList := make([]oid.Address, 0)
|
||||
uniqueMap := make(map[string]struct{})
|
||||
|
||||
|
@ -99,6 +98,7 @@ func (e *StorageEngine) _select(ctx context.Context, prm SelectPrm) (SelectRes,
|
|||
//
|
||||
// Returns an error if executions are blocked (see BlockExecution).
|
||||
func (e *StorageEngine) List(ctx context.Context, limit uint64) (res SelectRes, err error) {
|
||||
defer elapsed("List", e.metrics.AddMethodDuration)()
|
||||
err = e.execIfNotBlocked(func() error {
|
||||
res, err = e.list(ctx, limit)
|
||||
return err
|
||||
|
@ -108,8 +108,6 @@ func (e *StorageEngine) List(ctx context.Context, limit uint64) (res SelectRes,
|
|||
}
|
||||
|
||||
func (e *StorageEngine) list(ctx context.Context, limit uint64) (SelectRes, error) {
|
||||
defer elapsed("ListObjects", e.metrics.AddMethodDuration)()
|
||||
|
||||
addrList := make([]oid.Address, 0, limit)
|
||||
uniqueMap := make(map[string]struct{})
|
||||
ln := uint64(0)
|
||||
|
|
Loading…
Reference in a new issue