[#465] Handle cache cast failure

Signed-off-by: Denis Kirillov <denis@nspcc.ru>
This commit is contained in:
Denis Kirillov 2022-06-06 11:01:12 +03:00 committed by Alex Vanin
parent f00ca1b6c4
commit 300d4359d8
12 changed files with 307 additions and 41 deletions

View file

@ -1,23 +1,27 @@
package cache
import (
"fmt"
"time"
"github.com/bluele/gcache"
"github.com/nspcc-dev/neofs-s3-gw/creds/accessbox"
oid "github.com/nspcc-dev/neofs-sdk-go/object/id"
"go.uber.org/zap"
)
type (
// AccessBoxCache stores an access box by its address.
AccessBoxCache struct {
cache gcache.Cache
logger *zap.Logger
cache gcache.Cache
}
// Config stores expiration params for cache.
Config struct {
Size int
Lifetime time.Duration
Logger *zap.Logger
}
)
@ -29,15 +33,19 @@ const (
)
// DefaultAccessBoxConfig returns new default cache expiration values.
func DefaultAccessBoxConfig() *Config {
return &Config{Size: DefaultAccessBoxCacheSize, Lifetime: DefaultAccessBoxCacheLifetime}
func DefaultAccessBoxConfig(logger *zap.Logger) *Config {
return &Config{
Size: DefaultAccessBoxCacheSize,
Lifetime: DefaultAccessBoxCacheLifetime,
Logger: logger,
}
}
// NewAccessBoxCache creates an object of BucketCache.
func NewAccessBoxCache(config *Config) *AccessBoxCache {
gc := gcache.New(config.Size).LRU().Expiration(config.Lifetime).Build()
return &AccessBoxCache{cache: gc}
return &AccessBoxCache{cache: gc, logger: config.Logger}
}
// Get returns a cached object.
@ -49,6 +57,8 @@ func (o *AccessBoxCache) Get(address oid.Address) *accessbox.Box {
result, ok := entry.(*accessbox.Box)
if !ok {
o.logger.Warn("invalid cache entry type", zap.String("actual", fmt.Sprintf("%T", entry)),
zap.String("expected", "*accessbox.Box"))
return nil
}

17
api/cache/buckets.go vendored
View file

@ -1,15 +1,18 @@
package cache
import (
"fmt"
"time"
"github.com/bluele/gcache"
"github.com/nspcc-dev/neofs-s3-gw/api/data"
"go.uber.org/zap"
)
// BucketCache contains cache with objects and the lifetime of cache entries.
type BucketCache struct {
cache gcache.Cache
cache gcache.Cache
logger *zap.Logger
}
const (
@ -20,14 +23,18 @@ const (
)
// DefaultBucketConfig returns new default cache expiration values.
func DefaultBucketConfig() *Config {
return &Config{Size: DefaultBucketCacheSize, Lifetime: DefaultBucketCacheLifetime}
func DefaultBucketConfig(logger *zap.Logger) *Config {
return &Config{
Size: DefaultBucketCacheSize,
Lifetime: DefaultBucketCacheLifetime,
Logger: logger,
}
}
// NewBucketCache creates an object of BucketCache.
func NewBucketCache(config *Config) *BucketCache {
gc := gcache.New(config.Size).LRU().Expiration(config.Lifetime).Build()
return &BucketCache{cache: gc}
return &BucketCache{cache: gc, logger: config.Logger}
}
// Get returns a cached object.
@ -39,6 +46,8 @@ func (o *BucketCache) Get(key string) *data.BucketInfo {
result, ok := entry.(*data.BucketInfo)
if !ok {
o.logger.Warn("invalid cache entry type", zap.String("actual", fmt.Sprintf("%T", entry)),
zap.String("expected", "*data.BucketInfo"))
return nil
}

196
api/cache/cache_test.go vendored Normal file
View file

@ -0,0 +1,196 @@
package cache
import (
"testing"
"github.com/nspcc-dev/neofs-s3-gw/api/data"
"github.com/nspcc-dev/neofs-s3-gw/creds/accessbox"
cidtest "github.com/nspcc-dev/neofs-sdk-go/container/id/test"
"github.com/nspcc-dev/neofs-sdk-go/object"
oid "github.com/nspcc-dev/neofs-sdk-go/object/id"
oidtest "github.com/nspcc-dev/neofs-sdk-go/object/id/test"
"github.com/stretchr/testify/require"
"go.uber.org/zap"
"go.uber.org/zap/zaptest/observer"
)
func TestAccessBoxCacheType(t *testing.T) {
logger, observedLog := getObservedLogger()
cache := NewAccessBoxCache(DefaultAccessBoxConfig(logger))
addr := oidtest.Address()
box := &accessbox.Box{}
err := cache.Put(addr, box)
require.NoError(t, err)
val := cache.Get(addr)
require.Equal(t, box, val)
require.Equal(t, 0, observedLog.Len())
err = cache.cache.Set(addr.EncodeToString(), "tmp")
require.NoError(t, err)
assertInvalidCacheEntry(t, cache.Get(addr), observedLog)
}
func TestBucketsCacheType(t *testing.T) {
logger, observedLog := getObservedLogger()
cache := NewBucketCache(DefaultBucketConfig(logger))
bktInfo := &data.BucketInfo{Name: "bucket"}
err := cache.Put(bktInfo)
require.NoError(t, err)
val := cache.Get(bktInfo.Name)
require.Equal(t, bktInfo, val)
require.Equal(t, 0, observedLog.Len())
err = cache.cache.Set(bktInfo.Name, "tmp")
require.NoError(t, err)
assertInvalidCacheEntry(t, cache.Get(bktInfo.Name), observedLog)
}
func TestObjectNamesCacheType(t *testing.T) {
logger, observedLog := getObservedLogger()
cache := NewObjectsNameCache(DefaultObjectsNameConfig(logger))
key := "name"
addr := oidtest.Address()
err := cache.Put(key, addr)
require.NoError(t, err)
val := cache.Get(key)
require.Equal(t, addr, *val)
require.Equal(t, 0, observedLog.Len())
err = cache.cache.Set(key, "tmp")
require.NoError(t, err)
assertInvalidCacheEntry(t, cache.Get(key), observedLog)
}
func TestObjectCacheType(t *testing.T) {
logger, observedLog := getObservedLogger()
cache := New(DefaultObjectsConfig(logger))
addr := oidtest.Address()
obj := object.New()
obj.SetContainerID(addr.Container())
obj.SetID(addr.Object())
err := cache.Put(*obj)
require.NoError(t, err)
val := cache.Get(addr)
require.Equal(t, obj, val)
require.Equal(t, 0, observedLog.Len())
err = cache.cache.Set(addr.EncodeToString(), "tmp")
require.NoError(t, err)
assertInvalidCacheEntry(t, cache.Get(addr), observedLog)
}
func TestObjectsListCacheType(t *testing.T) {
logger, observedLog := getObservedLogger()
cache := NewObjectsListCache(DefaultObjectsListConfig(logger))
cnrID := cidtest.ID()
key := ObjectsListKey{cid: cnrID.EncodeToString(), prefix: "obj"}
objIDs := []oid.ID{oidtest.ID()}
err := cache.Put(key, objIDs)
require.NoError(t, err)
val := cache.Get(key)
require.Equal(t, objIDs, val)
require.Equal(t, 0, observedLog.Len())
err = cache.cache.Set(key, "tmp")
require.NoError(t, err)
assertInvalidCacheEntry(t, cache.Get(key), observedLog)
err = cache.cache.Set("key", "tmp")
require.NoError(t, err)
cache.CleanCacheEntriesContainingObject(key.prefix, cnrID)
require.Equal(t, 2, observedLog.Len())
require.Equal(t, observedLog.All()[1].Message, "invalid cache key type")
}
func TestObjectInfoCacheType(t *testing.T) {
logger, observedLog := getObservedLogger()
cache := NewSystemCache(DefaultSystemConfig(logger))
key := "key"
objInfo := &data.ObjectInfo{Name: key}
err := cache.PutObject(key, objInfo)
require.NoError(t, err)
val := cache.GetObject(key)
require.Equal(t, objInfo, val)
require.Equal(t, 0, observedLog.Len())
err = cache.cache.Set(key, "tmp")
require.NoError(t, err)
assertInvalidCacheEntry(t, cache.GetObject(key), observedLog)
}
func TestCORsCacheType(t *testing.T) {
logger, observedLog := getObservedLogger()
cache := NewSystemCache(DefaultSystemConfig(logger))
key := "key"
cors := &data.CORSConfiguration{}
err := cache.PutCORS(key, cors)
require.NoError(t, err)
val := cache.GetCORS(key)
require.Equal(t, cors, val)
require.Equal(t, 0, observedLog.Len())
err = cache.cache.Set(key, "tmp")
require.NoError(t, err)
assertInvalidCacheEntry(t, cache.GetCORS(key), observedLog)
}
func TestSettingsCacheType(t *testing.T) {
logger, observedLog := getObservedLogger()
cache := NewSystemCache(DefaultSystemConfig(logger))
key := "key"
settings := &data.BucketSettings{VersioningEnabled: true}
err := cache.PutSettings(key, settings)
require.NoError(t, err)
val := cache.GetSettings(key)
require.Equal(t, settings, val)
require.Equal(t, 0, observedLog.Len())
err = cache.cache.Set(key, "tmp")
require.NoError(t, err)
assertInvalidCacheEntry(t, cache.GetSettings(key), observedLog)
}
func TestNotificationConfigurationCacheType(t *testing.T) {
logger, observedLog := getObservedLogger()
cache := NewSystemCache(DefaultSystemConfig(logger))
key := "key"
notificationConfig := &data.NotificationConfiguration{}
err := cache.PutNotificationConfiguration(key, notificationConfig)
require.NoError(t, err)
val := cache.GetNotificationConfiguration(key)
require.Equal(t, notificationConfig, val)
require.Equal(t, 0, observedLog.Len())
err = cache.cache.Set(key, "tmp")
require.NoError(t, err)
assertInvalidCacheEntry(t, cache.GetNotificationConfiguration(key), observedLog)
}
func assertInvalidCacheEntry(t *testing.T, val interface{}, observedLog *observer.ObservedLogs) {
require.Nil(t, val)
require.Equal(t, 1, observedLog.Len())
require.Equal(t, observedLog.All()[0].Message, "invalid cache entry type")
}
func getObservedLogger() (*zap.Logger, *observer.ObservedLogs) {
loggerCore, observedLog := observer.New(zap.WarnLevel)
return zap.New(loggerCore), observedLog
}

17
api/cache/names.go vendored
View file

@ -1,17 +1,20 @@
package cache
import (
"fmt"
"time"
"github.com/bluele/gcache"
oid "github.com/nspcc-dev/neofs-sdk-go/object/id"
"go.uber.org/zap"
)
// ObjectsNameCache provides lru cache for objects.
// This cache contains mapping nice names to object addresses.
// Key is bucketName+objectName.
type ObjectsNameCache struct {
cache gcache.Cache
cache gcache.Cache
logger *zap.Logger
}
const (
@ -22,14 +25,18 @@ const (
)
// DefaultObjectsNameConfig returns new default cache expiration values.
func DefaultObjectsNameConfig() *Config {
return &Config{Size: DefaultObjectsNameCacheSize, Lifetime: DefaultObjectsNameCacheLifetime}
func DefaultObjectsNameConfig(logger *zap.Logger) *Config {
return &Config{
Size: DefaultObjectsNameCacheSize,
Lifetime: DefaultObjectsNameCacheLifetime,
Logger: logger,
}
}
// NewObjectsNameCache creates an object of ObjectsNameCache.
func NewObjectsNameCache(config *Config) *ObjectsNameCache {
gc := gcache.New(config.Size).LRU().Expiration(config.Lifetime).Build()
return &ObjectsNameCache{cache: gc}
return &ObjectsNameCache{cache: gc, logger: config.Logger}
}
// Get returns a cached object. Returns nil if value is missing.
@ -41,6 +48,8 @@ func (o *ObjectsNameCache) Get(key string) *oid.Address {
result, ok := entry.(oid.Address)
if !ok {
o.logger.Warn("invalid cache entry type", zap.String("actual", fmt.Sprintf("%T", entry)),
zap.String("expected", "oid.Address"))
return nil
}

27
api/cache/objects.go vendored
View file

@ -1,16 +1,19 @@
package cache
import (
"fmt"
"time"
"github.com/bluele/gcache"
"github.com/nspcc-dev/neofs-sdk-go/object"
oid "github.com/nspcc-dev/neofs-sdk-go/object/id"
"go.uber.org/zap"
)
// ObjectsCache provides lru cache for objects.
type ObjectsCache struct {
cache gcache.Cache
cache gcache.Cache
logger *zap.Logger
}
const (
@ -21,14 +24,18 @@ const (
)
// DefaultObjectsConfig returns new default cache expiration values.
func DefaultObjectsConfig() *Config {
return &Config{Size: DefaultObjectsCacheSize, Lifetime: DefaultObjectsCacheLifetime}
func DefaultObjectsConfig(logger *zap.Logger) *Config {
return &Config{
Size: DefaultObjectsCacheSize,
Lifetime: DefaultObjectsCacheLifetime,
Logger: logger,
}
}
// New creates an object of ObjectHeadersCache.
func New(config *Config) *ObjectsCache {
gc := gcache.New(config.Size).LRU().Expiration(config.Lifetime).Build()
return &ObjectsCache{cache: gc}
return &ObjectsCache{cache: gc, logger: config.Logger}
}
// Get returns a cached object.
@ -40,6 +47,8 @@ func (o *ObjectsCache) Get(address oid.Address) *object.Object {
result, ok := entry.(object.Object)
if !ok {
o.logger.Warn("invalid cache entry type", zap.String("actual", fmt.Sprintf("%T", entry)),
zap.String("expected", "object.Object"))
return nil
}
@ -48,8 +57,14 @@ func (o *ObjectsCache) Get(address oid.Address) *object.Object {
// Put puts an object to cache.
func (o *ObjectsCache) Put(obj object.Object) error {
cnrID, _ := obj.ContainerID()
objID, _ := obj.ID()
cnrID, ok := obj.ContainerID()
if !ok {
return fmt.Errorf("empty container id")
}
objID, ok := obj.ID()
if !ok {
return fmt.Errorf("empty object id")
}
var addr oid.Address
addr.SetContainer(cnrID)

View file

@ -8,6 +8,7 @@ import (
"github.com/bluele/gcache"
cid "github.com/nspcc-dev/neofs-sdk-go/container/id"
oid "github.com/nspcc-dev/neofs-sdk-go/object/id"
"go.uber.org/zap"
)
/*
@ -26,7 +27,8 @@ import (
type (
// ObjectsListCache contains cache for ListObjects and ListObjectVersions.
ObjectsListCache struct {
cache gcache.Cache
cache gcache.Cache
logger *zap.Logger
}
// ObjectsListKey is a key to find a ObjectsListCache's entry.
@ -44,14 +46,18 @@ const (
)
// DefaultObjectsListConfig returns new default cache expiration values.
func DefaultObjectsListConfig() *Config {
return &Config{Size: DefaultObjectsListCacheSize, Lifetime: DefaultObjectsListCacheLifetime}
func DefaultObjectsListConfig(logger *zap.Logger) *Config {
return &Config{
Size: DefaultObjectsListCacheSize,
Lifetime: DefaultObjectsListCacheLifetime,
Logger: logger,
}
}
// NewObjectsListCache is a constructor which creates an object of ListObjectsCache with the given lifetime of entries.
func NewObjectsListCache(config *Config) *ObjectsListCache {
gc := gcache.New(config.Size).LRU().Expiration(config.Lifetime).Build()
return &ObjectsListCache{cache: gc}
return &ObjectsListCache{cache: gc, logger: config.Logger}
}
// Get returns a list of ObjectInfo.
@ -63,6 +69,8 @@ func (l *ObjectsListCache) Get(key ObjectsListKey) []oid.ID {
result, ok := entry.([]oid.ID)
if !ok {
l.logger.Warn("invalid cache entry type", zap.String("actual", fmt.Sprintf("%T", entry)),
zap.String("expected", "[]oid.ID"))
return nil
}
@ -85,6 +93,8 @@ func (l *ObjectsListCache) CleanCacheEntriesContainingObject(objectName string,
for _, key := range keys {
k, ok := key.(ObjectsListKey)
if !ok {
l.logger.Warn("invalid cache key type", zap.String("actual", fmt.Sprintf("%T", key)),
zap.String("expected", "ObjectsListKey"))
continue
}
if cidStr == k.cid && strings.HasPrefix(objectName, k.prefix) {

23
api/cache/system.go vendored
View file

@ -1,17 +1,20 @@
package cache
import (
"fmt"
"time"
"github.com/bluele/gcache"
"github.com/nspcc-dev/neofs-s3-gw/api/data"
"go.uber.org/zap"
)
// SystemCache provides lru cache for objects.
// This cache contains "system" objects (bucket versioning settings, tagging object etc.).
// Key is bucketName+systemFileName.
type SystemCache struct {
cache gcache.Cache
cache gcache.Cache
logger *zap.Logger
}
const (
@ -22,14 +25,18 @@ const (
)
// DefaultSystemConfig returns new default cache expiration values.
func DefaultSystemConfig() *Config {
return &Config{Size: DefaultSystemCacheSize, Lifetime: DefaultSystemCacheLifetime}
func DefaultSystemConfig(logger *zap.Logger) *Config {
return &Config{
Size: DefaultSystemCacheSize,
Lifetime: DefaultSystemCacheLifetime,
Logger: logger,
}
}
// NewSystemCache creates an object of SystemCache.
func NewSystemCache(config *Config) *SystemCache {
gc := gcache.New(config.Size).LRU().Expiration(config.Lifetime).Build()
return &SystemCache{cache: gc}
return &SystemCache{cache: gc, logger: config.Logger}
}
// GetObject returns a cached object.
@ -41,6 +48,8 @@ func (o *SystemCache) GetObject(key string) *data.ObjectInfo {
result, ok := entry.(*data.ObjectInfo)
if !ok {
o.logger.Warn("invalid cache entry type", zap.String("actual", fmt.Sprintf("%T", entry)),
zap.String("expected", "*data.ObjectInfo"))
return nil
}
@ -55,6 +64,8 @@ func (o *SystemCache) GetCORS(key string) *data.CORSConfiguration {
result, ok := entry.(*data.CORSConfiguration)
if !ok {
o.logger.Warn("invalid cache entry type", zap.String("actual", fmt.Sprintf("%T", entry)),
zap.String("expected", "*data.CORSConfiguration"))
return nil
}
@ -69,6 +80,8 @@ func (o *SystemCache) GetSettings(key string) *data.BucketSettings {
result, ok := entry.(*data.BucketSettings)
if !ok {
o.logger.Warn("invalid cache entry type", zap.String("actual", fmt.Sprintf("%T", entry)),
zap.String("expected", "*data.BucketSettings"))
return nil
}
@ -83,6 +96,8 @@ func (o *SystemCache) GetNotificationConfiguration(key string) *data.Notificatio
result, ok := entry.(*data.NotificationConfiguration)
if !ok {
o.logger.Warn("invalid cache entry type", zap.String("actual", fmt.Sprintf("%T", entry)),
zap.String("expected", "*data.NotificationConfiguration"))
return nil
}

View file

@ -54,7 +54,7 @@ func prepareHandlerContext(t *testing.T) *handlerContext {
})
layerCfg := &layer.Config{
Caches: layer.DefaultCachesConfigs(),
Caches: layer.DefaultCachesConfigs(zap.NewExample()),
AnonKey: layer.AnonymousKey{Key: key},
Resolver: testResolver,
}

View file

@ -258,13 +258,13 @@ func (f MsgHandlerFunc) HandleMessage(ctx context.Context, msg *nats.Msg) error
}
// DefaultCachesConfigs returns filled configs.
func DefaultCachesConfigs() *CachesConfig {
func DefaultCachesConfigs(logger *zap.Logger) *CachesConfig {
return &CachesConfig{
Objects: cache.DefaultObjectsConfig(),
ObjectsList: cache.DefaultObjectsListConfig(),
Names: cache.DefaultObjectsNameConfig(),
Buckets: cache.DefaultBucketConfig(),
System: cache.DefaultSystemConfig(),
Objects: cache.DefaultObjectsConfig(logger),
ObjectsList: cache.DefaultObjectsListConfig(logger),
Names: cache.DefaultObjectsNameConfig(logger),
Buckets: cache.DefaultBucketConfig(logger),
System: cache.DefaultSystemConfig(logger),
}
}

View file

@ -136,6 +136,8 @@ type testContext struct {
}
func prepareContext(t *testing.T, cachesConfig ...*CachesConfig) *testContext {
logger := zap.NewExample()
key, err := keys.NewPrivateKey()
require.NoError(t, err)
@ -156,7 +158,7 @@ func prepareContext(t *testing.T, cachesConfig ...*CachesConfig) *testContext {
})
require.NoError(t, err)
config := DefaultCachesConfigs()
config := DefaultCachesConfigs(logger)
if len(cachesConfig) != 0 {
config = cachesConfig[0]
}
@ -168,7 +170,7 @@ func prepareContext(t *testing.T, cachesConfig ...*CachesConfig) *testContext {
return &testContext{
ctx: ctx,
layer: NewLayer(zap.NewNop(), tp, layerCfg),
layer: NewLayer(logger, tp, layerCfg),
bktInfo: &data.BucketInfo{
Name: bktName,
Owner: *usertest.ID(),
@ -607,7 +609,7 @@ func TestUpdateCRDT2PSetHeaders(t *testing.T) {
}
func TestSystemObjectsVersioning(t *testing.T) {
cacheConfig := DefaultCachesConfigs()
cacheConfig := DefaultCachesConfigs(zap.NewExample())
cacheConfig.System.Lifetime = 0
tc := prepareContext(t, cacheConfig)
@ -642,7 +644,7 @@ func TestSystemObjectsVersioning(t *testing.T) {
}
func TestDeleteSystemObjectsVersioning(t *testing.T) {
cacheConfig := DefaultCachesConfigs()
cacheConfig := DefaultCachesConfigs(zap.NewExample())
cacheConfig.System.Lifetime = 0
tc := prepareContext(t, cacheConfig)

View file

@ -249,7 +249,7 @@ func (a *Agent) IssueSecret(ctx context.Context, w io.Writer, options *IssueSecr
zap.Stringer("owner_tkn", idOwner))
addr, err := tokens.
New(a.neoFS, secrets.EphemeralKey, cache.DefaultAccessBoxConfig()).
New(a.neoFS, secrets.EphemeralKey, cache.DefaultAccessBoxConfig(a.log)).
Put(ctx, *id, idOwner, box, lifetime.Exp, options.GatesPublicKeys...)
if err != nil {
return fmt.Errorf("failed to put bearer token: %w", err)
@ -295,7 +295,7 @@ func (a *Agent) IssueSecret(ctx context.Context, w io.Writer, options *IssueSecr
// ObtainSecret receives an existing secret access key from NeoFS and
// writes to io.Writer the secret access key.
func (a *Agent) ObtainSecret(ctx context.Context, w io.Writer, options *ObtainSecretOptions) error {
bearerCreds := tokens.New(a.neoFS, options.GatePrivateKey, cache.DefaultAccessBoxConfig())
bearerCreds := tokens.New(a.neoFS, options.GatePrivateKey, cache.DefaultAccessBoxConfig(a.log))
var addr oid.Address
if err := addr.DecodeString(options.SecretAddress); err != nil {

View file

@ -297,7 +297,7 @@ func getNotificationsOptions(v *viper.Viper, l *zap.Logger) *notifications.Optio
}
func getCacheOptions(v *viper.Viper, l *zap.Logger) *layer.CachesConfig {
cacheCfg := layer.DefaultCachesConfigs()
cacheCfg := layer.DefaultCachesConfigs(l)
cacheCfg.Objects.Lifetime = getLifetime(v, l, cfgObjectsCacheLifetime, cacheCfg.Objects.Lifetime)
cacheCfg.Objects.Size = getSize(v, l, cfgObjectsCacheSize, cacheCfg.Objects.Size)
@ -348,7 +348,7 @@ func getSize(v *viper.Viper, l *zap.Logger, cfgEntry string, defaultValue int) i
}
func getAccessBoxCacheConfig(v *viper.Viper, l *zap.Logger) *cache.Config {
cacheCfg := cache.DefaultAccessBoxConfig()
cacheCfg := cache.DefaultAccessBoxConfig(l)
cacheCfg.Lifetime = getLifetime(v, l, cfgAccessBoxCacheLifetime, cacheCfg.Lifetime)
cacheCfg.Size = getSize(v, l, cfgAccessBoxCacheSize, cacheCfg.Size)