mirror of
https://github.com/netbirdio/netbird.git
synced 2026-10-13 17:09:07 +02:00
[management] Remove store locks 3 (#4390)
This commit is contained in:
@@ -492,8 +492,6 @@ func (am *DefaultAccountManager) peerLoginExpirationJob(ctx context.Context, acc
|
|||||||
ctx := context.WithValue(ctx, nbcontext.AccountIDKey, accountID)
|
ctx := context.WithValue(ctx, nbcontext.AccountIDKey, accountID)
|
||||||
//nolint
|
//nolint
|
||||||
ctx = context.WithValue(ctx, hook.ExecutionContextKey, fmt.Sprintf("%s-PEER-EXPIRATION", hook.SystemSource))
|
ctx = context.WithValue(ctx, hook.ExecutionContextKey, fmt.Sprintf("%s-PEER-EXPIRATION", hook.SystemSource))
|
||||||
unlock := am.Store.AcquireWriteLockByUID(ctx, accountID)
|
|
||||||
defer unlock()
|
|
||||||
|
|
||||||
expiredPeers, err := am.getExpiredPeers(ctx, accountID)
|
expiredPeers, err := am.getExpiredPeers(ctx, accountID)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
@@ -529,9 +527,6 @@ func (am *DefaultAccountManager) schedulePeerLoginExpiration(ctx context.Context
|
|||||||
// peerInactivityExpirationJob marks login expired for all inactive peers and returns the minimum duration in which the next peer of the account will expire by inactivity if found
|
// peerInactivityExpirationJob marks login expired for all inactive peers and returns the minimum duration in which the next peer of the account will expire by inactivity if found
|
||||||
func (am *DefaultAccountManager) peerInactivityExpirationJob(ctx context.Context, accountID string) func() (time.Duration, bool) {
|
func (am *DefaultAccountManager) peerInactivityExpirationJob(ctx context.Context, accountID string) func() (time.Duration, bool) {
|
||||||
return func() (time.Duration, bool) {
|
return func() (time.Duration, bool) {
|
||||||
unlock := am.Store.AcquireWriteLockByUID(ctx, accountID)
|
|
||||||
defer unlock()
|
|
||||||
|
|
||||||
inactivePeers, err := am.getInactivePeers(ctx, accountID)
|
inactivePeers, err := am.getInactivePeers(ctx, accountID)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
log.WithContext(ctx).Errorf("failed getting inactive peers for account %s", accountID)
|
log.WithContext(ctx).Errorf("failed getting inactive peers for account %s", accountID)
|
||||||
@@ -672,8 +667,6 @@ func (am *DefaultAccountManager) isCacheCold(ctx context.Context, store cacheSto
|
|||||||
|
|
||||||
// DeleteAccount deletes an account and all its users from local store and from the remote IDP if the requester is an admin and account owner
|
// DeleteAccount deletes an account and all its users from local store and from the remote IDP if the requester is an admin and account owner
|
||||||
func (am *DefaultAccountManager) DeleteAccount(ctx context.Context, accountID, userID string) error {
|
func (am *DefaultAccountManager) DeleteAccount(ctx context.Context, accountID, userID string) error {
|
||||||
unlock := am.Store.AcquireWriteLockByUID(ctx, accountID)
|
|
||||||
defer unlock()
|
|
||||||
account, err := am.Store.GetAccount(ctx, accountID)
|
account, err := am.Store.GetAccount(ctx, accountID)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
@@ -1345,13 +1338,6 @@ func (am *DefaultAccountManager) SyncUserJWTGroups(ctx context.Context, userAuth
|
|||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
unlockAccount := am.Store.AcquireWriteLockByUID(ctx, userAuth.AccountId)
|
|
||||||
defer func() {
|
|
||||||
if unlockAccount != nil {
|
|
||||||
unlockAccount()
|
|
||||||
}
|
|
||||||
}()
|
|
||||||
|
|
||||||
var addNewGroups []string
|
var addNewGroups []string
|
||||||
var removeOldGroups []string
|
var removeOldGroups []string
|
||||||
var hasChanges bool
|
var hasChanges bool
|
||||||
@@ -1414,8 +1400,6 @@ func (am *DefaultAccountManager) SyncUserJWTGroups(ctx context.Context, userAuth
|
|||||||
return fmt.Errorf("error incrementing network serial: %w", err)
|
return fmt.Errorf("error incrementing network serial: %w", err)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
unlockAccount()
|
|
||||||
unlockAccount = nil
|
|
||||||
|
|
||||||
return nil
|
return nil
|
||||||
})
|
})
|
||||||
|
|||||||
@@ -332,9 +332,6 @@ func (am *DefaultAccountManager) UpdatePeer(ctx context.Context, accountID, user
|
|||||||
|
|
||||||
// DeletePeer removes peer from the account by its IP
|
// DeletePeer removes peer from the account by its IP
|
||||||
func (am *DefaultAccountManager) DeletePeer(ctx context.Context, accountID, peerID, userID string) error {
|
func (am *DefaultAccountManager) DeletePeer(ctx context.Context, accountID, peerID, userID string) error {
|
||||||
unlock := am.Store.AcquireWriteLockByUID(ctx, accountID)
|
|
||||||
defer unlock()
|
|
||||||
|
|
||||||
allowed, err := am.permissionsManager.ValidateUserPermissions(ctx, accountID, userID, modules.Peers, operations.Delete)
|
allowed, err := am.permissionsManager.ValidateUserPermissions(ctx, accountID, userID, modules.Peers, operations.Delete)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return status.NewPermissionValidationError(err)
|
return status.NewPermissionValidationError(err)
|
||||||
|
|||||||
@@ -51,7 +51,6 @@ const (
|
|||||||
// SqlStore represents an account storage backed by a Sql DB persisted to disk
|
// SqlStore represents an account storage backed by a Sql DB persisted to disk
|
||||||
type SqlStore struct {
|
type SqlStore struct {
|
||||||
db *gorm.DB
|
db *gorm.DB
|
||||||
resourceLocks sync.Map
|
|
||||||
globalAccountLock sync.Mutex
|
globalAccountLock sync.Mutex
|
||||||
metrics telemetry.AppMetrics
|
metrics telemetry.AppMetrics
|
||||||
installationPK int
|
installationPK int
|
||||||
@@ -144,25 +143,6 @@ func (s *SqlStore) AcquireGlobalLock(ctx context.Context) (unlock func()) {
|
|||||||
return unlock
|
return unlock
|
||||||
}
|
}
|
||||||
|
|
||||||
// AcquireWriteLockByUID acquires an ID lock for writing to a resource and returns a function that releases the lock
|
|
||||||
func (s *SqlStore) AcquireWriteLockByUID(ctx context.Context, uniqueID string) (unlock func()) {
|
|
||||||
log.WithContext(ctx).Tracef("acquiring write lock for ID %s", uniqueID)
|
|
||||||
|
|
||||||
startWait := time.Now()
|
|
||||||
value, _ := s.resourceLocks.LoadOrStore(uniqueID, &sync.RWMutex{})
|
|
||||||
mtx := value.(*sync.RWMutex)
|
|
||||||
mtx.Lock()
|
|
||||||
log.WithContext(ctx).Tracef("waiting to acquire write lock for ID %s in %v", uniqueID, time.Since(startWait))
|
|
||||||
startHold := time.Now()
|
|
||||||
|
|
||||||
unlock = func() {
|
|
||||||
mtx.Unlock()
|
|
||||||
log.WithContext(ctx).Tracef("released write lock for ID %s in %v", uniqueID, time.Since(startHold))
|
|
||||||
}
|
|
||||||
|
|
||||||
return unlock
|
|
||||||
}
|
|
||||||
|
|
||||||
// Deprecated: Full account operations are no longer supported
|
// Deprecated: Full account operations are no longer supported
|
||||||
func (s *SqlStore) SaveAccount(ctx context.Context, account *types.Account) error {
|
func (s *SqlStore) SaveAccount(ctx context.Context, account *types.Account) error {
|
||||||
start := time.Now()
|
start := time.Now()
|
||||||
|
|||||||
@@ -168,8 +168,6 @@ type Store interface {
|
|||||||
GetInstallationID() string
|
GetInstallationID() string
|
||||||
SaveInstallationID(ctx context.Context, ID string) error
|
SaveInstallationID(ctx context.Context, ID string) error
|
||||||
|
|
||||||
// AcquireWriteLockByUID should attempt to acquire a lock for write purposes and return a function that releases the lock
|
|
||||||
AcquireWriteLockByUID(ctx context.Context, uniqueID string) func()
|
|
||||||
// AcquireGlobalLock should attempt to acquire a global lock and return a function that releases the lock
|
// AcquireGlobalLock should attempt to acquire a global lock and return a function that releases the lock
|
||||||
AcquireGlobalLock(ctx context.Context) func()
|
AcquireGlobalLock(ctx context.Context) func()
|
||||||
|
|
||||||
|
|||||||
@@ -73,9 +73,6 @@ func (am *DefaultAccountManager) CreateUser(ctx context.Context, accountID, user
|
|||||||
|
|
||||||
// inviteNewUser Invites a USer to a given account and creates reference in datastore
|
// inviteNewUser Invites a USer to a given account and creates reference in datastore
|
||||||
func (am *DefaultAccountManager) inviteNewUser(ctx context.Context, accountID, userID string, invite *types.UserInfo) (*types.UserInfo, error) {
|
func (am *DefaultAccountManager) inviteNewUser(ctx context.Context, accountID, userID string, invite *types.UserInfo) (*types.UserInfo, error) {
|
||||||
unlock := am.Store.AcquireWriteLockByUID(ctx, accountID)
|
|
||||||
defer unlock()
|
|
||||||
|
|
||||||
if am.idpManager == nil {
|
if am.idpManager == nil {
|
||||||
return nil, status.Errorf(status.PreconditionFailed, "IdP manager must be enabled to send user invites")
|
return nil, status.Errorf(status.PreconditionFailed, "IdP manager must be enabled to send user invites")
|
||||||
}
|
}
|
||||||
@@ -224,9 +221,6 @@ func (am *DefaultAccountManager) DeleteUser(ctx context.Context, accountID, init
|
|||||||
return status.Errorf(status.InvalidArgument, "self deletion is not allowed")
|
return status.Errorf(status.InvalidArgument, "self deletion is not allowed")
|
||||||
}
|
}
|
||||||
|
|
||||||
unlock := am.Store.AcquireWriteLockByUID(ctx, accountID)
|
|
||||||
defer unlock()
|
|
||||||
|
|
||||||
initiatorUser, err := am.Store.GetUserByUserID(ctx, store.LockingStrengthNone, initiatorUserID)
|
initiatorUser, err := am.Store.GetUserByUserID(ctx, store.LockingStrengthNone, initiatorUserID)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
@@ -282,9 +276,6 @@ func (am *DefaultAccountManager) DeleteUser(ctx context.Context, accountID, init
|
|||||||
|
|
||||||
// InviteUser resend invitations to users who haven't activated their accounts prior to the expiration period.
|
// InviteUser resend invitations to users who haven't activated their accounts prior to the expiration period.
|
||||||
func (am *DefaultAccountManager) InviteUser(ctx context.Context, accountID string, initiatorUserID string, targetUserID string) error {
|
func (am *DefaultAccountManager) InviteUser(ctx context.Context, accountID string, initiatorUserID string, targetUserID string) error {
|
||||||
unlock := am.Store.AcquireWriteLockByUID(ctx, accountID)
|
|
||||||
defer unlock()
|
|
||||||
|
|
||||||
if am.idpManager == nil {
|
if am.idpManager == nil {
|
||||||
return status.Errorf(status.PreconditionFailed, "IdP manager must be enabled to send user invites")
|
return status.Errorf(status.PreconditionFailed, "IdP manager must be enabled to send user invites")
|
||||||
}
|
}
|
||||||
@@ -325,9 +316,6 @@ func (am *DefaultAccountManager) InviteUser(ctx context.Context, accountID strin
|
|||||||
|
|
||||||
// CreatePAT creates a new PAT for the given user
|
// CreatePAT creates a new PAT for the given user
|
||||||
func (am *DefaultAccountManager) CreatePAT(ctx context.Context, accountID string, initiatorUserID string, targetUserID string, tokenName string, expiresIn int) (*types.PersonalAccessTokenGenerated, error) {
|
func (am *DefaultAccountManager) CreatePAT(ctx context.Context, accountID string, initiatorUserID string, targetUserID string, tokenName string, expiresIn int) (*types.PersonalAccessTokenGenerated, error) {
|
||||||
unlock := am.Store.AcquireWriteLockByUID(ctx, accountID)
|
|
||||||
defer unlock()
|
|
||||||
|
|
||||||
if tokenName == "" {
|
if tokenName == "" {
|
||||||
return nil, status.Errorf(status.InvalidArgument, "token name can't be empty")
|
return nil, status.Errorf(status.InvalidArgument, "token name can't be empty")
|
||||||
}
|
}
|
||||||
@@ -376,9 +364,6 @@ func (am *DefaultAccountManager) CreatePAT(ctx context.Context, accountID string
|
|||||||
|
|
||||||
// DeletePAT deletes a specific PAT from a user
|
// DeletePAT deletes a specific PAT from a user
|
||||||
func (am *DefaultAccountManager) DeletePAT(ctx context.Context, accountID string, initiatorUserID string, targetUserID string, tokenID string) error {
|
func (am *DefaultAccountManager) DeletePAT(ctx context.Context, accountID string, initiatorUserID string, targetUserID string, tokenID string) error {
|
||||||
unlock := am.Store.AcquireWriteLockByUID(ctx, accountID)
|
|
||||||
defer unlock()
|
|
||||||
|
|
||||||
allowed, err := am.permissionsManager.ValidateUserPermissions(ctx, accountID, initiatorUserID, modules.Pats, operations.Delete)
|
allowed, err := am.permissionsManager.ValidateUserPermissions(ctx, accountID, initiatorUserID, modules.Pats, operations.Delete)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return status.NewPermissionValidationError(err)
|
return status.NewPermissionValidationError(err)
|
||||||
@@ -478,9 +463,6 @@ func (am *DefaultAccountManager) SaveUser(ctx context.Context, accountID, initia
|
|||||||
// SaveOrAddUser updates the given user. If addIfNotExists is set to true it will add user when no exist
|
// SaveOrAddUser updates the given user. If addIfNotExists is set to true it will add user when no exist
|
||||||
// Only User.AutoGroups, User.Role, and User.Blocked fields are allowed to be updated for now.
|
// Only User.AutoGroups, User.Role, and User.Blocked fields are allowed to be updated for now.
|
||||||
func (am *DefaultAccountManager) SaveOrAddUser(ctx context.Context, accountID, initiatorUserID string, update *types.User, addIfNotExists bool) (*types.UserInfo, error) {
|
func (am *DefaultAccountManager) SaveOrAddUser(ctx context.Context, accountID, initiatorUserID string, update *types.User, addIfNotExists bool) (*types.UserInfo, error) {
|
||||||
unlock := am.Store.AcquireWriteLockByUID(ctx, accountID)
|
|
||||||
defer unlock()
|
|
||||||
|
|
||||||
updatedUsers, err := am.SaveOrAddUsers(ctx, accountID, initiatorUserID, []*types.User{update}, addIfNotExists)
|
updatedUsers, err := am.SaveOrAddUsers(ctx, accountID, initiatorUserID, []*types.User{update}, addIfNotExists)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
|
|||||||
Reference in New Issue
Block a user