TrueCloudLab/frostfs-node
20
1
Fork 27
Code Issues 105 Pull requests 13 Projects Releases 2 Activity Actions

Search service improvements #647

Merged
fyrchik merged 5 commits from fyrchik/frostfs-node:service-simplification-part6 into master 2024-09-04 19:51:03 +00:00
Conversation 0 Commits 5 Files changed 8 +155 -178
8 changed files with 155 additions and 178 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

41
pkg/services/object/search/container.go
View file

@ -3,6 +3,7 @@ package searchsvc
import (
"context"
"encoding/hex"
"fmt"
"sync"
"git.frostfs.info/TrueCloudLab/frostfs-node/internal/logs"
@ -10,12 +11,7 @@ import (
"go.uber.org/zap"
)
func (exec *execCtx) executeOnContainer(ctx context.Context) {
if exec.isLocal() {
exec.log.Debug(logs.SearchReturnResultDirectly)
return
}
func (exec *execCtx) executeOnContainer(ctx context.Context) error {
lookupDepth := exec.netmapLookupDepth()
exec.log.Debug(logs.TryingToExecuteInContainer,
@ -23,13 +19,12 @@ func (exec *execCtx) executeOnContainer(ctx context.Context) {
)
// initialize epoch number
ok := exec.initEpoch()
if !ok {
return
if err := exec.initEpoch(); err != nil {
return fmt.Errorf("%s: %w", logs.CouldNotGetCurrentEpochNumber, err)
}
for {
if exec.processCurrentEpoch(ctx) {
if err := exec.processCurrentEpoch(ctx); err != nil {
break
}
@ -44,18 +39,17 @@ func (exec *execCtx) executeOnContainer(ctx context.Context) {
exec.curProcEpoch--
}
exec.status = statusOK
exec.err = nil
return nil
}
func (exec *execCtx) processCurrentEpoch(ctx context.Context) bool {
func (exec *execCtx) processCurrentEpoch(ctx context.Context) error {
exec.log.Debug(logs.ProcessEpoch,
zap.Uint64("number", exec.curProcEpoch),
)
traverser, ok := exec.generateTraverser(exec.containerID())
if !ok {
return true
traverser, err := exec.svc.traverserGenerator.GenerateTraverser(exec.containerID(), nil, exec.curProcEpoch)
if err != nil {
return fmt.Errorf("%s: %w", logs.SearchCouldNotGenerateContainerTraverser, err)
}
ctx, cancel := context.WithCancel(ctx)
@ -91,12 +85,7 @@ func (exec *execCtx) processCurrentEpoch(ctx context.Context) bool {
c, err := exec.svc.clientConstructor.get(info)
if err != nil {
mtx.Lock()
exec.status = statusUndefined
exec.err = err
mtx.Unlock()
exec.log.Debug(logs.SearchCouldNotConstructRemoteNodeClient)
exec.log.Debug(logs.SearchCouldNotConstructRemoteNodeClient, zap.String("error", err.Error()))
return
}
@ -109,13 +98,17 @@ func (exec *execCtx) processCurrentEpoch(ctx context.Context) bool {
}
mtx.Lock()
exec.writeIDList(ids)
err = exec.writeIDList(ids)
mtx.Unlock()
if err != nil {
exec.log.Debug(logs.SearchCouldNotWriteObjectIdentifiers, zap.String("error", err.Error()))
return
}
}(i)
}
wg.Wait()
}
return false
return nil
}

80
pkg/services/object/search/exec.go
View file

@ -1,8 +1,6 @@
package searchsvc
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"
cid "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/container/id"
objectSDK "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/object"
@ -10,34 +8,16 @@ import (
"go.uber.org/zap"
)
type statusError struct {
status int
err error
}
type execCtx struct {
svc *Service
prm Prm
statusError
log *logger.Logger
curProcEpoch uint64
}
const (
statusUndefined int = iota
statusOK
)
func (exec *execCtx) prepare() {
if _, ok := exec.prm.writer.(*uniqueIDWriter); !ok {
exec.prm.writer = newUniqueAddressWriter(exec.prm.writer)
}
}
func (exec *execCtx) setLogger(l *logger.Logger) {
exec.log = &logger.Logger{Logger: l.With(
zap.String("request", "SEARCH"),
@ -68,64 +48,24 @@ func (exec *execCtx) netmapLookupDepth() uint64 {
return exec.prm.common.NetmapLookupDepth()
}
func (exec *execCtx) initEpoch() bool {
func (exec *execCtx) initEpoch() error {
exec.curProcEpoch = exec.netmapEpoch()
if exec.curProcEpoch > 0 {
return true
return nil
}
e, err := exec.svc.currentEpochReceiver.currentEpoch()
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
return true
e, err := exec.svc.currentEpochReceiver.Epoch()
if err != nil {
return err
}
exec.curProcEpoch = e
return nil
}
func (exec *execCtx) generateTraverser(cnr cid.ID) (*placement.Traverser, bool) {
t, err := exec.svc.traverserGenerator.generateTraverser(cnr, 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) {
func (exec *execCtx) writeIDList(ids []oid.ID) error {
ids = exec.filterAllowedObjectIDs(ids)
err := 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
}
return exec.prm.writer.WriteIDs(ids)
}
func (exec *execCtx) filterAllowedObjectIDs(objIDs []oid.ID) []oid.ID {

20
pkg/services/object/search/local.go
View file

@ -2,24 +2,22 @@ package searchsvc
import (
"context"
"fmt"
"git.frostfs.info/TrueCloudLab/frostfs-node/internal/logs"
"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)
if err != nil {
exec.status = statusUndefined
exec.err = err
exec.log.Debug(logs.SearchLocalOperationFailed,
zap.String("error", err.Error()),
)
return
exec.log.Debug(logs.SearchLocalOperationFailed, zap.String("error", err.Error()))
return err
}
exec.writeIDList(ids)
if err := exec.writeIDList(ids); err != nil {
return fmt.Errorf("%s: %w", logs.SearchCouldNotWriteObjectIdentifiers, err)
}
return nil
}

4
pkg/services/object/search/prm.go
View file

@ -12,7 +12,7 @@ import (
// Prm groups parameters of Get service call.
type Prm struct {
writer IDListWriter
writer *uniqueIDWriter
common *util.CommonPrm
@ -40,7 +40,7 @@ func (p *Prm) SetCommonParameters(common *util.CommonPrm) {
// SetWriter sets target component to write list of object identifiers.
func (p *Prm) SetWriter(w IDListWriter) {
p.writer = w
p.writer = newUniqueAddressWriter(w)
}
// SetRequestForwarder sets callback for forwarding

37
pkg/services/object/search/search.go
View file

@ -14,37 +14,32 @@ func (s *Service) Search(ctx context.Context, prm Prm) error {
prm: prm,
}
exec.prepare()
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)
err := 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) {
// analyze local result
switch exec.status {
func (exec *execCtx) logResult(err error) {
switch {
default:
exec.log.Debug(logs.OperationFinishedWithError,
zap.String("error", exec.err.Error()),
)
case statusOK:
exec.log.Debug(logs.OperationFinishedWithError, zap.String("error", err.Error()))
case err == nil:
exec.log.Debug(logs.OperationFinishedSuccessfully)
}
if execCnr {
exec.executeOnContainer(ctx)
exec.analyzeStatus(ctx, false)
}
}

117
pkg/services/object/search/search_test.go
View file

@ -22,6 +22,7 @@ import (
frostfsecdsa "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/crypto/ecdsa"
"git.frostfs.info/TrueCloudLab/frostfs-sdk-go/netmap"
oid "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/object/id"
oidtest "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/object/id/test"
sessionsdk "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/session"
"github.com/google/uuid"
"github.com/nspcc-dev/neo-go/pkg/crypto/keys"
@ -56,10 +57,18 @@ type simpleIDWriter struct {
type testEpochReceiver uint64
func (e testEpochReceiver) currentEpoch() (uint64, error) {
func (e testEpochReceiver) Epoch() (uint64, error) {
return uint64(e), nil
}
type errIDWriter struct {
err error
}
func (e errIDWriter) WriteIDs(ids []oid.ID) error {
return e.err
}
func (s *simpleIDWriter) WriteIDs(ids []oid.ID) error {
s.ids = append(s.ids, ids...)
return nil
@ -71,7 +80,7 @@ func newTestStorage() *testStorage {
}
}
func (g *testTraverserGenerator) generateTraverser(_ cid.ID, epoch uint64) (*placement.Traverser, error) {
func (g *testTraverserGenerator) GenerateTraverser(_ cid.ID, _ *oid.ID, epoch uint64) (*placement.Traverser, error) {
return placement.NewTraverser(
placement.ForContainer(g.c),
placement.UseBuilder(g.b[epoch]),
@ -194,6 +203,20 @@ func TestGetLocalOnly(t *testing.T) {
w := new(simpleIDWriter)
p := newPrm(cnr, w)
err := svc.Search(ctx, p)
require.ErrorIs(t, err, testErr)
})
t.Run("FAIL while writing ID", func(t *testing.T) {
storage := newTestStorage()
svc := newSvc(storage)
cnr := cidtest.ID()
storage.addResult(cnr, []oid.ID{oidtest.ID()}, nil)
testErr := errors.New("any error")
w := errIDWriter{testErr}
p := newPrm(cnr, w)
err := svc.Search(ctx, p)
require.ErrorIs(t, err, testErr)
})
@ -280,33 +303,34 @@ func TestGetRemoteSmall(t *testing.T) {
return p
}
var addr oid.Address
addr.SetContainer(id)
ns, as := testNodeMatrix(t, placementDim)
builder := &testPlacementBuilder{
vectors: map[string][][]netmap.NodeInfo{
addr.EncodeToString(): ns,
},
}
c1 := newTestStorage()
ids1 := generateIDs(10)
c2 := newTestStorage()
ids2 := generateIDs(10)
svc := newSvc(builder, &testClientCache{
clients: map[string]*testStorage{
as[0][0]: c1,
as[0][1]: c2,
},
})
t.Run("OK", func(t *testing.T) {
var addr oid.Address
addr.SetContainer(id)
ns, as := testNodeMatrix(t, placementDim)
builder := &testPlacementBuilder{
vectors: map[string][][]netmap.NodeInfo{
addr.EncodeToString(): ns,
},
}
c1 := newTestStorage()
ids1 := generateIDs(10)
c1.addResult(id, ids1, nil)
c2 := newTestStorage()
ids2 := generateIDs(10)
c2.addResult(id, ids2, nil)
svc := newSvc(builder, &testClientCache{
clients: map[string]*testStorage{
as[0][0]: c1,
as[0][1]: c2,
},
})
w := new(simpleIDWriter)
p := newPrm(id, w)
@ -319,6 +343,49 @@ func TestGetRemoteSmall(t *testing.T) {
require.Contains(t, w.ids, id)
}
})
t.Run("non-local fail is not a FAIL", func(t *testing.T) {
testErr := errors.New("opaque")
c1.addResult(id, ids1, nil)
c2.addResult(id, nil, testErr)
w := new(simpleIDWriter)
p := newPrm(id, w)
err := svc.Search(ctx, p)
require.NoError(t, err)
require.Equal(t, ids1, w.ids)
})
t.Run("client init fail is not a FAIL", func(t *testing.T) {
svc := newSvc(builder, &testClientCache{
clients: map[string]*testStorage{
as[0][0]: c1,
},
})
c1.addResult(id, ids1, nil)
c2.addResult(id, ids2, nil)
w := new(simpleIDWriter)
p := newPrm(id, w)
err := svc.Search(ctx, p)
require.NoError(t, err)
require.Equal(t, ids1, w.ids)
})
t.Run("context is respected", func(t *testing.T) {
c1.addResult(id, ids1, nil)
c2.addResult(id, ids2, nil)
w := new(simpleIDWriter)
p := newPrm(id, w)
ctx, cancel := context.WithCancel(context.Background())
cancel()
err := svc.Search(ctx, p)
require.NoError(t, err)
require.Empty(t, w.ids)
})
}
func TestGetFromPastEpoch(t *testing.T) {

12
pkg/services/object/search/service.go
View file

@ -45,11 +45,11 @@ type cfg struct {
}
traverserGenerator interface {
generateTraverser(cid.ID, uint64) (*placement.Traverser, error)
GenerateTraverser(cid.ID, *oid.ID, uint64) (*placement.Traverser, error)
}
currentEpochReceiver interface {
currentEpoch() (uint64, error)
Epoch() (uint64, error)
}
keyStore *util.KeyStorage
@ -71,11 +71,9 @@ func New(e *engine.StorageEngine,
localStorage: &storageEngineWrapper{
storage: e,
},
traverserGenerator: (*traverseGeneratorWrapper)(tg),
currentEpochReceiver: &nmSrcWrapper{
nmSrc: ns,
},
keyStore: ks,
traverserGenerator: tg,
currentEpochReceiver: ns,
keyStore: ks,
}
for i := range opts {

22
pkg/services/object/search/util.go
View file

@ -5,12 +5,9 @@ import (
"sync"
"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/local_object_storage/engine"
internalclient "git.frostfs.info/TrueCloudLab/frostfs-node/pkg/services/object/internal/client"
"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/services/object/util"
"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/services/object_manager/placement"
cid "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/container/id"
oid "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/object/id"
)
@ -34,13 +31,10 @@ type storageEngineWrapper struct {
storage *engine.StorageEngine
}
type traverseGeneratorWrapper util.TraverserGenerator
type nmSrcWrapper struct {
nmSrc netmap.Source
}
func newUniqueAddressWriter(w IDListWriter) IDListWriter {
func newUniqueAddressWriter(w IDListWriter) *uniqueIDWriter {
if w, ok := w.(*uniqueIDWriter); ok {
return w
}
return &uniqueIDWriter{
written: make(map[oid.ID]struct{}),
writer: w,
@ -139,11 +133,3 @@ func idsFromAddresses(addrs []oid.Address) []oid.ID {
return ids
}
func (e *traverseGeneratorWrapper) generateTraverser(cnr cid.ID, epoch uint64) (*placement.Traverser, error) {
return (*util.TraverserGenerator)(e).GenerateTraverser(cnr, nil, epoch)
}
func (n *nmSrcWrapper) currentEpoch() (uint64, error) {
return n.nmSrc.Epoch()
}