forked from TrueCloudLab/frostfs-node
[#647] objsvc/search: Simplify error handling
Signed-off-by: Evgenii Stratonikov <e.stratonikov@yadro.com>
This commit is contained in:
parent
56f841b022
commit
966ad22abf
4 changed files with 51 additions and 117 deletions
|
@ -3,6 +3,7 @@ package searchsvc
|
||||||
import (
|
import (
|
||||||
"context"
|
"context"
|
||||||
"encoding/hex"
|
"encoding/hex"
|
||||||
|
"fmt"
|
||||||
"sync"
|
"sync"
|
||||||
|
|
||||||
"git.frostfs.info/TrueCloudLab/frostfs-node/internal/logs"
|
"git.frostfs.info/TrueCloudLab/frostfs-node/internal/logs"
|
||||||
|
@ -10,12 +11,7 @@ import (
|
||||||
"go.uber.org/zap"
|
"go.uber.org/zap"
|
||||||
)
|
)
|
||||||
|
|
||||||
func (exec *execCtx) executeOnContainer(ctx context.Context) {
|
func (exec *execCtx) executeOnContainer(ctx context.Context) error {
|
||||||
if exec.isLocal() {
|
|
||||||
exec.log.Debug(logs.SearchReturnResultDirectly)
|
|
||||||
return
|
|
||||||
}
|
|
||||||
|
|
||||||
lookupDepth := exec.netmapLookupDepth()
|
lookupDepth := exec.netmapLookupDepth()
|
||||||
|
|
||||||
exec.log.Debug(logs.TryingToExecuteInContainer,
|
exec.log.Debug(logs.TryingToExecuteInContainer,
|
||||||
|
@ -23,13 +19,12 @@ func (exec *execCtx) executeOnContainer(ctx context.Context) {
|
||||||
)
|
)
|
||||||
|
|
||||||
// initialize epoch number
|
// initialize epoch number
|
||||||
ok := exec.initEpoch()
|
if err := exec.initEpoch(); err != nil {
|
||||||
if !ok {
|
return fmt.Errorf("%s: %w", logs.CouldNotGetCurrentEpochNumber, err)
|
||||||
return
|
|
||||||
}
|
}
|
||||||
|
|
||||||
for {
|
for {
|
||||||
if exec.processCurrentEpoch(ctx) {
|
if err := exec.processCurrentEpoch(ctx); err != nil {
|
||||||
break
|
break
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@ -44,18 +39,17 @@ func (exec *execCtx) executeOnContainer(ctx context.Context) {
|
||||||
exec.curProcEpoch--
|
exec.curProcEpoch--
|
||||||
}
|
}
|
||||||
|
|
||||||
exec.status = statusOK
|
return nil
|
||||||
exec.err = nil
|
|
||||||
}
|
}
|
||||||
|
|
||||||
func (exec *execCtx) processCurrentEpoch(ctx context.Context) bool {
|
func (exec *execCtx) processCurrentEpoch(ctx context.Context) error {
|
||||||
exec.log.Debug(logs.ProcessEpoch,
|
exec.log.Debug(logs.ProcessEpoch,
|
||||||
zap.Uint64("number", exec.curProcEpoch),
|
zap.Uint64("number", exec.curProcEpoch),
|
||||||
)
|
)
|
||||||
|
|
||||||
traverser, ok := exec.generateTraverser(exec.containerID())
|
traverser, err := exec.svc.traverserGenerator.GenerateTraverser(exec.containerID(), nil, exec.curProcEpoch)
|
||||||
if !ok {
|
if err != nil {
|
||||||
return true
|
return fmt.Errorf("%s: %w", logs.SearchCouldNotGenerateContainerTraverser, err)
|
||||||
}
|
}
|
||||||
|
|
||||||
ctx, cancel := context.WithCancel(ctx)
|
ctx, cancel := context.WithCancel(ctx)
|
||||||
|
@ -91,12 +85,7 @@ func (exec *execCtx) processCurrentEpoch(ctx context.Context) bool {
|
||||||
|
|
||||||
c, err := exec.svc.clientConstructor.get(info)
|
c, err := exec.svc.clientConstructor.get(info)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
mtx.Lock()
|
exec.log.Debug(logs.SearchCouldNotConstructRemoteNodeClient, zap.String("error", err.Error()))
|
||||||
exec.status = statusUndefined
|
|
||||||
exec.err = err
|
|
||||||
mtx.Unlock()
|
|
||||||
|
|
||||||
exec.log.Debug(logs.SearchCouldNotConstructRemoteNodeClient)
|
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@ -109,13 +98,17 @@ func (exec *execCtx) processCurrentEpoch(ctx context.Context) bool {
|
||||||
}
|
}
|
||||||
|
|
||||||
mtx.Lock()
|
mtx.Lock()
|
||||||
exec.writeIDList(ids)
|
err = exec.writeIDList(ids)
|
||||||
mtx.Unlock()
|
mtx.Unlock()
|
||||||
|
if err != nil {
|
||||||
|
exec.log.Debug(logs.SearchCouldNotWriteObjectIdentifiers, zap.String("error", err.Error()))
|
||||||
|
return
|
||||||
|
}
|
||||||
}(i)
|
}(i)
|
||||||
}
|
}
|
||||||
|
|
||||||
wg.Wait()
|
wg.Wait()
|
||||||
}
|
}
|
||||||
|
|
||||||
return false
|
return nil
|
||||||
}
|
}
|
||||||
|
|
|
@ -1,8 +1,6 @@
|
||||||
package searchsvc
|
package searchsvc
|
||||||
|
|
||||||
import (
|
import (
|
||||||
"git.frostfs.info/TrueCloudLab/frostfs-node/internal/logs"
|
|
||||||
"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"
|
||||||
objectSDK "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/object"
|
objectSDK "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/object"
|
||||||
|
@ -10,28 +8,16 @@ import (
|
||||||
"go.uber.org/zap"
|
"go.uber.org/zap"
|
||||||
)
|
)
|
||||||
|
|
||||||
type statusError struct {
|
|
||||||
status int
|
|
||||||
err error
|
|
||||||
}
|
|
||||||
|
|
||||||
type execCtx struct {
|
type execCtx struct {
|
||||||
svc *Service
|
svc *Service
|
||||||
|
|
||||||
prm Prm
|
prm Prm
|
||||||
|
|
||||||
statusError
|
|
||||||
|
|
||||||
log *logger.Logger
|
log *logger.Logger
|
||||||
|
|
||||||
curProcEpoch uint64
|
curProcEpoch uint64
|
||||||
}
|
}
|
||||||
|
|
||||||
const (
|
|
||||||
statusUndefined int = iota
|
|
||||||
statusOK
|
|
||||||
)
|
|
||||||
|
|
||||||
func (exec *execCtx) prepare() {
|
func (exec *execCtx) prepare() {
|
||||||
if _, ok := exec.prm.writer.(*uniqueIDWriter); !ok {
|
if _, ok := exec.prm.writer.(*uniqueIDWriter); !ok {
|
||||||
exec.prm.writer = newUniqueAddressWriter(exec.prm.writer)
|
exec.prm.writer = newUniqueAddressWriter(exec.prm.writer)
|
||||||
|
@ -68,64 +54,24 @@ func (exec *execCtx) netmapLookupDepth() uint64 {
|
||||||
return exec.prm.common.NetmapLookupDepth()
|
return exec.prm.common.NetmapLookupDepth()
|
||||||
}
|
}
|
||||||
|
|
||||||
func (exec *execCtx) initEpoch() bool {
|
func (exec *execCtx) initEpoch() error {
|
||||||
exec.curProcEpoch = exec.netmapEpoch()
|
exec.curProcEpoch = exec.netmapEpoch()
|
||||||
if exec.curProcEpoch > 0 {
|
if exec.curProcEpoch > 0 {
|
||||||
return true
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
e, err := exec.svc.currentEpochReceiver.Epoch()
|
e, err := exec.svc.currentEpochReceiver.Epoch()
|
||||||
|
if err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
|
||||||
switch {
|
|
||||||
default:
|
|
||||||
exec.status = statusUndefined
|
|
||||||
exec.err = err
|
|
||||||
|
|
||||||
exec.log.Debug(logs.CouldNotGetCurrentEpochNumber,
|
|
||||||
zap.String("error", err.Error()),
|
|
||||||
)
|
|
||||||
|
|
||||||
return false
|
|
||||||
case err == nil:
|
|
||||||
exec.curProcEpoch = e
|
exec.curProcEpoch = e
|
||||||
return true
|
return nil
|
||||||
}
|
|
||||||
}
|
}
|
||||||
|
|
||||||
func (exec *execCtx) generateTraverser(cnr cid.ID) (*placement.Traverser, bool) {
|
func (exec *execCtx) writeIDList(ids []oid.ID) error {
|
||||||
t, err := exec.svc.traverserGenerator.GenerateTraverser(cnr, nil, exec.curProcEpoch)
|
|
||||||
|
|
||||||
switch {
|
|
||||||
default:
|
|
||||||
exec.status = statusUndefined
|
|
||||||
exec.err = err
|
|
||||||
|
|
||||||
exec.log.Debug(logs.SearchCouldNotGenerateContainerTraverser,
|
|
||||||
zap.String("error", err.Error()),
|
|
||||||
)
|
|
||||||
|
|
||||||
return nil, false
|
|
||||||
case err == nil:
|
|
||||||
return t, true
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
func (exec *execCtx) writeIDList(ids []oid.ID) {
|
|
||||||
ids = exec.filterAllowedObjectIDs(ids)
|
ids = exec.filterAllowedObjectIDs(ids)
|
||||||
err := exec.prm.writer.WriteIDs(ids)
|
return exec.prm.writer.WriteIDs(ids)
|
||||||
|
|
||||||
switch {
|
|
||||||
default:
|
|
||||||
exec.status = statusUndefined
|
|
||||||
exec.err = err
|
|
||||||
|
|
||||||
exec.log.Debug(logs.SearchCouldNotWriteObjectIdentifiers,
|
|
||||||
zap.String("error", err.Error()),
|
|
||||||
)
|
|
||||||
case err == nil:
|
|
||||||
exec.status = statusOK
|
|
||||||
exec.err = nil
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
|
|
||||||
func (exec *execCtx) filterAllowedObjectIDs(objIDs []oid.ID) []oid.ID {
|
func (exec *execCtx) filterAllowedObjectIDs(objIDs []oid.ID) []oid.ID {
|
||||||
|
|
|
@ -2,24 +2,22 @@ package searchsvc
|
||||||
|
|
||||||
import (
|
import (
|
||||||
"context"
|
"context"
|
||||||
|
"fmt"
|
||||||
|
|
||||||
"git.frostfs.info/TrueCloudLab/frostfs-node/internal/logs"
|
"git.frostfs.info/TrueCloudLab/frostfs-node/internal/logs"
|
||||||
"go.uber.org/zap"
|
"go.uber.org/zap"
|
||||||
)
|
)
|
||||||
|
|
||||||
func (exec *execCtx) executeLocal(ctx context.Context) {
|
func (exec *execCtx) executeLocal(ctx context.Context) error {
|
||||||
ids, err := exec.svc.localStorage.search(ctx, exec)
|
ids, err := exec.svc.localStorage.search(ctx, exec)
|
||||||
|
|
||||||
if err != nil {
|
if err != nil {
|
||||||
exec.status = statusUndefined
|
exec.log.Debug(logs.SearchLocalOperationFailed, zap.String("error", err.Error()))
|
||||||
exec.err = err
|
return err
|
||||||
|
|
||||||
exec.log.Debug(logs.SearchLocalOperationFailed,
|
|
||||||
zap.String("error", err.Error()),
|
|
||||||
)
|
|
||||||
|
|
||||||
return
|
|
||||||
}
|
}
|
||||||
|
|
||||||
exec.writeIDList(ids)
|
if err := exec.writeIDList(ids); err != nil {
|
||||||
|
return fmt.Errorf("%s: %w", logs.SearchCouldNotWriteObjectIdentifiers, err)
|
||||||
|
}
|
||||||
|
|
||||||
|
return nil
|
||||||
}
|
}
|
||||||
|
|
|
@ -18,33 +18,30 @@ func (s *Service) Search(ctx context.Context, prm Prm) error {
|
||||||
|
|
||||||
exec.setLogger(s.log)
|
exec.setLogger(s.log)
|
||||||
|
|
||||||
exec.execute(ctx)
|
return exec.execute(ctx)
|
||||||
|
|
||||||
return exec.statusError.err
|
|
||||||
}
|
}
|
||||||
|
|
||||||
func (exec *execCtx) execute(ctx context.Context) {
|
func (exec *execCtx) execute(ctx context.Context) error {
|
||||||
exec.log.Debug(logs.ServingRequest)
|
exec.log.Debug(logs.ServingRequest)
|
||||||
|
|
||||||
// perform local operation
|
err := exec.executeLocal(ctx)
|
||||||
exec.executeLocal(ctx)
|
exec.logResult(err)
|
||||||
|
|
||||||
exec.analyzeStatus(ctx, true)
|
if exec.isLocal() {
|
||||||
|
exec.log.Debug(logs.SearchReturnResultDirectly)
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
|
||||||
|
err = exec.executeOnContainer(ctx)
|
||||||
|
exec.logResult(err)
|
||||||
|
return err
|
||||||
}
|
}
|
||||||
|
|
||||||
func (exec *execCtx) analyzeStatus(ctx context.Context, execCnr bool) {
|
func (exec *execCtx) logResult(err error) {
|
||||||
// analyze local result
|
switch {
|
||||||
switch exec.status {
|
|
||||||
default:
|
default:
|
||||||
exec.log.Debug(logs.OperationFinishedWithError,
|
exec.log.Debug(logs.OperationFinishedWithError, zap.String("error", err.Error()))
|
||||||
zap.String("error", exec.err.Error()),
|
case err == nil:
|
||||||
)
|
|
||||||
case statusOK:
|
|
||||||
exec.log.Debug(logs.OperationFinishedSuccessfully)
|
exec.log.Debug(logs.OperationFinishedSuccessfully)
|
||||||
}
|
}
|
||||||
|
|
||||||
if execCnr {
|
|
||||||
exec.executeOnContainer(ctx)
|
|
||||||
exec.analyzeStatus(ctx, false)
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
|
|
Loading…
Reference in a new issue