netmaker/logic/acls/common.go
Abhishek K 66069fbc34
NET-1082: Scale Testing Fixes (#2894)
* add additional mutex lock on node acls func

* increase verbosity

* disable acls on cloud emqx

* add emqx creds creation to go routine

* add debug log of mq client id

* comment port check

* uncomment port check

* check for connection mq connection open

* use username for client id

* add write mutex on acl is allowed

* add mq connection lost handler on server

* spin off zombie init as go routine

* get whole api path from config

* Revert "get whole api path from config"

This reverts commit 392f5f4c5f.

* update extclient acls async

* add additional mutex lock on node acls func

(cherry picked from commit 5325f0e7d7)

* increase verbosity

(cherry picked from commit 705b3cf0bf)

* add emqx creds creation to go routine

(cherry picked from commit c8e65f4820)

* add debug log of mq client id

(cherry picked from commit 29c5d6ceca)

* comment port check

(cherry picked from commit db8d6d95ea)

* check for connection mq connection open

(cherry picked from commit 13b11033b0)

* use username for client id

(cherry picked from commit e90c7386de)

* add write mutex on acl is allowed

(cherry picked from commit 4cae1b0bb4)

* add mq connection lost handler on server

(cherry picked from commit c82918ad35)

* spin off zombie init as go routine

(cherry picked from commit 6d65c44c43)

* update extclient acls async

(cherry picked from commit 6557ef1ebe)

* additionl logs for oauth user flow

(cherry picked from commit 61703038ae)

* add more debug logs

(cherry picked from commit 5980beacd1)

* add more debug logs

(cherry picked from commit 4d001f0d27)

* add set auth secret

(cherry picked from commit f41cef5da5)

* fix fetch pass

(cherry picked from commit 825caf4b60)

* make sure auth secret is set only once

(cherry picked from commit ba33ed02aa)

* make sure auth secret is set only once

(cherry picked from commit 920ac4c507)

* comment usage of emqx acls

* replace  read lock with write lock on acls

* replace  read lock with write lock on acls

(cherry picked from commit 808d2135c8)

* use deadlock pkg for visibility

* add additional mutex locks

* remove race flag

* on mq re-connecting donot exit if failed

* on mq re-connecting donot exit if failed

* revert mutex package change

* set mq clean session

* remove debug log

* go mod tidy

* revert on prem emqx acls del
2024-04-11 21:18:57 +05:30

198 lines
5.4 KiB
Go

package acls
import (
"encoding/json"
"sync"
"github.com/gravitl/netmaker/database"
"github.com/gravitl/netmaker/servercfg"
"golang.org/x/exp/slog"
)
var (
aclCacheMutex = &sync.RWMutex{}
aclCacheMap = make(map[ContainerID]ACLContainer)
AclMutex = &sync.RWMutex{}
)
func fetchAclContainerFromCache(containerID ContainerID) (aclCont ACLContainer, ok bool) {
aclCacheMutex.RLock()
aclCont, ok = aclCacheMap[containerID]
aclCacheMutex.RUnlock()
return
}
func storeAclContainerInCache(containerID ContainerID, aclContainer ACLContainer) {
aclCacheMutex.Lock()
aclCacheMap[containerID] = aclContainer
aclCacheMutex.Unlock()
}
func DeleteAclFromCache(containerID ContainerID) {
aclCacheMutex.Lock()
delete(aclCacheMap, containerID)
aclCacheMutex.Unlock()
}
// == type functions ==
// ACL.Allow - allows access by ID in memory
func (acl ACL) Allow(ID AclID) {
AclMutex.Lock()
defer AclMutex.Unlock()
acl[ID] = Allowed
}
// ACL.DisallowNode - disallows access by ID in memory
func (acl ACL) Disallow(ID AclID) {
AclMutex.Lock()
defer AclMutex.Unlock()
acl[ID] = NotAllowed
}
// ACL.Remove - removes a node from a ACL in memory
func (acl ACL) Remove(ID AclID) {
AclMutex.Lock()
defer AclMutex.Unlock()
delete(acl, ID)
}
// ACL.Update - updates a ACL in DB
func (acl ACL) Save(containerID ContainerID, ID AclID) (ACL, error) {
return upsertACL(containerID, ID, acl)
}
// ACL.IsAllowed - sees if ID is allowed in referring ACL
func (acl ACL) IsAllowed(ID AclID) (allowed bool) {
AclMutex.Lock()
allowed = acl[ID] == Allowed
AclMutex.Unlock()
return
}
// ACLContainer.UpdateACL - saves the state of a ACL in the ACLContainer in memory
func (aclContainer ACLContainer) UpdateACL(ID AclID, acl ACL) ACLContainer {
AclMutex.Lock()
defer AclMutex.Unlock()
aclContainer[ID] = acl
return aclContainer
}
// ACLContainer.RemoveACL - removes the state of a ACL in the ACLContainer in memory
func (aclContainer ACLContainer) RemoveACL(ID AclID) ACLContainer {
AclMutex.Lock()
defer AclMutex.Unlock()
delete(aclContainer, ID)
return aclContainer
}
// ACLContainer.ChangeAccess - changes the relationship between two nodes in memory
func (networkACL ACLContainer) ChangeAccess(ID1, ID2 AclID, value byte) {
AclMutex.Lock()
defer AclMutex.Unlock()
if _, ok := networkACL[ID1]; !ok {
slog.Error("ACL missing for ", "id", ID1)
return
}
if _, ok := networkACL[ID2]; !ok {
slog.Error("ACL missing for ", "id", ID2)
return
}
if _, ok := networkACL[ID1][ID2]; !ok {
slog.Error("ACL missing for ", "id1", ID1, "id2", ID2)
return
}
if _, ok := networkACL[ID2][ID1]; !ok {
slog.Error("ACL missing for ", "id2", ID2, "id1", ID1)
return
}
networkACL[ID1][ID2] = value
networkACL[ID2][ID1] = value
}
// ACLContainer.Save - saves the state of a ACLContainer to the db
func (aclContainer ACLContainer) Save(containerID ContainerID) (ACLContainer, error) {
return upsertACLContainer(containerID, aclContainer)
}
// ACLContainer.New - saves the state of a ACLContainer to the db
func (aclContainer ACLContainer) New(containerID ContainerID) (ACLContainer, error) {
return upsertACLContainer(containerID, nil)
}
// ACLContainer.Get - saves the state of a ACLContainer to the db
func (aclContainer ACLContainer) Get(containerID ContainerID) (ACLContainer, error) {
return fetchACLContainer(containerID)
}
// == private ==
// fetchACLContainer - fetches all current rules in given ACL container
func fetchACLContainer(containerID ContainerID) (ACLContainer, error) {
AclMutex.RLock()
defer AclMutex.RUnlock()
if servercfg.CacheEnabled() {
if aclContainer, ok := fetchAclContainerFromCache(containerID); ok {
return aclContainer, nil
}
}
aclJson, err := fetchACLContainerJson(ContainerID(containerID))
if err != nil {
return nil, err
}
var currentNetworkACL ACLContainer
if err := json.Unmarshal([]byte(aclJson), &currentNetworkACL); err != nil {
return nil, err
}
if servercfg.CacheEnabled() {
storeAclContainerInCache(containerID, currentNetworkACL)
}
return currentNetworkACL, nil
}
// fetchACLContainerJson - fetch the current ACL of given container except in json string
func fetchACLContainerJson(containerID ContainerID) (ACLJson, error) {
currentACLs, err := database.FetchRecord(database.NODE_ACLS_TABLE_NAME, string(containerID))
if err != nil {
return ACLJson(""), err
}
return ACLJson(currentACLs), nil
}
// upsertACL - applies a ACL to the db, overwrites or creates
func upsertACL(containerID ContainerID, ID AclID, acl ACL) (ACL, error) {
currentNetACL, err := fetchACLContainer(containerID)
if err != nil {
return acl, err
}
currentNetACL[ID] = acl
_, err = upsertACLContainer(containerID, currentNetACL)
return acl, err
}
// upsertACLContainer - Inserts or updates a network ACL given the json string of the ACL and the container ID
// if nil, create it
func upsertACLContainer(containerID ContainerID, aclContainer ACLContainer) (ACLContainer, error) {
AclMutex.Lock()
defer AclMutex.Unlock()
if aclContainer == nil {
aclContainer = make(ACLContainer)
}
err := database.Insert(string(containerID), string(convertNetworkACLtoACLJson(aclContainer)), database.NODE_ACLS_TABLE_NAME)
if err != nil {
return aclContainer, err
}
if servercfg.CacheEnabled() {
storeAclContainerInCache(containerID, aclContainer)
}
return aclContainer, nil
}
func convertNetworkACLtoACLJson(networkACL ACLContainer) ACLJson {
data, err := json.Marshal(networkACL)
if err != nil {
return ""
}
return ACLJson(data)
}