From 43f1d5964900e2662e70a76b305f3d0c5b9c359f Mon Sep 17 00:00:00 2001 From: Silvan Date: Tue, 9 Jan 2024 19:36:46 +0100 Subject: [PATCH] fix(auth): efficient user session projection (#7187) * fix(auth): cache users during session projection * fix(auth.user_sessions): add index for more efficient by user search --- cmd/setup/20.go | 26 +++++ cmd/setup/20.sql | 1 + cmd/setup/config.go | 1 + cmd/setup/setup.go | 3 + .../eventsourcing/handler/user_session.go | 21 ++++- .../handler/user_session_test.go | 94 +++++++++++++++++++ 6 files changed, 142 insertions(+), 4 deletions(-) create mode 100644 cmd/setup/20.go create mode 100644 cmd/setup/20.sql create mode 100644 internal/auth/repository/eventsourcing/handler/user_session_test.go diff --git a/cmd/setup/20.go b/cmd/setup/20.go new file mode 100644 index 000000000..f012140db --- /dev/null +++ b/cmd/setup/20.go @@ -0,0 +1,26 @@ +package setup + +import ( + "context" + _ "embed" + + "github.com/zitadel/zitadel/internal/database" +) + +var ( + //go:embed 20.sql + addByUserIndexToSession string +) + +type AddByUserIndexToSession struct { + dbClient *database.DB +} + +func (mig *AddByUserIndexToSession) Execute(ctx context.Context) error { + _, err := mig.dbClient.ExecContext(ctx, addByUserIndexToSession) + return err +} + +func (mig *AddByUserIndexToSession) String() string { + return "20_add_by_user_index_on_session" +} diff --git a/cmd/setup/20.sql b/cmd/setup/20.sql new file mode 100644 index 000000000..0eeae2cb0 --- /dev/null +++ b/cmd/setup/20.sql @@ -0,0 +1 @@ +CREATE INDEX CONCURRENTLY IF NOT EXISTS user_sessions_by_user ON auth.user_sessions (instance_id, user_id); \ No newline at end of file diff --git a/cmd/setup/config.go b/cmd/setup/config.go index 84e7903dc..098c12425 100644 --- a/cmd/setup/config.go +++ b/cmd/setup/config.go @@ -77,6 +77,7 @@ type Steps struct { s17AddOffsetToUniqueConstraints *AddOffsetToCurrentStates s18AddLowerFieldsToLoginNames *AddLowerFieldsToLoginNames s19AddCurrentStatesIndex *AddCurrentSequencesIndex + s20AddByUserSessionIndex *AddByUserIndexToSession } type encryptionKeyConfig struct { diff --git a/cmd/setup/setup.go b/cmd/setup/setup.go index 4fc3f6448..768a7ec70 100644 --- a/cmd/setup/setup.go +++ b/cmd/setup/setup.go @@ -110,6 +110,7 @@ func Setup(config *Config, steps *Steps, masterKey string) { steps.s17AddOffsetToUniqueConstraints = &AddOffsetToCurrentStates{dbClient: queryDBClient} steps.s18AddLowerFieldsToLoginNames = &AddLowerFieldsToLoginNames{dbClient: queryDBClient} steps.s19AddCurrentStatesIndex = &AddCurrentSequencesIndex{dbClient: queryDBClient} + steps.s20AddByUserSessionIndex = &AddByUserIndexToSession{dbClient: queryDBClient} err = projection.Create(ctx, projectionDBClient, eventstoreClient, config.Projections, nil, nil, nil) logging.OnError(err).Fatal("unable to start projections") @@ -156,6 +157,8 @@ func Setup(config *Config, steps *Steps, masterKey string) { logging.WithFields("name", steps.s17AddOffsetToUniqueConstraints.String()).OnError(err).Fatal("migration failed") err = migration.Migrate(ctx, eventstoreClient, steps.s19AddCurrentStatesIndex) logging.WithFields("name", steps.s19AddCurrentStatesIndex.String()).OnError(err).Fatal("migration failed") + err = migration.Migrate(ctx, eventstoreClient, steps.s20AddByUserSessionIndex) + logging.WithFields("name", steps.s20AddByUserSessionIndex.String()).OnError(err).Fatal("migration failed") for _, repeatableStep := range repeatableSteps { err = migration.Migrate(ctx, eventstoreClient, repeatableStep) diff --git a/internal/auth/repository/eventsourcing/handler/user_session.go b/internal/auth/repository/eventsourcing/handler/user_session.go index 65d1facf5..dfc2774c8 100644 --- a/internal/auth/repository/eventsourcing/handler/user_session.go +++ b/internal/auth/repository/eventsourcing/handler/user_session.go @@ -296,11 +296,24 @@ func (u *UserSession) Reduce(event eventstore.Event) (_ *handler.Statement, err } func (u *UserSession) appendEventOnSessions(sessions []*view_model.UserSessionView, event eventstore.Event) error { + users := make(map[string]*view_model.UserView) + usersByID := func(userID, instanceID string) (user *view_model.UserView, err error) { + user, ok := users[userID+"-"+instanceID] + if ok { + return user, nil + } + users[userID+"-"+instanceID], err = u.view.UserByID(userID, instanceID) + if err != nil { + return nil, err + } + + return users[userID+"-"+instanceID], nil + } for _, session := range sessions { if err := session.AppendEvent(event); err != nil { return err } - if err := u.fillUserInfo(session); err != nil { + if err := u.fillUserInfo(session, usersByID); err != nil { return err } } @@ -311,7 +324,7 @@ func (u *UserSession) updateSession(session *view_model.UserSessionView, event e if err := session.AppendEvent(event); err != nil { return err } - if err := u.fillUserInfo(session); err != nil { + if err := u.fillUserInfo(session, u.view.UserByID); err != nil { return err } if err := u.view.PutUserSession(session); err != nil { @@ -320,8 +333,8 @@ func (u *UserSession) updateSession(session *view_model.UserSessionView, event e return nil } -func (u *UserSession) fillUserInfo(session *view_model.UserSessionView) error { - user, err := u.view.UserByID(session.UserID, session.InstanceID) +func (u *UserSession) fillUserInfo(session *view_model.UserSessionView, getUserByID func(userID, instanceID string) (*view_model.UserView, error)) error { + user, err := getUserByID(session.UserID, session.InstanceID) if err != nil { return err } diff --git a/internal/auth/repository/eventsourcing/handler/user_session_test.go b/internal/auth/repository/eventsourcing/handler/user_session_test.go new file mode 100644 index 000000000..cafb3b471 --- /dev/null +++ b/internal/auth/repository/eventsourcing/handler/user_session_test.go @@ -0,0 +1,94 @@ +package handler + +import ( + "testing" + + view_model "github.com/zitadel/zitadel/internal/user/repository/view/model" +) + +// tests the proper working of the cache function +func TestUserSession_fillUserInfo(t *testing.T) { + type args struct { + sessions []*view_model.UserSessionView + } + tests := []struct { + name string + args args + cacheHits map[string]int + }{ + { + name: "one session", + args: args{ + sessions: []*view_model.UserSessionView{ + { + UserID: "user", + InstanceID: "instance", + }, + }, + }, + cacheHits: map[string]int{ + "user-instance": 1, + }, + }, + { + name: "same user", + args: args{ + sessions: []*view_model.UserSessionView{ + { + UserID: "user", + InstanceID: "instance", + }, + { + UserID: "user", + InstanceID: "instance", + }, + }, + }, + cacheHits: map[string]int{ + "user-instance": 2, + }, + }, + { + name: "different users", + args: args{ + sessions: []*view_model.UserSessionView{ + { + UserID: "user", + InstanceID: "instance", + }, + { + UserID: "user2", + InstanceID: "instance", + }, + }, + }, + cacheHits: map[string]int{ + "user-instance": 1, + "user2-instance": 1, + }, + }, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + cache := map[string]int{} + getUserByID := func(userID, instanceID string) (*view_model.UserView, error) { + cache[userID+"-"+instanceID]++ + return &view_model.UserView{HumanView: &view_model.HumanView{}}, nil + } + for _, session := range tt.args.sessions { + if err := new(UserSession).fillUserInfo(session, getUserByID); err != nil { + t.Errorf("UserSession.fillUserInfo() unexpected error = %v", err) + } + } + if len(cache) != len(tt.cacheHits) { + t.Errorf("unexpected length of cache hits: want %d, got %d", len(tt.cacheHits), len(cache)) + return + } + for key, count := range tt.cacheHits { + if cache[key] != count { + t.Errorf("unexpected cache hits on %s: want %d, got %d", key, count, cache[key]) + } + } + }) + } +}