mirrors/mostlymatter
1
0
Форкнуть 0
Код Задачи Запросы на слияние Действия Пакеты Проекты Релизы Вики Активность

MM-56876: Redis: first introduction (#27752)

Просмотр исходного кода 
```release-note
NONE
```

---------

Co-authored-by: Jesús Espino <jespinog@gmail.com>
Co-authored-by: Mattermost Build <build@mattermost.com>
Этот коммит содержится в:
Agniva De Sarker
2024-08-06 09:28:41 +05:30
коммит произвёл GitHub
родитель c3ed07e679
Коммит 540febd866
45 изменённых файлов: 1113 добавлений и 278 удалений
Скачать Patch файл Скачать Diff файл Показать все файлы Свернуть все файлы

2
server/platform/services/cache/cache.go поставляемый
Просмотреть файл

@@ -34,6 +34,8 @@ type Cache interface {
// Return ErrKeyNotFound if the key is missing from the cache
Get(key string, value any) error
GetMulti(keys []string, values []any) []error
// Remove deletes the value for a given key.
Remove(key string) error

22
server/platform/services/cache/lru.go поставляемый
Просмотреть файл

@@ -27,17 +27,6 @@ type LRU struct {
invalidateClusterEvent model.ClusterEvent
}
// LRUOptions contains options for initializing LRU cache
type LRUOptions struct {
Name string
Size int
DefaultExpiry time.Duration
InvalidateClusterEvent model.ClusterEvent
// StripedBuckets is used only by LRUStriped and shouldn't be greater than the number
// of CPUs available on the machine running this cache.
StripedBuckets int
}
// entry is used to hold a value in the evictList.
type entry struct {
key string
@@ -47,7 +36,7 @@ type entry struct {
}
// NewLRU creates an LRU of the given size.
func NewLRU(opts LRUOptions) Cache {
func NewLRU(opts *CacheOptions) Cache {
return &LRU{
name: opts.Name,
size: opts.Size,
@@ -92,6 +81,15 @@ func (l *LRU) Get(key string, value any) error {
return l.get(key, value)
}
func (l *LRU) GetMulti(keys []string, values []any) []error {
errs := make([]error, 0, len(values))
for i, key := range keys {
errs = append(errs, l.get(key, values[i]))
}
return errs
}
// Remove deletes the value for a key.
func (l *LRU) Remove(key string) error {
l.lock.Lock()

14
server/platform/services/cache/lru_striped.go поставляемый
Просмотреть файл

@@ -82,6 +82,14 @@ func (L LRUStriped) Get(key string, value any) error {
return L.keyBucket(key).Get(key, value)
}
func (L LRUStriped) GetMulti(keys []string, values []any) []error {
errs := make([]error, 0, len(values))
for i, key := range keys {
errs = append(errs, L.keyBucket(key).Get(key, values[i]))
}
return errs
}
// Remove does the same as LRU.Remove
func (L LRUStriped) Remove(key string) error {
return L.keyBucket(key).Remove(key)
@@ -119,12 +127,12 @@ func (L LRUStriped) Name() string {
return L.name
}
// NewLRUStriped creates a striped LRU cache using the special LRUOptions.StripedBuckets value.
// See LRUStriped and LRUOptions for more details.
// NewLRUStriped creates a striped LRU cache using the special CacheOptions.StripedBuckets value.
// See LRUStriped and CacheOptions for more details.
//
// Not that in order to prevent false eviction, this LRU cache adds 10% (computation is rounded up) of the
// requested size to the total cache size.
func NewLRUStriped(opts LRUOptions) (Cache, error) {
func NewLRUStriped(opts *CacheOptions) (Cache, error) {
if opts.StripedBuckets == 0 {
return nil, fmt.Errorf("number of buckets is mandatory")
}

4
server/platform/services/cache/lru_striped_bench_test.go поставляемый
Просмотреть файл

@@ -19,7 +19,7 @@ const (
)
func BenchmarkLRUStriped(b *testing.B) {
opts := cache.LRUOptions{
opts := cache.CacheOptions{
Name: "",
Size: 128,
DefaultExpiry: 0,
@@ -27,7 +27,7 @@ func BenchmarkLRUStriped(b *testing.B) {
StripedBuckets: runtime.NumCPU() - 1,
}
cache, err := cache.NewLRUStriped(opts)
cache, err := cache.NewLRUStriped(&opts)
if err != nil {
panic(err)
}

10
server/platform/services/cache/lru_striped_test.go поставляемый
Просмотреть файл

@@ -27,7 +27,7 @@ func makeLRUPredictableTestData(num int) [][2]string {
}
func TestNewLRUStriped(t *testing.T) {
scache, err := NewLRUStriped(LRUOptions{StripedBuckets: 3, Size: 20})
scache, err := NewLRUStriped(&CacheOptions{StripedBuckets: 3, Size: 20})
require.NoError(t, err)
cache := scache.(LRUStriped)
@@ -41,7 +41,7 @@ func TestNewLRUStriped(t *testing.T) {
func TestLRUStripedKeyDistribution(t *testing.T) {
dataset := makeLRUPredictableTestData(100)
scache, err := NewLRUStriped(LRUOptions{StripedBuckets: 4, Size: len(dataset)})
scache, err := NewLRUStriped(&CacheOptions{StripedBuckets: 4, Size: len(dataset)})
require.NoError(t, err)
cache := scache.(LRUStriped)
for _, kv := range dataset {
@@ -66,7 +66,7 @@ func TestLRUStripedKeyDistribution(t *testing.T) {
}
func TestLRUStriped_Size(t *testing.T) {
scache, err := NewLRUStriped(LRUOptions{StripedBuckets: 2, Size: 128})
scache, err := NewLRUStriped(&CacheOptions{StripedBuckets: 2, Size: 128})
require.NoError(t, err)
cache := scache.(LRUStriped)
acc := 0
@@ -77,7 +77,7 @@ func TestLRUStriped_Size(t *testing.T) {
}
func TestLRUStriped_HashKey(t *testing.T) {
scache, err := NewLRUStriped(LRUOptions{StripedBuckets: 2, Size: 128})
scache, err := NewLRUStriped(&CacheOptions{StripedBuckets: 2, Size: 128})
require.NoError(t, err)
cache := scache.(LRUStriped)
first := cache.hashkeyMapHash("key")
@@ -87,7 +87,7 @@ func TestLRUStriped_HashKey(t *testing.T) {
}
func TestLRUStriped_Get(t *testing.T) {
cache, err := NewLRUStriped(LRUOptions{StripedBuckets: 4, Size: 128})
cache, err := NewLRUStriped(&CacheOptions{StripedBuckets: 4, Size: 128})
require.NoError(t, err)
var out string
require.Equal(t, ErrKeyNotFound, cache.Get("key", &out))

22
server/platform/services/cache/lru_test.go поставляемый
Просмотреть файл

@@ -16,7 +16,7 @@ import (
)
func TestLRU(t *testing.T) {
l := NewLRU(LRUOptions{
l := NewLRU(&CacheOptions{
Size: 128,
DefaultExpiry: 0,
InvalidateClusterEvent: "",
@@ -84,7 +84,7 @@ func TestLRU(t *testing.T) {
}
func TestLRUExpire(t *testing.T) {
l := NewLRU(LRUOptions{
l := NewLRU(&CacheOptions{
Size: 128,
DefaultExpiry: 1 * time.Second,
InvalidateClusterEvent: "",
@@ -106,7 +106,7 @@ func TestLRUExpire(t *testing.T) {
}
func TestLRUMarshalUnMarshal(t *testing.T) {
l := NewLRU(LRUOptions{
l := NewLRU(&CacheOptions{
Size: 1,
DefaultExpiry: 0,
InvalidateClusterEvent: "",
@@ -295,7 +295,7 @@ func BenchmarkLRU(b *testing.B) {
b.Run("simple=new", func(b *testing.B) {
for i := 0; i < b.N; i++ {
l2 := NewLRU(LRUOptions{
l2 := NewLRU(&CacheOptions{
Size: 1,
DefaultExpiry: 0,
InvalidateClusterEvent: "",
@@ -345,7 +345,7 @@ func BenchmarkLRU(b *testing.B) {
b.Run("complex=new", func(b *testing.B) {
for i := 0; i < b.N; i++ {
l2 := NewLRU(LRUOptions{
l2 := NewLRU(&CacheOptions{
Size: 1,
DefaultExpiry: 0,
InvalidateClusterEvent: "",
@@ -428,7 +428,7 @@ func BenchmarkLRU(b *testing.B) {
b.Run("User=new", func(b *testing.B) {
for i := 0; i < b.N; i++ {
l2 := NewLRU(LRUOptions{
l2 := NewLRU(&CacheOptions{
Size: 1,
DefaultExpiry: 0,
InvalidateClusterEvent: "",
@@ -461,7 +461,7 @@ func BenchmarkLRU(b *testing.B) {
b.Run("UserMap=new", func(b *testing.B) {
for i := 0; i < b.N; i++ {
l2 := NewLRU(LRUOptions{
l2 := NewLRU(&CacheOptions{
Size: 1,
DefaultExpiry: 0,
InvalidateClusterEvent: "",
@@ -540,7 +540,7 @@ func BenchmarkLRU(b *testing.B) {
b.Run("Post=new", func(b *testing.B) {
for i := 0; i < b.N; i++ {
l2 := NewLRU(LRUOptions{
l2 := NewLRU(&CacheOptions{
Size: 1,
DefaultExpiry: 0,
InvalidateClusterEvent: "",
@@ -564,7 +564,7 @@ func BenchmarkLRU(b *testing.B) {
b.Run("Status=new", func(b *testing.B) {
for i := 0; i < b.N; i++ {
l2 := NewLRU(LRUOptions{
l2 := NewLRU(&CacheOptions{
Size: 1,
DefaultExpiry: 0,
InvalidateClusterEvent: "",
@@ -600,7 +600,7 @@ func BenchmarkLRU(b *testing.B) {
b.Run("Session=new", func(b *testing.B) {
for i := 0; i < b.N; i++ {
l2 := NewLRU(LRUOptions{
l2 := NewLRU(&CacheOptions{
Size: 1,
DefaultExpiry: 0,
InvalidateClusterEvent: "",
@@ -616,7 +616,7 @@ func BenchmarkLRU(b *testing.B) {
}
func TestLRURace(t *testing.T) {
l2 := NewLRU(LRUOptions{
l2 := NewLRU(&CacheOptions{
Size: 1,
DefaultExpiry: 0,
InvalidateClusterEvent: "",

125
server/platform/services/cache/mocks/Cache.go поставляемый
Просмотреть файл

@@ -1,10 +1,13 @@
// Code generated by mockery v1.0.0. DO NOT EDIT.
// Code generated by mockery v2.42.2. DO NOT EDIT.
// Regenerate this file using `make cache-mocks`.
package mocks
import (
time "time"
model "github.com/mattermost/mattermost/server/public/model"
mock "github.com/stretchr/testify/mock"
)
@@ -14,11 +17,15 @@ type Cache struct {
}
// Get provides a mock function with given fields: key, value
func (_m *Cache) Get(key string, value any) error {
func (_m *Cache) Get(key string, value interface{}) error {
ret := _m.Called(key, value)
if len(ret) == 0 {
panic("no return value specified for Get")
}
var r0 error
if rf, ok := ret.Get(0).(func(string, any) error); ok {
if rf, ok := ret.Get(0).(func(string, interface{}) error); ok {
r0 = rf(key, value)
} else {
r0 = ret.Error(0)
@@ -28,14 +35,38 @@ func (_m *Cache) Get(key string, value any) error {
}
// GetInvalidateClusterEvent provides a mock function with given fields:
func (_m *Cache) GetInvalidateClusterEvent() string {
func (_m *Cache) GetInvalidateClusterEvent() model.ClusterEvent {
ret := _m.Called()
var r0 string
if rf, ok := ret.Get(0).(func() string); ok {
if len(ret) == 0 {
panic("no return value specified for GetInvalidateClusterEvent")
}
var r0 model.ClusterEvent
if rf, ok := ret.Get(0).(func() model.ClusterEvent); ok {
r0 = rf()
} else {
r0 = ret.Get(0).(string)
r0 = ret.Get(0).(model.ClusterEvent)
}
return r0
}
// GetMulti provides a mock function with given fields: keys, values
func (_m *Cache) GetMulti(keys []string, values []interface{}) []error {
ret := _m.Called(keys, values)
if len(ret) == 0 {
panic("no return value specified for GetMulti")
}
var r0 []error
if rf, ok := ret.Get(0).(func([]string, []interface{}) []error); ok {
r0 = rf(keys, values)
} else {
if ret.Get(0) != nil {
r0 = ret.Get(0).([]error)
}
}
return r0
@@ -45,7 +76,15 @@ func (_m *Cache) GetInvalidateClusterEvent() string {
func (_m *Cache) Keys() ([]string, error) {
ret := _m.Called()
if len(ret) == 0 {
panic("no return value specified for Keys")
}
var r0 []string
var r1 error
if rf, ok := ret.Get(0).(func() ([]string, error)); ok {
return rf()
}
if rf, ok := ret.Get(0).(func() []string); ok {
r0 = rf()
} else {
@@ -54,28 +93,6 @@ func (_m *Cache) Keys() ([]string, error) {
}
}
var r1 error
if rf, ok := ret.Get(1).(func() error); ok {
r1 = rf()
} else {
r1 = ret.Error(1)
}
return r0, r1
}
// Len provides a mock function with given fields:
func (_m *Cache) Len() (int, error) {
ret := _m.Called()
var r0 int
if rf, ok := ret.Get(0).(func() int); ok {
r0 = rf()
} else {
r0 = ret.Get(0).(int)
}
var r1 error
if rf, ok := ret.Get(1).(func() error); ok {
r1 = rf()
} else {
@@ -89,6 +106,10 @@ func (_m *Cache) Len() (int, error) {
func (_m *Cache) Name() string {
ret := _m.Called()
if len(ret) == 0 {
panic("no return value specified for Name")
}
var r0 string
if rf, ok := ret.Get(0).(func() string); ok {
r0 = rf()
@@ -103,6 +124,10 @@ func (_m *Cache) Name() string {
func (_m *Cache) Purge() error {
ret := _m.Called()
if len(ret) == 0 {
panic("no return value specified for Purge")
}
var r0 error
if rf, ok := ret.Get(0).(func() error); ok {
r0 = rf()
@@ -117,6 +142,10 @@ func (_m *Cache) Purge() error {
func (_m *Cache) Remove(key string) error {
ret := _m.Called(key)
if len(ret) == 0 {
panic("no return value specified for Remove")
}
var r0 error
if rf, ok := ret.Get(0).(func(string) error); ok {
r0 = rf(key)
@@ -128,11 +157,15 @@ func (_m *Cache) Remove(key string) error {
}
// Set provides a mock function with given fields: key, value
func (_m *Cache) Set(key string, value any) error {
func (_m *Cache) Set(key string, value interface{}) error {
ret := _m.Called(key, value)
if len(ret) == 0 {
panic("no return value specified for Set")
}
var r0 error
if rf, ok := ret.Get(0).(func(string, any) error); ok {
if rf, ok := ret.Get(0).(func(string, interface{}) error); ok {
r0 = rf(key, value)
} else {
r0 = ret.Error(0)
@@ -142,11 +175,15 @@ func (_m *Cache) Set(key string, value any) error {
}
// SetWithDefaultExpiry provides a mock function with given fields: key, value
func (_m *Cache) SetWithDefaultExpiry(key string, value any) error {
func (_m *Cache) SetWithDefaultExpiry(key string, value interface{}) error {
ret := _m.Called(key, value)
if len(ret) == 0 {
panic("no return value specified for SetWithDefaultExpiry")
}
var r0 error
if rf, ok := ret.Get(0).(func(string, any) error); ok {
if rf, ok := ret.Get(0).(func(string, interface{}) error); ok {
r0 = rf(key, value)
} else {
r0 = ret.Error(0)
@@ -156,11 +193,15 @@ func (_m *Cache) SetWithDefaultExpiry(key string, value any) error {
}
// SetWithExpiry provides a mock function with given fields: key, value, ttl
func (_m *Cache) SetWithExpiry(key string, value any, ttl time.Duration) error {
func (_m *Cache) SetWithExpiry(key string, value interface{}, ttl time.Duration) error {
ret := _m.Called(key, value, ttl)
if len(ret) == 0 {
panic("no return value specified for SetWithExpiry")
}
var r0 error
if rf, ok := ret.Get(0).(func(string, any, time.Duration) error); ok {
if rf, ok := ret.Get(0).(func(string, interface{}, time.Duration) error); ok {
r0 = rf(key, value, ttl)
} else {
r0 = ret.Error(0)
@@ -168,3 +209,17 @@ func (_m *Cache) SetWithExpiry(key string, value any, ttl time.Duration) error {
return r0
}
// NewCache creates a new instance of Cache. It also registers a testing interface on the mock and a cleanup function to assert the mocks expectations.
// The first argument is typically a *testing.T value.
func NewCache(t interface {
mock.TestingT
Cleanup(func())
}) *Cache {
mock := &Cache{}
mock.Mock.Test(t)
t.Cleanup(func() { mock.AssertExpectations(t) })
return mock
}

87
server/platform/services/cache/mocks/Provider.go поставляемый
Просмотреть файл

@@ -1,11 +1,14 @@
// Code generated by mockery v1.0.0. DO NOT EDIT.
// Code generated by mockery v2.42.2. DO NOT EDIT.
// Regenerate this file using `make cache-mocks`.
package mocks
import (
mock "github.com/stretchr/testify/mock"
einterfaces "github.com/mattermost/mattermost/server/v8/einterfaces"
cache "github.com/mattermost/mattermost/server/v8/platform/services/cache"
mock "github.com/stretchr/testify/mock"
)
// Provider is an autogenerated mock type for the Provider type
@@ -17,6 +20,10 @@ type Provider struct {
func (_m *Provider) Close() error {
ret := _m.Called()
if len(ret) == 0 {
panic("no return value specified for Close")
}
var r0 error
if rf, ok := ret.Get(0).(func() error); ok {
r0 = rf()
@@ -28,24 +35,46 @@ func (_m *Provider) Close() error {
}
// Connect provides a mock function with given fields:
func (_m *Provider) Connect() error {
func (_m *Provider) Connect() (string, error) {
ret := _m.Called()
var r0 error
if rf, ok := ret.Get(0).(func() error); ok {
r0 = rf()
} else {
r0 = ret.Error(0)
if len(ret) == 0 {
panic("no return value specified for Connect")
}
return r0
var r0 string
var r1 error
if rf, ok := ret.Get(0).(func() (string, error)); ok {
return rf()
}
if rf, ok := ret.Get(0).(func() string); ok {
r0 = rf()
} else {
r0 = ret.Get(0).(string)
}
if rf, ok := ret.Get(1).(func() error); ok {
r1 = rf()
} else {
r1 = ret.Error(1)
}
return r0, r1
}
// NewCache provides a mock function with given fields: opts
func (_m *Provider) NewCache(opts *cache.CacheOptions) (cache.Cache, error) {
ret := _m.Called(opts)
if len(ret) == 0 {
panic("no return value specified for NewCache")
}
var r0 cache.Cache
var r1 error
if rf, ok := ret.Get(0).(func(*cache.CacheOptions) (cache.Cache, error)); ok {
return rf(opts)
}
if rf, ok := ret.Get(0).(func(*cache.CacheOptions) cache.Cache); ok {
r0 = rf(opts)
} else {
@@ -54,7 +83,6 @@ func (_m *Provider) NewCache(opts *cache.CacheOptions) (cache.Cache, error) {
}
}
var r1 error
if rf, ok := ret.Get(1).(func(*cache.CacheOptions) error); ok {
r1 = rf(opts)
} else {
@@ -63,3 +91,40 @@ func (_m *Provider) NewCache(opts *cache.CacheOptions) (cache.Cache, error) {
return r0, r1
}
// SetMetrics provides a mock function with given fields: metrics
func (_m *Provider) SetMetrics(metrics einterfaces.MetricsInterface) {
_m.Called(metrics)
}
// Type provides a mock function with given fields:
func (_m *Provider) Type() string {
ret := _m.Called()
if len(ret) == 0 {
panic("no return value specified for Type")
}
var r0 string
if rf, ok := ret.Get(0).(func() string); ok {
r0 = rf()
} else {
r0 = ret.Get(0).(string)
}
return r0
}
// NewProvider creates a new instance of Provider. It also registers a testing interface on the mock and a cleanup function to assert the mocks expectations.
// The first argument is typically a *testing.T value.
func NewProvider(t interface {
mock.TestingT
Cleanup(func())
}) *Provider {
mock := &Provider{}
mock.Mock.Test(t)
t.Cleanup(func() { mock.AssertExpectations(t) })
return mock
}

98
server/platform/services/cache/provider.go поставляемый
Просмотреть файл

@@ -4,9 +4,13 @@
package cache
import (
"context"
"fmt"
"time"
"github.com/mattermost/mattermost/server/public/model"
"github.com/mattermost/mattermost/server/v8/einterfaces"
"github.com/redis/rueidis"
)
// CacheOptions contains options for initializing a cache
@@ -16,7 +20,9 @@ type CacheOptions struct {
Name string
InvalidateClusterEvent model.ClusterEvent
Striped bool
StripedBuckets int
// StripedBuckets is used only by LRUStriped and shouldn't be greater than the number
// of CPUs available on the machine running this cache.
StripedBuckets int
}
// Provider is a provider for Cache
@@ -24,9 +30,14 @@ type Provider interface {
// NewCache creates a new cache with given options.
NewCache(opts *CacheOptions) (Cache, error)
// Connect opens a new connection to the cache using specific provider parameters.
Connect() error
// The returned string contains the status of the response from the cache backend.
Connect() (string, error)
// SetMetrics
SetMetrics(metrics einterfaces.MetricsInterface)
// Close releases any resources used by the cache provider.
Close() error
// Type returns what type of cache it generates.
Type() string
}
type cacheProvider struct {
@@ -40,28 +51,81 @@ func NewProvider() Provider {
// NewCache creates a new cache with given opts
func (c *cacheProvider) NewCache(opts *CacheOptions) (Cache, error) {
if opts.Striped {
return NewLRUStriped(LRUOptions{
Name: opts.Name,
Size: opts.Size,
DefaultExpiry: opts.DefaultExpiry,
InvalidateClusterEvent: opts.InvalidateClusterEvent,
StripedBuckets: opts.StripedBuckets,
})
return NewLRUStriped(opts)
}
return NewLRU(LRUOptions{
Name: opts.Name,
Size: opts.Size,
DefaultExpiry: opts.DefaultExpiry,
InvalidateClusterEvent: opts.InvalidateClusterEvent,
}), nil
return NewLRU(opts), nil
}
// Connect opens a new connection to the cache using specific provider parameters.
func (c *cacheProvider) Connect() error {
return nil
func (c *cacheProvider) Connect() (string, error) {
return "OK", nil
}
func (c *cacheProvider) SetMetrics(metrics einterfaces.MetricsInterface) {
}
// Close releases any resources used by the cache provider.
func (c *cacheProvider) Close() error {
return nil
}
func (c *cacheProvider) Type() string {
return model.CacheTypeLRU
}
type redisProvider struct {
client rueidis.Client
metrics einterfaces.MetricsInterface
}
type RedisOptions struct {
RedisAddr string
RedisPassword string
RedisDB int
}
// NewProvider creates a new CacheProvider
func NewRedisProvider(opts *RedisOptions) (Provider, error) {
client, err := rueidis.NewClient(rueidis.ClientOption{
InitAddress: []string{opts.RedisAddr},
Password: opts.RedisPassword,
SelectDB: opts.RedisDB,
ForceSingleClient: true,
CacheSizeEachConn: 16 * (1 << 20), // 16MiB local cache size
//TODO: look into MaxFlushDelay
})
if err != nil {
return nil, err
}
return &redisProvider{client: client}, nil
}
// NewCache creates a new cache with given opts
func (r *redisProvider) NewCache(opts *CacheOptions) (Cache, error) {
rr, err := NewRedis(opts, r.client)
rr.metrics = r.metrics
return rr, err
}
// Connect opens a new connection to the cache using specific provider parameters.
func (r *redisProvider) Connect() (string, error) {
res, err := r.client.Do(context.Background(), r.client.B().Ping().Build()).ToString()
if err != nil {
return "", fmt.Errorf("unable to establish connection with redis: %v", err)
}
return res, nil
}
func (r *redisProvider) SetMetrics(metrics einterfaces.MetricsInterface) {
r.metrics = metrics
}
func (r *redisProvider) Type() string {
return model.CacheTypeRedis
}
// Close releases any resources used by the cache provider.
func (r *redisProvider) Close() error {
r.client.Close()
return nil
}

2
server/platform/services/cache/provider_test.go поставляемый
Просмотреть файл

@@ -161,7 +161,7 @@ func TestNewCache_Striped(t *testing.T) {
func TestConnectClose(t *testing.T) {
p := NewProvider()
err := p.Connect()
_, err := p.Connect()
require.NoError(t, err)
err = p.Close()

287
server/platform/services/cache/redis.go поставляемый Обычный файл
Просмотреть файл

@@ -0,0 +1,287 @@
// Copyright (c) 2015-present Mattermost, Inc. All Rights Reserved.
// See LICENSE.txt for license information.
package cache
import (
"context"
"errors"
"fmt"
"time"
"github.com/mattermost/mattermost/server/public/model"
"github.com/mattermost/mattermost/server/v8/einterfaces"
"github.com/redis/rueidis"
"github.com/tinylib/msgp/msgp"
"github.com/vmihailenco/msgpack/v5"
)
const clientSideTTL = 5 * time.Minute
type Redis struct {
name string
client rueidis.Client
defaultExpiry time.Duration
metrics einterfaces.MetricsInterface
}
func NewRedis(opts *CacheOptions, client rueidis.Client) (*Redis, error) {
if opts.Name == "" {
return nil, errors.New("no name specified for cache")
}
return &Redis{
name: opts.Name,
defaultExpiry: opts.DefaultExpiry,
client: client,
}, nil
}
func (r *Redis) Purge() error {
// TODO: move to scan
keys, err := r.Keys()
if err != nil {
return err
}
return r.client.Do(context.Background(),
r.client.B().Del().
Key(keys...).
Build(),
).Error()
}
func (r *Redis) Set(key string, value any) error {
return r.SetWithExpiry(key, value, 0)
}
// SetWithDefaultExpiry adds the given key and value to the store with the default expiry. If
// the key already exists, it will overwrite the previous value
func (r *Redis) SetWithDefaultExpiry(key string, value any) error {
return r.SetWithExpiry(key, value, r.defaultExpiry)
}
// SetWithExpiry adds the given key and value to the cache with the given expiry. If the key
// already exists, it will overwrite the previous value
func (r *Redis) SetWithExpiry(key string, value any, ttl time.Duration) error {
now := time.Now()
defer func() {
if r.metrics != nil {
elapsed := time.Since(now).Seconds()
r.metrics.ObserveRedisEndpointDuration(r.name, "Set", elapsed)
}
}()
var buf []byte
var err error
// We use a fast path for hot structs.
if msgpVal, ok := value.(msgp.Marshaler); ok {
buf, err = msgpVal.MarshalMsg(nil)
} else {
// Slow path for other structs.
buf, err = msgpack.Marshal(value)
}
if err != nil {
return err
}
return r.client.Do(context.Background(),
r.client.B().Set().
Key(r.name+":"+key).
Value(rueidis.BinaryString(buf)).
Ex(ttl).
Build(),
).Error()
}
// Get the content stored in the cache for the given key, and decode it into the value interface.
// Return ErrKeyNotFound if the key is missing from the cache
func (r *Redis) Get(key string, value any) error {
now := time.Now()
defer func() {
if r.metrics != nil {
elapsed := time.Since(now).Seconds()
r.metrics.ObserveRedisEndpointDuration(r.name, "Get", elapsed)
}
}()
val, err := r.client.DoCache(context.Background(),
r.client.B().Get().
Key(r.name+":"+key).
Cache(),
clientSideTTL,
).AsBytes()
if err != nil {
if rueidis.IsRedisNil(err) {
return ErrKeyNotFound
}
return err
}
// We use a fast path for hot structs.
if msgpVal, ok := value.(msgp.Unmarshaler); ok {
_, err := msgpVal.UnmarshalMsg(val)
return err
}
// This is ugly and makes the cache package aware of the model package.
// But this is due to 2 things.
// 1. The msgp package works on methods on structs rather than functions.
// 2. Our cache interface passes pointers to empty pointers, and not pointers
// to values. This is mainly how all our model structs are passed around.
// It might be technically possible to use values _just_ for hot structs
// like these and then return a pointer while returning from the cache function,
// but it will make the codebase inconsistent, and has some edge-cases to take care of.
switch v := value.(type) {
case **model.User:
var u model.User
_, err := u.UnmarshalMsg(val)
*v = &u
return err
case *map[string]*model.User:
var u model.UserMap
_, err := u.UnmarshalMsg(val)
*v = u
return err
}
// Slow path for other structs.
return msgpack.Unmarshal(val, value)
}
func (r *Redis) GetMulti(keys []string, values []any) []error {
now := time.Now()
defer func() {
if r.metrics != nil {
elapsed := time.Since(now).Seconds()
r.metrics.ObserveRedisEndpointDuration(r.name, "GetMulti", elapsed)
}
}()
errs := make([]error, len(keys))
newKeys := make([]string, len(keys))
for i := range keys {
newKeys[i] = r.name + ":" + keys[i]
}
vals, err := r.client.DoCache(context.Background(),
r.client.B().Mget().
Key(newKeys...).
Cache(),
clientSideTTL,
).ToArray()
if err != nil {
for i := range errs {
errs[i] = err
}
return errs
}
if len(vals) != len(keys) {
for i := range errs {
errs[i] = fmt.Errorf("length of returned vals %d, does not match length of keys %d", len(vals), len(keys))
}
return errs
}
for i, val := range vals {
if val.IsNil() {
errs[i] = ErrKeyNotFound
continue
}
buf, err := val.AsBytes()
if err != nil {
errs[i] = err
continue
}
// We use a fast path for hot structs.
if msgpVal, ok := values[i].(msgp.Unmarshaler); ok {
_, err := msgpVal.UnmarshalMsg(buf)
errs[i] = err
continue
}
switch v := values[i].(type) {
case **model.User:
var u model.User
_, err := u.UnmarshalMsg(buf)
*v = &u
errs[i] = err
continue
case *map[string]*model.User:
var u model.UserMap
_, err := u.UnmarshalMsg(buf)
*v = u
errs[i] = err
continue
}
// Slow path for other structs.
errs[i] = msgpack.Unmarshal(buf, values[i])
}
return errs
}
// Remove deletes the value for a given key.
func (r *Redis) Remove(key string) error {
now := time.Now()
defer func() {
if r.metrics != nil {
elapsed := time.Since(now).Seconds()
r.metrics.ObserveRedisEndpointDuration(r.name, "Del", elapsed)
}
}()
return r.client.Do(context.Background(),
r.client.B().Del().
Key(r.name+":"+key).
Build(),
).Error()
}
// Keys returns a slice of the keys in the cache.
func (r *Redis) Keys() ([]string, error) {
now := time.Now()
defer func() {
if r.metrics != nil {
elapsed := time.Since(now).Seconds()
r.metrics.ObserveRedisEndpointDuration(r.name, "Keys", elapsed)
}
}()
// TODO: migrate to a function that works on a batch of keys.
return r.client.Do(context.Background(),
r.client.B().Keys().
Pattern(r.name+":*").
Build(),
).AsStrSlice()
}
// Len returns the number of items in the cache.
func (r *Redis) Len() (int, error) {
now := time.Now()
defer func() {
if r.metrics != nil {
elapsed := time.Since(now).Seconds()
r.metrics.ObserveRedisEndpointDuration(r.name, "Len", elapsed)
}
}()
// TODO: migrate to scan
keys, err := r.client.Do(context.Background(),
r.client.B().Keys().
Pattern(r.name+":*").
Build(),
).AsStrSlice()
if err != nil {
return 0, err
}
return len(keys), nil
}
// GetInvalidateClusterEvent returns the cluster event configured when this cache was created.
func (r *Redis) GetInvalidateClusterEvent() model.ClusterEvent {
return model.ClusterEventNone
}
func (r *Redis) Name() string {
return r.name
}