mirror of
https://github.com/minio/minio.git
synced 2025-11-07 12:52:58 -05:00
load credential for in-flights requests as singleflight (#19920)
avoid concurrent callers for LoadUser() to even initiate object read() requests, if an on-going operation is in progress. this avoids many callers hitting the drives causing I/O spikes, also allows for loading credentials faster.
This commit is contained in:
Harshavardhana
GitHub
121
cmd/iam-store.go
121
cmd/iam-store.go
@@ -33,8 +33,10 @@ import (
|
||||
"github.com/minio/minio/internal/auth"
|
||||
"github.com/minio/minio/internal/config/identity/openid"
|
||||
"github.com/minio/minio/internal/jwt"
|
||||
"github.com/minio/pkg/v3/console"
|
||||
"github.com/minio/pkg/v3/policy"
|
||||
"github.com/puzpuzpuz/xsync/v3"
|
||||
"golang.org/x/sync/singleflight"
|
||||
)
|
||||
|
||||
const (
|
||||
@@ -633,6 +635,8 @@ func (store *IAMStoreSys) LoadIAMCache(ctx context.Context, firstTime bool) erro
|
||||
// layer.
|
||||
type IAMStoreSys struct {
|
||||
IAMStorageAPI
|
||||
|
||||
group *singleflight.Group
|
||||
}
|
||||
|
||||
// HasWatcher - returns if the storage system has a watcher.
|
||||
@@ -1800,6 +1804,7 @@ func (store *IAMStoreSys) DeleteUser(ctx context.Context, accessKey string, user
|
||||
err = nil
|
||||
}
|
||||
delete(cache.iamUsersMap, accessKey)
|
||||
store.group.Forget(accessKey)
|
||||
|
||||
cache.updatedAt = time.Now()
|
||||
|
||||
@@ -1875,6 +1880,7 @@ func (store *IAMStoreSys) DeleteUsers(ctx context.Context, users []string) error
|
||||
err := store.deleteUserIdentity(ctx, user, userType)
|
||||
iamLogIf(GlobalContext, err)
|
||||
delete(cache.iamUsersMap, user)
|
||||
store.group.Forget(user)
|
||||
|
||||
deleted = true
|
||||
}
|
||||
@@ -2563,67 +2569,78 @@ func (store *IAMStoreSys) UpdateUserIdentity(ctx context.Context, cred auth.Cred
|
||||
|
||||
// LoadUser - attempts to load user info from storage and updates cache.
|
||||
func (store *IAMStoreSys) LoadUser(ctx context.Context, accessKey string) {
|
||||
cache := store.lock()
|
||||
defer store.unlock()
|
||||
// We use singleflight to de-duplicate requests when server
|
||||
// is coming up and loading accessKey and its associated assets
|
||||
val, err, shared := store.group.Do(accessKey, func() (interface{}, error) {
|
||||
cache := store.lock()
|
||||
defer func() {
|
||||
cache.updatedAt = time.Now()
|
||||
store.unlock()
|
||||
}()
|
||||
|
||||
cache.updatedAt = time.Now()
|
||||
_, found := cache.iamUsersMap[accessKey]
|
||||
|
||||
_, found := cache.iamUsersMap[accessKey]
|
||||
|
||||
// Check for regular user access key
|
||||
if !found {
|
||||
store.loadUser(ctx, accessKey, regUser, cache.iamUsersMap)
|
||||
if _, found = cache.iamUsersMap[accessKey]; found {
|
||||
// load mapped policies
|
||||
store.loadMappedPolicyWithRetry(ctx, accessKey, regUser, false, cache.iamUserPolicyMap, 3)
|
||||
}
|
||||
}
|
||||
|
||||
// Check for service account
|
||||
if !found {
|
||||
store.loadUser(ctx, accessKey, svcUser, cache.iamUsersMap)
|
||||
var svc UserIdentity
|
||||
svc, found = cache.iamUsersMap[accessKey]
|
||||
if found {
|
||||
// Load parent user and mapped policies.
|
||||
if store.getUsersSysType() == MinIOUsersSysType {
|
||||
store.loadUser(ctx, svc.Credentials.ParentUser, regUser, cache.iamUsersMap)
|
||||
store.loadMappedPolicyWithRetry(ctx, svc.Credentials.ParentUser, regUser, false, cache.iamUserPolicyMap, 3)
|
||||
} else {
|
||||
// In case of LDAP the parent user's policy mapping needs to be
|
||||
// loaded into sts map
|
||||
store.loadMappedPolicyWithRetry(ctx, svc.Credentials.ParentUser, stsUser, false, cache.iamSTSPolicyMap, 3)
|
||||
// Check for regular user access key
|
||||
if !found {
|
||||
store.loadUser(ctx, accessKey, regUser, cache.iamUsersMap)
|
||||
if _, found = cache.iamUsersMap[accessKey]; found {
|
||||
// load mapped policies
|
||||
store.loadMappedPolicyWithRetry(ctx, accessKey, regUser, false, cache.iamUserPolicyMap, 3)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Check for STS account
|
||||
stsAccountFound := false
|
||||
var stsUserCred UserIdentity
|
||||
if !found {
|
||||
store.loadUser(ctx, accessKey, stsUser, cache.iamSTSAccountsMap)
|
||||
if stsUserCred, found = cache.iamSTSAccountsMap[accessKey]; found {
|
||||
// Load mapped policy
|
||||
store.loadMappedPolicyWithRetry(ctx, stsUserCred.Credentials.ParentUser, stsUser, false, cache.iamSTSPolicyMap, 3)
|
||||
stsAccountFound = true
|
||||
}
|
||||
}
|
||||
|
||||
// Load any associated policy definitions
|
||||
if !stsAccountFound {
|
||||
pols, _ := cache.iamUserPolicyMap.Load(accessKey)
|
||||
for _, policy := range pols.toSlice() {
|
||||
if _, found = cache.iamPolicyDocsMap[policy]; !found {
|
||||
store.loadPolicyDocWithRetry(ctx, policy, cache.iamPolicyDocsMap, 3)
|
||||
// Check for service account
|
||||
if !found {
|
||||
store.loadUser(ctx, accessKey, svcUser, cache.iamUsersMap)
|
||||
var svc UserIdentity
|
||||
svc, found = cache.iamUsersMap[accessKey]
|
||||
if found {
|
||||
// Load parent user and mapped policies.
|
||||
if store.getUsersSysType() == MinIOUsersSysType {
|
||||
store.loadUser(ctx, svc.Credentials.ParentUser, regUser, cache.iamUsersMap)
|
||||
store.loadMappedPolicyWithRetry(ctx, svc.Credentials.ParentUser, regUser, false, cache.iamUserPolicyMap, 3)
|
||||
} else {
|
||||
// In case of LDAP the parent user's policy mapping needs to be
|
||||
// loaded into sts map
|
||||
store.loadMappedPolicyWithRetry(ctx, svc.Credentials.ParentUser, stsUser, false, cache.iamSTSPolicyMap, 3)
|
||||
}
|
||||
}
|
||||
}
|
||||
} else {
|
||||
pols, _ := cache.iamSTSPolicyMap.Load(stsUserCred.Credentials.AccessKey)
|
||||
for _, policy := range pols.toSlice() {
|
||||
if _, found = cache.iamPolicyDocsMap[policy]; !found {
|
||||
store.loadPolicyDocWithRetry(ctx, policy, cache.iamPolicyDocsMap, 3)
|
||||
|
||||
// Check for STS account
|
||||
stsAccountFound := false
|
||||
var stsUserCred UserIdentity
|
||||
if !found {
|
||||
store.loadUser(ctx, accessKey, stsUser, cache.iamSTSAccountsMap)
|
||||
if stsUserCred, found = cache.iamSTSAccountsMap[accessKey]; found {
|
||||
// Load mapped policy
|
||||
store.loadMappedPolicyWithRetry(ctx, stsUserCred.Credentials.ParentUser, stsUser, false, cache.iamSTSPolicyMap, 3)
|
||||
stsAccountFound = true
|
||||
}
|
||||
}
|
||||
|
||||
// Load any associated policy definitions
|
||||
if !stsAccountFound {
|
||||
pols, _ := cache.iamUserPolicyMap.Load(accessKey)
|
||||
for _, policy := range pols.toSlice() {
|
||||
if _, found = cache.iamPolicyDocsMap[policy]; !found {
|
||||
store.loadPolicyDocWithRetry(ctx, policy, cache.iamPolicyDocsMap, 3)
|
||||
}
|
||||
}
|
||||
} else {
|
||||
pols, _ := cache.iamSTSPolicyMap.Load(stsUserCred.Credentials.AccessKey)
|
||||
for _, policy := range pols.toSlice() {
|
||||
if _, found = cache.iamPolicyDocsMap[policy]; !found {
|
||||
store.loadPolicyDocWithRetry(ctx, policy, cache.iamPolicyDocsMap, 3)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
return "done", nil
|
||||
})
|
||||
|
||||
if serverDebugLog {
|
||||
console.Debugln("loadUser: loading shared", val, err, shared)
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user