cache v2 interface - store serialized cache (#14118)

Automatic Merge

services/cache2/cache.go

@@ -0,0 +1,46 @@
+// Copyright (c) 2015-present Mattermost, Inc. All Rights Reserved.
+// See LICENSE.txt for license information.
+
+package cache2
+
+import (
+	"errors"
+	"time"
+)
+
+// ErrKeyNotFound is the error when the given key is not found
+var ErrKeyNotFound = errors.New("key not found")
+
+// Cache (under package cache2) is a representation of a cache store that aims to replace cache.Cache
+type Cache interface {
+	// Purge is used to completely clear the cache.
+	Purge() error
+
+	// Set adds the given key and value to the store without an expiry. If the key already exists,
+	// it will overwrite the previous value.
+	Set(key string, value interface{}) error
+
+	// SetWithDefaultExpiry adds the given key and value to the store with the default expiry. If
+	// the key already exists, it will overwrite the previoous value
+	SetWithDefaultExpiry(key string, value interface{}) error
+
+	// SetWithExpiry adds the given key and value to the cache with the given expiry. If the key
+	// already exists, it will overwrite the previoous value
+	SetWithExpiry(key string, value interface{}, ttl time.Duration) 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
+	Get(key string, value interface{}) error
+
+	// Remove deletes the value for a given key.
+	Remove(key string) error
+
+	// Keys returns a slice of the keys in the cache.
+	Keys() ([]string, error)
+
+	// Len returns the number of items in the cache.
+	Len() (int, error)
+
+	// GetInvalidateClusterEvent returns the cluster event configured when this cache was created.
+	GetInvalidateClusterEvent() string
+}

services/cache2/lru.go

@@ -0,0 +1,189 @@
+// Copyright (c) 2015-present Mattermost, Inc. All Rights Reserved.
+// See LICENSE.txt for license information.
+
+package cache2
+
+import (
+	"bytes"
+	"container/list"
+	"encoding/gob"
+	"sync"
+	"time"
+)
+
+// LRU is a thread-safe fixed size LRU cache.
+type LRU struct {
+	size                   int
+	evictList              *list.List
+	items                  map[string]*list.Element
+	lock                   sync.RWMutex
+	defaultExpiry          time.Duration
+	invalidateClusterEvent string
+	currentGeneration      int64
+	len                    int
+}
+
+// LRUOptions contains options for initializing LRU cache
+type LRUOptions struct {
+	Size                   int
+	DefaultExpiry          time.Duration
+	InvalidateClusterEvent string
+}
+
+// entry is used to hold a value in the evictList.
+type entry struct {
+	key        string
+	value      []byte
+	expires    time.Time
+	generation int64
+}
+
+// NewLRU creates an LRU of the given size.
+func NewLRU(opts *LRUOptions) Cache {
+	return &LRU{
+		size:                   opts.Size,
+		evictList:              list.New(),
+		items:                  make(map[string]*list.Element, opts.Size),
+		defaultExpiry:          opts.DefaultExpiry,
+		invalidateClusterEvent: opts.InvalidateClusterEvent,
+	}
+}
+
+// Purge is used to completely clear the cache.
+func (l *LRU) Purge() error {
+	l.lock.Lock()
+	defer l.lock.Unlock()
+
+	l.len = 0
+	l.currentGeneration++
+	return nil
+}
+
+// Set adds the given key and value to the store without an expiry. If the key already exists,
+// it will overwrite the previous value.
+func (l *LRU) Set(key string, value interface{}) error {
+	return l.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 previoous value
+func (l *LRU) SetWithDefaultExpiry(key string, value interface{}) error {
+	return l.SetWithExpiry(key, value, l.defaultExpiry)
+}
+
+// SetWithExpiry adds the given key and value to the cache with the given expiry. If the key
+// already exists, it will overwrite the previoous value
+func (l *LRU) SetWithExpiry(key string, value interface{}, ttl time.Duration) error {
+	l.lock.Lock()
+	defer l.lock.Unlock()
+	return l.set(key, value, ttl)
+}
+
+// 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 (l *LRU) Get(key string, value interface{}) error {
+	l.lock.Lock()
+	defer l.lock.Unlock()
+	return l.get(key, value)
+}
+
+// Remove deletes the value for a key.
+func (l *LRU) Remove(key string) error {
+	l.lock.Lock()
+	defer l.lock.Unlock()
+
+	if ent, ok := l.items[key]; ok {
+		l.removeElement(ent)
+	}
+	return nil
+}
+
+// Keys returns a slice of the keys in the cache.
+func (l *LRU) Keys() ([]string, error) {
+	l.lock.RLock()
+	defer l.lock.RUnlock()
+
+	keys := make([]string, l.len)
+	i := 0
+	for ent := l.evictList.Back(); ent != nil; ent = ent.Prev() {
+		e := ent.Value.(*entry)
+		if e.generation == l.currentGeneration {
+			keys[i] = e.key
+			i++
+		}
+	}
+	return keys, nil
+}
+
+// Len returns the number of items in the cache.
+func (l *LRU) Len() (int, error) {
+	l.lock.RLock()
+	defer l.lock.RUnlock()
+	return l.len, nil
+}
+
+// GetInvalidateClusterEvent returns the cluster event configured when this cache was created.
+func (l *LRU) GetInvalidateClusterEvent() string {
+	return l.invalidateClusterEvent
+}
+
+func (l *LRU) set(key string, value interface{}, ttl time.Duration) error {
+	var expires time.Time
+	if ttl > 0 {
+		expires = time.Now().Add(ttl)
+	}
+
+	var buffer bytes.Buffer
+	err := gob.NewEncoder(&buffer).Encode(value)
+	if err != nil {
+		return err
+	}
+
+	// Check for existing item, ignoring expiry since we'd update anyway.
+	if ent, ok := l.items[key]; ok {
+		l.evictList.MoveToFront(ent)
+		e := ent.Value.(*entry)
+		e.value = buffer.Bytes()
+		e.expires = expires
+		if e.generation != l.currentGeneration {
+			e.generation = l.currentGeneration
+			l.len++
+		}
+		return nil
+	}
+
+	// Add new item
+	ent := &entry{key, buffer.Bytes(), expires, l.currentGeneration}
+	entry := l.evictList.PushFront(ent)
+	l.items[key] = entry
+	l.len++
+
+	if l.evictList.Len() > l.size {
+		l.removeElement(l.evictList.Back())
+	}
+	return nil
+}
+
+func (l *LRU) get(key string, value interface{}) error {
+	if ent, ok := l.items[key]; ok {
+		e := ent.Value.(*entry)
+
+		if e.generation != l.currentGeneration || (!e.expires.IsZero() && time.Now().After(e.expires)) {
+			l.removeElement(ent)
+			return ErrKeyNotFound
+		}
+
+		l.evictList.MoveToFront(ent)
+		return gob.NewDecoder(bytes.NewBuffer(e.value)).Decode(value)
+	}
+	return ErrKeyNotFound
+}
+
+func (l *LRU) removeElement(e *list.Element) {
+	l.evictList.Remove(e)
+	kv := e.Value.(*entry)
+	if kv.generation == l.currentGeneration {
+		l.len--
+	}
+	delete(l.items, kv.key)
+}

services/cache2/lru_test.go

@@ -0,0 +1,504 @@
+// Copyright (c) 2015-present Mattermost, Inc. All Rights Reserved.
+// See LICENSE.txt for license information.
+
+package cache2
+
+import (
+	"fmt"
+	"testing"
+	"time"
+
+	"github.com/mattermost/mattermost-server/v5/model"
+	"github.com/mattermost/mattermost-server/v5/services/cache/lru"
+	"github.com/stretchr/testify/require"
+)
+
+func TestLRU(t *testing.T) {
+	l := NewLRU(&LRUOptions{
+		Size:                   128,
+		DefaultExpiry:          0,
+		InvalidateClusterEvent: "",
+	})
+
+	for i := 0; i < 256; i++ {
+		err := l.Set(fmt.Sprintf("%d", i), i)
+		require.Nil(t, err)
+	}
+	size, err := l.Len()
+	require.Nil(t, err)
+	require.Equalf(t, size, 128, "bad len: %v", size)
+
+	keys, err := l.Keys()
+	require.Nil(t, err)
+	for i, k := range keys {
+		var v int
+		err = l.Get(k, &v)
+		require.Nil(t, err, "bad key: %v", k)
+		require.Equalf(t, fmt.Sprintf("%d", v), k, "bad key: %v", k)
+		require.Equalf(t, i+128, v, "bad value: %v", k)
+	}
+	for i := 0; i < 128; i++ {
+		var v int
+		err = l.Get(fmt.Sprintf("%d", i), &v)
+		require.Equal(t, ErrKeyNotFound, err, "should be evicted %v: %v", i, err)
+	}
+	for i := 128; i < 256; i++ {
+		var v int
+		err = l.Get(fmt.Sprintf("%d", i), &v)
+		require.Nil(t, err, "should not be evicted %v: %v", i, err)
+	}
+	for i := 128; i < 192; i++ {
+		l.Remove(fmt.Sprintf("%d", i))
+		var v int
+		err = l.Get(fmt.Sprintf("%d", i), &v)
+		require.Equal(t, ErrKeyNotFound, err, "should be deleted %v: %v", i, err)
+	}
+
+	var v int
+	err = l.Get("192", &v) // expect 192 to be last key in l.Keys()
+	require.Nil(t, err, "should exist")
+	require.Equalf(t, 192, v, "bad value: %v", v)
+
+	keys, err = l.Keys()
+	require.Nil(t, err)
+	for i, k := range keys {
+		require.Falsef(t, i < 63 && k != fmt.Sprintf("%d", i+193), "out of order key: %v", k)
+		require.Falsef(t, i == 63 && k != "192", "out of order key: %v", k)
+	}
+
+	l.Purge()
+	size, err = l.Len()
+	require.Nil(t, err)
+	require.Equalf(t, size, 0, "bad len: %v", size)
+	err = l.Get("200", &v)
+	require.Equal(t, err, ErrKeyNotFound, "should contain nothing")
+
+	err = l.Set("201", 301)
+	require.Nil(t, err)
+	err = l.Get("201", &v)
+	require.Nil(t, err)
+	require.Equal(t, 301, v)
+
+}
+
+func TestLRUExpire(t *testing.T) {
+	l := NewLRU(&LRUOptions{
+		Size:                   128,
+		DefaultExpiry:          1 * time.Second,
+		InvalidateClusterEvent: "",
+	})
+
+	l.SetWithDefaultExpiry("1", 1)
+	l.SetWithExpiry("3", 3, 0*time.Second)
+
+	time.Sleep(time.Second * 2)
+
+	var r1 int
+	err := l.Get("1", &r1)
+	require.Equal(t, err, ErrKeyNotFound, "should not exist")
+
+	var r2 int
+	err2 := l.Get("3", &r2)
+	require.Nil(t, err2, "should exist")
+	require.Equal(t, 3, r2)
+}
+
+func TestLRUMarshalUnMarshal(t *testing.T) {
+	l := NewLRU(&LRUOptions{
+		Size:                   1,
+		DefaultExpiry:          0,
+		InvalidateClusterEvent: "",
+	})
+
+	value1 := map[string]interface{}{
+		"key1": 1,
+		"key2": "value2",
+	}
+	err := l.Set("test", value1)
+
+	require.Nil(t, err)
+
+	var value2 map[string]interface{}
+	err = l.Get("test", &value2)
+	require.Nil(t, err)
+
+	v1, ok := value2["key1"].(int)
+	require.True(t, ok, "unable to cast value")
+	require.Equal(t, 1, v1)
+
+	v2, ok := value2["key2"].(string)
+	require.True(t, ok, "unable to cast value")
+	require.Equal(t, "value2", v2)
+
+	post := model.Post{
+		Id:            "id",
+		CreateAt:      11111,
+		UpdateAt:      11111,
+		DeleteAt:      11111,
+		EditAt:        111111,
+		IsPinned:      true,
+		UserId:        "UserId",
+		ChannelId:     "ChannelId",
+		RootId:        "RootId",
+		ParentId:      "ParentId",
+		OriginalId:    "OriginalId",
+		Message:       "OriginalId",
+		MessageSource: "MessageSource",
+		Type:          "Type",
+		Props: map[string]interface{}{
+			"key": "val",
+		},
+		Hashtags:      "Hashtags",
+		Filenames:     []string{"item1", "item2"},
+		FileIds:       []string{"item1", "item2"},
+		PendingPostId: "PendingPostId",
+		HasReactions:  true,
+		ReplyCount:    11111,
+		Metadata: &model.PostMetadata{
+			Embeds: []*model.PostEmbed{
+				{
+					Type: "Type",
+					URL:  "URL",
+					Data: "some data",
+				},
+				{
+					Type: "Type 2",
+					URL:  "URL 2",
+					Data: "some data 2",
+				},
+			},
+			Emojis: []*model.Emoji{
+				{
+					Id:   "id",
+					Name: "name",
+				},
+			},
+			Files: nil,
+			Images: map[string]*model.PostImage{
+				"key": {
+					Width:      1,
+					Height:     1,
+					Format:     "format",
+					FrameCount: 1,
+				},
+				"key2": {
+					Width:      999,
+					Height:     888,
+					Format:     "format 2",
+					FrameCount: 1000,
+				},
+			},
+			Reactions: []*model.Reaction{
+				{
+					UserId:    "user_id",
+					PostId:    "post_id",
+					EmojiName: "emoji_name",
+					CreateAt:  111,
+				},
+			},
+		},
+	}
+	err = l.Set("post", post.Clone())
+	require.Nil(t, err)
+
+	var p model.Post
+	err = l.Get("post", &p)
+	require.Nil(t, err)
+	require.Equal(t, post.Clone(), p.Clone())
+}
+
+func BenchmarkLRU(b *testing.B) {
+
+	value1 := "simplestring"
+	b.Run("simple=old", func(b *testing.B) {
+		for i := 0; i < b.N; i++ {
+			l := lru.New(1)
+			l.Add("test", value1)
+			_, ok := l.Get("test")
+			require.True(b, ok)
+		}
+	})
+
+	b.Run("simple=new", func(b *testing.B) {
+		for i := 0; i < b.N; i++ {
+			l2 := NewLRU(&LRUOptions{
+				Size:                   1,
+				DefaultExpiry:          0,
+				InvalidateClusterEvent: "",
+			})
+			err := l2.Set("test", value1)
+			require.Nil(b, err)
+
+			var val string
+			err = l2.Get("test", &val)
+			require.Nil(b, err)
+		}
+	})
+
+	type obj struct {
+		Field1 int
+		Field2 string
+		Field3 struct {
+			Field4 int
+			Field5 string
+		}
+		Field6 map[string]string
+	}
+
+	value2 := obj{
+		1,
+		"field2",
+		struct {
+			Field4 int
+			Field5 string
+		}{
+			6,
+			"field5 is a looooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooong string",
+		},
+		map[string]string{
+			"key0": "value0",
+			"key1": "value value1",
+			"key2": "value value value2",
+			"key3": "value value value value3",
+			"key4": "value value value value value4",
+			"key5": "value value value value value value5",
+			"key6": "value value value value value value value6",
+			"key7": "value value value value value value value value7",
+			"key8": "value value value value value value value value value8",
+			"key9": "value value value value value value value value value value9",
+		},
+	}
+	b.Run("complex=old", func(b *testing.B) {
+		for i := 0; i < b.N; i++ {
+			l := lru.New(1)
+			l.Add("test", value2)
+			_, ok := l.Get("test")
+			require.True(b, ok)
+		}
+	})
+	b.Run("complex=new", func(b *testing.B) {
+		for i := 0; i < b.N; i++ {
+			l2 := NewLRU(&LRUOptions{
+				Size:                   1,
+				DefaultExpiry:          0,
+				InvalidateClusterEvent: "",
+			})
+			err := l2.Set("test", value2)
+			require.Nil(b, err)
+
+			var val obj
+			err = l2.Get("test", &val)
+			require.Nil(b, err)
+		}
+	})
+
+	user := &model.User{
+		Id:             "id",
+		CreateAt:       11111,
+		UpdateAt:       11111,
+		DeleteAt:       11111,
+		Username:       "username",
+		Password:       "password",
+		AuthService:    "AuthService",
+		AuthData:       nil,
+		Email:          "Email",
+		EmailVerified:  true,
+		Nickname:       "Nickname",
+		FirstName:      "FirstName",
+		LastName:       "LastName",
+		Position:       "Position",
+		Roles:          "Roles",
+		AllowMarketing: true,
+		Props: map[string]string{
+			"key0": "value0",
+			"key1": "value value1",
+			"key2": "value value value2",
+			"key3": "value value value value3",
+			"key4": "value value value value value4",
+			"key5": "value value value value value value5",
+			"key6": "value value value value value value value6",
+			"key7": "value value value value value value value value7",
+			"key8": "value value value value value value value value value8",
+			"key9": "value value value value value value value value value value9",
+		},
+		NotifyProps: map[string]string{
+			"key0": "value0",
+			"key1": "value value1",
+			"key2": "value value value2",
+			"key3": "value value value value3",
+			"key4": "value value value value value4",
+			"key5": "value value value value value value5",
+			"key6": "value value value value value value value6",
+			"key7": "value value value value value value value value7",
+			"key8": "value value value value value value value value value8",
+			"key9": "value value value value value value value value value value9",
+		},
+		LastPasswordUpdate: 111111,
+		LastPictureUpdate:  111111,
+		FailedAttempts:     111111,
+		Locale:             "Locale",
+		Timezone: map[string]string{
+			"key0": "value0",
+			"key1": "value value1",
+			"key2": "value value value2",
+			"key3": "value value value value3",
+			"key4": "value value value value value4",
+			"key5": "value value value value value value5",
+			"key6": "value value value value value value value6",
+			"key7": "value value value value value value value value7",
+			"key8": "value value value value value value value value value8",
+			"key9": "value value value value value value value value value value9",
+		},
+		MfaActive:              true,
+		MfaSecret:              "MfaSecret",
+		LastActivityAt:         111111,
+		IsBot:                  true,
+		BotDescription:         "field5 is a looooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooong string",
+		BotLastIconUpdate:      111111,
+		TermsOfServiceId:       "TermsOfServiceId",
+		TermsOfServiceCreateAt: 111111,
+	}
+
+	b.Run("User=old", func(b *testing.B) {
+		for i := 0; i < b.N; i++ {
+			l := lru.New(1)
+			l.Add("test", user)
+			_, ok := l.Get("test")
+			require.True(b, ok)
+		}
+	})
+	b.Run("User=new", func(b *testing.B) {
+		for i := 0; i < b.N; i++ {
+			l2 := NewLRU(&LRUOptions{
+				Size:                   1,
+				DefaultExpiry:          0,
+				InvalidateClusterEvent: "",
+			})
+			err := l2.Set("test", user)
+			require.Nil(b, err)
+
+			var val model.User
+			err = l2.Get("test", &val)
+			require.Nil(b, err)
+		}
+	})
+
+	post := &model.Post{
+		Id:            "id",
+		CreateAt:      11111,
+		UpdateAt:      11111,
+		DeleteAt:      11111,
+		EditAt:        111111,
+		IsPinned:      true,
+		UserId:        "UserId",
+		ChannelId:     "ChannelId",
+		RootId:        "RootId",
+		ParentId:      "ParentId",
+		OriginalId:    "OriginalId",
+		Message:       "OriginalId",
+		MessageSource: "MessageSource",
+		Type:          "Type",
+		Props: map[string]interface{}{
+			"key": "val",
+		},
+		Hashtags:      "Hashtags",
+		Filenames:     []string{"item1", "item2"},
+		FileIds:       []string{"item1", "item2"},
+		PendingPostId: "PendingPostId",
+		HasReactions:  true,
+
+		// Transient data populated before sending a post to the client
+		ReplyCount: 11111,
+		Metadata: &model.PostMetadata{
+			Embeds: []*model.PostEmbed{
+				{
+					Type: "Type",
+					URL:  "URL",
+					Data: "some data",
+				},
+				{
+					Type: "Type 2",
+					URL:  "URL 2",
+					Data: "some data 2",
+				},
+			},
+			Emojis: []*model.Emoji{
+				{
+					Id:   "id",
+					Name: "name",
+				},
+			},
+			Files: nil,
+			Images: map[string]*model.PostImage{
+				"key": {
+					Width:      1,
+					Height:     1,
+					Format:     "format",
+					FrameCount: 1,
+				},
+				"key2": {
+					Width:      999,
+					Height:     888,
+					Format:     "format 2",
+					FrameCount: 1000,
+				},
+			},
+			Reactions: []*model.Reaction{},
+		},
+	}
+
+	b.Run("Post=old", func(b *testing.B) {
+		for i := 0; i < b.N; i++ {
+			l := lru.New(1)
+			l.Add("test", post)
+			_, ok := l.Get("test")
+			require.True(b, ok)
+		}
+	})
+	b.Run("Post=new", func(b *testing.B) {
+		for i := 0; i < b.N; i++ {
+			l2 := NewLRU(&LRUOptions{
+				Size:                   1,
+				DefaultExpiry:          0,
+				InvalidateClusterEvent: "",
+			})
+			err := l2.Set("test", post)
+			require.Nil(b, err)
+
+			var val model.Post
+			err = l2.Get("test", &val)
+			require.Nil(b, err)
+		}
+	})
+
+	status := model.Status{
+		UserId:         "UserId",
+		Status:         "Status",
+		Manual:         true,
+		LastActivityAt: 111111,
+		ActiveChannel:  "ActiveChannel",
+	}
+	b.Run("Status=old", func(b *testing.B) {
+		for i := 0; i < b.N; i++ {
+			l := lru.New(1)
+			l.Add("test", status)
+			_, ok := l.Get("test")
+			require.True(b, ok)
+		}
+	})
+	b.Run("Status=new", func(b *testing.B) {
+		for i := 0; i < b.N; i++ {
+			l2 := NewLRU(&LRUOptions{
+				Size:                   1,
+				DefaultExpiry:          0,
+				InvalidateClusterEvent: "",
+			})
+			err := l2.Set("test", status)
+			require.Nil(b, err)
+
+			var val model.Status
+			err = l2.Get("test", &val)
+			require.Nil(b, err)
+		}
+	})
+}