[#676] services/container: Cache the results of read operations
In previous implementation Container service handlers didn't cache the results of `Get` / `GetEACL` / `List` operations. As a consequence of this, high load on the service caused neo-go client's connection errors. To avoid this there is a need to use cache. Object service already uses `Get` and `GetEACL` caches. Implement cache of `List` results. Share already implemented cache of Object service with the Container one. Provide new instance of read-only container storage (defined as an interface)to morph executor's constructor on which container service is based. Write operations remained unchanged. Signed-off-by: Leonard Lyubich <leonard@nspcc.ru>
This commit is contained in:
parent
c54f524df9
commit
e738699fcc
5 changed files with 115 additions and 30 deletions
|
@ -9,8 +9,10 @@ import (
|
|||
containerSDK "github.com/nspcc-dev/neofs-api-go/pkg/container"
|
||||
cid "github.com/nspcc-dev/neofs-api-go/pkg/container/id"
|
||||
netmapSDK "github.com/nspcc-dev/neofs-api-go/pkg/netmap"
|
||||
"github.com/nspcc-dev/neofs-api-go/pkg/owner"
|
||||
"github.com/nspcc-dev/neofs-node/pkg/core/container"
|
||||
"github.com/nspcc-dev/neofs-node/pkg/core/netmap"
|
||||
"github.com/nspcc-dev/neofs-node/pkg/morph/client/container/wrapper"
|
||||
"github.com/nspcc-dev/neofs-node/pkg/services/object/acl/eacl"
|
||||
)
|
||||
|
||||
|
@ -227,3 +229,49 @@ func (s *lruNetmapSource) getNetMapByEpoch(epoch uint64) (*netmapSDK.Netmap, err
|
|||
func (s *lruNetmapSource) Epoch() (uint64, error) {
|
||||
return s.netState.CurrentEpoch(), nil
|
||||
}
|
||||
|
||||
// wrapper over TTL cache of values read from the network
|
||||
// that implements container lister.
|
||||
type ttlContainerLister ttlNetCache
|
||||
|
||||
func newCachedContainerLister(w *wrapper.Wrapper) *ttlContainerLister {
|
||||
const (
|
||||
containerListerCacheSize = 100
|
||||
containerListerCacheTTL = 30 * time.Second
|
||||
)
|
||||
|
||||
lruCnrListerCache := newNetworkTTLCache(containerListerCacheSize, containerListerCacheTTL, func(key interface{}) (interface{}, error) {
|
||||
var (
|
||||
id *owner.ID
|
||||
strID = key.(string)
|
||||
)
|
||||
|
||||
if strID != "" {
|
||||
id = owner.NewID()
|
||||
|
||||
err := id.Parse(strID)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
}
|
||||
|
||||
return w.List(id)
|
||||
})
|
||||
|
||||
return (*ttlContainerLister)(lruCnrListerCache)
|
||||
}
|
||||
|
||||
func (s *ttlContainerLister) List(id *owner.ID) ([]*cid.ID, error) {
|
||||
var str string
|
||||
|
||||
if id != nil {
|
||||
str = id.String()
|
||||
}
|
||||
|
||||
val, err := (*ttlNetCache)(s).get(str)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
return val.([]*cid.ID), nil
|
||||
}
|
||||
|
|
|
@ -31,7 +31,6 @@ import (
|
|||
"github.com/nspcc-dev/neofs-node/pkg/local_object_storage/writecache"
|
||||
"github.com/nspcc-dev/neofs-node/pkg/metrics"
|
||||
"github.com/nspcc-dev/neofs-node/pkg/morph/client"
|
||||
cntwrapper "github.com/nspcc-dev/neofs-node/pkg/morph/client/container/wrapper"
|
||||
nmwrapper "github.com/nspcc-dev/neofs-node/pkg/morph/client/netmap/wrapper"
|
||||
"github.com/nspcc-dev/neofs-node/pkg/morph/event"
|
||||
netmap2 "github.com/nspcc-dev/neofs-node/pkg/morph/event/netmap"
|
||||
|
@ -39,6 +38,7 @@ import (
|
|||
"github.com/nspcc-dev/neofs-node/pkg/network"
|
||||
"github.com/nspcc-dev/neofs-node/pkg/network/cache"
|
||||
"github.com/nspcc-dev/neofs-node/pkg/services/control"
|
||||
"github.com/nspcc-dev/neofs-node/pkg/services/object/acl/eacl"
|
||||
trustcontroller "github.com/nspcc-dev/neofs-node/pkg/services/reputation/local/controller"
|
||||
truststorage "github.com/nspcc-dev/neofs-node/pkg/services/reputation/local/storage"
|
||||
tokenStorage "github.com/nspcc-dev/neofs-node/pkg/services/session/storage"
|
||||
|
@ -171,7 +171,7 @@ type cfgObject struct {
|
|||
|
||||
cnrSource container.Source
|
||||
|
||||
cnrClient *cntwrapper.Wrapper
|
||||
eaclSource eacl.Source
|
||||
|
||||
pool cfgObjectRoutines
|
||||
|
||||
|
|
|
@ -8,10 +8,12 @@ import (
|
|||
"fmt"
|
||||
"strconv"
|
||||
|
||||
eaclSDK "github.com/nspcc-dev/neofs-api-go/pkg/acl/eacl"
|
||||
apiClient "github.com/nspcc-dev/neofs-api-go/pkg/client"
|
||||
containerSDK "github.com/nspcc-dev/neofs-api-go/pkg/container"
|
||||
cid "github.com/nspcc-dev/neofs-api-go/pkg/container/id"
|
||||
"github.com/nspcc-dev/neofs-api-go/pkg/netmap"
|
||||
"github.com/nspcc-dev/neofs-api-go/pkg/owner"
|
||||
containerV2 "github.com/nspcc-dev/neofs-api-go/v2/container"
|
||||
containerGRPC "github.com/nspcc-dev/neofs-api-go/v2/container/grpc"
|
||||
containerCore "github.com/nspcc-dev/neofs-node/pkg/core/container"
|
||||
|
@ -28,6 +30,7 @@ import (
|
|||
placementrouter "github.com/nspcc-dev/neofs-node/pkg/services/container/announcement/load/route/placement"
|
||||
loadstorage "github.com/nspcc-dev/neofs-node/pkg/services/container/announcement/load/storage"
|
||||
containerMorph "github.com/nspcc-dev/neofs-node/pkg/services/container/morph"
|
||||
"github.com/nspcc-dev/neofs-node/pkg/services/object/acl/eacl"
|
||||
"github.com/nspcc-dev/neofs-node/pkg/util/logger"
|
||||
"go.uber.org/zap"
|
||||
)
|
||||
|
@ -43,16 +46,26 @@ func initContainerService(c *cfg) {
|
|||
|
||||
cnrSrc := wrapper.AsContainerSource(wrap)
|
||||
|
||||
var containerSource containerCore.Source
|
||||
|
||||
if c.cfgMorph.disableCache {
|
||||
containerSource = cnrSrc
|
||||
} else {
|
||||
containerSource = newCachedContainerStorage(cnrSrc) // use RPC node as source of containers (with caching)
|
||||
eACLFetcher := &morphEACLFetcher{
|
||||
w: wrap,
|
||||
}
|
||||
|
||||
c.cfgObject.cnrSource = containerSource
|
||||
c.cfgObject.cnrClient = wrap
|
||||
cnrRdr := new(morphContainerReader)
|
||||
|
||||
if c.cfgMorph.disableCache {
|
||||
c.cfgObject.eaclSource = eACLFetcher
|
||||
cnrRdr.eacl = eACLFetcher
|
||||
c.cfgObject.cnrSource = cnrSrc
|
||||
cnrRdr.get = cnrSrc
|
||||
cnrRdr.lister = wrap
|
||||
} else {
|
||||
// use RPC node as source of Container contract items (with caching)
|
||||
c.cfgObject.eaclSource = newCachedEACLStorage(eACLFetcher)
|
||||
c.cfgObject.cnrSource = newCachedContainerStorage(cnrSrc)
|
||||
cnrRdr.lister = newCachedContainerLister(wrap)
|
||||
cnrRdr.eacl = c.cfgObject.eaclSource
|
||||
cnrRdr.get = c.cfgObject.cnrSource
|
||||
}
|
||||
|
||||
localMetrics := &localStorageLoad{
|
||||
log: c.log,
|
||||
|
@ -122,7 +135,7 @@ func initContainerService(c *cfg) {
|
|||
&c.key.PrivateKey,
|
||||
containerService.NewResponseService(
|
||||
&usedSpaceService{
|
||||
Server: containerService.NewExecutionService(containerMorph.NewExecutor(wrap)),
|
||||
Server: containerService.NewExecutionService(containerMorph.NewExecutor(wrap, cnrRdr)),
|
||||
loadWriterProvider: loadRouter,
|
||||
loadPlacementBuilder: loadPlacementBuilder,
|
||||
routeBuilder: routeBuilder,
|
||||
|
@ -481,3 +494,26 @@ func (c *usedSpaceService) processLoadValue(ctx context.Context, a containerSDK.
|
|||
|
||||
return nil
|
||||
}
|
||||
|
||||
// implements interface required by container service provided by morph executor.
|
||||
type morphContainerReader struct {
|
||||
eacl eacl.Source
|
||||
|
||||
get containerCore.Source
|
||||
|
||||
lister interface {
|
||||
List(*owner.ID) ([]*cid.ID, error)
|
||||
}
|
||||
}
|
||||
|
||||
func (x *morphContainerReader) Get(id *cid.ID) (*containerSDK.Container, error) {
|
||||
return x.get.Get(id)
|
||||
}
|
||||
|
||||
func (x *morphContainerReader) GetEACL(id *cid.ID) (*eaclSDK.Table, error) {
|
||||
return x.eacl.GetEACL(id)
|
||||
}
|
||||
|
||||
func (x *morphContainerReader) List(id *owner.ID) ([]*cid.ID, error) {
|
||||
return x.lister.List(id)
|
||||
}
|
||||
|
|
|
@ -364,20 +364,6 @@ func initObjectService(c *cfg) {
|
|||
respSvc,
|
||||
)
|
||||
|
||||
var (
|
||||
eACLSource eacl.Source
|
||||
eACLFetcher = &morphEACLFetcher{
|
||||
w: c.cfgObject.cnrClient,
|
||||
}
|
||||
)
|
||||
|
||||
if c.cfgMorph.disableCache {
|
||||
eACLSource = eACLFetcher
|
||||
} else {
|
||||
// use RPC node as source of eACL (with caching)
|
||||
eACLSource = newCachedEACLStorage(eACLFetcher)
|
||||
}
|
||||
|
||||
aclSvc := acl.New(
|
||||
acl.WithSenderClassifier(
|
||||
acl.NewSenderClassifier(
|
||||
|
@ -392,7 +378,7 @@ func initObjectService(c *cfg) {
|
|||
acl.WithNextService(signSvc),
|
||||
acl.WithLocalStorage(ls),
|
||||
acl.WithEACLValidatorOptions(
|
||||
eacl.WithEACLSource(eACLSource),
|
||||
eacl.WithEACLSource(c.cfgObject.eaclSource),
|
||||
eacl.WithLogger(c.log),
|
||||
),
|
||||
acl.WithNetmapState(c.cfgNetmap.state),
|
||||
|
|
|
@ -14,15 +14,30 @@ import (
|
|||
containercore "github.com/nspcc-dev/neofs-node/pkg/core/container"
|
||||
"github.com/nspcc-dev/neofs-node/pkg/morph/client/container/wrapper"
|
||||
containerSvc "github.com/nspcc-dev/neofs-node/pkg/services/container"
|
||||
"github.com/nspcc-dev/neofs-node/pkg/services/object/acl/eacl"
|
||||
)
|
||||
|
||||
type morphExecutor struct {
|
||||
wrapper *wrapper.Wrapper
|
||||
|
||||
rdr Reader
|
||||
}
|
||||
|
||||
func NewExecutor(w *wrapper.Wrapper) containerSvc.ServiceExecutor {
|
||||
// Reader is an interface of read-only container storage.
|
||||
type Reader interface {
|
||||
containercore.Source
|
||||
eacl.Source
|
||||
|
||||
// List returns a list of container identifiers belonging
|
||||
// to the specified owner of NeoFS system. Returns the identifiers
|
||||
// of all NeoFS containers if pointer to owner identifier is nil.
|
||||
List(*owner.ID) ([]*cid.ID, error)
|
||||
}
|
||||
|
||||
func NewExecutor(w *wrapper.Wrapper, rdr Reader) containerSvc.ServiceExecutor {
|
||||
return &morphExecutor{
|
||||
wrapper: w,
|
||||
rdr: rdr,
|
||||
}
|
||||
}
|
||||
|
||||
|
@ -70,7 +85,7 @@ func (s *morphExecutor) Delete(ctx containerSvc.ContextWithToken, body *containe
|
|||
func (s *morphExecutor) Get(ctx context.Context, body *container.GetRequestBody) (*container.GetResponseBody, error) {
|
||||
id := cid.NewFromV2(body.GetContainerID())
|
||||
|
||||
cnr, err := wrapper.Get(s.wrapper, id)
|
||||
cnr, err := s.rdr.Get(id)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
@ -86,7 +101,7 @@ func (s *morphExecutor) Get(ctx context.Context, body *container.GetRequestBody)
|
|||
func (s *morphExecutor) List(ctx context.Context, body *container.ListRequestBody) (*container.ListResponseBody, error) {
|
||||
oid := owner.NewIDFromV2(body.GetOwnerID())
|
||||
|
||||
cnrs, err := s.wrapper.List(oid)
|
||||
cnrs, err := s.rdr.List(oid)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
@ -120,7 +135,7 @@ func (s *morphExecutor) SetExtendedACL(ctx containerSvc.ContextWithToken, body *
|
|||
func (s *morphExecutor) GetExtendedACL(ctx context.Context, body *container.GetExtendedACLRequestBody) (*container.GetExtendedACLResponseBody, error) {
|
||||
id := cid.NewFromV2(body.GetContainerID())
|
||||
|
||||
table, err := s.wrapper.GetEACL(id)
|
||||
table, err := s.rdr.GetEACL(id)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
|
Loading…
Reference in a new issue