Skip to content

services/object: use SearchV2 for search objects #3312

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
wants to merge 1 commit into
base: master
Choose a base branch
from
Draft

services/object: use SearchV2 for search objects #3312

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
92 changes: 81 additions & 11 deletions cmd/neofs-node/object.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,10 @@
"crypto/ecdsa"
"errors"
"fmt"
"github.com/nspcc-dev/neofs-node/pkg/network"
neofsecdsa "github.com/nspcc-dev/neofs-sdk-go/crypto/ecdsa"
"slices"
"sync"
"sync/atomic"

"github.com/google/uuid"
Expand Down Expand Up @@ -257,7 +261,13 @@
putsvc.WithWorkerPools(c.cfgObject.pool.putRemote, c.cfgObject.pool.putLocal),
putsvc.WithLogger(c.log),
putsvc.WithSplitChainVerifier(split.NewVerifier(sGet)),
putsvc.WithTombstoneVerifier(tombstone.NewVerifier(objectSource{sGet, sSearch})),
putsvc.WithTombstoneVerifier(
tombstone.NewVerifier(objectSource{
sGet,
coreConstructor,
newRemoteContainerNodes(cnrNodes, c.IsLocalKey),
&c.key.PrivateKey,
})),

Check warning on line 270 in cmd/neofs-node/object.go

View check run for this annotation

Codecov / codecov/patch

cmd/neofs-node/object.go#L264-L270 

Added lines #L264 - L270 were not covered by tests
)

sDelete := deletesvc.New(
Expand Down Expand Up @@ -638,8 +648,14 @@
}

type objectSource struct {
get *getsvc.Service
search *searchsvc.Service
get *getsvc.Service
clientConstractor interface {
Get(coreclient.NodeInfo) (coreclient.MultiAddressClient, error)
}
containers interface {
ForEachRemoteContainerNode(cnr cid.ID, f func(netmapsdk.NodeInfo)) error
}
key *ecdsa.PrivateKey
}

type searchWriter struct {
Expand All @@ -665,19 +681,73 @@
}

func (o objectSource) Search(ctx context.Context, cnr cid.ID, filters objectSDK.SearchFilters) ([]oid.ID, error) {
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This API can be reworked to be sv2-compatible as well. It is important to have it being used properly. For example, tombstone verifier for v2 doesn't need all oids, it can set lower limit. For v1 it can avoid subsequent link object HEAD by requesting payload size (see #3324, it's not as reliable as children check, but it's good enough for old v1 objects).

var sw searchWriter
mProcessedNodes := make(map[string]struct{})
var wg sync.WaitGroup
var mu sync.Mutex

result := make([]oid.ID, 0)
err := o.containers.ForEachRemoteContainerNode(cnr, func(node netmapsdk.NodeInfo) {
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Not sure why you have to do it this way, it should be a regular SearchV2 call just like it was with the old Search call.

pubKey := node.PublicKey()
strKey := string(pubKey)
if _, ok := mProcessedNodes[strKey]; ok {
return
}
mProcessedNodes[strKey] = struct{}{}

var endpoints network.AddressGroup
err := endpoints.FromIterator(network.NodeEndpointsIterator(node))
if err != nil {
// critical error that may ultimately block the storage service. Normally it
// should not appear because entry into the network map under strict control.
return
}

Check warning on line 703 in cmd/neofs-node/object.go

View check run for this annotation

Codecov / codecov/patch

cmd/neofs-node/object.go#L684-L703 

Added lines #L684 - L703 were not covered by tests

var sPrm searchsvc.Prm
sPrm.SetWriter(&sw)
sPrm.WithSearchFilters(filters)
sPrm.WithContainerID(cnr)
var info coreclient.NodeInfo
info.SetAddressGroup(endpoints)
info.SetPublicKey(pubKey)

wg.Add(1)
go func() {
defer wg.Done()
select {
case <-ctx.Done():
return
default:

Check warning on line 715 in cmd/neofs-node/object.go

View check run for this annotation

Codecov / codecov/patch

cmd/neofs-node/object.go#L705-L715 

Added lines #L705 - L715 were not covered by tests
}

c, err := o.clientConstractor.Get(info)
if err != nil {
return
}

Check warning on line 721 in cmd/neofs-node/object.go

View check run for this annotation

Codecov / codecov/patch

cmd/neofs-node/object.go#L718-L721 

Added lines #L718 - L721 were not covered by tests

err := o.search.Search(ctx, sPrm)
var opts client.SearchObjectsOptions
var cursor string
var next []client.SearchResultItem
var res []oid.ID
for {
next, cursor, err = c.SearchObjects(ctx, cnr, filters, nil, cursor, (*neofsecdsa.Signer)(o.key), opts)
if err != nil {
return
}
res = slices.Grow(res, len(res)+len(next))
for i := range next {
res = append(res, next[i].ID)
}
if cursor == "" {
mu.Lock()
result = append(result, res...)
mu.Unlock()
return
}

Check warning on line 741 in cmd/neofs-node/object.go

View check run for this annotation

Codecov / codecov/patch

cmd/neofs-node/object.go#L723-L741 

Added lines #L723 - L741 were not covered by tests
}
}()
})
wg.Wait()

Check warning on line 745 in cmd/neofs-node/object.go

View check run for this annotation

Codecov / codecov/patch

cmd/neofs-node/object.go#L745 

Added line #L745 was not covered by tests
if err != nil {
return nil, err
return nil, fmt.Errorf("iteration over container nodes: %w", err)

Check warning on line 747 in cmd/neofs-node/object.go

View check run for this annotation

Codecov / codecov/patch

cmd/neofs-node/object.go#L747 

Added line #L747 was not covered by tests
}

return sw.ids, nil
return result, nil

Check warning on line 750 in cmd/neofs-node/object.go

View check run for this annotation

Codecov / codecov/patch

cmd/neofs-node/object.go#L750 

Added line #L750 was not covered by tests
}

// IsLocalNodePublicKey checks whether given binary-encoded public key is
Expand Down
Loading