TrueCloudLab/frostfs-node
20
1
Fork 27
Code Issues 101 Pull requests 5 Projects Releases 1 Activity Actions 1

[#242] engine: Do not iterate over shards if cid is not set

Browse source 
Signed-off-by: Alex Vanin <alexey@nspcc.ru>
This commit is contained in:
Alex Vanin 2020-12-10 17:12:26 +03:00
parent 9fb7190358
commit 63bc3aab82
1 changed files with 20 additions and 6 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

26
pkg/local_object_storage/engine/select.go
View file

@ -1,8 +1,11 @@
package engine package engine
import ( import (
"errors"
"github.com/nspcc-dev/neofs-api-go/pkg/container" "github.com/nspcc-dev/neofs-api-go/pkg/container"
"github.com/nspcc-dev/neofs-api-go/pkg/object" "github.com/nspcc-dev/neofs-api-go/pkg/object"
meta "github.com/nspcc-dev/neofs-node/pkg/local_object_storage/metabase"
"github.com/nspcc-dev/neofs-node/pkg/local_object_storage/shard" "github.com/nspcc-dev/neofs-node/pkg/local_object_storage/shard"
"go.uber.org/zap" "go.uber.org/zap"
) )
@ -48,6 +51,8 @@ func (e *StorageEngine) Select(prm *SelectPrm) (*SelectRes, error) {
addrList := make([]*object.Address, 0) addrList := make([]*object.Address, 0)
uniqueMap := make(map[string]struct{}) uniqueMap := make(map[string]struct{})
var outError error
shPrm := new(shard.SelectPrm). shPrm := new(shard.SelectPrm).
WithContainerID(prm.cid). WithContainerID(prm.cid).
WithFilters(prm.filters) WithFilters(prm.filters)
@ -55,11 +60,20 @@ func (e *StorageEngine) Select(prm *SelectPrm) (*SelectRes, error) {
e.iterateOverUnsortedShards(func(sh *shard.Shard) (stop bool) { e.iterateOverUnsortedShards(func(sh *shard.Shard) (stop bool) {
res, err := sh.Select(shPrm) res, err := sh.Select(shPrm)
if err != nil { if err != nil {
// TODO: smth wrong with shard, need to be processed switch {
e.log.Warn("could not select objects from shard", case errors.Is(err, meta.ErrMissingContainerID): // should never happen
zap.Stringer("shard", sh.ID()), e.log.Error("missing container ID parameter")
zap.String("error", err.Error()), outError = err
)
return true
default:
// TODO: smth wrong with shard, need to be processed
e.log.Warn("could not select objects from shard",
zap.Stringer("shard", sh.ID()),
zap.String("error", err.Error()),
)
return false
}
} else { } else {
for _, addr := range res.AddressList() { // save only unique values for _, addr := range res.AddressList() { // save only unique values
if _, ok := uniqueMap[addr.String()]; !ok { if _, ok := uniqueMap[addr.String()]; !ok {
@ -74,7 +88,7 @@ func (e *StorageEngine) Select(prm *SelectPrm) (*SelectRes, error) {
return &SelectRes{ return &SelectRes{
addrList: addrList, addrList: addrList,
}, nil }, outError
} }
// List returns `limit` available physically storage object addresses in engine. // List returns `limit` available physically storage object addresses in engine.