NET-1778: scale test code changes (#3203)

* comment ACL call and add debug message

* add cache for network nodes

* fix load node to network cache issue

* add peerUpdate call 1 min limit

* add debug log for scale test

* release maps

* avoid default policy for node

* 1 min limit for peerUpdate trigger

* mq options

* Revert "mq options"

This reverts commit 10b93d0118.

* set peerUpdate run in sequence

* update for emqx 5.8.2

* remove batch peer update

* change the sleep to 10 millisec to avoid timeout

* add compress and change encrypt for peerUpdate message

* add mem profiling and automaxprocs

* add failover ctx mutex

* ignore request to failover peer

* remove code without called

* remove debug logs

* update emqx to v5.8.2

* change broker keepalive

* add OLD_ACL_SUPPORT setting

* add host version check for message encrypt

* remove debug message

* remove peerUpdate call control

---------

Co-authored-by: abhishek9686 <abhi281342@gmail.com>
This commit is contained in:
Yabin Ma
2024-12-10 07:15:31 +01:00
committed by GitHub
parent c56f1cab15
commit 5f21c8bb1d
21 changed files with 273 additions and 82 deletions

View File

@@ -5,7 +5,9 @@ import (
"encoding/json"
"errors"
"fmt"
"maps"
"net"
"slices"
"sort"
"sync"
"time"
@@ -24,8 +26,10 @@ import (
)
var (
nodeCacheMutex = &sync.RWMutex{}
nodesCacheMap = make(map[string]models.Node)
nodeCacheMutex = &sync.RWMutex{}
nodeNetworkCacheMutex = &sync.RWMutex{}
nodesCacheMap = make(map[string]models.Node)
nodesNetworkCacheMap = make(map[string]map[string]models.Node)
)
func getNodeFromCache(nodeID string) (node models.Node, ok bool) {
@@ -48,12 +52,37 @@ func deleteNodeFromCache(nodeID string) {
delete(nodesCacheMap, nodeID)
nodeCacheMutex.Unlock()
}
func deleteNodeFromNetworkCache(nodeID string, network string) {
nodeNetworkCacheMutex.Lock()
delete(nodesNetworkCacheMap[network], nodeID)
nodeNetworkCacheMutex.Unlock()
}
func storeNodeInNetworkCache(node models.Node, network string) {
nodeNetworkCacheMutex.Lock()
if nodesNetworkCacheMap[network] == nil {
nodesNetworkCacheMap[network] = make(map[string]models.Node)
}
nodesNetworkCacheMap[network][node.ID.String()] = node
nodeNetworkCacheMutex.Unlock()
}
func storeNodeInCache(node models.Node) {
nodeCacheMutex.Lock()
nodesCacheMap[node.ID.String()] = node
nodeCacheMutex.Unlock()
}
func loadNodesIntoNetworkCache(nMap map[string]models.Node) {
nodeNetworkCacheMutex.Lock()
for _, v := range nMap {
network := v.Network
if nodesNetworkCacheMap[network] == nil {
nodesNetworkCacheMap[network] = make(map[string]models.Node)
}
nodesNetworkCacheMap[network][v.ID.String()] = v
}
nodeNetworkCacheMutex.Unlock()
}
func loadNodesIntoCache(nMap map[string]models.Node) {
nodeCacheMutex.Lock()
@@ -63,6 +92,7 @@ func loadNodesIntoCache(nMap map[string]models.Node) {
func ClearNodeCache() {
nodeCacheMutex.Lock()
nodesCacheMap = make(map[string]models.Node)
nodesNetworkCacheMap = make(map[string]map[string]models.Node)
nodeCacheMutex.Unlock()
}
@@ -77,6 +107,12 @@ const (
// GetNetworkNodes - gets the nodes of a network
func GetNetworkNodes(network string) ([]models.Node, error) {
if networkNodes, ok := nodesNetworkCacheMap[network]; ok {
nodeNetworkCacheMutex.Lock()
defer nodeNetworkCacheMutex.Unlock()
return slices.Collect(maps.Values(networkNodes)), nil
}
allnodes, err := GetAllNodes()
if err != nil {
return []models.Node{}, err
@@ -99,6 +135,12 @@ func GetHostNodes(host *models.Host) []models.Node {
// GetNetworkNodesMemory - gets all nodes belonging to a network from list in memory
func GetNetworkNodesMemory(allNodes []models.Node, network string) []models.Node {
if networkNodes, ok := nodesNetworkCacheMap[network]; ok {
nodeNetworkCacheMutex.Lock()
defer nodeNetworkCacheMutex.Unlock()
return slices.Collect(maps.Values(networkNodes))
}
var nodes = []models.Node{}
for i := range allNodes {
node := allNodes[i]
@@ -123,6 +165,7 @@ func UpdateNodeCheckin(node *models.Node) error {
}
if servercfg.CacheEnabled() {
storeNodeInCache(*node)
storeNodeInNetworkCache(*node, node.Network)
}
return nil
}
@@ -140,6 +183,7 @@ func UpsertNode(newNode *models.Node) error {
}
if servercfg.CacheEnabled() {
storeNodeInCache(*newNode)
storeNodeInNetworkCache(*newNode, newNode.Network)
}
return nil
}
@@ -179,6 +223,7 @@ func UpdateNode(currentNode *models.Node, newNode *models.Node) error {
}
if servercfg.CacheEnabled() {
storeNodeInCache(*newNode)
storeNodeInNetworkCache(*newNode, newNode.Network)
if _, ok := allocatedIpMap[newNode.Network]; ok {
if newNode.Address.IP != nil && !newNode.Address.IP.Equal(currentNode.Address.IP) {
AddIpToAllocatedIpMap(newNode.Network, newNode.Address.IP)
@@ -298,6 +343,7 @@ func DeleteNodeByID(node *models.Node) error {
}
if servercfg.CacheEnabled() {
deleteNodeFromCache(node.ID.String())
deleteNodeFromNetworkCache(node.ID.String(), node.Network)
}
if servercfg.IsDNSMode() {
SetDNS()
@@ -360,6 +406,7 @@ func GetAllNodes() ([]models.Node, error) {
nodesMap := make(map[string]models.Node)
if servercfg.CacheEnabled() {
defer loadNodesIntoCache(nodesMap)
defer loadNodesIntoNetworkCache(nodesMap)
}
collection, err := database.FetchRecords(database.NODES_TABLE_NAME)
if err != nil {
@@ -469,6 +516,7 @@ func GetNodeByID(uuid string) (models.Node, error) {
}
if servercfg.CacheEnabled() {
storeNodeInCache(node)
storeNodeInNetworkCache(node, node.Network)
}
return node, nil
}
@@ -622,6 +670,7 @@ func createNode(node *models.Node) error {
}
if servercfg.CacheEnabled() {
storeNodeInCache(*node)
storeNodeInNetworkCache(*node, node.Network)
}
if _, ok := allocatedIpMap[node.Network]; ok {
if node.Address.IP != nil {