138 lines
4.2 KiB
Go
138 lines
4.2 KiB
Go
package getsvc
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
|
|
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() {
|
|
if !exec.canAssemble() {
|
|
exec.log.Debug("can not assemble the object")
|
|
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("trying to assemble the object...")
|
|
|
|
assembler := newAssembler(exec.address(), exec.splitInfo(), exec.ctxRange(), exec)
|
|
|
|
exec.log.Debug("assembling splitted object...",
|
|
zap.Stringer("address", exec.address()),
|
|
zap.Uint64("range_offset", exec.ctxRange().GetOffset()),
|
|
zap.Uint64("range_length", exec.ctxRange().GetLength()),
|
|
)
|
|
defer exec.log.Debug("assembling splitted object completed",
|
|
zap.Stringer("address", exec.address()),
|
|
zap.Uint64("range_offset", exec.ctxRange().GetOffset()),
|
|
zap.Uint64("range_length", exec.ctxRange().GetLength()),
|
|
)
|
|
|
|
obj, err := assembler.Assemble(exec.context(), exec.prm.objWriter)
|
|
if err != nil {
|
|
exec.log.Warn("failed to assemble splitted object",
|
|
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) {
|
|
p := exec.prm
|
|
p.common = p.common.WithLocalOnly(false)
|
|
p.addr.SetContainer(exec.containerID())
|
|
p.addr.SetObject(id)
|
|
|
|
prm := HeadPrm{
|
|
commonPrm: p.commonPrm,
|
|
}
|
|
|
|
w := NewSimpleObjectWriter()
|
|
prm.SetHeaderWriter(w)
|
|
//nolint: contextcheck
|
|
err := exec.svc.Head(exec.context(), prm)
|
|
|
|
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.SetRange(rng)
|
|
|
|
p.addr.SetContainer(exec.containerID())
|
|
p.addr.SetObject(id)
|
|
|
|
//nolint: contextcheck
|
|
statusError := exec.svc.get(exec.context(), p.commonPrm, withPayloadRange(rng))
|
|
|
|
if statusError.err != nil {
|
|
return nil, statusError.err
|
|
}
|
|
return w.Object(), nil
|
|
}
|