forked from TrueCloudLab/frostfs-node
[#6] services/object: Remove useless helpers
We have lots of small _private_ methods on `execCtx` whose sole purpose is to just return a struct field. Signed-off-by: Evgenii Stratonikov <e.stratonikov@yadro.com>
This commit is contained in:
parent
9a4f40626c
commit
882d010870
15 changed files with 88 additions and 182 deletions
|
@ -5,9 +5,7 @@ import (
|
||||||
"strconv"
|
"strconv"
|
||||||
|
|
||||||
objectV2 "git.frostfs.info/TrueCloudLab/frostfs-api-go/v2/object"
|
objectV2 "git.frostfs.info/TrueCloudLab/frostfs-api-go/v2/object"
|
||||||
"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/services/object/util"
|
|
||||||
"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/util/logger"
|
"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/util/logger"
|
||||||
cid "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/container/id"
|
|
||||||
"git.frostfs.info/TrueCloudLab/frostfs-sdk-go/object"
|
"git.frostfs.info/TrueCloudLab/frostfs-sdk-go/object"
|
||||||
oid "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/object/id"
|
oid "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/object/id"
|
||||||
"go.uber.org/zap"
|
"go.uber.org/zap"
|
||||||
|
@ -45,37 +43,17 @@ const (
|
||||||
func (exec *execCtx) setLogger(l *logger.Logger) {
|
func (exec *execCtx) setLogger(l *logger.Logger) {
|
||||||
exec.log = &logger.Logger{Logger: l.With(
|
exec.log = &logger.Logger{Logger: l.With(
|
||||||
zap.String("request", "DELETE"),
|
zap.String("request", "DELETE"),
|
||||||
zap.Stringer("address", exec.address()),
|
zap.Stringer("address", exec.prm.addr),
|
||||||
zap.Bool("local", exec.isLocal()),
|
zap.Bool("local", exec.prm.common.LocalOnly()),
|
||||||
zap.Bool("with session", exec.prm.common.SessionToken() != nil),
|
zap.Bool("with session", exec.prm.common.SessionToken() != nil),
|
||||||
zap.Bool("with bearer", exec.prm.common.BearerToken() != nil),
|
zap.Bool("with bearer", exec.prm.common.BearerToken() != nil),
|
||||||
)}
|
)}
|
||||||
}
|
}
|
||||||
|
|
||||||
func (exec execCtx) context() context.Context {
|
|
||||||
return exec.ctx
|
|
||||||
}
|
|
||||||
|
|
||||||
func (exec execCtx) isLocal() bool {
|
|
||||||
return exec.prm.common.LocalOnly()
|
|
||||||
}
|
|
||||||
|
|
||||||
func (exec *execCtx) address() oid.Address {
|
|
||||||
return exec.prm.addr
|
|
||||||
}
|
|
||||||
|
|
||||||
func (exec *execCtx) containerID() cid.ID {
|
|
||||||
return exec.prm.addr.Container()
|
|
||||||
}
|
|
||||||
|
|
||||||
func (exec *execCtx) commonParameters() *util.CommonPrm {
|
|
||||||
return exec.prm.common
|
|
||||||
}
|
|
||||||
|
|
||||||
func (exec *execCtx) newAddress(id oid.ID) oid.Address {
|
func (exec *execCtx) newAddress(id oid.ID) oid.Address {
|
||||||
var a oid.Address
|
var a oid.Address
|
||||||
a.SetObject(id)
|
a.SetObject(id)
|
||||||
a.SetContainer(exec.containerID())
|
a.SetContainer(exec.prm.addr.Container())
|
||||||
|
|
||||||
return a
|
return a
|
||||||
}
|
}
|
||||||
|
@ -241,11 +219,11 @@ func (exec *execCtx) initTombstoneObject() bool {
|
||||||
}
|
}
|
||||||
|
|
||||||
exec.tombstoneObj = object.New()
|
exec.tombstoneObj = object.New()
|
||||||
exec.tombstoneObj.SetContainerID(exec.containerID())
|
exec.tombstoneObj.SetContainerID(exec.prm.addr.Container())
|
||||||
exec.tombstoneObj.SetType(object.TypeTombstone)
|
exec.tombstoneObj.SetType(object.TypeTombstone)
|
||||||
exec.tombstoneObj.SetPayload(payload)
|
exec.tombstoneObj.SetPayload(payload)
|
||||||
|
|
||||||
tokenSession := exec.commonParameters().SessionToken()
|
tokenSession := exec.prm.common.SessionToken()
|
||||||
if tokenSession != nil {
|
if tokenSession != nil {
|
||||||
issuer := tokenSession.Issuer()
|
issuer := tokenSession.Issuer()
|
||||||
exec.tombstoneObj.SetOwnerID(&issuer)
|
exec.tombstoneObj.SetOwnerID(&issuer)
|
||||||
|
|
|
@ -36,7 +36,7 @@ func (exec *execCtx) formTombstone() (ok bool) {
|
||||||
exec.tombstone.SetExpirationEpoch(
|
exec.tombstone.SetExpirationEpoch(
|
||||||
exec.svc.netInfo.CurrentEpoch() + tsLifetime,
|
exec.svc.netInfo.CurrentEpoch() + tsLifetime,
|
||||||
)
|
)
|
||||||
exec.addMembers([]oid.ID{exec.address().Object()})
|
exec.addMembers([]oid.ID{exec.prm.addr.Object()})
|
||||||
|
|
||||||
exec.log.Debug("forming split info...")
|
exec.log.Debug("forming split info...")
|
||||||
|
|
||||||
|
|
|
@ -24,12 +24,12 @@ func (w *headSvcWrapper) headAddress(exec *execCtx, addr oid.Address) (*object.O
|
||||||
wr := getsvc.NewSimpleObjectWriter()
|
wr := getsvc.NewSimpleObjectWriter()
|
||||||
|
|
||||||
p := getsvc.HeadPrm{}
|
p := getsvc.HeadPrm{}
|
||||||
p.SetCommonParameters(exec.commonParameters())
|
p.SetCommonParameters(exec.prm.common)
|
||||||
p.SetHeaderWriter(wr)
|
p.SetHeaderWriter(wr)
|
||||||
p.WithRawFlag(true)
|
p.WithRawFlag(true)
|
||||||
p.WithAddress(addr)
|
p.WithAddress(addr)
|
||||||
|
|
||||||
err := (*getsvc.Service)(w).Head(exec.context(), p)
|
err := (*getsvc.Service)(w).Head(exec.ctx, p)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
|
@ -38,7 +38,7 @@ func (w *headSvcWrapper) headAddress(exec *execCtx, addr oid.Address) (*object.O
|
||||||
}
|
}
|
||||||
|
|
||||||
func (w *headSvcWrapper) splitInfo(exec *execCtx) (*object.SplitInfo, error) {
|
func (w *headSvcWrapper) splitInfo(exec *execCtx) (*object.SplitInfo, error) {
|
||||||
_, err := w.headAddress(exec, exec.address())
|
_, err := w.headAddress(exec, exec.prm.addr)
|
||||||
|
|
||||||
var errSplitInfo *object.SplitInfoError
|
var errSplitInfo *object.SplitInfoError
|
||||||
|
|
||||||
|
@ -89,11 +89,11 @@ func (w *searchSvcWrapper) splitMembers(exec *execCtx) ([]oid.ID, error) {
|
||||||
|
|
||||||
p := searchsvc.Prm{}
|
p := searchsvc.Prm{}
|
||||||
p.SetWriter(wr)
|
p.SetWriter(wr)
|
||||||
p.SetCommonParameters(exec.commonParameters())
|
p.SetCommonParameters(exec.prm.common)
|
||||||
p.WithContainerID(exec.containerID())
|
p.WithContainerID(exec.prm.addr.Container())
|
||||||
p.WithSearchFilters(fs)
|
p.WithSearchFilters(fs)
|
||||||
|
|
||||||
err := (*searchsvc.Service)(w).Search(exec.context(), p)
|
err := (*searchsvc.Service)(w).Search(exec.ctx, p)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
|
@ -108,7 +108,7 @@ func (s *simpleIDWriter) WriteIDs(ids []oid.ID) error {
|
||||||
}
|
}
|
||||||
|
|
||||||
func (w *putSvcWrapper) put(exec *execCtx) (*oid.ID, error) {
|
func (w *putSvcWrapper) put(exec *execCtx) (*oid.ID, error) {
|
||||||
streamer, err := (*putsvc.Service)(w).Put(exec.context())
|
streamer, err := (*putsvc.Service)(w).Put(exec.ctx)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
|
@ -116,7 +116,7 @@ func (w *putSvcWrapper) put(exec *execCtx) (*oid.ID, error) {
|
||||||
payload := exec.tombstoneObj.Payload()
|
payload := exec.tombstoneObj.Payload()
|
||||||
|
|
||||||
initPrm := new(putsvc.PutInitPrm).
|
initPrm := new(putsvc.PutInitPrm).
|
||||||
WithCommonPrm(exec.commonParameters()).
|
WithCommonPrm(exec.prm.common).
|
||||||
WithObject(exec.tombstoneObj.CutPayload())
|
WithObject(exec.tombstoneObj.CutPayload())
|
||||||
|
|
||||||
err = streamer.Init(initPrm)
|
err = streamer.Init(initPrm)
|
||||||
|
|
|
@ -36,26 +36,26 @@ func (exec *execCtx) assemble() {
|
||||||
|
|
||||||
exec.log.Debug("trying to assemble the object...")
|
exec.log.Debug("trying to assemble the object...")
|
||||||
|
|
||||||
assembler := newAssembler(exec.address(), exec.splitInfo(), exec.ctxRange(), exec)
|
assembler := newAssembler(exec.prm.addr, exec.splitInfo, exec.prm.rng, exec)
|
||||||
|
|
||||||
exec.log.Debug("assembling splitted object...",
|
exec.log.Debug("assembling splitted object...",
|
||||||
zap.Stringer("address", exec.address()),
|
zap.Stringer("address", exec.prm.addr),
|
||||||
zap.Uint64("range_offset", exec.ctxRange().GetOffset()),
|
zap.Uint64("range_offset", exec.prm.rng.GetOffset()),
|
||||||
zap.Uint64("range_length", exec.ctxRange().GetLength()),
|
zap.Uint64("range_length", exec.prm.rng.GetLength()),
|
||||||
)
|
)
|
||||||
defer exec.log.Debug("assembling splitted object completed",
|
defer exec.log.Debug("assembling splitted object completed",
|
||||||
zap.Stringer("address", exec.address()),
|
zap.Stringer("address", exec.prm.addr),
|
||||||
zap.Uint64("range_offset", exec.ctxRange().GetOffset()),
|
zap.Uint64("range_offset", exec.prm.rng.GetOffset()),
|
||||||
zap.Uint64("range_length", exec.ctxRange().GetLength()),
|
zap.Uint64("range_length", exec.prm.rng.GetLength()),
|
||||||
)
|
)
|
||||||
|
|
||||||
obj, err := assembler.Assemble(exec.context(), exec.prm.objWriter)
|
obj, err := assembler.Assemble(exec.ctx, exec.prm.objWriter)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
exec.log.Warn("failed to assemble splitted object",
|
exec.log.Warn("failed to assemble splitted object",
|
||||||
zap.Error(err),
|
zap.Error(err),
|
||||||
zap.Stringer("address", exec.address()),
|
zap.Stringer("address", exec.prm.addr),
|
||||||
zap.Uint64("range_offset", exec.ctxRange().GetOffset()),
|
zap.Uint64("range_offset", exec.prm.rng.GetOffset()),
|
||||||
zap.Uint64("range_length", exec.ctxRange().GetLength()),
|
zap.Uint64("range_length", exec.prm.rng.GetLength()),
|
||||||
)
|
)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@ -98,7 +98,7 @@ func equalAddresses(a, b oid.Address) bool {
|
||||||
func (exec *execCtx) HeadObject(ctx context.Context, id oid.ID) (*objectSDK.Object, error) {
|
func (exec *execCtx) HeadObject(ctx context.Context, id oid.ID) (*objectSDK.Object, error) {
|
||||||
p := exec.prm
|
p := exec.prm
|
||||||
p.common = p.common.WithLocalOnly(false)
|
p.common = p.common.WithLocalOnly(false)
|
||||||
p.addr.SetContainer(exec.containerID())
|
p.addr.SetContainer(exec.prm.addr.Container())
|
||||||
p.addr.SetObject(id)
|
p.addr.SetObject(id)
|
||||||
|
|
||||||
prm := HeadPrm{
|
prm := HeadPrm{
|
||||||
|
@ -108,7 +108,7 @@ func (exec *execCtx) HeadObject(ctx context.Context, id oid.ID) (*objectSDK.Obje
|
||||||
w := NewSimpleObjectWriter()
|
w := NewSimpleObjectWriter()
|
||||||
prm.SetHeaderWriter(w)
|
prm.SetHeaderWriter(w)
|
||||||
//nolint: contextcheck
|
//nolint: contextcheck
|
||||||
err := exec.svc.Head(exec.context(), prm)
|
err := exec.svc.Head(exec.ctx, prm)
|
||||||
|
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
|
@ -125,11 +125,11 @@ func (exec *execCtx) GetObject(ctx context.Context, id oid.ID, rng *objectSDK.Ra
|
||||||
p.objWriter = w
|
p.objWriter = w
|
||||||
p.SetRange(rng)
|
p.SetRange(rng)
|
||||||
|
|
||||||
p.addr.SetContainer(exec.containerID())
|
p.addr.SetContainer(exec.prm.addr.Container())
|
||||||
p.addr.SetObject(id)
|
p.addr.SetObject(id)
|
||||||
|
|
||||||
//nolint: contextcheck
|
//nolint: contextcheck
|
||||||
statusError := exec.svc.get(exec.context(), p.commonPrm, withPayloadRange(rng))
|
statusError := exec.svc.get(exec.ctx, p.commonPrm, withPayloadRange(rng))
|
||||||
|
|
||||||
if statusError.err != nil {
|
if statusError.err != nil {
|
||||||
return nil, statusError.err
|
return nil, statusError.err
|
||||||
|
|
|
@ -8,12 +8,12 @@ import (
|
||||||
)
|
)
|
||||||
|
|
||||||
func (exec *execCtx) executeOnContainer() {
|
func (exec *execCtx) executeOnContainer() {
|
||||||
if exec.isLocal() {
|
if exec.prm.common.LocalOnly() {
|
||||||
exec.log.Debug("return result directly")
|
exec.log.Debug("return result directly")
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|
||||||
lookupDepth := exec.netmapLookupDepth()
|
lookupDepth := exec.prm.common.NetmapLookupDepth()
|
||||||
|
|
||||||
exec.log.Debug("trying to execute in container...",
|
exec.log.Debug("trying to execute in container...",
|
||||||
zap.Uint64("netmap lookup depth", lookupDepth),
|
zap.Uint64("netmap lookup depth", lookupDepth),
|
||||||
|
@ -47,12 +47,12 @@ func (exec *execCtx) processCurrentEpoch() bool {
|
||||||
zap.Uint64("number", exec.curProcEpoch),
|
zap.Uint64("number", exec.curProcEpoch),
|
||||||
)
|
)
|
||||||
|
|
||||||
traverser, ok := exec.generateTraverser(exec.address())
|
traverser, ok := exec.generateTraverser(exec.prm.addr)
|
||||||
if !ok {
|
if !ok {
|
||||||
return true
|
return true
|
||||||
}
|
}
|
||||||
|
|
||||||
ctx, cancel := context.WithCancel(exec.context())
|
ctx, cancel := context.WithCancel(exec.ctx)
|
||||||
defer cancel()
|
defer cancel()
|
||||||
|
|
||||||
exec.status = statusUndefined
|
exec.status = statusUndefined
|
||||||
|
|
|
@ -8,7 +8,6 @@ import (
|
||||||
"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/services/object/util"
|
"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/services/object/util"
|
||||||
"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/services/object_manager/placement"
|
"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/services/object_manager/placement"
|
||||||
"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/util/logger"
|
"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/util/logger"
|
||||||
cid "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/container/id"
|
|
||||||
objectSDK "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/object"
|
objectSDK "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/object"
|
||||||
oid "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/object/id"
|
oid "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/object/id"
|
||||||
"go.uber.org/zap"
|
"go.uber.org/zap"
|
||||||
|
@ -29,13 +28,13 @@ type execCtx struct {
|
||||||
|
|
||||||
statusError
|
statusError
|
||||||
|
|
||||||
infoSplit *objectSDK.SplitInfo
|
splitInfo *objectSDK.SplitInfo
|
||||||
|
|
||||||
log *logger.Logger
|
log *logger.Logger
|
||||||
|
|
||||||
collectedObject *objectSDK.Object
|
collectedObject *objectSDK.Object
|
||||||
|
|
||||||
head bool
|
headOnly bool
|
||||||
|
|
||||||
curProcEpoch uint64
|
curProcEpoch uint64
|
||||||
}
|
}
|
||||||
|
@ -52,7 +51,7 @@ const (
|
||||||
|
|
||||||
func headOnly() execOption {
|
func headOnly() execOption {
|
||||||
return func(c *execCtx) {
|
return func(c *execCtx) {
|
||||||
c.head = true
|
c.headOnly = true
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@ -64,38 +63,22 @@ func withPayloadRange(r *objectSDK.Range) execOption {
|
||||||
|
|
||||||
func (exec *execCtx) setLogger(l *logger.Logger) {
|
func (exec *execCtx) setLogger(l *logger.Logger) {
|
||||||
req := "GET"
|
req := "GET"
|
||||||
if exec.headOnly() {
|
if exec.headOnly {
|
||||||
req = "HEAD"
|
req = "HEAD"
|
||||||
} else if exec.ctxRange() != nil {
|
} else if exec.prm.rng != nil {
|
||||||
req = "GET_RANGE"
|
req = "GET_RANGE"
|
||||||
}
|
}
|
||||||
|
|
||||||
exec.log = &logger.Logger{Logger: l.With(
|
exec.log = &logger.Logger{Logger: l.With(
|
||||||
zap.String("request", req),
|
zap.String("request", req),
|
||||||
zap.Stringer("address", exec.address()),
|
zap.Stringer("address", exec.prm.addr),
|
||||||
zap.Bool("raw", exec.isRaw()),
|
zap.Bool("raw", exec.prm.raw),
|
||||||
zap.Bool("local", exec.isLocal()),
|
zap.Bool("local", exec.prm.common.LocalOnly()),
|
||||||
zap.Bool("with session", exec.prm.common.SessionToken() != nil),
|
zap.Bool("with session", exec.prm.common.SessionToken() != nil),
|
||||||
zap.Bool("with bearer", exec.prm.common.BearerToken() != nil),
|
zap.Bool("with bearer", exec.prm.common.BearerToken() != nil),
|
||||||
)}
|
)}
|
||||||
}
|
}
|
||||||
|
|
||||||
func (exec execCtx) context() context.Context {
|
|
||||||
return exec.ctx
|
|
||||||
}
|
|
||||||
|
|
||||||
func (exec execCtx) isLocal() bool {
|
|
||||||
return exec.prm.common.LocalOnly()
|
|
||||||
}
|
|
||||||
|
|
||||||
func (exec execCtx) isRaw() bool {
|
|
||||||
return exec.prm.raw
|
|
||||||
}
|
|
||||||
|
|
||||||
func (exec execCtx) address() oid.Address {
|
|
||||||
return exec.prm.addr
|
|
||||||
}
|
|
||||||
|
|
||||||
func (exec execCtx) key() (*ecdsa.PrivateKey, error) {
|
func (exec execCtx) key() (*ecdsa.PrivateKey, error) {
|
||||||
if exec.prm.signerKey != nil {
|
if exec.prm.signerKey != nil {
|
||||||
// the key has already been requested and
|
// the key has already been requested and
|
||||||
|
@ -116,35 +99,11 @@ func (exec execCtx) key() (*ecdsa.PrivateKey, error) {
|
||||||
}
|
}
|
||||||
|
|
||||||
func (exec *execCtx) canAssemble() bool {
|
func (exec *execCtx) canAssemble() bool {
|
||||||
return !exec.isRaw() && !exec.headOnly()
|
return !exec.prm.raw && !exec.headOnly
|
||||||
}
|
|
||||||
|
|
||||||
func (exec *execCtx) splitInfo() *objectSDK.SplitInfo {
|
|
||||||
return exec.infoSplit
|
|
||||||
}
|
|
||||||
|
|
||||||
func (exec *execCtx) containerID() cid.ID {
|
|
||||||
return exec.address().Container()
|
|
||||||
}
|
|
||||||
|
|
||||||
func (exec *execCtx) ctxRange() *objectSDK.Range {
|
|
||||||
return exec.prm.rng
|
|
||||||
}
|
|
||||||
|
|
||||||
func (exec *execCtx) headOnly() bool {
|
|
||||||
return exec.head
|
|
||||||
}
|
|
||||||
|
|
||||||
func (exec *execCtx) netmapEpoch() uint64 {
|
|
||||||
return exec.prm.common.NetmapEpoch()
|
|
||||||
}
|
|
||||||
|
|
||||||
func (exec *execCtx) netmapLookupDepth() uint64 {
|
|
||||||
return exec.prm.common.NetmapLookupDepth()
|
|
||||||
}
|
}
|
||||||
|
|
||||||
func (exec *execCtx) initEpoch() bool {
|
func (exec *execCtx) initEpoch() bool {
|
||||||
exec.curProcEpoch = exec.netmapEpoch()
|
exec.curProcEpoch = exec.prm.common.NetmapEpoch()
|
||||||
if exec.curProcEpoch > 0 {
|
if exec.curProcEpoch > 0 {
|
||||||
return true
|
return true
|
||||||
}
|
}
|
||||||
|
@ -218,12 +177,12 @@ func mergeSplitInfo(dst, src *objectSDK.SplitInfo) {
|
||||||
}
|
}
|
||||||
|
|
||||||
func (exec *execCtx) writeCollectedHeader() bool {
|
func (exec *execCtx) writeCollectedHeader() bool {
|
||||||
if exec.ctxRange() != nil {
|
if exec.prm.rng != nil {
|
||||||
return true
|
return true
|
||||||
}
|
}
|
||||||
|
|
||||||
err := exec.prm.objWriter.WriteHeader(
|
err := exec.prm.objWriter.WriteHeader(
|
||||||
exec.context(),
|
exec.ctx,
|
||||||
exec.collectedObject.CutPayload(),
|
exec.collectedObject.CutPayload(),
|
||||||
)
|
)
|
||||||
|
|
||||||
|
@ -244,11 +203,11 @@ func (exec *execCtx) writeCollectedHeader() bool {
|
||||||
}
|
}
|
||||||
|
|
||||||
func (exec *execCtx) writeObjectPayload(obj *objectSDK.Object) bool {
|
func (exec *execCtx) writeObjectPayload(obj *objectSDK.Object) bool {
|
||||||
if exec.headOnly() {
|
if exec.headOnly {
|
||||||
return true
|
return true
|
||||||
}
|
}
|
||||||
|
|
||||||
err := exec.prm.objWriter.WriteChunk(exec.context(), obj.Payload())
|
err := exec.prm.objWriter.WriteChunk(exec.ctx, obj.Payload())
|
||||||
|
|
||||||
switch {
|
switch {
|
||||||
default:
|
default:
|
||||||
|
@ -272,12 +231,6 @@ func (exec *execCtx) writeCollectedObject() {
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// isForwardingEnabled returns true if common execution
|
|
||||||
// parameters has request forwarding closure set.
|
|
||||||
func (exec execCtx) isForwardingEnabled() bool {
|
|
||||||
return exec.prm.forwarder != nil
|
|
||||||
}
|
|
||||||
|
|
||||||
// disableForwarding removes request forwarding closure from common
|
// disableForwarding removes request forwarding closure from common
|
||||||
// parameters, so it won't be inherited in new execution contexts.
|
// parameters, so it won't be inherited in new execution contexts.
|
||||||
func (exec *execCtx) disableForwarding() {
|
func (exec *execCtx) disableForwarding() {
|
||||||
|
|
|
@ -69,7 +69,7 @@ func (s *Service) get(ctx context.Context, prm commonPrm, opts ...execOption) st
|
||||||
prm: RangePrm{
|
prm: RangePrm{
|
||||||
commonPrm: prm,
|
commonPrm: prm,
|
||||||
},
|
},
|
||||||
infoSplit: object.NewSplitInfo(),
|
splitInfo: object.NewSplitInfo(),
|
||||||
}
|
}
|
||||||
|
|
||||||
for i := range opts {
|
for i := range opts {
|
||||||
|
|
|
@ -118,7 +118,7 @@ func newTestClient() *testClient {
|
||||||
}
|
}
|
||||||
|
|
||||||
func (c *testClient) getObject(exec *execCtx, _ client.NodeInfo) (*objectSDK.Object, error) {
|
func (c *testClient) getObject(exec *execCtx, _ client.NodeInfo) (*objectSDK.Object, error) {
|
||||||
v, ok := c.results[exec.address().EncodeToString()]
|
v, ok := c.results[exec.prm.addr.EncodeToString()]
|
||||||
if !ok {
|
if !ok {
|
||||||
var errNotFound apistatus.ObjectNotFound
|
var errNotFound apistatus.ObjectNotFound
|
||||||
|
|
||||||
|
@ -129,7 +129,7 @@ func (c *testClient) getObject(exec *execCtx, _ client.NodeInfo) (*objectSDK.Obj
|
||||||
return nil, v.err
|
return nil, v.err
|
||||||
}
|
}
|
||||||
|
|
||||||
return cutToRange(v.obj, exec.ctxRange()), nil
|
return cutToRange(v.obj, exec.prm.rng), nil
|
||||||
}
|
}
|
||||||
|
|
||||||
func (c *testClient) addResult(addr oid.Address, obj *objectSDK.Object, err error) {
|
func (c *testClient) addResult(addr oid.Address, obj *objectSDK.Object, err error) {
|
||||||
|
@ -143,7 +143,7 @@ func (s *testStorage) get(exec *execCtx) (*objectSDK.Object, error) {
|
||||||
var (
|
var (
|
||||||
ok bool
|
ok bool
|
||||||
obj *objectSDK.Object
|
obj *objectSDK.Object
|
||||||
sAddr = exec.address().EncodeToString()
|
sAddr = exec.prm.addr.EncodeToString()
|
||||||
)
|
)
|
||||||
|
|
||||||
if _, ok = s.inhumed[sAddr]; ok {
|
if _, ok = s.inhumed[sAddr]; ok {
|
||||||
|
@ -157,7 +157,7 @@ func (s *testStorage) get(exec *execCtx) (*objectSDK.Object, error) {
|
||||||
}
|
}
|
||||||
|
|
||||||
if obj, ok = s.phy[sAddr]; ok {
|
if obj, ok = s.phy[sAddr]; ok {
|
||||||
return cutToRange(obj, exec.ctxRange()), nil
|
return cutToRange(obj, exec.prm.rng), nil
|
||||||
}
|
}
|
||||||
|
|
||||||
var errNotFound apistatus.ObjectNotFound
|
var errNotFound apistatus.ObjectNotFound
|
||||||
|
|
|
@ -34,8 +34,8 @@ func (exec *execCtx) executeLocal() {
|
||||||
exec.err = errRemoved
|
exec.err = errRemoved
|
||||||
case errors.As(err, &errSplitInfo):
|
case errors.As(err, &errSplitInfo):
|
||||||
exec.status = statusVIRTUAL
|
exec.status = statusVIRTUAL
|
||||||
mergeSplitInfo(exec.splitInfo(), errSplitInfo.SplitInfo())
|
mergeSplitInfo(exec.splitInfo, errSplitInfo.SplitInfo())
|
||||||
exec.err = objectSDK.NewSplitInfoError(exec.infoSplit)
|
exec.err = objectSDK.NewSplitInfoError(exec.splitInfo)
|
||||||
case errors.As(err, &errOutOfRange):
|
case errors.As(err, &errOutOfRange):
|
||||||
exec.status = statusOutOfRange
|
exec.status = statusOutOfRange
|
||||||
exec.err = errOutOfRange
|
exec.err = errOutOfRange
|
||||||
|
|
|
@ -54,8 +54,8 @@ func (exec *execCtx) processNode(ctx context.Context, info client.NodeInfo) bool
|
||||||
exec.err = errOutOfRange
|
exec.err = errOutOfRange
|
||||||
case errors.As(err, &errSplitInfo):
|
case errors.As(err, &errSplitInfo):
|
||||||
exec.status = statusVIRTUAL
|
exec.status = statusVIRTUAL
|
||||||
mergeSplitInfo(exec.splitInfo(), errSplitInfo.SplitInfo())
|
mergeSplitInfo(exec.splitInfo, errSplitInfo.SplitInfo())
|
||||||
exec.err = objectSDK.NewSplitInfoError(exec.infoSplit)
|
exec.err = objectSDK.NewSplitInfoError(exec.splitInfo)
|
||||||
}
|
}
|
||||||
|
|
||||||
return exec.status != statusUndefined
|
return exec.status != statusUndefined
|
||||||
|
|
|
@ -89,7 +89,7 @@ func (c *clientCacheWrapper) get(info coreclient.NodeInfo) (getClient, error) {
|
||||||
|
|
||||||
// nolint: funlen
|
// nolint: funlen
|
||||||
func (c *clientWrapper) getObject(exec *execCtx, info coreclient.NodeInfo) (*object.Object, error) {
|
func (c *clientWrapper) getObject(exec *execCtx, info coreclient.NodeInfo) (*object.Object, error) {
|
||||||
if exec.isForwardingEnabled() {
|
if exec.prm.forwarder != nil {
|
||||||
return exec.prm.forwarder(info, c.client)
|
return exec.prm.forwarder(info, c.client)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@ -98,20 +98,20 @@ func (c *clientWrapper) getObject(exec *execCtx, info coreclient.NodeInfo) (*obj
|
||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
|
|
||||||
if exec.headOnly() {
|
if exec.headOnly {
|
||||||
var prm internalclient.HeadObjectPrm
|
var prm internalclient.HeadObjectPrm
|
||||||
|
|
||||||
prm.SetContext(exec.context())
|
prm.SetContext(exec.ctx)
|
||||||
prm.SetClient(c.client)
|
prm.SetClient(c.client)
|
||||||
prm.SetTTL(exec.prm.common.TTL())
|
prm.SetTTL(exec.prm.common.TTL())
|
||||||
prm.SetNetmapEpoch(exec.curProcEpoch)
|
prm.SetNetmapEpoch(exec.curProcEpoch)
|
||||||
prm.SetAddress(exec.address())
|
prm.SetAddress(exec.prm.addr)
|
||||||
prm.SetPrivateKey(key)
|
prm.SetPrivateKey(key)
|
||||||
prm.SetSessionToken(exec.prm.common.SessionToken())
|
prm.SetSessionToken(exec.prm.common.SessionToken())
|
||||||
prm.SetBearerToken(exec.prm.common.BearerToken())
|
prm.SetBearerToken(exec.prm.common.BearerToken())
|
||||||
prm.SetXHeaders(exec.prm.common.XHeaders())
|
prm.SetXHeaders(exec.prm.common.XHeaders())
|
||||||
|
|
||||||
if exec.isRaw() {
|
if exec.prm.raw {
|
||||||
prm.SetRawFlag()
|
prm.SetRawFlag()
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@ -124,21 +124,21 @@ func (c *clientWrapper) getObject(exec *execCtx, info coreclient.NodeInfo) (*obj
|
||||||
}
|
}
|
||||||
// we don't specify payload writer because we accumulate
|
// we don't specify payload writer because we accumulate
|
||||||
// the object locally (even huge).
|
// the object locally (even huge).
|
||||||
if rng := exec.ctxRange(); rng != nil {
|
if rng := exec.prm.rng; rng != nil {
|
||||||
var prm internalclient.PayloadRangePrm
|
var prm internalclient.PayloadRangePrm
|
||||||
|
|
||||||
prm.SetContext(exec.context())
|
prm.SetContext(exec.ctx)
|
||||||
prm.SetClient(c.client)
|
prm.SetClient(c.client)
|
||||||
prm.SetTTL(exec.prm.common.TTL())
|
prm.SetTTL(exec.prm.common.TTL())
|
||||||
prm.SetNetmapEpoch(exec.curProcEpoch)
|
prm.SetNetmapEpoch(exec.curProcEpoch)
|
||||||
prm.SetAddress(exec.address())
|
prm.SetAddress(exec.prm.addr)
|
||||||
prm.SetPrivateKey(key)
|
prm.SetPrivateKey(key)
|
||||||
prm.SetSessionToken(exec.prm.common.SessionToken())
|
prm.SetSessionToken(exec.prm.common.SessionToken())
|
||||||
prm.SetBearerToken(exec.prm.common.BearerToken())
|
prm.SetBearerToken(exec.prm.common.BearerToken())
|
||||||
prm.SetXHeaders(exec.prm.common.XHeaders())
|
prm.SetXHeaders(exec.prm.common.XHeaders())
|
||||||
prm.SetRange(rng)
|
prm.SetRange(rng)
|
||||||
|
|
||||||
if exec.isRaw() {
|
if exec.prm.raw {
|
||||||
prm.SetRawFlag()
|
prm.SetRawFlag()
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@ -175,17 +175,17 @@ func (c *clientWrapper) getObject(exec *execCtx, info coreclient.NodeInfo) (*obj
|
||||||
func (c *clientWrapper) get(exec *execCtx, key *ecdsa.PrivateKey) (*object.Object, error) {
|
func (c *clientWrapper) get(exec *execCtx, key *ecdsa.PrivateKey) (*object.Object, error) {
|
||||||
var prm internalclient.GetObjectPrm
|
var prm internalclient.GetObjectPrm
|
||||||
|
|
||||||
prm.SetContext(exec.context())
|
prm.SetContext(exec.ctx)
|
||||||
prm.SetClient(c.client)
|
prm.SetClient(c.client)
|
||||||
prm.SetTTL(exec.prm.common.TTL())
|
prm.SetTTL(exec.prm.common.TTL())
|
||||||
prm.SetNetmapEpoch(exec.curProcEpoch)
|
prm.SetNetmapEpoch(exec.curProcEpoch)
|
||||||
prm.SetAddress(exec.address())
|
prm.SetAddress(exec.prm.addr)
|
||||||
prm.SetPrivateKey(key)
|
prm.SetPrivateKey(key)
|
||||||
prm.SetSessionToken(exec.prm.common.SessionToken())
|
prm.SetSessionToken(exec.prm.common.SessionToken())
|
||||||
prm.SetBearerToken(exec.prm.common.BearerToken())
|
prm.SetBearerToken(exec.prm.common.BearerToken())
|
||||||
prm.SetXHeaders(exec.prm.common.XHeaders())
|
prm.SetXHeaders(exec.prm.common.XHeaders())
|
||||||
|
|
||||||
if exec.isRaw() {
|
if exec.prm.raw {
|
||||||
prm.SetRawFlag()
|
prm.SetRawFlag()
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@ -198,10 +198,10 @@ func (c *clientWrapper) get(exec *execCtx, key *ecdsa.PrivateKey) (*object.Objec
|
||||||
}
|
}
|
||||||
|
|
||||||
func (e *storageEngineWrapper) get(exec *execCtx) (*object.Object, error) {
|
func (e *storageEngineWrapper) get(exec *execCtx) (*object.Object, error) {
|
||||||
if exec.headOnly() {
|
if exec.headOnly {
|
||||||
var headPrm engine.HeadPrm
|
var headPrm engine.HeadPrm
|
||||||
headPrm.WithAddress(exec.address())
|
headPrm.WithAddress(exec.prm.addr)
|
||||||
headPrm.WithRaw(exec.isRaw())
|
headPrm.WithRaw(exec.prm.raw)
|
||||||
|
|
||||||
r, err := e.engine.Head(headPrm)
|
r, err := e.engine.Head(headPrm)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
|
@ -209,9 +209,9 @@ func (e *storageEngineWrapper) get(exec *execCtx) (*object.Object, error) {
|
||||||
}
|
}
|
||||||
|
|
||||||
return r.Header(), nil
|
return r.Header(), nil
|
||||||
} else if rng := exec.ctxRange(); rng != nil {
|
} else if rng := exec.prm.rng; rng != nil {
|
||||||
var getRange engine.RngPrm
|
var getRange engine.RngPrm
|
||||||
getRange.WithAddress(exec.address())
|
getRange.WithAddress(exec.prm.addr)
|
||||||
getRange.WithPayloadRange(rng)
|
getRange.WithPayloadRange(rng)
|
||||||
|
|
||||||
r, err := e.engine.GetRange(getRange)
|
r, err := e.engine.GetRange(getRange)
|
||||||
|
@ -222,7 +222,7 @@ func (e *storageEngineWrapper) get(exec *execCtx) (*object.Object, error) {
|
||||||
return r.Object(), nil
|
return r.Object(), nil
|
||||||
} else {
|
} else {
|
||||||
var getPrm engine.GetPrm
|
var getPrm engine.GetPrm
|
||||||
getPrm.WithAddress(exec.address())
|
getPrm.WithAddress(exec.prm.addr)
|
||||||
|
|
||||||
r, err := e.engine.Get(getPrm)
|
r, err := e.engine.Get(getPrm)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
|
|
|
@ -10,12 +10,12 @@ import (
|
||||||
)
|
)
|
||||||
|
|
||||||
func (exec *execCtx) executeOnContainer() {
|
func (exec *execCtx) executeOnContainer() {
|
||||||
if exec.isLocal() {
|
if exec.prm.common.LocalOnly() {
|
||||||
exec.log.Debug("return result directly")
|
exec.log.Debug("return result directly")
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|
||||||
lookupDepth := exec.netmapLookupDepth()
|
lookupDepth := exec.prm.common.NetmapLookupDepth()
|
||||||
|
|
||||||
exec.log.Debug("trying to execute in container...",
|
exec.log.Debug("trying to execute in container...",
|
||||||
zap.Uint64("netmap lookup depth", lookupDepth),
|
zap.Uint64("netmap lookup depth", lookupDepth),
|
||||||
|
@ -52,12 +52,12 @@ func (exec *execCtx) processCurrentEpoch() bool {
|
||||||
zap.Uint64("number", exec.curProcEpoch),
|
zap.Uint64("number", exec.curProcEpoch),
|
||||||
)
|
)
|
||||||
|
|
||||||
traverser, ok := exec.generateTraverser(exec.containerID())
|
traverser, ok := exec.generateTraverser(exec.prm.cnr)
|
||||||
if !ok {
|
if !ok {
|
||||||
return true
|
return true
|
||||||
}
|
}
|
||||||
|
|
||||||
ctx, cancel := context.WithCancel(exec.context())
|
ctx, cancel := context.WithCancel(exec.ctx)
|
||||||
defer cancel()
|
defer cancel()
|
||||||
|
|
||||||
for {
|
for {
|
||||||
|
|
|
@ -6,7 +6,6 @@ import (
|
||||||
"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/services/object_manager/placement"
|
"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/services/object_manager/placement"
|
||||||
"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/util/logger"
|
"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/util/logger"
|
||||||
cid "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/container/id"
|
cid "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/container/id"
|
||||||
"git.frostfs.info/TrueCloudLab/frostfs-sdk-go/object"
|
|
||||||
oid "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/object/id"
|
oid "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/object/id"
|
||||||
"go.uber.org/zap"
|
"go.uber.org/zap"
|
||||||
)
|
)
|
||||||
|
@ -45,39 +44,15 @@ func (exec *execCtx) prepare() {
|
||||||
func (exec *execCtx) setLogger(l *logger.Logger) {
|
func (exec *execCtx) setLogger(l *logger.Logger) {
|
||||||
exec.log = &logger.Logger{Logger: l.With(
|
exec.log = &logger.Logger{Logger: l.With(
|
||||||
zap.String("request", "SEARCH"),
|
zap.String("request", "SEARCH"),
|
||||||
zap.Stringer("container", exec.containerID()),
|
zap.Stringer("container", exec.prm.cnr),
|
||||||
zap.Bool("local", exec.isLocal()),
|
zap.Bool("local", exec.prm.common.LocalOnly()),
|
||||||
zap.Bool("with session", exec.prm.common.SessionToken() != nil),
|
zap.Bool("with session", exec.prm.common.SessionToken() != nil),
|
||||||
zap.Bool("with bearer", exec.prm.common.BearerToken() != nil),
|
zap.Bool("with bearer", exec.prm.common.BearerToken() != nil),
|
||||||
)}
|
)}
|
||||||
}
|
}
|
||||||
|
|
||||||
func (exec execCtx) context() context.Context {
|
|
||||||
return exec.ctx
|
|
||||||
}
|
|
||||||
|
|
||||||
func (exec execCtx) isLocal() bool {
|
|
||||||
return exec.prm.common.LocalOnly()
|
|
||||||
}
|
|
||||||
|
|
||||||
func (exec *execCtx) containerID() cid.ID {
|
|
||||||
return exec.prm.cnr
|
|
||||||
}
|
|
||||||
|
|
||||||
func (exec *execCtx) searchFilters() object.SearchFilters {
|
|
||||||
return exec.prm.filters
|
|
||||||
}
|
|
||||||
|
|
||||||
func (exec *execCtx) netmapEpoch() uint64 {
|
|
||||||
return exec.prm.common.NetmapEpoch()
|
|
||||||
}
|
|
||||||
|
|
||||||
func (exec *execCtx) netmapLookupDepth() uint64 {
|
|
||||||
return exec.prm.common.NetmapLookupDepth()
|
|
||||||
}
|
|
||||||
|
|
||||||
func (exec *execCtx) initEpoch() bool {
|
func (exec *execCtx) initEpoch() bool {
|
||||||
exec.curProcEpoch = exec.netmapEpoch()
|
exec.curProcEpoch = exec.prm.common.NetmapEpoch()
|
||||||
if exec.curProcEpoch > 0 {
|
if exec.curProcEpoch > 0 {
|
||||||
return true
|
return true
|
||||||
}
|
}
|
||||||
|
|
|
@ -103,7 +103,7 @@ func (c *testClientCache) get(info clientcore.NodeInfo) (searchClient, error) {
|
||||||
}
|
}
|
||||||
|
|
||||||
func (s *testStorage) search(exec *execCtx) ([]oid.ID, error) {
|
func (s *testStorage) search(exec *execCtx) ([]oid.ID, error) {
|
||||||
v, ok := s.items[exec.containerID().EncodeToString()]
|
v, ok := s.items[exec.prm.cnr.EncodeToString()]
|
||||||
if !ok {
|
if !ok {
|
||||||
return nil, nil
|
return nil, nil
|
||||||
}
|
}
|
||||||
|
@ -112,7 +112,7 @@ func (s *testStorage) search(exec *execCtx) ([]oid.ID, error) {
|
||||||
}
|
}
|
||||||
|
|
||||||
func (c *testStorage) searchObjects(exec *execCtx, _ clientcore.NodeInfo) ([]oid.ID, error) {
|
func (c *testStorage) searchObjects(exec *execCtx, _ clientcore.NodeInfo) ([]oid.ID, error) {
|
||||||
v, ok := c.items[exec.containerID().EncodeToString()]
|
v, ok := c.items[exec.prm.cnr.EncodeToString()]
|
||||||
if !ok {
|
if !ok {
|
||||||
return nil, nil
|
return nil, nil
|
||||||
}
|
}
|
||||||
|
|
|
@ -98,7 +98,7 @@ func (c *clientWrapper) searchObjects(exec *execCtx, info client.NodeInfo) ([]oi
|
||||||
|
|
||||||
var prm internalclient.SearchObjectsPrm
|
var prm internalclient.SearchObjectsPrm
|
||||||
|
|
||||||
prm.SetContext(exec.context())
|
prm.SetContext(exec.ctx)
|
||||||
prm.SetClient(c.client)
|
prm.SetClient(c.client)
|
||||||
prm.SetPrivateKey(key)
|
prm.SetPrivateKey(key)
|
||||||
prm.SetSessionToken(exec.prm.common.SessionToken())
|
prm.SetSessionToken(exec.prm.common.SessionToken())
|
||||||
|
@ -106,8 +106,8 @@ func (c *clientWrapper) searchObjects(exec *execCtx, info client.NodeInfo) ([]oi
|
||||||
prm.SetTTL(exec.prm.common.TTL())
|
prm.SetTTL(exec.prm.common.TTL())
|
||||||
prm.SetXHeaders(exec.prm.common.XHeaders())
|
prm.SetXHeaders(exec.prm.common.XHeaders())
|
||||||
prm.SetNetmapEpoch(exec.curProcEpoch)
|
prm.SetNetmapEpoch(exec.curProcEpoch)
|
||||||
prm.SetContainerID(exec.containerID())
|
prm.SetContainerID(exec.prm.cnr)
|
||||||
prm.SetFilters(exec.searchFilters())
|
prm.SetFilters(exec.prm.filters)
|
||||||
|
|
||||||
res, err := internalclient.SearchObjects(prm)
|
res, err := internalclient.SearchObjects(prm)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
|
@ -119,8 +119,8 @@ func (c *clientWrapper) searchObjects(exec *execCtx, info client.NodeInfo) ([]oi
|
||||||
|
|
||||||
func (e *storageEngineWrapper) search(exec *execCtx) ([]oid.ID, error) {
|
func (e *storageEngineWrapper) search(exec *execCtx) ([]oid.ID, error) {
|
||||||
var selectPrm engine.SelectPrm
|
var selectPrm engine.SelectPrm
|
||||||
selectPrm.WithFilters(exec.searchFilters())
|
selectPrm.WithFilters(exec.prm.filters)
|
||||||
selectPrm.WithContainerID(exec.containerID())
|
selectPrm.WithContainerID(exec.prm.cnr)
|
||||||
|
|
||||||
r, err := e.storage.Select(selectPrm)
|
r, err := e.storage.Select(selectPrm)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
|
|
Loading…
Reference in a new issue