frostfs-node/pkg/services/object/get/assemble.go
Dmitrii Stepanov 1440450606 [#277] getsvc: Drop cyclic struct dependency
Drop cyclic dependency between execCtx and Service.

Signed-off-by: Dmitrii Stepanov <d.stepanov@yadro.com>
2023-04-28 14:03:12 +00:00

151 lines
4.6 KiB
Go

package getsvc
import (
"context"
"errors"
"git.frostfs.info/TrueCloudLab/frostfs-node/internal/logs"
apistatus "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/client/status"
objectSDK "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/object"
oid "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/object/id"
"go.uber.org/zap"
)
func (exec *execCtx) assemble(ctx context.Context) {
if !exec.canAssemble() {
exec.log.Debug(logs.GetCanNotAssembleTheObject)
return
}
// Any access tokens are not expected to be used in the assembly process:
// - there is no requirement to specify child objects in session/bearer
// token for `GET`/`GETRANGE`/`RANGEHASH` requests in the API protocol,
// and, therefore, their missing in the original request should not be
// considered as error; on the other hand, without session for every child
// object, it is impossible to attach bearer token in the new generated
// requests correctly because the token has not been issued for that node's
// key;
// - the assembly process is expected to be handled on a container node
// only since the requests forwarding mechanism presentation; such the
// node should have enough rights for getting any child object by design.
exec.prm.common.ForgetTokens()
// Do not use forwarding during assembly stage.
// Request forwarding closure inherited in produced
// `execCtx` so it should be disabled there.
exec.disableForwarding()
exec.log.Debug(logs.GetTryingToAssembleTheObject)
assembler := newAssembler(exec.address(), exec.splitInfo(), exec.ctxRange(), exec)
exec.log.Debug(logs.GetAssemblingSplittedObject,
zap.Stringer("address", exec.address()),
zap.Uint64("range_offset", exec.ctxRange().GetOffset()),
zap.Uint64("range_length", exec.ctxRange().GetLength()),
)
defer exec.log.Debug(logs.GetAssemblingSplittedObjectCompleted,
zap.Stringer("address", exec.address()),
zap.Uint64("range_offset", exec.ctxRange().GetOffset()),
zap.Uint64("range_length", exec.ctxRange().GetLength()),
)
obj, err := assembler.Assemble(ctx, exec.prm.objWriter)
if err != nil {
exec.log.Warn(logs.GetFailedToAssembleSplittedObject,
zap.Error(err),
zap.Stringer("address", exec.address()),
zap.Uint64("range_offset", exec.ctxRange().GetOffset()),
zap.Uint64("range_length", exec.ctxRange().GetLength()),
)
}
var errSplitInfo *objectSDK.SplitInfoError
var errRemovedRemote *apistatus.ObjectAlreadyRemoved
var errOutOfRangeRemote *apistatus.ObjectOutOfRange
var errRemovedLocal apistatus.ObjectAlreadyRemoved
var errOutOfRangeLocal apistatus.ObjectOutOfRange
switch {
default:
exec.status = statusUndefined
exec.err = err
case err == nil:
exec.status = statusOK
exec.err = nil
exec.collectedObject = obj
case errors.As(err, &errRemovedRemote):
exec.status = statusINHUMED
exec.err = errRemovedRemote
case errors.As(err, &errRemovedLocal):
exec.status = statusINHUMED
exec.err = errRemovedLocal
case errors.As(err, &errSplitInfo):
exec.status = statusVIRTUAL
exec.err = errSplitInfo
case errors.As(err, &errOutOfRangeRemote):
exec.status = statusOutOfRange
exec.err = errOutOfRangeRemote
case errors.As(err, &errOutOfRangeLocal):
exec.status = statusOutOfRange
exec.err = errOutOfRangeLocal
}
}
func equalAddresses(a, b oid.Address) bool {
return a.Container().Equals(b.Container()) && a.Object().Equals(b.Object())
}
func (exec *execCtx) HeadObject(ctx context.Context, id oid.ID) (*objectSDK.Object, error) {
w := NewSimpleObjectWriter()
p := RequestParameters{}
p.common = p.common.WithLocalOnly(false)
p.addr.SetContainer(exec.containerID())
p.addr.SetObject(id)
p.head = true
p.SetHeaderWriter(w)
err := exec.getDetached(ctx, p)
if err != nil {
return nil, err
}
return w.Object(), nil
}
func (exec *execCtx) GetObject(ctx context.Context, id oid.ID, rng *objectSDK.Range) (*objectSDK.Object, error) {
w := NewSimpleObjectWriter()
p := exec.prm
p.common = p.common.WithLocalOnly(false)
p.objWriter = w
p.rng = rng
p.addr.SetContainer(exec.containerID())
p.addr.SetObject(id)
if err := exec.getDetached(ctx, p); err != nil {
return nil, err
}
return w.Object(), nil
}
func (exec *execCtx) getDetached(ctx context.Context, prm RequestParameters) error {
detachedExecutor := &execCtx{
keyStore: exec.keyStore,
traverserGenerator: exec.traverserGenerator,
remoteStorageConstructor: exec.remoteStorageConstructor,
epochSource: exec.epochSource,
localStorage: exec.localStorage,
prm: prm,
infoSplit: objectSDK.NewSplitInfo(),
log: exec.log,
}
detachedExecutor.execute(ctx)
return detachedExecutor.statusError.err
}