Add cache to store UID to UserID mapping in EOS (#1340)

cs3org · Dec 2, 2020 · 0ff299d · 0ff299d
1 parent 0aad381
commit 0ff299d
Show file tree

Hide file tree

Showing 5 changed files with 27 additions and 6 deletions.
diff --git a/changelog/unreleased/eos-user-cache.md b/changelog/unreleased/eos-user-cache.md
@@ -0,0 +1,7 @@
+Enhancement: Add cache to store UID to UserID mapping in EOS
+
+Previously, we used to send an RPC to the user provider service for every lookup
+of user IDs from the UID stored in EOS. This PR adds an in-memory lock-protected
+cache to store this mapping.
+
+https://github.com/cs3org/reva/pull/1340
diff --git a/internal/grpc/services/gateway/storageprovider.go b/internal/grpc/services/gateway/storageprovider.go
@@ -2016,7 +2016,7 @@ func (s *svc) split(ctx context.Context, p string, i int) bool {
 
 	// validate that we have always at least two elements
 	if len(parts) < 2 {
-		panic(fmt.Sprintf("split: len(parts) < 2: path:%s parts:%+v", p, parts))
+		return false
 	}
 
 	// validate the share folder is always the second element, first element is always the hardcoded value of "home"

diff --git a/pkg/publicshare/manager/json/json.go b/pkg/publicshare/manager/json/json.go
@@ -363,16 +363,19 @@ func (m *manager) ListPublicShares(ctx context.Context, u *user.User, filters []
 // RevokePublicShare undocumented.
 func (m *manager) RevokePublicShare(ctx context.Context, u *user.User, ref *link.PublicShareReference) error {
 	m.mutex.Lock()
-	defer m.mutex.Unlock()
-
 	db, err := m.readDb()
 	if err != nil {
 		return err
 	}
+	m.mutex.Unlock()
 
 	switch {
-	case ref.GetId().OpaqueId != "":
-		delete(db, ref.GetId().OpaqueId)
+	case ref.GetId() != nil && ref.GetId().OpaqueId != "":
+		if _, ok := db[ref.GetId().OpaqueId]; ok {
+			delete(db, ref.GetId().OpaqueId)
+		} else {
+			return errors.New("reference does not exist")
+		}
 	case ref.GetToken() != "":
 		share, err := m.getByToken(ctx, ref.GetToken())
 		if err != nil {
@@ -383,6 +386,8 @@ func (m *manager) RevokePublicShare(ctx context.Context, u *user.User, ref *link
 		return errors.New("reference does not exist")
 	}
 
+	m.mutex.Lock()
+	defer m.mutex.Unlock()
 	return m.writeDb(db)
 }
 

diff --git a/pkg/publicshare/manager/memory/memory.go b/pkg/publicshare/manager/memory/memory.go
@@ -194,7 +194,7 @@ func (m *manager) ListPublicShares(ctx context.Context, u *user.User, filters []
 func (m *manager) RevokePublicShare(ctx context.Context, u *user.User, ref *link.PublicShareReference) error {
 	// check whether the reference exists
 	switch {
-	case ref.GetId().OpaqueId != "":
+	case ref.GetId() != nil && ref.GetId().OpaqueId != "":
 		s, err := m.getPublicShareByTokenID(ctx, *ref.GetId())
 		if err != nil {
 			return errors.New("reference does not exist")

diff --git a/pkg/storage/utils/eosfs/eosfs.go b/pkg/storage/utils/eosfs/eosfs.go
@@ -29,6 +29,7 @@ import (
 	"regexp"
 	"strconv"
 	"strings"
+	"sync"
 
 	userpb "github.com/cs3org/go-cs3apis/cs3/identity/user/v1beta1"
 	rpc "github.com/cs3org/go-cs3apis/cs3/rpc/v1beta1"
@@ -179,6 +180,7 @@ type eosfs struct {
 	chunkHandler  *chunking.ChunkHandler
 	singleUserUID string
 	singleUserGID string
+	userIDCache   sync.Map
 }
 
 // NewEOSFS returns a storage.FS interface implementation that connects to an
@@ -213,6 +215,7 @@ func NewEOSFS(c *Config) (storage.FS, error) {
 		c:            eosClient,
 		conf:         c,
 		chunkHandler: chunking.NewChunkHandler(c.CacheDirectory),
+		userIDCache:  sync.Map{},
 	}
 
 	return eosfs, nil
@@ -1455,6 +1458,10 @@ func (fs *eosfs) getUIDGateway(ctx context.Context, u *userpb.UserId) (string, s
 }
 
 func (fs *eosfs) getUserIDGateway(ctx context.Context, uid string) (*userpb.UserId, error) {
+	if userIDInterface, ok := fs.userIDCache.Load(uid); ok {
+		return userIDInterface.(*userpb.UserId), nil
+	}
+
 	client, err := pool.GetGatewayServiceClient(fs.conf.GatewaySvc)
 	if err != nil {
 		return nil, errors.Wrap(err, "eos: error getting gateway grpc client")
@@ -1469,6 +1476,8 @@ func (fs *eosfs) getUserIDGateway(ctx context.Context, uid string) (*userpb.User
 	if getUserResp.Status.Code != rpc.Code_CODE_OK {
 		return nil, errors.Wrap(err, "eos: grpc get user failed")
 	}
+
+	fs.userIDCache.Store(uid, getUserResp.User.Id)
 	return getUserResp.User.Id, nil
 }