jan/openreplay
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Sessions cache layer (#789)

Browse source 
* feat(backend/db): cache layer for sessions with auto deleting logic
This commit is contained in:
Alexander 2022-11-02 17:00:41 +01:00 • committed by GitHub
parent eaa2dfc720
commit 2abf063ba2
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
9 changed files with 136 additions and 50 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

3
backend/cmd/db/main.go
View file

@ -30,7 +30,8 @@ func main() {
cfg := db.New()
// Init database
pg := cache.NewPGCache(postgres.NewConn(cfg.Postgres, cfg.BatchQueueLimit, cfg.BatchSizeLimit, metrics), cfg.ProjectExpirationTimeoutMs)
pg := cache.NewPGCache(
postgres.NewConn(cfg.Postgres, cfg.BatchQueueLimit, cfg.BatchSizeLimit, metrics), cfg.ProjectExpirationTimeoutMs)
defer pg.Close()
// HandlersFabric returns the list of message handlers we want to be applied to each incoming message.

73
backend/pkg/db/cache/cache.go vendored Normal file
View file

@ -0,0 +1,73 @@
package cache
import (
"log"
"openreplay/backend/pkg/db/postgres"
"openreplay/backend/pkg/db/types"
"sync"
"time"
)
type SessionMeta struct {
*types.Session
lastUse time.Time
}
type ProjectMeta struct {
*types.Project
expirationTime time.Time
}
type Cache interface {
SetSession(sess *types.Session)
HasSession(sessID uint64) bool
GetSession(sessionID uint64) (*types.Session, error)
GetProject(projectID uint32) (*types.Project, error)
GetProjectByKey(projectKey string) (*types.Project, error)
}
type cacheImpl struct {
conn *postgres.Conn
mutex sync.RWMutex
sessions map[uint64]*SessionMeta
projects map[uint32]*ProjectMeta
projectsByKeys sync.Map
projectExpirationTimeout time.Duration
}
func NewCache(conn *postgres.Conn, projectExpirationTimeoutMs int64) Cache {
newCache := &cacheImpl{
conn: conn,
sessions: make(map[uint64]*SessionMeta),
projects: make(map[uint32]*ProjectMeta),
projectExpirationTimeout: time.Duration(1000 * projectExpirationTimeoutMs),
}
go newCache.cleaner()
return newCache
}
func (c *cacheImpl) cleaner() {
cleanTick := time.Tick(time.Minute * 5)
for {
select {
case <-cleanTick:
c.clearCache()
}
}
}
func (c *cacheImpl) clearCache() {
c.mutex.Lock()
defer c.mutex.Unlock()
now := time.Now()
cacheSize := len(c.sessions)
deleted := 0
for id, sess := range c.sessions {
if now.Sub(sess.lastUse).Minutes() > 3 {
deleted++
delete(c.sessions, id)
}
}
log.Printf("cache cleaner: deleted %d/%d sessions", deleted, cacheSize)
}

1
backend/pkg/db/cache/messages-common.go vendored
View file

@ -19,7 +19,6 @@ func (c *PGCache) HandleSessionEnd(sessionID uint64) error {
if err := c.Conn.HandleSessionEnd(sessionID); err != nil {
log.Printf("can't handle session end: %s", err)
}
c.DeleteSession(sessionID)
return nil
}

14
backend/pkg/db/cache/messages-ios.go vendored
View file

@ -1,16 +1,16 @@
package cache
import (
"errors"
"fmt"
. "openreplay/backend/pkg/db/types"
. "openreplay/backend/pkg/messages"
)
func (c *PGCache) InsertIOSSessionStart(sessionID uint64, s *IOSSessionStart) error {
if c.sessions[sessionID] != nil {
return errors.New("This session already in cache!")
if c.cache.HasSession(sessionID) {
return fmt.Errorf("session %d already in cache", sessionID)
}
c.sessions[sessionID] = &Session{
newSess := &Session{
SessionID: sessionID,
Platform: "ios",
Timestamp: s.Timestamp,
@ -24,8 +24,10 @@ func (c *PGCache) InsertIOSSessionStart(sessionID uint64, s *IOSSessionStart) er
UserCountry: s.UserCountry,
UserDeviceType: s.UserDeviceType,
}
if err := c.Conn.InsertSessionStart(sessionID, c.sessions[sessionID]); err != nil {
c.sessions[sessionID] = nil
c.cache.SetSession(newSess)
if err := c.Conn.InsertSessionStart(sessionID, newSess); err != nil {
// don't know why?
c.cache.SetSession(nil)
return err
}
return nil

14
backend/pkg/db/cache/messages-web.go vendored
View file

@ -1,7 +1,7 @@
package cache
import (
"errors"
"fmt"
. "openreplay/backend/pkg/db/types"
. "openreplay/backend/pkg/messages"
)
@ -31,10 +31,10 @@ func (c *PGCache) InsertWebSessionStart(sessionID uint64, s *SessionStart) error
}
func (c *PGCache) HandleWebSessionStart(sessionID uint64, s *SessionStart) error {
if c.sessions[sessionID] != nil {
return errors.New("This session already in cache!")
if c.cache.HasSession(sessionID) {
return fmt.Errorf("session %d already in cache", sessionID)
}
c.sessions[sessionID] = &Session{
newSess := &Session{
SessionID: sessionID,
Platform: "web",
Timestamp: s.Timestamp,
@ -55,8 +55,10 @@ func (c *PGCache) HandleWebSessionStart(sessionID uint64, s *SessionStart) error
UserDeviceHeapSize: s.UserDeviceHeapSize,
UserID: &s.UserID,
}
if err := c.Conn.HandleSessionStart(sessionID, c.sessions[sessionID]); err != nil {
c.sessions[sessionID] = nil
c.cache.SetSession(newSess)
if err := c.Conn.HandleSessionStart(sessionID, newSess); err != nil {
// don't know why?
c.cache.SetSession(nil)
return err
}
return nil

25
backend/pkg/db/cache/pg-cache.go vendored
View file

@ -1,31 +1,20 @@
package cache
import (
"sync"
"time"
"openreplay/backend/pkg/db/postgres"
. "openreplay/backend/pkg/db/types"
)
type ProjectMeta struct {
*Project
expirationTime time.Time
}
type PGCache struct {
*postgres.Conn
sessions map[uint64]*Session
projects map[uint32]*ProjectMeta
projectsByKeys sync.Map // map[string]*ProjectMeta
projectExpirationTimeout time.Duration
cache Cache
}
func NewPGCache(pgConn *postgres.Conn, projectExpirationTimeoutMs int64) *PGCache {
func NewPGCache(conn *postgres.Conn, projectExpirationTimeoutMs int64) *PGCache {
// Create in-memory cache layer for sessions and projects
c := NewCache(conn, projectExpirationTimeoutMs)
// Return PG wrapper with integrated cache layer
return &PGCache{
Conn: pgConn,
sessions: make(map[uint64]*Session),
projects: make(map[uint32]*ProjectMeta),
projectExpirationTimeout: time.Duration(1000 * projectExpirationTimeoutMs),
Conn: conn,
cache: c,
}
}

8
backend/pkg/db/cache/project.go vendored
View file

@ -5,7 +5,7 @@ import (
"time"
)
func (c *PGCache) GetProjectByKey(projectKey string) (*Project, error) {
func (c *cacheImpl) GetProjectByKey(projectKey string) (*Project, error) {
pmInterface, found := c.projectsByKeys.Load(projectKey)
if found {
if pm, ok := pmInterface.(*ProjectMeta); ok {
@ -15,7 +15,7 @@ func (c *PGCache) GetProjectByKey(projectKey string) (*Project, error) {
}
}
p, err := c.Conn.GetProjectByKey(projectKey)
p, err := c.conn.GetProjectByKey(projectKey)
if err != nil {
return nil, err
}
@ -24,12 +24,12 @@ func (c *PGCache) GetProjectByKey(projectKey string) (*Project, error) {
return p, nil
}
func (c *PGCache) GetProject(projectID uint32) (*Project, error) {
func (c *cacheImpl) GetProject(projectID uint32) (*Project, error) {
if c.projects[projectID] != nil &&
time.Now().Before(c.projects[projectID].expirationTime) {
return c.projects[projectID].Project, nil
}
p, err := c.Conn.GetProject(projectID)
p, err := c.conn.GetProject(projectID)
if err != nil {
return nil, err
}

46
backend/pkg/db/cache/session.go vendored
View file

@ -4,28 +4,48 @@ import (
"errors"
"github.com/jackc/pgx/v4"
. "openreplay/backend/pkg/db/types"
"time"
)
var NilSessionInCacheError = errors.New("nil session in error")
func (c *PGCache) GetSession(sessionID uint64) (*Session, error) {
if s, inCache := c.sessions[sessionID]; inCache {
if s == nil {
return s, NilSessionInCacheError
}
return s, nil
func (c *cacheImpl) SetSession(sess *Session) {
c.mutex.Lock()
defer c.mutex.Unlock()
if meta, ok := c.sessions[sess.SessionID]; ok {
meta.Session = sess
meta.lastUse = time.Now()
} else {
c.sessions[sess.SessionID] = &SessionMeta{sess, time.Now()}
}
s, err := c.Conn.GetSession(sessionID)
}
func (c *cacheImpl) HasSession(sessID uint64) bool {
c.mutex.RLock()
defer c.mutex.RUnlock()
sess, ok := c.sessions[sessID]
return ok && sess.Session != nil
}
func (c *cacheImpl) GetSession(sessionID uint64) (*Session, error) {
c.mutex.Lock()
defer c.mutex.Unlock()
if s, inCache := c.sessions[sessionID]; inCache {
if s.Session == nil {
return nil, NilSessionInCacheError
}
return s.Session, nil
}
s, err := c.conn.GetSession(sessionID)
if err == pgx.ErrNoRows {
c.sessions[sessionID] = nil
c.sessions[sessionID] = &SessionMeta{nil, time.Now()}
}
if err != nil {
return nil, err
}
c.sessions[sessionID] = s
c.sessions[sessionID] = &SessionMeta{s, time.Now()}
return s, nil
}
func (c *PGCache) DeleteSession(sessionID uint64) {
delete(c.sessions, sessionID)
}

2
backend/pkg/db/types/session.go
View file

@ -20,7 +20,7 @@ type Session struct {
IssueTypes []string
IssueScore int
UserID *string // pointer??
UserID *string
UserAnonymousID *string
Metadata1 *string
Metadata2 *string