[#648] objsvc/delete: Handle errors in Go style

Signed-off-by: Evgenii Stratonikov <e.stratonikov@yadro.com>
This commit is contained in:
Evgenii Stratonikov 2023-08-25 10:57:31 +03:00 committed by Evgenii Stratonikov
parent d2084ece41
commit f8ba60aa0c
4 changed files with 62 additions and 181 deletions

View file

@ -82,19 +82,12 @@ const (
PersistentCouldNotCleanUpExpiredTokens = "could not clean up expired tokens"
ControllerReportIsAlreadyStarted = "report is already started"
TombstoneCouldNotGetTheTombstoneTheSource = "tombstone getter: could not get the tombstone the source"
DeleteCouldNotComposeSplitInfo = "could not compose split info"
DeleteNoSplitInfoObjectIsPHY = "no split info, object is PHY"
DeleteAssemblingChain = "assembling chain..."
DeleteCouldNotGetPreviousSplitElement = "could not get previous split element"
DeleteCollectingChildren = "collecting children..."
DeleteCouldNotCollectObjectChildren = "could not collect object children"
DeleteSupplementBySplitID = "supplement by split ID"
DeleteCouldNotSearchForSplitChainMembers = "could not search for split chain members"
DeleteCouldNotMarshalTombstoneStructure = "could not marshal tombstone structure"
DeleteCouldNotSaveTheTombstone = "could not save the tombstone"
DeleteFormingTombstoneStructure = "forming tombstone structure..."
DeleteTombstoneStructureSuccessfullyFormedSaving = "tombstone structure successfully formed, saving..."
DeleteCouldNotReadTombstoneLifetimeConfig = "could not read tombstone lifetime config"
DeleteFormingSplitInfo = "forming split info..."
DeleteSplitInfoSuccessfullyFormedCollectingMembers = "split info successfully formed, collecting members..."
DeleteMembersSuccessfullyCollected = "members successfully collected"

View file

@ -29,27 +29,17 @@ func (s *Service) Delete(ctx context.Context, prm Prm) error {
exec.setLogger(s.log)
exec.execute(ctx)
return exec.statusError.err
return exec.execute(ctx)
}
func (exec *execCtx) execute(ctx context.Context) {
func (exec *execCtx) execute(ctx context.Context) error {
exec.log.Debug(logs.ServingRequest)
// perform local operation
exec.executeLocal(ctx)
exec.analyzeStatus()
}
func (exec *execCtx) analyzeStatus() {
// analyze local result
switch exec.status {
case statusOK:
exec.log.Debug(logs.OperationFinishedSuccessfully)
default:
exec.log.Debug(logs.OperationFinishedWithError,
zap.String("error", exec.err.Error()),
)
if err := exec.executeLocal(ctx); err != nil {
exec.log.Debug(logs.OperationFinishedWithError, zap.String("error", err.Error()))
return err
}
exec.log.Debug(logs.OperationFinishedSuccessfully)
return nil
}

View file

@ -2,6 +2,7 @@ package deletesvc
import (
"context"
"fmt"
"strconv"
objectV2 "git.frostfs.info/TrueCloudLab/frostfs-api-go/v2/object"
@ -15,18 +16,11 @@ import (
"go.uber.org/zap"
)
type statusError struct {
status int
err error
}
type execCtx struct {
svc *Service
prm Prm
statusError
log *logger.Logger
tombstone *objectSDK.Tombstone
@ -36,11 +30,6 @@ type execCtx struct {
tombstoneObj *objectSDK.Object
}
const (
statusUndefined int = iota
statusOK
)
func (exec *execCtx) setLogger(l *logger.Logger) {
exec.log = &logger.Logger{Logger: l.With(
zap.String("request", "DELETE"),
@ -75,48 +64,34 @@ func (exec *execCtx) newAddress(id oid.ID) oid.Address {
return a
}
func (exec *execCtx) formSplitInfo(ctx context.Context) bool {
success := false
func (exec *execCtx) formSplitInfo(ctx context.Context) error {
var err error
exec.splitInfo, err = exec.svc.header.splitInfo(ctx, exec)
switch {
default:
exec.status = statusUndefined
exec.err = err
exec.log.Debug(logs.DeleteCouldNotComposeSplitInfo,
zap.String("error", err.Error()),
)
case err == nil, apiclient.IsErrObjectAlreadyRemoved(err):
if err != nil && !apiclient.IsErrObjectAlreadyRemoved(err) {
// IsErrObjectAlreadyRemoved check is required because splitInfo
// implicitly performs Head request that may return ObjectAlreadyRemoved
// status that is not specified for Delete
exec.status = statusOK
exec.err = nil
success = true
// status that is not specified for Delete.
return err
}
return success
return nil
}
func (exec *execCtx) collectMembers(ctx context.Context) (ok bool) {
func (exec *execCtx) collectMembers(ctx context.Context) error {
if exec.splitInfo == nil {
exec.log.Debug(logs.DeleteNoSplitInfoObjectIsPHY)
return true
return nil
}
var err error
if _, withLink := exec.splitInfo.Link(); withLink {
ok = exec.collectChildren(ctx)
err = exec.collectChildren(ctx)
}
if !ok {
if err != nil {
if _, withLast := exec.splitInfo.LastPart(); withLast {
ok = exec.collectChain(ctx)
if !ok {
return
if err := exec.collectChain(ctx); err != nil {
return err
}
}
} // may be fail if neither right nor linking ID is set?
@ -124,7 +99,7 @@ func (exec *execCtx) collectMembers(ctx context.Context) (ok bool) {
return exec.supplementBySplitID(ctx)
}
func (exec *execCtx) collectChain(ctx context.Context) bool {
func (exec *execCtx) collectChain(ctx context.Context) error {
var chain []oid.ID
exec.log.Debug(logs.DeleteAssemblingChain)
@ -133,84 +108,43 @@ func (exec *execCtx) collectChain(ctx context.Context) bool {
chain = append(chain, prev)
p, err := exec.svc.header.previous(ctx, exec, prev)
switch {
default:
exec.status = statusUndefined
exec.err = err
exec.log.Debug(logs.DeleteCouldNotGetPreviousSplitElement,
zap.Stringer("id", prev),
zap.String("error", err.Error()),
)
return false
case err == nil:
exec.status = statusOK
exec.err = nil
if err != nil {
return fmt.Errorf("get previous split element for %s: %w", prev, err)
}
withPrev = p != nil
if withPrev {
prev = *p
}
}
}
exec.addMembers(chain)
return true
return nil
}
func (exec *execCtx) collectChildren(ctx context.Context) bool {
func (exec *execCtx) collectChildren(ctx context.Context) error {
exec.log.Debug(logs.DeleteCollectingChildren)
children, err := exec.svc.header.children(ctx, exec)
switch {
default:
exec.status = statusUndefined
exec.err = err
exec.log.Debug(logs.DeleteCouldNotCollectObjectChildren,
zap.String("error", err.Error()),
)
return false
case err == nil:
exec.status = statusOK
exec.err = nil
if err != nil {
return fmt.Errorf("collect children: %w", err)
}
link, _ := exec.splitInfo.Link()
exec.addMembers(append(children, link))
return true
}
return nil
}
func (exec *execCtx) supplementBySplitID(ctx context.Context) bool {
func (exec *execCtx) supplementBySplitID(ctx context.Context) error {
exec.log.Debug(logs.DeleteSupplementBySplitID)
chain, err := exec.svc.searcher.splitMembers(ctx, exec)
switch {
default:
exec.status = statusUndefined
exec.err = err
exec.log.Debug(logs.DeleteCouldNotSearchForSplitChainMembers,
zap.String("error", err.Error()),
)
return false
case err == nil:
exec.status = statusOK
exec.err = nil
if err != nil {
return fmt.Errorf("search split chain members: %w", err)
}
exec.addMembers(chain)
return true
}
return nil
}
func (exec *execCtx) addMembers(incoming []oid.ID) {
@ -228,17 +162,10 @@ func (exec *execCtx) addMembers(incoming []oid.ID) {
exec.tombstone.SetMembers(append(members, incoming...))
}
func (exec *execCtx) initTombstoneObject() bool {
func (exec *execCtx) initTombstoneObject() error {
payload, err := exec.tombstone.Marshal()
if err != nil {
exec.status = statusUndefined
exec.err = err
exec.log.Debug(logs.DeleteCouldNotMarshalTombstoneStructure,
zap.String("error", err.Error()),
)
return false
return fmt.Errorf("marshal tombstone: %w", err)
}
exec.tombstoneObj = objectSDK.New()
@ -262,29 +189,15 @@ func (exec *execCtx) initTombstoneObject() bool {
exec.tombstoneObj.SetAttributes(a)
return true
return nil
}
func (exec *execCtx) saveTombstone(ctx context.Context) bool {
func (exec *execCtx) saveTombstone(ctx context.Context) error {
id, err := exec.svc.placer.put(ctx, exec)
switch {
default:
exec.status = statusUndefined
exec.err = err
exec.log.Debug(logs.DeleteCouldNotSaveTheTombstone,
zap.String("error", err.Error()),
)
return false
case err == nil:
exec.status = statusOK
exec.err = nil
exec.prm.tombAddrWriter.
SetAddress(exec.newAddress(*id))
if err != nil {
return fmt.Errorf("save tombstone: %w", err)
}
return true
exec.prm.tombAddrWriter.SetAddress(exec.newAddress(*id))
return nil
}

View file

@ -2,37 +2,29 @@ package deletesvc
import (
"context"
"fmt"
"git.frostfs.info/TrueCloudLab/frostfs-node/internal/logs"
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) executeLocal(ctx context.Context) {
func (exec *execCtx) executeLocal(ctx context.Context) error {
exec.log.Debug(logs.DeleteFormingTombstoneStructure)
ok := exec.formTombstone(ctx)
if !ok {
return
if err := exec.formTombstone(ctx); err != nil {
return err
}
exec.log.Debug(logs.DeleteTombstoneStructureSuccessfullyFormedSaving)
exec.saveTombstone(ctx)
return exec.saveTombstone(ctx)
}
func (exec *execCtx) formTombstone(ctx context.Context) (ok bool) {
func (exec *execCtx) formTombstone(ctx context.Context) error {
tsLifetime, err := exec.svc.netInfo.TombstoneLifetime()
if err != nil {
exec.status = statusUndefined
exec.err = err
exec.log.Debug(logs.DeleteCouldNotReadTombstoneLifetimeConfig,
zap.String("error", err.Error()),
)
return false
return fmt.Errorf("fetch tombstone lifetime: %w", err)
}
exec.tombstone = objectSDK.NewTombstone()
@ -43,26 +35,19 @@ func (exec *execCtx) formTombstone(ctx context.Context) (ok bool) {
exec.log.Debug(logs.DeleteFormingSplitInfo)
ok = exec.formSplitInfo(ctx)
if !ok {
return
if err := exec.formSplitInfo(ctx); err != nil {
return fmt.Errorf("form split info: %w", err)
}
exec.log.Debug(logs.DeleteSplitInfoSuccessfullyFormedCollectingMembers)
exec.tombstone.SetSplitID(exec.splitInfo.SplitID())
ok = exec.collectMembers(ctx)
if !ok {
return
if err := exec.collectMembers(ctx); err != nil {
return err
}
exec.log.Debug(logs.DeleteMembersSuccessfullyCollected)
ok = exec.initTombstoneObject()
if !ok {
return
}
return true
return exec.initTombstoneObject()
}