forked from TrueCloudLab/frostfs-node
[#193] getsvc: Resolve context linters
Resolve containedctx and contextcheck linters. Signed-off-by: Dmitrii Stepanov <d.stepanov@yadro.com>
This commit is contained in:
parent
206458c841
commit
1bf21dbb47
9 changed files with 39 additions and 48 deletions
|
@ -10,7 +10,7 @@ import (
|
||||||
"go.uber.org/zap"
|
"go.uber.org/zap"
|
||||||
)
|
)
|
||||||
|
|
||||||
func (exec *execCtx) assemble() {
|
func (exec *execCtx) assemble(ctx context.Context) {
|
||||||
if !exec.canAssemble() {
|
if !exec.canAssemble() {
|
||||||
exec.log.Debug("can not assemble the object")
|
exec.log.Debug("can not assemble the object")
|
||||||
return
|
return
|
||||||
|
@ -49,7 +49,7 @@ func (exec *execCtx) assemble() {
|
||||||
zap.Uint64("range_length", exec.ctxRange().GetLength()),
|
zap.Uint64("range_length", exec.ctxRange().GetLength()),
|
||||||
)
|
)
|
||||||
|
|
||||||
obj, err := assembler.Assemble(exec.context(), exec.prm.objWriter)
|
obj, err := assembler.Assemble(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),
|
||||||
|
@ -107,8 +107,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
|
err := exec.svc.Head(ctx, prm)
|
||||||
err := exec.svc.Head(exec.context(), prm)
|
|
||||||
|
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
|
@ -128,8 +127,7 @@ func (exec *execCtx) GetObject(ctx context.Context, id oid.ID, rng *objectSDK.Ra
|
||||||
p.addr.SetContainer(exec.containerID())
|
p.addr.SetContainer(exec.containerID())
|
||||||
p.addr.SetObject(id)
|
p.addr.SetObject(id)
|
||||||
|
|
||||||
//nolint: contextcheck
|
statusError := exec.svc.get(ctx, p.commonPrm, withPayloadRange(rng))
|
||||||
statusError := exec.svc.get(exec.context(), p.commonPrm, withPayloadRange(rng))
|
|
||||||
|
|
||||||
if statusError.err != nil {
|
if statusError.err != nil {
|
||||||
return nil, statusError.err
|
return nil, statusError.err
|
||||||
|
|
|
@ -7,7 +7,7 @@ import (
|
||||||
"go.uber.org/zap"
|
"go.uber.org/zap"
|
||||||
)
|
)
|
||||||
|
|
||||||
func (exec *execCtx) executeOnContainer() {
|
func (exec *execCtx) executeOnContainer(ctx context.Context) {
|
||||||
if exec.isLocal() {
|
if exec.isLocal() {
|
||||||
exec.log.Debug("return result directly")
|
exec.log.Debug("return result directly")
|
||||||
return
|
return
|
||||||
|
@ -26,7 +26,7 @@ func (exec *execCtx) executeOnContainer() {
|
||||||
}
|
}
|
||||||
|
|
||||||
for {
|
for {
|
||||||
if exec.processCurrentEpoch() {
|
if exec.processCurrentEpoch(ctx) {
|
||||||
break
|
break
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@ -42,7 +42,7 @@ func (exec *execCtx) executeOnContainer() {
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
func (exec *execCtx) processCurrentEpoch() bool {
|
func (exec *execCtx) processCurrentEpoch(ctx context.Context) bool {
|
||||||
exec.log.Debug("process epoch",
|
exec.log.Debug("process epoch",
|
||||||
zap.Uint64("number", exec.curProcEpoch),
|
zap.Uint64("number", exec.curProcEpoch),
|
||||||
)
|
)
|
||||||
|
@ -52,7 +52,7 @@ func (exec *execCtx) processCurrentEpoch() bool {
|
||||||
return true
|
return true
|
||||||
}
|
}
|
||||||
|
|
||||||
ctx, cancel := context.WithCancel(exec.context())
|
ctx, cancel := context.WithCancel(ctx)
|
||||||
defer cancel()
|
defer cancel()
|
||||||
|
|
||||||
exec.status = statusUndefined
|
exec.status = statusUndefined
|
||||||
|
|
|
@ -19,12 +19,9 @@ type statusError struct {
|
||||||
err error
|
err error
|
||||||
}
|
}
|
||||||
|
|
||||||
// nolint: containedctx
|
|
||||||
type execCtx struct {
|
type execCtx struct {
|
||||||
svc *Service
|
svc *Service
|
||||||
|
|
||||||
ctx context.Context
|
|
||||||
|
|
||||||
prm RangePrm
|
prm RangePrm
|
||||||
|
|
||||||
statusError
|
statusError
|
||||||
|
@ -80,10 +77,6 @@ func (exec *execCtx) setLogger(l *logger.Logger) {
|
||||||
)}
|
)}
|
||||||
}
|
}
|
||||||
|
|
||||||
func (exec execCtx) context() context.Context {
|
|
||||||
return exec.ctx
|
|
||||||
}
|
|
||||||
|
|
||||||
func (exec execCtx) isLocal() bool {
|
func (exec execCtx) isLocal() bool {
|
||||||
return exec.prm.common.LocalOnly()
|
return exec.prm.common.LocalOnly()
|
||||||
}
|
}
|
||||||
|
@ -217,13 +210,13 @@ func mergeSplitInfo(dst, src *objectSDK.SplitInfo) {
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
func (exec *execCtx) writeCollectedHeader() bool {
|
func (exec *execCtx) writeCollectedHeader(ctx context.Context) bool {
|
||||||
if exec.ctxRange() != nil {
|
if exec.ctxRange() != nil {
|
||||||
return true
|
return true
|
||||||
}
|
}
|
||||||
|
|
||||||
err := exec.prm.objWriter.WriteHeader(
|
err := exec.prm.objWriter.WriteHeader(
|
||||||
exec.context(),
|
ctx,
|
||||||
exec.collectedObject.CutPayload(),
|
exec.collectedObject.CutPayload(),
|
||||||
)
|
)
|
||||||
|
|
||||||
|
@ -243,12 +236,12 @@ func (exec *execCtx) writeCollectedHeader() bool {
|
||||||
return exec.status == statusOK
|
return exec.status == statusOK
|
||||||
}
|
}
|
||||||
|
|
||||||
func (exec *execCtx) writeObjectPayload(obj *objectSDK.Object) bool {
|
func (exec *execCtx) writeObjectPayload(ctx context.Context, 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(ctx, obj.Payload())
|
||||||
|
|
||||||
switch {
|
switch {
|
||||||
default:
|
default:
|
||||||
|
@ -266,9 +259,9 @@ func (exec *execCtx) writeObjectPayload(obj *objectSDK.Object) bool {
|
||||||
return err == nil
|
return err == nil
|
||||||
}
|
}
|
||||||
|
|
||||||
func (exec *execCtx) writeCollectedObject() {
|
func (exec *execCtx) writeCollectedObject(ctx context.Context) {
|
||||||
if ok := exec.writeCollectedHeader(); ok {
|
if ok := exec.writeCollectedHeader(ctx); ok {
|
||||||
exec.writeObjectPayload(exec.collectedObject)
|
exec.writeObjectPayload(ctx, exec.collectedObject)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
|
@ -65,7 +65,6 @@ func (s *Service) Head(ctx context.Context, prm HeadPrm) error {
|
||||||
func (s *Service) get(ctx context.Context, prm commonPrm, opts ...execOption) statusError {
|
func (s *Service) get(ctx context.Context, prm commonPrm, opts ...execOption) statusError {
|
||||||
exec := &execCtx{
|
exec := &execCtx{
|
||||||
svc: s,
|
svc: s,
|
||||||
ctx: ctx,
|
|
||||||
prm: RangePrm{
|
prm: RangePrm{
|
||||||
commonPrm: prm,
|
commonPrm: prm,
|
||||||
},
|
},
|
||||||
|
@ -78,22 +77,21 @@ func (s *Service) get(ctx context.Context, prm commonPrm, opts ...execOption) st
|
||||||
|
|
||||||
exec.setLogger(s.log)
|
exec.setLogger(s.log)
|
||||||
|
|
||||||
//nolint: contextcheck
|
exec.execute(ctx)
|
||||||
exec.execute()
|
|
||||||
|
|
||||||
return exec.statusError
|
return exec.statusError
|
||||||
}
|
}
|
||||||
|
|
||||||
func (exec *execCtx) execute() {
|
func (exec *execCtx) execute(ctx context.Context) {
|
||||||
exec.log.Debug("serving request...")
|
exec.log.Debug("serving request...")
|
||||||
|
|
||||||
// perform local operation
|
// perform local operation
|
||||||
exec.executeLocal()
|
exec.executeLocal(ctx)
|
||||||
|
|
||||||
exec.analyzeStatus(true)
|
exec.analyzeStatus(ctx, true)
|
||||||
}
|
}
|
||||||
|
|
||||||
func (exec *execCtx) analyzeStatus(execCnr bool) {
|
func (exec *execCtx) analyzeStatus(ctx context.Context, execCnr bool) {
|
||||||
// analyze local result
|
// analyze local result
|
||||||
switch exec.status {
|
switch exec.status {
|
||||||
case statusOK:
|
case statusOK:
|
||||||
|
@ -102,7 +100,7 @@ func (exec *execCtx) analyzeStatus(execCnr bool) {
|
||||||
exec.log.Debug("requested object was marked as removed")
|
exec.log.Debug("requested object was marked as removed")
|
||||||
case statusVIRTUAL:
|
case statusVIRTUAL:
|
||||||
exec.log.Debug("requested object is virtual")
|
exec.log.Debug("requested object is virtual")
|
||||||
exec.assemble()
|
exec.assemble(ctx)
|
||||||
case statusOutOfRange:
|
case statusOutOfRange:
|
||||||
exec.log.Debug("requested range is out of object bounds")
|
exec.log.Debug("requested range is out of object bounds")
|
||||||
default:
|
default:
|
||||||
|
@ -111,8 +109,8 @@ func (exec *execCtx) analyzeStatus(execCnr bool) {
|
||||||
)
|
)
|
||||||
|
|
||||||
if execCnr {
|
if execCnr {
|
||||||
exec.executeOnContainer()
|
exec.executeOnContainer(ctx)
|
||||||
exec.analyzeStatus(false)
|
exec.analyzeStatus(ctx, false)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
|
@ -117,7 +117,7 @@ func newTestClient() *testClient {
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
func (c *testClient) getObject(exec *execCtx, _ client.NodeInfo) (*objectSDK.Object, error) {
|
func (c *testClient) getObject(ctx context.Context, exec *execCtx, _ client.NodeInfo) (*objectSDK.Object, error) {
|
||||||
v, ok := c.results[exec.address().EncodeToString()]
|
v, ok := c.results[exec.address().EncodeToString()]
|
||||||
if !ok {
|
if !ok {
|
||||||
var errNotFound apistatus.ObjectNotFound
|
var errNotFound apistatus.ObjectNotFound
|
||||||
|
|
|
@ -1,6 +1,7 @@
|
||||||
package getsvc
|
package getsvc
|
||||||
|
|
||||||
import (
|
import (
|
||||||
|
"context"
|
||||||
"errors"
|
"errors"
|
||||||
|
|
||||||
apistatus "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/client/status"
|
apistatus "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/client/status"
|
||||||
|
@ -8,7 +9,7 @@ import (
|
||||||
"go.uber.org/zap"
|
"go.uber.org/zap"
|
||||||
)
|
)
|
||||||
|
|
||||||
func (exec *execCtx) executeLocal() {
|
func (exec *execCtx) executeLocal(ctx context.Context) {
|
||||||
var err error
|
var err error
|
||||||
|
|
||||||
exec.collectedObject, err = exec.svc.localStorage.get(exec)
|
exec.collectedObject, err = exec.svc.localStorage.get(exec)
|
||||||
|
@ -28,7 +29,7 @@ func (exec *execCtx) executeLocal() {
|
||||||
case err == nil:
|
case err == nil:
|
||||||
exec.status = statusOK
|
exec.status = statusOK
|
||||||
exec.err = nil
|
exec.err = nil
|
||||||
exec.writeCollectedObject()
|
exec.writeCollectedObject(ctx)
|
||||||
case errors.As(err, &errRemoved):
|
case errors.As(err, &errRemoved):
|
||||||
exec.status = statusINHUMED
|
exec.status = statusINHUMED
|
||||||
exec.err = errRemoved
|
exec.err = errRemoved
|
||||||
|
|
|
@ -18,7 +18,7 @@ func (exec *execCtx) processNode(ctx context.Context, info client.NodeInfo) bool
|
||||||
return true
|
return true
|
||||||
}
|
}
|
||||||
|
|
||||||
obj, err := client.getObject(exec, info)
|
obj, err := client.getObject(ctx, exec, info)
|
||||||
|
|
||||||
var errSplitInfo *objectSDK.SplitInfoError
|
var errSplitInfo *objectSDK.SplitInfoError
|
||||||
var errRemoved *apistatus.ObjectAlreadyRemoved
|
var errRemoved *apistatus.ObjectAlreadyRemoved
|
||||||
|
@ -43,8 +43,7 @@ func (exec *execCtx) processNode(ctx context.Context, info client.NodeInfo) bool
|
||||||
// has already been streamed to the requesting party
|
// has already been streamed to the requesting party
|
||||||
if obj != nil {
|
if obj != nil {
|
||||||
exec.collectedObject = obj
|
exec.collectedObject = obj
|
||||||
//nolint: contextcheck
|
exec.writeCollectedObject(ctx)
|
||||||
exec.writeCollectedObject()
|
|
||||||
}
|
}
|
||||||
case errors.As(err, &errRemoved):
|
case errors.As(err, &errRemoved):
|
||||||
exec.status = statusINHUMED
|
exec.status = statusINHUMED
|
||||||
|
|
|
@ -1,6 +1,8 @@
|
||||||
package getsvc
|
package getsvc
|
||||||
|
|
||||||
import (
|
import (
|
||||||
|
"context"
|
||||||
|
|
||||||
"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/core/client"
|
"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/core/client"
|
||||||
"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/core/netmap"
|
"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/core/netmap"
|
||||||
"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/local_object_storage/engine"
|
"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/local_object_storage/engine"
|
||||||
|
@ -22,7 +24,7 @@ type Service struct {
|
||||||
type Option func(*cfg)
|
type Option func(*cfg)
|
||||||
|
|
||||||
type getClient interface {
|
type getClient interface {
|
||||||
getObject(*execCtx, client.NodeInfo) (*object.Object, error)
|
getObject(context.Context, *execCtx, client.NodeInfo) (*object.Object, error)
|
||||||
}
|
}
|
||||||
|
|
||||||
type cfg struct {
|
type cfg struct {
|
||||||
|
|
|
@ -88,7 +88,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(ctx context.Context, exec *execCtx, info coreclient.NodeInfo) (*object.Object, error) {
|
||||||
if exec.isForwardingEnabled() {
|
if exec.isForwardingEnabled() {
|
||||||
return exec.prm.forwarder(info, c.client)
|
return exec.prm.forwarder(info, c.client)
|
||||||
}
|
}
|
||||||
|
@ -101,7 +101,7 @@ func (c *clientWrapper) getObject(exec *execCtx, info coreclient.NodeInfo) (*obj
|
||||||
if exec.headOnly() {
|
if exec.headOnly() {
|
||||||
var prm internalclient.HeadObjectPrm
|
var prm internalclient.HeadObjectPrm
|
||||||
|
|
||||||
prm.SetContext(exec.context())
|
prm.SetContext(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)
|
||||||
|
@ -127,7 +127,7 @@ func (c *clientWrapper) getObject(exec *execCtx, info coreclient.NodeInfo) (*obj
|
||||||
if rng := exec.ctxRange(); rng != nil {
|
if rng := exec.ctxRange(); rng != nil {
|
||||||
var prm internalclient.PayloadRangePrm
|
var prm internalclient.PayloadRangePrm
|
||||||
|
|
||||||
prm.SetContext(exec.context())
|
prm.SetContext(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)
|
||||||
|
@ -148,7 +148,7 @@ func (c *clientWrapper) getObject(exec *execCtx, info coreclient.NodeInfo) (*obj
|
||||||
if errors.As(err, &errAccessDenied) {
|
if errors.As(err, &errAccessDenied) {
|
||||||
// Current spec allows other storage node to deny access,
|
// Current spec allows other storage node to deny access,
|
||||||
// fallback to GET here.
|
// fallback to GET here.
|
||||||
obj, err := c.get(exec, key)
|
obj, err := c.get(ctx, exec, key)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
|
@ -169,13 +169,13 @@ func (c *clientWrapper) getObject(exec *execCtx, info coreclient.NodeInfo) (*obj
|
||||||
return payloadOnlyObject(res.PayloadRange()), nil
|
return payloadOnlyObject(res.PayloadRange()), nil
|
||||||
}
|
}
|
||||||
|
|
||||||
return c.get(exec, key)
|
return c.get(ctx, exec, key)
|
||||||
}
|
}
|
||||||
|
|
||||||
func (c *clientWrapper) get(exec *execCtx, key *ecdsa.PrivateKey) (*object.Object, error) {
|
func (c *clientWrapper) get(ctx context.Context, exec *execCtx, key *ecdsa.PrivateKey) (*object.Object, error) {
|
||||||
var prm internalclient.GetObjectPrm
|
var prm internalclient.GetObjectPrm
|
||||||
|
|
||||||
prm.SetContext(exec.context())
|
prm.SetContext(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)
|
||||||
|
|
Loading…
Reference in a new issue