[#275] innerring: Add storage group search timeout
Signed-off-by: Alex Vanin <alexey@nspcc.ru>
This commit is contained in:
parent
6836975272
commit
310a4c3e4d
4 changed files with 11 additions and 3 deletions
|
@ -96,4 +96,5 @@ func defaultConfiguration(cfg *viper.Viper) {
|
||||||
cfg.SetDefault("audit.timeout.get", "5s")
|
cfg.SetDefault("audit.timeout.get", "5s")
|
||||||
cfg.SetDefault("audit.timeout.head", "5s")
|
cfg.SetDefault("audit.timeout.head", "5s")
|
||||||
cfg.SetDefault("audit.timeout.rangehash", "5s")
|
cfg.SetDefault("audit.timeout.rangehash", "5s")
|
||||||
|
cfg.SetDefault("audit.timeout.search", "10s")
|
||||||
}
|
}
|
||||||
|
|
|
@ -242,6 +242,7 @@ func New(ctx context.Context, log *zap.Logger, cfg *viper.Viper) (*Server, error
|
||||||
MorphClient: server.morphClient,
|
MorphClient: server.morphClient,
|
||||||
IRList: server,
|
IRList: server,
|
||||||
ClientCache: clientCache,
|
ClientCache: clientCache,
|
||||||
|
RPCSearchTimeout: cfg.GetDuration("audit.timeout.search"),
|
||||||
TaskManager: auditTaskManager,
|
TaskManager: auditTaskManager,
|
||||||
Reporter: server,
|
Reporter: server,
|
||||||
})
|
})
|
||||||
|
|
|
@ -2,7 +2,6 @@ package audit
|
||||||
|
|
||||||
import (
|
import (
|
||||||
"context"
|
"context"
|
||||||
"time"
|
|
||||||
|
|
||||||
"github.com/nspcc-dev/neofs-api-go/pkg/client"
|
"github.com/nspcc-dev/neofs-api-go/pkg/client"
|
||||||
"github.com/nspcc-dev/neofs-api-go/pkg/container"
|
"github.com/nspcc-dev/neofs-api-go/pkg/container"
|
||||||
|
@ -134,8 +133,7 @@ func (ap *Processor) findStorageGroups(cid *container.ID, shuffled netmap.Nodes)
|
||||||
sgSearchParams.WithContainerID(cid)
|
sgSearchParams.WithContainerID(cid)
|
||||||
sgSearchParams.WithSearchFilters(sgFilter)
|
sgSearchParams.WithSearchFilters(sgFilter)
|
||||||
|
|
||||||
// fixme: timeout from config
|
ctx, cancel := context.WithTimeout(context.Background(), ap.searchTimeout)
|
||||||
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
|
|
||||||
result, err := cli.SearchObject(ctx, sgSearchParams)
|
result, err := cli.SearchObject(ctx, sgSearchParams)
|
||||||
cancel()
|
cancel()
|
||||||
|
|
||||||
|
|
|
@ -2,6 +2,7 @@ package audit
|
||||||
|
|
||||||
import (
|
import (
|
||||||
"context"
|
"context"
|
||||||
|
"time"
|
||||||
|
|
||||||
"github.com/nspcc-dev/neo-go/pkg/util"
|
"github.com/nspcc-dev/neo-go/pkg/util"
|
||||||
SDKClient "github.com/nspcc-dev/neofs-api-go/pkg/client"
|
SDKClient "github.com/nspcc-dev/neofs-api-go/pkg/client"
|
||||||
|
@ -45,6 +46,7 @@ type (
|
||||||
morphClient *client.Client
|
morphClient *client.Client
|
||||||
irList Indexer
|
irList Indexer
|
||||||
clientCache NeoFSClientCache
|
clientCache NeoFSClientCache
|
||||||
|
searchTimeout time.Duration
|
||||||
|
|
||||||
containerClient *wrapContainer.Wrapper
|
containerClient *wrapContainer.Wrapper
|
||||||
netmapClient *wrapNetmap.Wrapper
|
netmapClient *wrapNetmap.Wrapper
|
||||||
|
@ -63,6 +65,7 @@ type (
|
||||||
MorphClient *client.Client
|
MorphClient *client.Client
|
||||||
IRList Indexer
|
IRList Indexer
|
||||||
ClientCache NeoFSClientCache
|
ClientCache NeoFSClientCache
|
||||||
|
RPCSearchTimeout time.Duration
|
||||||
TaskManager TaskManager
|
TaskManager TaskManager
|
||||||
Reporter audit.Reporter
|
Reporter audit.Reporter
|
||||||
}
|
}
|
||||||
|
@ -90,6 +93,10 @@ func New(p *Params) (*Processor, error) {
|
||||||
return nil, errors.New("ir/audit: global state is not set")
|
return nil, errors.New("ir/audit: global state is not set")
|
||||||
case p.ClientCache == nil:
|
case p.ClientCache == nil:
|
||||||
return nil, errors.New("ir/audit: neofs RPC client cache is not set")
|
return nil, errors.New("ir/audit: neofs RPC client cache is not set")
|
||||||
|
case p.TaskManager == nil:
|
||||||
|
return nil, errors.New("ir/audit: audit task manager is not set")
|
||||||
|
case p.Reporter == nil:
|
||||||
|
return nil, errors.New("ir/audit: audit result reporter is not set")
|
||||||
}
|
}
|
||||||
|
|
||||||
pool, err := ants.NewPool(ProcessorPoolSize, ants.WithNonblocking(true))
|
pool, err := ants.NewPool(ProcessorPoolSize, ants.WithNonblocking(true))
|
||||||
|
@ -117,6 +124,7 @@ func New(p *Params) (*Processor, error) {
|
||||||
morphClient: p.MorphClient,
|
morphClient: p.MorphClient,
|
||||||
irList: p.IRList,
|
irList: p.IRList,
|
||||||
clientCache: p.ClientCache,
|
clientCache: p.ClientCache,
|
||||||
|
searchTimeout: p.RPCSearchTimeout,
|
||||||
containerClient: containerClient,
|
containerClient: containerClient,
|
||||||
netmapClient: netmapClient,
|
netmapClient: netmapClient,
|
||||||
taskManager: p.TaskManager,
|
taskManager: p.TaskManager,
|
||||||
|
|
Loading…
Reference in a new issue