increase grpc timeouts

management/client/grpc.go

@@ -272,9 +272,9 @@ func (c *GrpcClient) GetServerPublicKey() (*wgtypes.Key, error) {
 		return nil, errors.New(errMsgNoMgmtConnection)
 	}
 
-	mgmCtx, cancel := context.WithTimeout(c.ctx, 5*time.Second)
-	defer cancel()
-	resp, err := c.realClient.GetServerKey(mgmCtx, &proto.Empty{})
+	// mgmCtx, cancel := context.WithTimeout(c.ctx, 30*time.Second)
+	// defer cancel()
+	resp, err := c.realClient.GetServerKey(c.ctx, &proto.Empty{})
 	if err != nil {
 		log.Errorf("failed while getting Management Service public key: %v", err)
 		return nil, fmt.Errorf("failed while getting Management Service public key")
@@ -327,11 +327,9 @@ func (c *GrpcClient) login(serverKey wgtypes.Key, req *proto.LoginRequest) (*pro
 
 	var resp *proto.EncryptedMessage
 	operation := func() error {
-		mgmCtx, cancel := context.WithTimeout(context.Background(), ConnectTimeout)
-		defer cancel()
 
 		var err error
-		resp, err = c.realClient.Login(mgmCtx, &proto.EncryptedMessage{
+		resp, err = c.realClient.Login(context.Background(), &proto.EncryptedMessage{
 			WgPubKey: c.key.PublicKey().String(),
 			Body:     loginReq,
 		})

management/cmd/management.go

@@ -11,6 +11,7 @@ import (
 	"io/fs"
 	"net"
 	"net/http"
+	_ "net/http/pprof"
 	"net/netip"
 	"net/url"
 	"os"
@@ -34,6 +35,7 @@ import (
 	"github.com/grpc-ecosystem/go-grpc-middleware/v2/interceptors/realip"
 
 	"github.com/netbirdio/management-integrations/integrations"
+
 	"github.com/netbirdio/netbird/management/server/peers"
 
 	"github.com/netbirdio/netbird/encryption"
@@ -90,6 +92,11 @@ var (
 		PreRunE: func(cmd *cobra.Command, args []string) error {
 			flag.Parse()
 
+			go func() {
+				log.Infof("Starting pprof on :6060")
+				http.ListenAndServe("localhost:6060", nil)
+			}()
+
 			//nolint
 			ctx := context.WithValue(cmd.Context(), hook.ExecutionContextKey, hook.SystemSource)
 
@@ -132,6 +139,7 @@ var (
 			return nil
 		},
 		RunE: func(cmd *cobra.Command, args []string) error {
+
 			flag.Parse()
 
 			ctx, cancel := context.WithCancel(cmd.Context())

management/server/account.go

@@ -68,7 +68,8 @@ type DefaultAccountManager struct {
 	eventStore           activity.Store
 	geo                  geolocation.Geolocation
 
-	requestBuffer *AccountRequestBuffer
+	accountUpdateLocks sync.Map
+	requestBuffer      *AccountRequestBuffer
 
 	proxyController port_forwarding.Controller
 	settingsManager settings.Manager
@@ -1222,7 +1223,7 @@ func (am *DefaultAccountManager) SyncUserJWTGroups(ctx context.Context, userAuth
 
 		if removedGroupAffectsPeers || newGroupsAffectsPeers {
 			log.WithContext(ctx).Tracef("user %s: JWT group membership changed, updating account peers", userAuth.UserId)
-			am.UpdateAccountPeers(ctx, userAuth.AccountId)
+			am.BufferUpdateAccountPeers(ctx, userAuth.AccountId)
 		}
 	}
 
@@ -1461,7 +1462,7 @@ func (am *DefaultAccountManager) GetDNSDomain() string {
 
 func (am *DefaultAccountManager) onPeersInvalidated(ctx context.Context, accountID string) {
 	log.WithContext(ctx).Debugf("validated peers has been invalidated for account %s", accountID)
-	am.UpdateAccountPeers(ctx, accountID)
+	am.BufferUpdateAccountPeers(ctx, accountID)
 }
 
 func (am *DefaultAccountManager) FindExistingPostureCheck(accountID string, checks *posture.ChecksDefinition) (*posture.Checks, error) {

management/server/account_test.go

@@ -385,7 +385,7 @@ func TestAccount_GetPeerNetworkMap(t *testing.T) {
 		}
 
 		customZone := account.GetPeersCustomZone(context.Background(), "netbird.io")
-		networkMap := account.GetPeerNetworkMap(context.Background(), testCase.peerID, customZone, validatedPeers, account.GetResourcePoliciesMap(), account.GetResourceRoutersMap(), account.GetPeersGroupsMap(), account.GetGroupsPolicyMap(), nil)
+		networkMap := account.GetPeerNetworkMap(context.Background(), testCase.peerID, customZone, validatedPeers, account.GetResourcePoliciesMap(), account.GetResourceRoutersMap(), nil)
 		assert.Len(t, networkMap.Peers, len(testCase.expectedPeers))
 		assert.Len(t, networkMap.OfflinePeers, len(testCase.expectedOfflinePeers))
 	}

management/server/grpcserver.go

@@ -10,16 +10,14 @@ import (
 	"time"
 
 	pb "github.com/golang/protobuf/proto" // nolint
-	"github.com/golang/protobuf/ptypes/timestamp"
 	"github.com/grpc-ecosystem/go-grpc-middleware/v2/interceptors/realip"
 	log "github.com/sirupsen/logrus"
 	"golang.zx2c4.com/wireguard/wgctrl/wgtypes"
 	"google.golang.org/grpc/codes"
-	"google.golang.org/grpc/peer"
 	"google.golang.org/grpc/status"
+	"google.golang.org/protobuf/types/known/timestamppb"
 
 	integrationsConfig "github.com/netbirdio/management-integrations/integrations/config"
-
 	"github.com/netbirdio/netbird/encryption"
 	"github.com/netbirdio/netbird/management/proto"
 	"github.com/netbirdio/netbird/management/server/account"
@@ -39,6 +37,7 @@ type GRPCServer struct {
 	accountManager  account.Manager
 	settingsManager settings.Manager
 	wgKey           wgtypes.Key
+	wgPubKeySting   string
 	proto.UnimplementedManagementServiceServer
 	peersUpdateManager *PeersUpdateManager
 	config             *Config
@@ -77,7 +76,8 @@ func NewServer(
 	}
 
 	return &GRPCServer{
-		wgKey: key,
+		wgKey:         key,
+		wgPubKeySting: key.PublicKey().String(),
 		// peerKey -> event channel
 		peersUpdateManager: peersUpdateManager,
 		accountManager:     accountManager,
@@ -91,30 +91,16 @@ func NewServer(
 }
 
 func (s *GRPCServer) GetServerKey(ctx context.Context, req *proto.Empty) (*proto.ServerKeyResponse, error) {
-	ip := ""
-	p, ok := peer.FromContext(ctx)
-	if ok {
-		ip = p.Addr.String()
-	}
-
-	log.WithContext(ctx).Tracef("GetServerKey request from %s", ip)
-	start := time.Now()
-	defer func() {
-		log.WithContext(ctx).Tracef("GetServerKey from %s took %v", ip, time.Since(start))
-	}()
-
 	// todo introduce something more meaningful with the key expiration/rotation
 	if s.appMetrics != nil {
 		s.appMetrics.GRPCMetrics().CountGetKeyRequest()
 	}
-	now := time.Now().Add(24 * time.Hour)
-	secs := int64(now.Second())
-	nanos := int32(now.Nanosecond())
-	expiresAt := &timestamp.Timestamp{Seconds: secs, Nanos: nanos}
+
+	expiresAt := time.Now().Add(24 * time.Hour)
 
 	return &proto.ServerKeyResponse{
-		Key:       s.wgKey.PublicKey().String(),
-		ExpiresAt: expiresAt,
+		Key:       s.wgPubKeySting,
+		ExpiresAt: timestamppb.New(expiresAt),
 	}, nil
 }
 
@@ -243,7 +229,7 @@ func (s *GRPCServer) sendUpdate(ctx context.Context, accountID string, peerKey w
 		return status.Errorf(codes.Internal, "failed processing update message")
 	}
 	err = srv.SendMsg(&proto.EncryptedMessage{
-		WgPubKey: s.wgKey.PublicKey().String(),
+		WgPubKey: s.wgPubKeySting,
 		Body:     encryptedResp,
 	})
 	if err != nil {
@@ -501,7 +487,7 @@ func (s *GRPCServer) Login(ctx context.Context, req *proto.EncryptedMessage) (*p
 	}
 
 	return &proto.EncryptedMessage{
-		WgPubKey: s.wgKey.PublicKey().String(),
+		WgPubKey: s.wgPubKeySting,
 		Body:     encryptedResp,
 	}, nil
 }
@@ -636,13 +622,14 @@ func toSyncResponse(ctx context.Context, config *Config, peer *nbpeer.Peer, turn
 
 	response.NetworkMap.PeerConfig = response.PeerConfig
 
-	allPeers := appendRemotePeerConfig(networkMap.Peers, dnsName)
+	allPeers := make([]*proto.RemotePeerConfig, 0, len(networkMap.Peers)+len(networkMap.OfflinePeers))
+	allPeers = appendRemotePeerConfig(allPeers, networkMap.Peers, dnsName)
 	response.RemotePeers = allPeers
 	response.NetworkMap.RemotePeers = allPeers
 	response.RemotePeersIsEmpty = len(allPeers) == 0
 	response.NetworkMap.RemotePeersIsEmpty = response.RemotePeersIsEmpty
 
-	response.NetworkMap.OfflinePeers = appendRemotePeerConfig(networkMap.OfflinePeers, dnsName)
+	response.NetworkMap.OfflinePeers = appendRemotePeerConfig(nil, networkMap.OfflinePeers, dnsName)
 
 	firewallRules := toProtocolFirewallRules(networkMap.FirewallRules)
 	response.NetworkMap.FirewallRules = firewallRules
@@ -663,18 +650,15 @@ func toSyncResponse(ctx context.Context, config *Config, peer *nbpeer.Peer, turn
 	return response
 }
 
-func appendRemotePeerConfig(peers []*nbpeer.Peer, dnsName string) []*proto.RemotePeerConfig {
-	dst := make([]*proto.RemotePeerConfig, len(peers))
-
-	for i, rPeer := range peers {
-		dst[i] = &proto.RemotePeerConfig{
+func appendRemotePeerConfig(dst []*proto.RemotePeerConfig, peers []*nbpeer.Peer, dnsName string) []*proto.RemotePeerConfig {
+	for _, rPeer := range peers {
+		dst = append(dst, &proto.RemotePeerConfig{
 			WgPubKey:   rPeer.Key,
 			AllowedIps: []string{rPeer.IP.String() + "/32"},
 			SshConfig:  &proto.SSHConfig{SshPubKey: []byte(rPeer.SSHKey)},
 			Fqdn:       rPeer.FQDN(dnsName),
-		}
+		})
 	}
-
 	return dst
 }
 
@@ -716,7 +700,7 @@ func (s *GRPCServer) sendInitialSync(ctx context.Context, peerKey wgtypes.Key, p
 	}
 
 	err = srv.Send(&proto.EncryptedMessage{
-		WgPubKey: s.wgKey.PublicKey().String(),
+		WgPubKey: s.wgPubKeySting,
 		Body:     encryptedResp,
 	})
 
@@ -781,7 +765,7 @@ func (s *GRPCServer) GetDeviceAuthorizationFlow(ctx context.Context, req *proto.
 	}
 
 	return &proto.EncryptedMessage{
-		WgPubKey: s.wgKey.PublicKey().String(),
+		WgPubKey: s.wgPubKeySting,
 		Body:     encryptedResp,
 	}, nil
 }
@@ -833,7 +817,7 @@ func (s *GRPCServer) GetPKCEAuthorizationFlow(ctx context.Context, req *proto.En
 	}
 
 	return &proto.EncryptedMessage{
-		WgPubKey: s.wgKey.PublicKey().String(),
+		WgPubKey: s.wgPubKeySting,
 		Body:     encryptedResp,
 	}, nil
 }

management/server/http/handlers/peers/peers_handler.go

@@ -281,7 +281,7 @@ func (h *Handler) GetAccessiblePeers(w http.ResponseWriter, r *http.Request) {
 	dnsDomain := h.accountManager.GetDNSDomain()
 
 	customZone := account.GetPeersCustomZone(r.Context(), dnsDomain)
-	netMap := account.GetPeerNetworkMap(r.Context(), peerID, customZone, validPeers, account.GetResourcePoliciesMap(), account.GetResourceRoutersMap(), account.GetPeersGroupsMap(), account.GetGroupsPolicyMap(), nil)
+	netMap := account.GetPeerNetworkMap(r.Context(), peerID, customZone, validPeers, account.GetResourcePoliciesMap(), account.GetResourceRoutersMap(), nil)
 
 	util.WriteJSONObject(r.Context(), w, toAccessiblePeers(netMap, dnsDomain))
 }

management/server/peer.go

@@ -6,7 +6,9 @@ import (
 	b64 "encoding/base64"
 	"fmt"
 	"net"
+	"os"
 	"slices"
+	"strconv"
 	"strings"
 	"sync"
 	"time"
@@ -83,7 +85,7 @@ func (am *DefaultAccountManager) GetPeers(ctx context.Context, accountID, userID
 
 	// fetch all the peers that have access to the user's peers
 	for _, peer := range peers {
-		aclPeers, _ := account.GetPeerConnectionResources(ctx, peer.ID, approvedPeersMap, account.GetPeersGroupsMap(), account.GetGroupsPolicyMap())
+		aclPeers, _ := account.GetPeerConnectionResources(ctx, peer.ID, approvedPeersMap)
 		for _, p := range aclPeers {
 			peersMap[p.ID] = p
 		}
@@ -135,7 +137,7 @@ func (am *DefaultAccountManager) MarkPeerConnected(ctx context.Context, peerPubK
 	if expired {
 		// we need to update other peers because when peer login expires all other peers are notified to disconnect from
 		// the expired one. Here we notify them that connection is now allowed again.
-		am.UpdateAccountPeers(ctx, accountID)
+		am.BufferUpdateAccountPeers(ctx, accountID)
 	}
 
 	return nil
@@ -302,7 +304,7 @@ func (am *DefaultAccountManager) UpdatePeer(ctx context.Context, accountID, user
 	}
 
 	if peerLabelChanged || requiresPeerUpdates {
-		am.UpdateAccountPeers(ctx, accountID)
+		am.BufferUpdateAccountPeers(ctx, accountID)
 	} else if sshChanged {
 		am.UpdateAccountPeer(ctx, accountID, peer.ID)
 	}
@@ -383,7 +385,7 @@ func (am *DefaultAccountManager) DeletePeer(ctx context.Context, accountID, peer
 	}
 
 	if updateAccountPeers {
-		am.UpdateAccountPeers(ctx, accountID)
+		am.BufferUpdateAccountPeers(ctx, accountID)
 	}
 
 	return nil
@@ -418,7 +420,7 @@ func (am *DefaultAccountManager) GetNetworkMap(ctx context.Context, peerID strin
 		return nil, err
 	}
 
-	networkMap := account.GetPeerNetworkMap(ctx, peer.ID, customZone, validatedPeers, account.GetResourcePoliciesMap(), account.GetResourceRoutersMap(), account.GetPeersGroupsMap(), account.GetGroupsPolicyMap(), nil)
+	networkMap := account.GetPeerNetworkMap(ctx, peer.ID, customZone, validatedPeers, account.GetResourcePoliciesMap(), account.GetResourceRoutersMap(), nil)
 
 	proxyNetworkMap, ok := proxyNetworkMaps[peer.ID]
 	if ok {
@@ -653,7 +655,7 @@ func (am *DefaultAccountManager) AddPeer(ctx context.Context, setupKey, userID s
 	unlock = nil
 
 	if updateAccountPeers {
-		am.UpdateAccountPeers(ctx, accountID)
+		am.BufferUpdateAccountPeers(ctx, accountID)
 	}
 
 	return am.getValidatedPeerWithMap(ctx, false, accountID, newPeer)
@@ -748,7 +750,7 @@ func (am *DefaultAccountManager) SyncPeer(ctx context.Context, sync types.PeerSy
 	}
 
 	if isStatusChanged || sync.UpdateAccountPeers || (updated && len(postureChecks) > 0) {
-		am.UpdateAccountPeers(ctx, accountID)
+		am.BufferUpdateAccountPeers(ctx, accountID)
 	}
 
 	return am.getValidatedPeerWithMap(ctx, peerNotValid, accountID, peer)
@@ -893,7 +895,7 @@ func (am *DefaultAccountManager) LoginPeer(ctx context.Context, login types.Peer
 	unlockPeer = nil
 
 	if updateRemotePeers || isStatusChanged || (isPeerUpdated && len(postureChecks) > 0) {
-		am.UpdateAccountPeers(ctx, accountID)
+		am.BufferUpdateAccountPeers(ctx, accountID)
 	}
 
 	return am.getValidatedPeerWithMap(ctx, isRequiresApproval, accountID, peer)
@@ -1029,7 +1031,7 @@ func (am *DefaultAccountManager) getValidatedPeerWithMap(ctx context.Context, is
 		return nil, nil, nil, err
 	}
 
-	networkMap := account.GetPeerNetworkMap(ctx, peer.ID, customZone, approvedPeersMap, account.GetResourcePoliciesMap(), account.GetResourceRoutersMap(), account.GetPeersGroupsMap(), account.GetGroupsPolicyMap(), am.metrics.AccountManagerMetrics())
+	networkMap := account.GetPeerNetworkMap(ctx, peer.ID, customZone, approvedPeersMap, account.GetResourcePoliciesMap(), account.GetResourceRoutersMap(), am.metrics.AccountManagerMetrics())
 
 	proxyNetworkMap, ok := proxyNetworkMaps[peer.ID]
 	if ok {
@@ -1140,7 +1142,7 @@ func (am *DefaultAccountManager) GetPeer(ctx context.Context, accountID, peerID,
 	}
 
 	for _, p := range userPeers {
-		aclPeers, _ := account.GetPeerConnectionResources(ctx, p.ID, approvedPeersMap, account.GetPeersGroupsMap(), account.GetGroupsPolicyMap())
+		aclPeers, _ := account.GetPeerConnectionResources(ctx, p.ID, approvedPeersMap)
 		for _, aclPeer := range aclPeers {
 			if aclPeer.ID == peerID {
 				return peer, nil
@@ -1175,8 +1177,6 @@ func (am *DefaultAccountManager) UpdateAccountPeers(ctx context.Context, account
 	customZone := account.GetPeersCustomZone(ctx, am.dnsDomain)
 	resourcePolicies := account.GetResourcePoliciesMap()
 	routers := account.GetResourceRoutersMap()
-	peersGroups := account.GetPeersGroupsMap()
-	groupsPolicies := account.GetGroupsPolicyMap()
 
 	proxyNetworkMaps, err := am.proxyController.GetProxyNetworkMaps(ctx, accountID)
 	if err != nil {
@@ -1202,7 +1202,7 @@ func (am *DefaultAccountManager) UpdateAccountPeers(ctx context.Context, account
 				return
 			}
 
-			remotePeerNetworkMap := account.GetPeerNetworkMap(ctx, p.ID, customZone, approvedPeersMap, resourcePolicies, routers, peersGroups, groupsPolicies, am.metrics.AccountManagerMetrics())
+			remotePeerNetworkMap := account.GetPeerNetworkMap(ctx, p.ID, customZone, approvedPeersMap, resourcePolicies, routers, am.metrics.AccountManagerMetrics())
 
 			proxyNetworkMap, ok := proxyNetworkMaps[p.ID]
 			if ok {
@@ -1228,6 +1228,28 @@ func (am *DefaultAccountManager) UpdateAccountPeers(ctx context.Context, account
 	}
 }
 
+func (am *DefaultAccountManager) BufferUpdateAccountPeers(ctx context.Context, accountID string) {
+	mu, _ := am.accountUpdateLocks.LoadOrStore(accountID, &sync.Mutex{})
+	lock := mu.(*sync.Mutex)
+
+	if !lock.TryLock() {
+		return
+	}
+
+	go func() {
+		intervalStr := os.Getenv("PEER_UPDATE_INTERVAL_MS")
+		interval, err := strconv.Atoi(intervalStr)
+		if err != nil {
+			interval = 10000
+		}
+		time.Sleep(time.Duration(interval) * time.Millisecond)
+		am.UpdateAccountPeers(ctx, accountID)
+
+		am.accountUpdateLocks.Delete(accountID)
+		lock.Unlock()
+	}()
+}
+
 // UpdateAccountPeer updates a single peer that belongs to an account.
 // Should be called when changes need to be synced to a specific peer only.
 func (am *DefaultAccountManager) UpdateAccountPeer(ctx context.Context, accountId string, peerId string) {
@@ -1271,7 +1293,7 @@ func (am *DefaultAccountManager) UpdateAccountPeer(ctx context.Context, accountI
 		return
 	}
 
-	remotePeerNetworkMap := account.GetPeerNetworkMap(ctx, peerId, customZone, approvedPeersMap, resourcePolicies, routers, account.GetPeersGroupsMap(), account.GetGroupsPolicyMap(), am.metrics.AccountManagerMetrics())
+	remotePeerNetworkMap := account.GetPeerNetworkMap(ctx, peerId, customZone, approvedPeersMap, resourcePolicies, routers, am.metrics.AccountManagerMetrics())
 
 	proxyNetworkMap, ok := proxyNetworkMaps[peer.ID]
 	if ok {

management/server/peer_test.go

@@ -934,13 +934,13 @@ func BenchmarkUpdateAccountPeers(b *testing.B) {
 		minMsPerOpCICD  float64
 		maxMsPerOpCICD  float64
 	}{
-		// {"Small", 50, 5, 90, 120, 90, 120},
-		// {"Medium", 500, 100, 110, 150, 120, 260},
-		// {"Large", 5000, 200, 800, 1700, 2500, 5000},
-		// {"Small single", 50, 10, 90, 120, 90, 120},
-		// {"Medium single", 500, 10, 110, 170, 120, 200},
+		{"Small", 50, 5, 90, 120, 90, 120},
+		{"Medium", 500, 100, 110, 150, 120, 260},
+		{"Large", 5000, 200, 800, 1700, 2500, 5000},
+		{"Small single", 50, 10, 90, 120, 90, 120},
+		{"Medium single", 500, 10, 110, 170, 120, 200},
 		{"Large 5", 5000, 15, 1300, 2100, 4900, 7000},
-		// {"Extra Large", 5000, 2000, 1300, 2400, 3000, 6400},
+		{"Extra Large", 2000, 2000, 1300, 2400, 3000, 6400},
 	}
 
 	log.SetOutput(io.Discard)
@@ -948,7 +948,6 @@ func BenchmarkUpdateAccountPeers(b *testing.B) {
 
 	for _, bc := range benchCases {
 		b.Run(bc.name, func(b *testing.B) {
-			b.Setenv("NB_GET_ACCOUNT_BUFFER_INTERVAL", "0")
 			manager, accountID, _, err := setupTestAccountManager(b, bc.peers, bc.groups)
 			if err != nil {
 				b.Fatalf("Failed to setup test account manager: %v", err)

management/server/policy_test.go

@@ -158,14 +158,14 @@ func TestAccount_getPeersByPolicy(t *testing.T) {
 
 	t.Run("check that all peers get map", func(t *testing.T) {
 		for _, p := range account.Peers {
-			peers, firewallRules := account.GetPeerConnectionResources(context.Background(), p.ID, validatedPeers, account.GetPeersGroupsMap(), account.GetGroupsPolicyMap())
+			peers, firewallRules := account.GetPeerConnectionResources(context.Background(), p.ID, validatedPeers)
 			assert.GreaterOrEqual(t, len(peers), 2, "minimum number peers should present")
 			assert.GreaterOrEqual(t, len(firewallRules), 2, "minimum number of firewall rules should present")
 		}
 	})
 
 	t.Run("check first peer map details", func(t *testing.T) {
-		peers, firewallRules := account.GetPeerConnectionResources(context.Background(), "peerB", validatedPeers, account.GetPeersGroupsMap(), account.GetGroupsPolicyMap())
+		peers, firewallRules := account.GetPeerConnectionResources(context.Background(), "peerB", validatedPeers)
 		assert.Len(t, peers, 7)
 		assert.Contains(t, peers, account.Peers["peerA"])
 		assert.Contains(t, peers, account.Peers["peerC"])
@@ -394,7 +394,7 @@ func TestAccount_getPeersByPolicyDirect(t *testing.T) {
 	}
 
 	t.Run("check first peer map", func(t *testing.T) {
-		peers, firewallRules := account.GetPeerConnectionResources(context.Background(), "peerB", approvedPeers, account.GetPeersGroupsMap(), account.GetGroupsPolicyMap())
+		peers, firewallRules := account.GetPeerConnectionResources(context.Background(), "peerB", approvedPeers)
 		assert.Contains(t, peers, account.Peers["peerC"])
 
 		epectedFirewallRules := []*types.FirewallRule{
@@ -422,7 +422,7 @@ func TestAccount_getPeersByPolicyDirect(t *testing.T) {
 	})
 
 	t.Run("check second peer map", func(t *testing.T) {
-		peers, firewallRules := account.GetPeerConnectionResources(context.Background(), "peerC", approvedPeers, account.GetPeersGroupsMap(), account.GetGroupsPolicyMap())
+		peers, firewallRules := account.GetPeerConnectionResources(context.Background(), "peerC", approvedPeers)
 		assert.Contains(t, peers, account.Peers["peerB"])
 
 		epectedFirewallRules := []*types.FirewallRule{
@@ -452,7 +452,7 @@ func TestAccount_getPeersByPolicyDirect(t *testing.T) {
 	account.Policies[1].Rules[0].Bidirectional = false
 
 	t.Run("check first peer map directional only", func(t *testing.T) {
-		peers, firewallRules := account.GetPeerConnectionResources(context.Background(), "peerB", approvedPeers, account.GetPeersGroupsMap(), account.GetGroupsPolicyMap())
+		peers, firewallRules := account.GetPeerConnectionResources(context.Background(), "peerB", approvedPeers)
 		assert.Contains(t, peers, account.Peers["peerC"])
 
 		epectedFirewallRules := []*types.FirewallRule{
@@ -473,7 +473,7 @@ func TestAccount_getPeersByPolicyDirect(t *testing.T) {
 	})
 
 	t.Run("check second peer map directional only", func(t *testing.T) {
-		peers, firewallRules := account.GetPeerConnectionResources(context.Background(), "peerC", approvedPeers, account.GetPeersGroupsMap(), account.GetGroupsPolicyMap())
+		peers, firewallRules := account.GetPeerConnectionResources(context.Background(), "peerC", approvedPeers)
 		assert.Contains(t, peers, account.Peers["peerB"])
 
 		epectedFirewallRules := []*types.FirewallRule{
@@ -670,7 +670,7 @@ func TestAccount_getPeersByPolicyPostureChecks(t *testing.T) {
 	t.Run("verify peer's network map with default group peer list", func(t *testing.T) {
 		// peerB doesn't fulfill the NB posture check but is included in the destination group Swarm,
 		// will establish a connection with all source peers satisfying the NB posture check.
-		peers, firewallRules := account.GetPeerConnectionResources(context.Background(), "peerB", approvedPeers, account.GetPeersGroupsMap(), account.GetGroupsPolicyMap())
+		peers, firewallRules := account.GetPeerConnectionResources(context.Background(), "peerB", approvedPeers)
 		assert.Len(t, peers, 4)
 		assert.Len(t, firewallRules, 4)
 		assert.Contains(t, peers, account.Peers["peerA"])
@@ -680,7 +680,7 @@ func TestAccount_getPeersByPolicyPostureChecks(t *testing.T) {
 
 		// peerC satisfy the NB posture check, should establish connection to all destination group peer's
 		// We expect a single permissive firewall rule which all outgoing connections
-		peers, firewallRules = account.GetPeerConnectionResources(context.Background(), "peerC", approvedPeers, account.GetPeersGroupsMap(), account.GetGroupsPolicyMap())
+		peers, firewallRules = account.GetPeerConnectionResources(context.Background(), "peerC", approvedPeers)
 		assert.Len(t, peers, len(account.Groups["GroupSwarm"].Peers))
 		assert.Len(t, firewallRules, 1)
 		expectedFirewallRules := []*types.FirewallRule{
@@ -696,7 +696,7 @@ func TestAccount_getPeersByPolicyPostureChecks(t *testing.T) {
 
 		// peerE doesn't fulfill the NB posture check and exists in only destination group Swarm,
 		// all source group peers satisfying the NB posture check should establish connection
-		peers, firewallRules = account.GetPeerConnectionResources(context.Background(), "peerE", approvedPeers, account.GetPeersGroupsMap(), account.GetGroupsPolicyMap())
+		peers, firewallRules = account.GetPeerConnectionResources(context.Background(), "peerE", approvedPeers)
 		assert.Len(t, peers, 4)
 		assert.Len(t, firewallRules, 4)
 		assert.Contains(t, peers, account.Peers["peerA"])
@@ -706,7 +706,7 @@ func TestAccount_getPeersByPolicyPostureChecks(t *testing.T) {
 
 		// peerI doesn't fulfill the OS version posture check and exists in only destination group Swarm,
 		// all source group peers satisfying the NB posture check should establish connection
-		peers, firewallRules = account.GetPeerConnectionResources(context.Background(), "peerI", approvedPeers, account.GetPeersGroupsMap(), account.GetGroupsPolicyMap())
+		peers, firewallRules = account.GetPeerConnectionResources(context.Background(), "peerI", approvedPeers)
 		assert.Len(t, peers, 4)
 		assert.Len(t, firewallRules, 4)
 		assert.Contains(t, peers, account.Peers["peerA"])
@@ -721,19 +721,19 @@ func TestAccount_getPeersByPolicyPostureChecks(t *testing.T) {
 
 		// peerB doesn't satisfy the NB posture check, and doesn't exist in destination group peer's
 		// no connection should be established to any peer of destination group
-		peers, firewallRules := account.GetPeerConnectionResources(context.Background(), "peerB", approvedPeers, account.GetPeersGroupsMap(), account.GetGroupsPolicyMap())
+		peers, firewallRules := account.GetPeerConnectionResources(context.Background(), "peerB", approvedPeers)
 		assert.Len(t, peers, 0)
 		assert.Len(t, firewallRules, 0)
 
 		// peerI doesn't satisfy the OS version posture check, and doesn't exist in destination group peer's
 		// no connection should be established to any peer of destination group
-		peers, firewallRules = account.GetPeerConnectionResources(context.Background(), "peerI", approvedPeers, account.GetPeersGroupsMap(), account.GetGroupsPolicyMap())
+		peers, firewallRules = account.GetPeerConnectionResources(context.Background(), "peerI", approvedPeers)
 		assert.Len(t, peers, 0)
 		assert.Len(t, firewallRules, 0)
 
 		// peerC satisfy the NB posture check, should establish connection to all destination group peer's
 		// We expect a single permissive firewall rule which all outgoing connections
-		peers, firewallRules = account.GetPeerConnectionResources(context.Background(), "peerC", approvedPeers, account.GetPeersGroupsMap(), account.GetGroupsPolicyMap())
+		peers, firewallRules = account.GetPeerConnectionResources(context.Background(), "peerC", approvedPeers)
 		assert.Len(t, peers, len(account.Groups["GroupSwarm"].Peers))
 		assert.Len(t, firewallRules, len(account.Groups["GroupSwarm"].Peers))
 
@@ -748,14 +748,14 @@ func TestAccount_getPeersByPolicyPostureChecks(t *testing.T) {
 
 		// peerE doesn't fulfill the NB posture check and exists in only destination group Swarm,
 		// all source group peers satisfying the NB posture check should establish connection
-		peers, firewallRules = account.GetPeerConnectionResources(context.Background(), "peerE", approvedPeers, account.GetPeersGroupsMap(), account.GetGroupsPolicyMap())
+		peers, firewallRules = account.GetPeerConnectionResources(context.Background(), "peerE", approvedPeers)
 		assert.Len(t, peers, 3)
 		assert.Len(t, firewallRules, 3)
 		assert.Contains(t, peers, account.Peers["peerA"])
 		assert.Contains(t, peers, account.Peers["peerC"])
 		assert.Contains(t, peers, account.Peers["peerD"])
 
-		peers, firewallRules = account.GetPeerConnectionResources(context.Background(), "peerA", approvedPeers, account.GetPeersGroupsMap(), account.GetGroupsPolicyMap())
+		peers, firewallRules = account.GetPeerConnectionResources(context.Background(), "peerA", approvedPeers)
 		assert.Len(t, peers, 5)
 		// assert peers from Group Swarm
 		assert.Contains(t, peers, account.Peers["peerD"])

management/server/types/account.go

@@ -225,8 +225,6 @@ func (a *Account) GetPeerNetworkMap(
 	validatedPeersMap map[string]struct{},
 	resourcePolicies map[string][]*Policy,
 	routers map[string]map[string]*routerTypes.NetworkRouter,
-	peersGroups map[string][]string,
-	groupsPolicies map[string]map[string]*Policy,
 	metrics *telemetry.AccountManagerMetrics,
 ) *NetworkMap {
 	start := time.Now()
@@ -244,7 +242,7 @@ func (a *Account) GetPeerNetworkMap(
 		}
 	}
 
-	aclPeers, firewallRules := a.GetPeerConnectionResources(ctx, peerID, validatedPeersMap, peersGroups, groupsPolicies)
+	aclPeers, firewallRules := a.GetPeerConnectionResources(ctx, peerID, validatedPeersMap)
 	// exclude expired peers
 	var peersToConnect []*nbpeer.Peer
 	var expiredPeers []*nbpeer.Peer
@@ -947,54 +945,37 @@ func (a *Account) UserGroupsRemoveFromPeers(userID string, groups ...string) map
 // GetPeerConnectionResources for a given peer
 //
 // This function returns the list of peers and firewall rules that are applicable to a given peer.
-func (a *Account) GetPeerConnectionResources(
-	ctx context.Context,
-	peerID string,
-	validatedPeersMap map[string]struct{},
-	peersGroups map[string][]string,
-	groupsPolicies map[string]map[string]*Policy,
-) ([]*nbpeer.Peer, []*FirewallRule) {
+func (a *Account) GetPeerConnectionResources(ctx context.Context, peerID string, validatedPeersMap map[string]struct{}) ([]*nbpeer.Peer, []*FirewallRule) {
 	generateResources, getAccumulatedResources := a.connResourcesGenerator(ctx)
-	groups, ok := peersGroups[peerID]
-	if !ok {
-		return nil, nil
-	}
-
-	for _, group := range groups {
-		policiesPerGroup, ok := groupsPolicies[group]
-		if !ok {
+	for _, policy := range a.Policies {
+		if !policy.Enabled {
 			continue
 		}
-		for _, policy := range policiesPerGroup {
-			if !policy.Enabled {
+
+		for _, rule := range policy.Rules {
+			if !rule.Enabled {
 				continue
 			}
 
-			for _, rule := range policy.Rules {
-				if !rule.Enabled {
-					continue
-				}
-
-				sourcePeers, peerInSources := a.getAllPeersFromGroups(ctx, rule.Sources, peerID, policy.SourcePostureChecks, validatedPeersMap)
-				destinationPeers, peerInDestinations := a.getAllPeersFromGroups(ctx, rule.Destinations, peerID, nil, validatedPeersMap)
-
-				if rule.Bidirectional {
-					if peerInSources {
-						generateResources(rule, destinationPeers, FirewallRuleDirectionIN)
-					}
-					if peerInDestinations {
-						generateResources(rule, sourcePeers, FirewallRuleDirectionOUT)
-					}
-				}
+			sourcePeers, peerInSources := a.getAllPeersFromGroups(ctx, rule.Sources, peerID, policy.SourcePostureChecks, validatedPeersMap)
+			destinationPeers, peerInDestinations := a.getAllPeersFromGroups(ctx, rule.Destinations, peerID, nil, validatedPeersMap)
 
+			if rule.Bidirectional {
 				if peerInSources {
-					generateResources(rule, destinationPeers, FirewallRuleDirectionOUT)
+					generateResources(rule, destinationPeers, FirewallRuleDirectionIN)
 				}
-
 				if peerInDestinations {
-					generateResources(rule, sourcePeers, FirewallRuleDirectionIN)
+					generateResources(rule, sourcePeers, FirewallRuleDirectionOUT)
 				}
 			}
+
+			if peerInSources {
+				generateResources(rule, destinationPeers, FirewallRuleDirectionOUT)
+			}
+
+			if peerInDestinations {
+				generateResources(rule, sourcePeers, FirewallRuleDirectionIN)
+			}
 		}
 	}
 
@@ -1006,7 +987,7 @@ func (a *Account) GetPeerConnectionResources(
 // The generator function is used to generate the list of peers and firewall rules that are applicable to a given peer.
 // It safe to call the generator function multiple times for same peer and different rules no duplicates will be
 // generated. The accumulator function returns the result of all the generator calls.
-func (a *Account) connResourcesGenerator(ctx context.Context) (func(*PolicyRule, map[string]*nbpeer.Peer, int), func() ([]*nbpeer.Peer, []*FirewallRule)) {
+func (a *Account) connResourcesGenerator(ctx context.Context) (func(*PolicyRule, []*nbpeer.Peer, int), func() ([]*nbpeer.Peer, []*FirewallRule)) {
 	rulesExists := make(map[string]struct{})
 	peersExists := make(map[string]struct{})
 	rules := make([]*FirewallRule, 0)
@@ -1018,7 +999,7 @@ func (a *Account) connResourcesGenerator(ctx context.Context) (func(*PolicyRule,
 		all = &Group{}
 	}
 
-	return func(rule *PolicyRule, groupPeers map[string]*nbpeer.Peer, direction int) {
+	return func(rule *PolicyRule, groupPeers []*nbpeer.Peer, direction int) {
 			isAll := (len(all.Peers) - 1) == len(groupPeers)
 			for _, peer := range groupPeers {
 				if peer == nil {
@@ -1071,38 +1052,32 @@ func (a *Account) connResourcesGenerator(ctx context.Context) (func(*PolicyRule,
 //
 // Important: Posture checks are applicable only to source group peers,
 // for destination group peers, call this method with an empty list of sourcePostureChecksIDs
-func (a *Account) getAllPeersFromGroups(ctx context.Context, groups []string, peerID string, sourcePostureChecksIDs []string, validatedPeersMap map[string]struct{}) (map[string]*nbpeer.Peer, bool) {
+func (a *Account) getAllPeersFromGroups(ctx context.Context, groups []string, peerID string, sourcePostureChecksIDs []string, validatedPeersMap map[string]struct{}) ([]*nbpeer.Peer, bool) {
 	peerInGroups := false
-	filteredPeers := make(map[string]*nbpeer.Peer)
-	for _, groupID := range groups {
-		group := a.GetGroup(groupID)
-		for _, p := range group.Peers {
-			peer, ok := a.Peers[p]
-			if !ok || peer == nil {
-				continue
-			}
-
-			if _, ok := filteredPeers[p]; ok {
-				continue
-			}
-
-			// validate the peer based on policy posture checks applied
-			isValid := a.validatePostureChecksOnPeer(ctx, sourcePostureChecksIDs, peer.ID)
-			if !isValid {
-				continue
-			}
-
-			if _, ok := validatedPeersMap[peer.ID]; !ok {
-				continue
-			}
-
-			if peer.ID == peerID {
-				peerInGroups = true
-				continue
-			}
-
-			filteredPeers[p] = peer
+	uniquePeerIDs := a.getUniquePeerIDsFromGroupsIDs(ctx, groups)
+	filteredPeers := make([]*nbpeer.Peer, 0, len(uniquePeerIDs))
+	for _, p := range uniquePeerIDs {
+		peer, ok := a.Peers[p]
+		if !ok || peer == nil {
+			continue
 		}
+
+		// validate the peer based on policy posture checks applied
+		isValid := a.validatePostureChecksOnPeer(ctx, sourcePostureChecksIDs, peer.ID)
+		if !isValid {
+			continue
+		}
+
+		if _, ok := validatedPeersMap[peer.ID]; !ok {
+			continue
+		}
+
+		if peer.ID == peerID {
+			peerInGroups = true
+			continue
+		}
+
+		filteredPeers = append(filteredPeers, peer)
 	}
 
 	return filteredPeers, peerInGroups
@@ -1343,7 +1318,7 @@ func (a *Account) GetResourcePoliciesMap() map[string][]*Policy {
 func (a *Account) GetNetworkResourcesRoutesToSync(ctx context.Context, peerID string, resourcePolicies map[string][]*Policy, routers map[string]map[string]*routerTypes.NetworkRouter) (bool, []*route.Route, map[string]struct{}) {
 	var isRoutingPeer bool
 	var routes []*route.Route
-	allSourcePeers := make(map[string]struct{})
+	allSourcePeers := make(map[string]struct{}, len(a.Peers))
 
 	for _, resource := range a.NetworkResources {
 		if !resource.Enabled {
@@ -1367,7 +1342,7 @@ func (a *Account) GetNetworkResourcesRoutesToSync(ctx context.Context, peerID st
 				for _, pID := range a.getPostureValidPeers(peers, policy.SourcePostureChecks) {
 					allSourcePeers[pID] = struct{}{}
 				}
-			} else if _, ok := peers[peerID]; ok && a.validatePostureChecksOnPeer(ctx, policy.SourcePostureChecks, peerID) {
+			} else if slices.Contains(peers, peerID) && a.validatePostureChecksOnPeer(ctx, policy.SourcePostureChecks, peerID) {
 				// add routes for the resource if the peer is in the distribution group
 				for peerId, router := range networkRoutingPeers {
 					routes = append(routes, a.getNetworkResourcesRoutes(resource, peerId, router, resourcePolicies)...)
@@ -1383,9 +1358,9 @@ func (a *Account) GetNetworkResourcesRoutesToSync(ctx context.Context, peerID st
 	return isRoutingPeer, routes, allSourcePeers
 }
 
-func (a *Account) getPostureValidPeers(inputPeers map[string]struct{}, postureChecksIDs []string) []string {
+func (a *Account) getPostureValidPeers(inputPeers []string, postureChecksIDs []string) []string {
 	var dest []string
-	for peerID := range inputPeers {
+	for _, peerID := range inputPeers {
 		if a.validatePostureChecksOnPeer(context.Background(), postureChecksIDs, peerID) {
 			dest = append(dest, peerID)
 		}
@@ -1393,7 +1368,7 @@ func (a *Account) getPostureValidPeers(inputPeers map[string]struct{}, postureCh
 	return dest
 }
 
-func (a *Account) getUniquePeerIDsFromGroupsIDs(ctx context.Context, groups []string) map[string]struct{} {
+func (a *Account) getUniquePeerIDsFromGroupsIDs(ctx context.Context, groups []string) []string {
 	peerIDs := make(map[string]struct{}, len(groups)) // we expect at least one peer per group as initial capacity
 	for _, groupID := range groups {
 		group := a.GetGroup(groupID)
@@ -1402,21 +1377,21 @@ func (a *Account) getUniquePeerIDsFromGroupsIDs(ctx context.Context, groups []st
 			continue
 		}
 
-		// if group.IsGroupAll() || len(groups) == 1 {
-		// 	return group.Peers
-		// }
+		if group.IsGroupAll() || len(groups) == 1 {
+			return group.Peers
+		}
 
 		for _, peerID := range group.Peers {
 			peerIDs[peerID] = struct{}{}
 		}
 	}
 
-	// ids := make([]string, 0, len(peerIDs))
-	// for peerID := range peerIDs {
-	// 	ids = append(ids, peerID)
-	// }
+	ids := make([]string, 0, len(peerIDs))
+	for peerID := range peerIDs {
+		ids = append(ids, peerID)
+	}
 
-	return peerIDs
+	return ids
 }
 
 // getNetworkResources filters and returns a list of network resources associated with the given network ID.
@@ -1591,35 +1566,3 @@ func (a *Account) AddAllGroup() error {
 	}
 	return nil
 }
-
-func (a *Account) GetPeersGroupsMap() map[string][]string {
-	groups := make(map[string][]string, len(a.Groups))
-	for _, group := range a.Groups {
-		for _, peerID := range group.Peers {
-			groups[peerID] = append(groups[peerID], group.ID)
-		}
-	}
-	return groups
-}
-
-func (a *Account) GetGroupsPolicyMap() map[string]map[string]*Policy {
-	policies := make(map[string]map[string]*Policy, len(a.Groups))
-	for _, policy := range a.Policies {
-		for _, rules := range policy.Rules {
-			for _, src := range rules.Sources {
-				if policies[src] == nil {
-					policies[src] = make(map[string]*Policy)
-				}
-				policies[src][policy.ID] = policy
-			}
-			for _, dest := range rules.Destinations {
-				if policies[dest] == nil {
-					policies[dest] = make(map[string]*Policy)
-				}
-				policies[dest][policy.ID] = policy
-			}
-		}
-	}
-
-	return policies
-}

management/server/updatechannel.go

@@ -57,8 +57,13 @@ func (p *PeersUpdateManager) SendUpdate(ctx context.Context, peerID string, upda
 		case channel <- update:
 			log.WithContext(ctx).Debugf("update was sent to channel for peer %s", peerID)
 		default:
-			dropped = true
-			log.WithContext(ctx).Warnf("channel for peer %s is %d full or closed", peerID, len(channel))
+			select {
+			case <-channel:
+				log.WithContext(ctx).Trace("dropped oldest message from channel for peer %s", peerID)
+			default:
+				channel <- update
+				log.WithContext(ctx).Debugf("update was sent to channel for peer %s", peerID)
+			}
 		}
 	} else {
 		log.WithContext(ctx).Debugf("peer %s has no channel", peerID)
@@ -85,7 +90,7 @@ func (p *PeersUpdateManager) CreateChannel(ctx context.Context, peerID string) c
 		close(channel)
 	}
 	// mbragin: todo shouldn't it be more? or configurable?
-	channel := make(chan *UpdateMessage, channelBufferSize)
+	channel := make(chan *UpdateMessage, 2)
 	p.peerChannels[peerID] = channel
 
 	log.WithContext(ctx).Debugf("opened updates channel for a peer %s", peerID)

util/grpc/dialer.go

@@ -71,18 +71,15 @@ func CreateConnection(addr string, tlsEnabled bool) (*grpc.ClientConn, error) {
 		}))
 	}
 
-	connCtx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
-	defer cancel()
-
 	conn, err := grpc.DialContext(
-		connCtx,
+		context.Background(),
 		addr,
 		transportOption,
 		WithCustomDialer(),
 		grpc.WithBlock(),
 		grpc.WithKeepaliveParams(keepalive.ClientParameters{
-			Time:    30 * time.Second,
-			Timeout: 10 * time.Second,
+			Time:    10 * time.Minute,
+			Timeout: 5 * time.Minute,
 		}),
 	)
 	if err != nil {